Spaces:
Sleeping
Sleeping
File size: 7,883 Bytes
cb922bf | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 | """
Episode grading module for the Cloud Resource Allocation RL environment.
This module implements the EpisodeGrader class that evaluates complete episode
trajectories based on stability, efficiency, and overall performance. The grader
calculates a composite score and determines pass/fail status against thresholds.
"""
import numpy as np
class EpisodeGrader:
"""
Evaluates agent performance across complete episodes.
Metrics:
- Average reward (overall performance)
- Stability score (utilization variance)
- Efficiency score (resource usage patterns)
The grader combines these metrics into a weighted composite score and
determines whether the episode meets minimum performance thresholds.
"""
def __init__(
self,
pass_threshold: float = 0.0,
stability_weight: float = 0.3,
efficiency_weight: float = 0.3,
reward_weight: float = 0.4
):
"""
Initialize grader with scoring weights and thresholds.
Args:
pass_threshold: Minimum score to pass evaluation
stability_weight: Weight for stability component (0-1)
efficiency_weight: Weight for efficiency component (0-1)
reward_weight: Weight for cumulative reward component (0-1)
"""
self.pass_threshold = pass_threshold
self.stability_weight = stability_weight
self.efficiency_weight = efficiency_weight
self.reward_weight = reward_weight
def grade_episode(
self,
states: list[np.ndarray],
actions: list[int],
rewards: list[float]
) -> dict:
"""
Evaluate a complete episode trajectory.
The grader analyzes the full trajectory to assess:
1. Stability: How consistent were utilization levels?
2. Efficiency: How well were resources managed?
3. Performance: What was the average reward?
Args:
states: List of state observations throughout episode
actions: List of actions taken throughout episode
rewards: List of rewards received throughout episode
Returns:
Dictionary containing:
- score: Overall performance score
- passed: Boolean pass/fail status
- stability_score: System stability metric
- efficiency_score: Resource efficiency metric
- avg_reward: Average reward per step
- avg_cpu: Average CPU utilization
- avg_memory: Average memory utilization
- avg_latency: Average latency
- final_score: Alias for score (compatibility)
Raises:
ValueError: If trajectory data is empty or mismatched lengths
"""
# Validate inputs
if not states or not actions or not rewards:
raise ValueError("Trajectory data cannot be empty")
if not (len(states) == len(actions) == len(rewards)):
raise ValueError(
f"Mismatched trajectory lengths: states={len(states)}, "
f"actions={len(actions)}, rewards={len(rewards)}"
)
# Calculate component metrics
stability_score = self._calculate_stability(states)
efficiency_score = self._calculate_efficiency(states)
avg_reward = float(np.mean(rewards))
# Calculate average CPU and memory utilization
cpu_utils = [state[0] for state in states]
memory_utils = [state[1] for state in states]
avg_cpu = float(np.mean(cpu_utils))
avg_memory = float(np.mean(memory_utils))
# Calculate average latency from states
# Latency = request_rate / allocated_resources
request_rates = [state[2] for state in states]
allocated_resources = [state[3] for state in states]
latencies = [req / max(res, 1) for req, res in zip(request_rates, allocated_resources)]
avg_latency = float(np.mean(latencies))
# Normalize average reward to [0, 1] range for scoring
# Assuming rewards typically range from -5 to +2 based on reward module
# Map this to [0, 1] where 0 is worst (-5) and 1 is best (+2)
normalized_reward = (avg_reward + 5.0) / 7.0
normalized_reward = max(0.0, min(1.0, normalized_reward))
# Calculate weighted composite score
score = (
self.stability_weight * stability_score +
self.efficiency_weight * efficiency_score +
self.reward_weight * normalized_reward
)
# Determine pass/fail status
passed = score >= self.pass_threshold
return {
'score': score,
'passed': passed,
'stability_score': stability_score,
'efficiency_score': efficiency_score,
'avg_reward': avg_reward,
'avg_cpu': avg_cpu,
'avg_memory': avg_memory,
'avg_latency': avg_latency,
'final_score': score # Alias for compatibility
}
def _calculate_stability(self, states: list[np.ndarray]) -> float:
"""
Calculate stability score based on utilization variance.
Stability is measured by how consistent CPU and memory utilization
levels are throughout the episode. Lower variance indicates better
stability and receives a higher score.
Args:
states: List of state observations throughout episode
Returns:
Stability score in range [0, 1] where 1 is most stable
"""
# Extract CPU and memory utilization from states
# State format: [cpu_util, memory_util, request_rate, allocated_resources]
cpu_utils = [state[0] for state in states]
memory_utils = [state[1] for state in states]
# Calculate variance for both metrics
cpu_variance = float(np.var(cpu_utils))
memory_variance = float(np.var(memory_utils))
# Average variance
avg_variance = (cpu_variance + memory_variance) / 2.0
# Convert variance to stability score [0, 1]
# Lower variance = higher stability
# Use exponential decay: score = exp(-variance / scale)
# Scale factor of 100 means variance of 100 gives score ~0.37
stability_score = np.exp(-avg_variance / 100.0)
return float(stability_score)
def _calculate_efficiency(self, states: list[np.ndarray]) -> float:
"""
Calculate efficiency score based on resource allocation patterns.
Efficiency is measured by how well resources are utilized. Lower
average resource allocation (while maintaining acceptable utilization)
indicates better efficiency and receives a higher score.
Args:
states: List of state observations throughout episode
Returns:
Efficiency score in range [0, 1] where 1 is most efficient
"""
# Extract allocated resources from states
# State format: [cpu_util, memory_util, request_rate, allocated_resources]
allocated_resources = [state[3] for state in states]
# Calculate average resource allocation
avg_resources = float(np.mean(allocated_resources))
# Convert to efficiency score [0, 1]
# Lower average resources = higher efficiency
# Assume typical range is 1-10 instances
# Map: 1 instance -> 1.0, 10 instances -> 0.1
# Use formula: score = 1 / avg_resources, capped at [0, 1]
efficiency_score = 1.0 / avg_resources
efficiency_score = min(1.0, efficiency_score)
return float(efficiency_score)
|