Spaces:
Sleeping
Sleeping
File size: 6,177 Bytes
9c1645f | 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 | """
Reward calculation system for SupportOps Arena.
Implements dense reward shaping with detailed breakdown tracking.
"""
from typing import Any
from env.state import EnvState, TaskLevel
from env.actions import ActionType
# ─── Reward Events ────────────────────────────────────────────
REWARD_EVENTS = {
# Positive rewards
"DIAGNOSTIC_ACTION": +0.10,
"CORRECT_TARGETED_DIAGNOSTIC": +0.20,
"EVIDENCE_BEFORE_RISKY": +0.05,
"CORRECT_REMEDIATION": +0.40,
"OPTIMAL_RESOLUTION": +0.20, # Only if steps_used <= min_required_steps
"JUSTIFIED_ESCALATION": +0.30,
"TERMINAL_CORRECT": +1.00,
# Negative rewards
"REDUNDANT_ACTION": -0.05,
"RISKY_WITHOUT_DIAGNOSTIC": -0.15,
"INCORRECT_REMEDIATION": -0.25,
"HARMFUL_ACTION": -0.35,
"PREMATURE_CLOSURE": -0.40,
"BUDGET_EXCEEDED_PER_STEP": -0.02,
"PREMATURE_ESCALATION": -0.20, # Escalate before unlock
"UNJUSTIFIED_ESCALATION": -0.20, # Escalate without enough evidence
}
# ─── Configuration ────────────────────────────────────────────
MIN_STEPS_FOR_OPTIMAL = {
TaskLevel.EASY: 4,
TaskLevel.MEDIUM: 7,
TaskLevel.HARD: 12,
}
MAX_POSSIBLE_REWARD = {
TaskLevel.EASY: 2.05,
TaskLevel.MEDIUM: 2.35,
TaskLevel.HARD: 2.75,
}
class RewardCalculator:
"""
Calculates rewards for agent actions with dense reward shaping.
Tracks individual reward events for interpretability.
"""
def calculate(
self,
events: list[str],
state: EnvState,
action: ActionType
) -> tuple[float, dict[str, Any]]:
"""
Calculate total reward from a list of reward event keys.
Args:
events: List of reward event keys (e.g., ["DIAGNOSTIC_ACTION", "CORRECT_TARGETED_DIAGNOSTIC"])
state: Current environment state
action: Action that was taken
Returns:
Tuple of (total_reward, breakdown_dict)
- total_reward: Sum of all event rewards
- breakdown_dict: Dictionary mapping event keys to their reward values
"""
breakdown = {}
total_reward = 0.0
for event_key in events:
if event_key not in REWARD_EVENTS:
raise ValueError(f"Unknown reward event: {event_key}")
reward_value = REWARD_EVENTS[event_key]
breakdown[event_key] = reward_value
total_reward += reward_value
return total_reward, breakdown
def calculate_episode_score(
self,
cumulative_reward: float,
task_level: TaskLevel
) -> float:
"""
Normalize cumulative reward to episode score in [0, 1].
Args:
cumulative_reward: Total reward accumulated during episode
task_level: Difficulty level of the task
Returns:
Normalized score clamped to [0, 1]
"""
max_reward = MAX_POSSIBLE_REWARD[task_level]
score = cumulative_reward / max_reward
return max(0.0, min(1.0, score))
def is_optimal_resolution(
self,
step_count: int,
task_level: TaskLevel
) -> bool:
"""
Check if resolution was achieved within optimal step count.
Args:
step_count: Number of steps taken
task_level: Difficulty level
Returns:
True if step_count <= minimum optimal steps for this level
"""
return step_count <= MIN_STEPS_FOR_OPTIMAL[task_level]
def get_max_possible_reward(self, task_level: TaskLevel) -> float:
"""Get maximum possible reward for a task level."""
return MAX_POSSIBLE_REWARD[task_level]
def validate_events(self, events: list[str]) -> bool:
"""
Validate that all event keys are recognized.
Args:
events: List of event keys to validate
Returns:
True if all events are valid
Raises:
ValueError: If any event key is not recognized
"""
for event in events:
if event not in REWARD_EVENTS:
raise ValueError(f"Unknown reward event: {event}")
return True
def get_reward_event_description(event_key: str) -> str:
"""Get human-readable description of a reward event."""
descriptions = {
"DIAGNOSTIC_ACTION": "Performed a diagnostic action to gather information",
"CORRECT_TARGETED_DIAGNOSTIC": "Performed diagnostic action that targets the actual root cause",
"EVIDENCE_BEFORE_RISKY": "Gathered evidence before taking risky action",
"CORRECT_REMEDIATION": "Applied the correct fix for the root cause",
"OPTIMAL_RESOLUTION": "Resolved issue within optimal step count",
"JUSTIFIED_ESCALATION": "Escalated with sufficient evidence",
"TERMINAL_CORRECT": "Episode terminated with correct resolution",
"REDUNDANT_ACTION": "Repeated an action already performed",
"RISKY_WITHOUT_DIAGNOSTIC": "Took risky action without prior diagnostics",
"INCORRECT_REMEDIATION": "Applied wrong fix for the root cause",
"HARMFUL_ACTION": "Action caused harm or disruption",
"PREMATURE_CLOSURE": "Closed ticket without proper resolution",
"BUDGET_EXCEEDED_PER_STEP": "Exceeded step budget (per step penalty)",
"PREMATURE_ESCALATION": "Escalated before gathering minimum evidence",
"UNJUSTIFIED_ESCALATION": "Escalated without sufficient diagnostic evidence",
}
return descriptions.get(event_key, f"Unknown event: {event_key}")
def get_all_positive_events() -> list[str]:
"""Get list of all positive reward event keys."""
return [k for k, v in REWARD_EVENTS.items() if v > 0]
def get_all_negative_events() -> list[str]:
"""Get list of all negative reward event keys."""
return [k for k, v in REWARD_EVENTS.items() if v < 0]
|