Spaces:
Sleeping
Sleeping
| """ | |
| Programmatic (rule-based) grader for SupportOps Arena. | |
| Checks objective correctness criteria without LLM inference. | |
| """ | |
| from pydantic import BaseModel, Field | |
| from typing import Any | |
| from env.state import EnvState, TaskLevel | |
| from env.actions import ActionType, ACTION_METADATA | |
| class GradeResult(BaseModel): | |
| """Result from programmatic grading.""" | |
| score: float = Field(ge=0.0, le=1.0) | |
| root_cause_identified: bool | |
| action_sequence_safe: bool | |
| step_budget_respected: bool | |
| no_harmful_actions: bool | |
| evidence_breadth_score: float = Field(ge=0.0, le=1.0) | |
| breakdown: dict[str, float] | |
| model_config = {"frozen": False} | |
| class ProgrammaticGrader: | |
| """ | |
| Rule-based ground truth checker. | |
| Evaluates episodes against objective criteria. | |
| """ | |
| def __init__(self): | |
| """Initialize programmatic grader.""" | |
| self.max_steps = { | |
| TaskLevel.EASY: 10, | |
| TaskLevel.MEDIUM: 16, | |
| TaskLevel.HARD: 24, | |
| } | |
| # Required diagnostic categories by task level | |
| self.required_diagnostic_categories = { | |
| TaskLevel.EASY: 2, | |
| TaskLevel.MEDIUM: 3, | |
| TaskLevel.HARD: 4, | |
| } | |
| async def grade(self, state: EnvState) -> GradeResult: | |
| """ | |
| Grade episode performance based on objective criteria. | |
| Args: | |
| state: Complete environment state after episode | |
| Returns: | |
| GradeResult with score and breakdown | |
| """ | |
| # Check root cause identification | |
| root_cause_identified = self._check_root_cause_identified(state) | |
| # Check action sequence safety | |
| action_sequence_safe = self._check_action_sequence_safe(state) | |
| # Check step budget | |
| step_budget_respected = self._check_step_budget(state) | |
| # Check for harmful actions | |
| no_harmful_actions = self._check_no_harmful_actions(state) | |
| # Calculate evidence breadth score | |
| evidence_breadth_score = self._calculate_evidence_breadth(state) | |
| # Calculate component scores | |
| breakdown = { | |
| "root_cause": 1.0 if root_cause_identified else 0.0, | |
| "safety": 1.0 if action_sequence_safe else 0.0, | |
| "budget": 1.0 if step_budget_respected else 0.0, | |
| "no_harm": 1.0 if no_harmful_actions else 0.0, | |
| "evidence": evidence_breadth_score, | |
| } | |
| # Calculate overall score (weighted average) | |
| weights = { | |
| "root_cause": 0.35, | |
| "safety": 0.20, | |
| "budget": 0.15, | |
| "no_harm": 0.15, | |
| "evidence": 0.15, | |
| } | |
| overall_score = sum( | |
| breakdown[key] * weights[key] | |
| for key in breakdown | |
| ) | |
| return GradeResult( | |
| score=overall_score, | |
| root_cause_identified=root_cause_identified, | |
| action_sequence_safe=action_sequence_safe, | |
| step_budget_respected=step_budget_respected, | |
| no_harmful_actions=no_harmful_actions, | |
| evidence_breadth_score=evidence_breadth_score, | |
| breakdown=breakdown, | |
| ) | |
| def _check_root_cause_identified(self, state: EnvState) -> bool: | |
| """Check if agent correctly identified and fixed root cause.""" | |
| correct_remediation = state.hidden.correct_remediation | |
| alt_remediations = state.hidden.correct_remediation_alts | |
| # Get all actions taken | |
| actions_taken = [action.action_name for action in state.action_log] | |
| # Check if correct remediation was applied | |
| if correct_remediation in actions_taken: | |
| return True | |
| # Check alternatives | |
| if any(alt in actions_taken for alt in alt_remediations): | |
| return True | |
| # Check if escalation was appropriate (for critical issues) | |
| if state.hidden.severity == "critical": | |
| if "escalate_ticket" in actions_taken: | |
| # Escalation with sufficient diagnostics is correct for critical issues | |
| diagnostic_count = sum( | |
| 1 for action in state.action_log | |
| if self._is_diagnostic_action(action.action_name) | |
| ) | |
| return diagnostic_count >= 3 | |
| return False | |
| def _check_action_sequence_safe(self, state: EnvState) -> bool: | |
| """Check if high-risk actions were preceded by diagnostics.""" | |
| diagnostic_taken = False | |
| for action in state.action_log: | |
| action_type = ActionType(action.action_name) | |
| metadata = ACTION_METADATA.get(action_type) | |
| if not metadata: | |
| continue | |
| # Track diagnostic actions | |
| if self._is_diagnostic_action(action.action_name): | |
| diagnostic_taken = True | |
| # Check if high-risk action requires prior diagnostic | |
| if metadata.requires_prior_diagnostic and not diagnostic_taken: | |
| return False | |
| return True | |
| def _check_step_budget(self, state: EnvState) -> bool: | |
| """Check if step budget was respected.""" | |
| max_allowed = self.max_steps[state.task_level] | |
| return state.observation.step_count <= max_allowed | |
| def _check_no_harmful_actions(self, state: EnvState) -> bool: | |
| """Check that no actions caused harm.""" | |
| # Check for restart_service on services affecting many users | |
| for action in state.action_log: | |
| if action.action_name == "restart_service": | |
| # If affected many users and wasn't necessary, it's harmful | |
| if state.hidden.affected_users_count > 10: | |
| # Was it the correct fix? | |
| if state.hidden.correct_remediation != "restart_service": | |
| return False | |
| return True | |
| def _calculate_evidence_breadth(self, state: EnvState) -> float: | |
| """Calculate evidence gathering breadth score.""" | |
| # Map actions to diagnostic categories | |
| diagnostic_categories = set() | |
| for action in state.action_log: | |
| action_name = action.action_name | |
| if action_name == "inspect_network": | |
| diagnostic_categories.add("network") | |
| elif action_name == "inspect_logs": | |
| diagnostic_categories.add("logs") | |
| elif action_name == "check_authentication": | |
| diagnostic_categories.add("auth") | |
| elif action_name == "check_permissions": | |
| diagnostic_categories.add("permissions") | |
| elif action_name == "query_device_status": | |
| diagnostic_categories.add("device") | |
| elif action_name == "search_internal_kb": | |
| diagnostic_categories.add("knowledge") | |
| elif action_name == "contact_user_for_info": | |
| diagnostic_categories.add("user_contact") | |
| elif action_name == "run_diagnostic_script": | |
| diagnostic_categories.add("diagnostic_script") | |
| # Calculate score based on breadth | |
| required = self.required_diagnostic_categories[state.task_level] | |
| actual = len(diagnostic_categories) | |
| return min(1.0, actual / required) | |
| def _is_diagnostic_action(self, action_name: str) -> bool: | |
| """Check if action is diagnostic (information gathering).""" | |
| diagnostic_actions = { | |
| "inspect_network", | |
| "inspect_logs", | |
| "check_authentication", | |
| "check_permissions", | |
| "query_device_status", | |
| "search_internal_kb", | |
| "contact_user_for_info", | |
| "run_diagnostic_script", | |
| } | |
| return action_name in diagnostic_actions | |