Spaces:
Sleeping
Sleeping
| """ | |
| Adversarial grader for testing policy robustness. | |
| Generates variant episodes and tests if agent's strategy would still work. | |
| """ | |
| import logging | |
| from typing import Any | |
| from env.state import EnvState | |
| from adversary.adversary import AdaptiveAdversary | |
| logger = logging.getLogger(__name__) | |
| class AdversarialGrader: | |
| """ | |
| Tests policy robustness by generating harder variants. | |
| Used for hard task only. | |
| """ | |
| def __init__(self): | |
| """Initialize adversarial grader.""" | |
| pass | |
| async def grade( | |
| self, | |
| state: EnvState, | |
| adversary: AdaptiveAdversary | None = None | |
| ) -> float: | |
| """ | |
| Grade policy robustness by testing on variant episodes. | |
| Strategy: | |
| 1. Generate a variant of the same ticket (different surface presentation) | |
| 2. Simulate if agent's action strategy would still work | |
| 3. Return robustness score (0.0-1.0) | |
| Args: | |
| state: Complete environment state after episode | |
| adversary: Optional adaptive adversary for variant generation | |
| Returns: | |
| Robustness score (0.0-1.0) | |
| """ | |
| # For now, use heuristics to estimate robustness | |
| # A full implementation would actually run the variant episode | |
| try: | |
| robustness_score = self._estimate_robustness(state) | |
| return robustness_score | |
| except Exception as e: | |
| logger.error(f"Adversarial grading failed: {e}", exc_info=True) | |
| return 0.5 # Neutral fallback | |
| def _estimate_robustness(self, state: EnvState) -> float: | |
| """ | |
| Estimate policy robustness using heuristics. | |
| A robust policy: | |
| - Uses diverse diagnostic actions (not always same pattern) | |
| - Adapts to observations (doesn't follow fixed sequence) | |
| - Gathers evidence before high-risk actions | |
| - Makes decisions based on evidence, not just templates | |
| """ | |
| score_components = [] | |
| # 1. Action diversity score | |
| unique_actions = len(set(a.action_name for a in state.action_log)) | |
| total_actions = len(state.action_log) | |
| diversity_score = unique_actions / total_actions if total_actions > 0 else 0.0 | |
| score_components.append(diversity_score) | |
| # 2. Diagnostic breadth score | |
| diagnostic_actions = { | |
| "inspect_network", | |
| "inspect_logs", | |
| "check_authentication", | |
| "check_permissions", | |
| "query_device_status", | |
| "search_internal_kb", | |
| "contact_user_for_info", | |
| "run_diagnostic_script", | |
| } | |
| diagnostics_used = sum( | |
| 1 for a in state.action_log | |
| if a.action_name in diagnostic_actions | |
| ) | |
| unique_diagnostics_used = len(set( | |
| a.action_name for a in state.action_log | |
| if a.action_name in diagnostic_actions | |
| )) | |
| diagnostic_breadth = ( | |
| unique_diagnostics_used / len(diagnostic_actions) | |
| if diagnostic_actions else 0.0 | |
| ) | |
| score_components.append(diagnostic_breadth) | |
| # 3. Evidence-before-action score | |
| # Check if high-risk actions were preceded by diagnostics | |
| diagnostic_count = 0 | |
| risky_without_diagnostic = 0 | |
| for action in state.action_log: | |
| if action.action_name in diagnostic_actions: | |
| diagnostic_count += 1 | |
| elif action.action_name in ["restart_service", "reset_credentials"]: | |
| if diagnostic_count < 2: | |
| risky_without_diagnostic += 1 | |
| if risky_without_diagnostic == 0: | |
| evidence_score = 1.0 | |
| else: | |
| evidence_score = max(0.0, 1.0 - (risky_without_diagnostic * 0.3)) | |
| score_components.append(evidence_score) | |
| # 4. Adaptive behavior score | |
| # Check if agent repeated same action multiple times (template behavior) | |
| action_counts = {} | |
| for action in state.action_log: | |
| action_counts[action.action_name] = action_counts.get(action.action_name, 0) + 1 | |
| max_repeats = max(action_counts.values()) if action_counts else 1 | |
| if max_repeats >= 4: | |
| adaptive_score = 0.5 # Too much repetition | |
| elif max_repeats >= 3: | |
| adaptive_score = 0.7 | |
| else: | |
| adaptive_score = 1.0 | |
| score_components.append(adaptive_score) | |
| # 5. Success under pressure score | |
| # Check if agent succeeded despite complexity | |
| task_multiplier = { | |
| "easy": 0.8, | |
| "medium": 1.0, | |
| "hard": 1.2, | |
| } | |
| # If agent succeeded on hard task, boost robustness | |
| if state.done and state.cumulative_reward > 0: | |
| success_score = 1.0 * task_multiplier.get(state.task_level.value, 1.0) | |
| else: | |
| success_score = 0.5 | |
| score_components.append(min(1.0, success_score)) | |
| # Calculate weighted average | |
| weights = [0.15, 0.20, 0.25, 0.20, 0.20] | |
| robustness_score = sum( | |
| score * weight | |
| for score, weight in zip(score_components, weights) | |
| ) | |
| return min(1.0, max(0.0, robustness_score)) | |
| def aggregate_scores( | |
| programmatic: float, | |
| llm: float | None, | |
| adversarial: float | None, | |
| task_level: str | |
| ) -> float: | |
| """ | |
| Aggregate scores from multiple graders. | |
| Args: | |
| programmatic: Score from programmatic grader | |
| llm: Score from LLM grader (optional) | |
| adversarial: Score from adversarial grader (optional) | |
| task_level: Task difficulty level | |
| Returns: | |
| Aggregated score (0.0-1.0) | |
| """ | |
| if task_level == "easy": | |
| return programmatic | |
| elif task_level == "medium": | |
| llm_score = llm if llm is not None else 0.5 | |
| return 0.6 * programmatic + 0.4 * llm_score | |
| else: # hard | |
| llm_score = llm if llm is not None else 0.5 | |
| adv_score = adversarial if adversarial is not None else 0.5 | |
| return 0.5 * programmatic + 0.3 * llm_score + 0.2 * adv_score | |
| # Benchmark score weights by task level | |
| OVERALL_BENCHMARK_SCORE_WEIGHTS = { | |
| "easy": 0.2, | |
| "medium": 0.3, | |
| "hard": 0.5, | |
| } | |
| def calculate_benchmark_score(scores_by_level: dict[str, float]) -> float: | |
| """ | |
| Calculate overall benchmark score across task levels. | |
| Args: | |
| scores_by_level: Dictionary mapping task level to score | |
| Returns: | |
| Weighted benchmark score | |
| """ | |
| return sum( | |
| scores_by_level.get(level, 0.0) * weight | |
| for level, weight in OVERALL_BENCHMARK_SCORE_WEIGHTS.items() | |
| ) | |