Spaces:
Sleeping
Sleeping
| """ | |
| 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) | |