""" Deterministic graders for SupportEnv tasks. Each grader inspects the agent's action_history against ground-truth data and returns (score, breakdown, feedback) where score is in (0.0, 1.0). Task 1 — Classification: category match (0.50) + priority match (0.40) + efficiency (0.10) Task 2 — Extraction: entity coverage (0.60) + action coverage (0.30) + no hallucination (0.10) Task 3 — Resolution: keyword coverage (0.30) + step coverage (0.30) + tone (0.25) + length (0.10) + non-empty steps (0.05) """ from __future__ import annotations import math from typing import Any, Dict, List, Optional, Tuple SCORE_EPSILON = 0.01 def _strict_score(score: float) -> float: """Map any score into the strict open interval (0, 1).""" try: value = float(score) except (TypeError, ValueError): value = SCORE_EPSILON # Guard against NaN, which would bypass numeric comparisons. if math.isnan(value): value = SCORE_EPSILON value = min(max(value, 0.0), 1.0) return round(value, 4) def grade_task( task_id: str, episode_state: Dict[str, Any] ) -> Tuple[float, Dict[str, float], str]: if task_id == "task1": return _grade_classification(episode_state) elif task_id == "task2": return _grade_extraction(episode_state) elif task_id == "task3": return _grade_resolution(episode_state) return _strict_score(0.01), {}, "Unknown task" # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _last_action_of_type( history: List[Dict[str, Any]], action_type: str ) -> Optional[Dict[str, Any]]: """Return the last action matching *action_type*, or None.""" for action in reversed(history): if action.get("action_type") == action_type: return action return None def _normalize(s: Any) -> str: return str(s).strip().lower() if s is not None else "" # --------------------------------------------------------------------------- # Task 1 — Classification # --------------------------------------------------------------------------- def _grade_classification(ep: Dict[str, Any]) -> Tuple[float, Dict[str, float], str]: """ Score breakdown: category_correct 0.50 — exact match priority_correct 0.40 — exact match efficiency 0.10 — 1 step = full, degrades linearly """ gt = ep["ticket_data"]["ground_truth"] history = ep.get("action_history", []) breakdown: Dict[str, float] = { "baseline": 0.01, "category_correct": 0.0, "priority_correct": 0.0, "efficiency": 0.0, } classify_action = _last_action_of_type(history, "classify") if classify_action is None: return _strict_score(0.0), breakdown, "No classify action found." # Category if _normalize(classify_action.get("category")) == _normalize(gt["category"]): breakdown["category_correct"] = 0.49 # Priority if _normalize(classify_action.get("priority")) == _normalize(gt["priority"]): breakdown["priority_correct"] = 0.40 # Efficiency: full marks if classified in 1 step, degrades linearly max_steps = ep.get("max_steps", 3) steps_used = ep.get("step_number", max_steps) if steps_used <= 1: breakdown["efficiency"] = 0.09 else: breakdown["efficiency"] = round(max(0.0, 0.09 * (1 - (steps_used - 1) / max_steps)), 4) score = _strict_score(sum(breakdown.values())) parts = ", ".join(f"{k}={v:.2f}" for k, v in breakdown.items()) return score, breakdown, f"Task 1: {parts}" # --------------------------------------------------------------------------- # Task 2 — Information Extraction # --------------------------------------------------------------------------- def _grade_extraction(ep: Dict[str, Any]) -> Tuple[float, Dict[str, float], str]: """ Score breakdown: entity_coverage 0.60 — fraction of ground-truth entities matched action_coverage 0.30 — fraction of required actions matched no_hallucination 0.10 — penalty for extra entities not in ground truth """ gt = ep["ticket_data"]["ground_truth"] history = ep.get("action_history", []) breakdown: Dict[str, float] = { "baseline": 0.01, "entity_coverage": 0.0, "action_coverage": 0.0, "no_hallucination": 0.09, # start with full marks, deduct } extract_action = _last_action_of_type(history, "extract") if extract_action is None: breakdown["no_hallucination"] = 0.0 return _strict_score(0.01), breakdown, "No extract action found." # --- Entity coverage --- gt_entities: Dict[str, Any] = gt.get("entities", {}) pred_entities: Dict[str, Any] = extract_action.get("extracted_entities") or {} if gt_entities: matched = 0 for key, gt_val in gt_entities.items(): pred_val = pred_entities.get(key) if pred_val is not None and _entity_matches(gt_val, pred_val): matched += 1 breakdown["entity_coverage"] = round(0.59 * matched / len(gt_entities), 4) # --- Action coverage --- gt_actions: List[str] = gt.get("required_actions", []) pred_actions: List[str] = extract_action.get("required_actions") or [] pred_actions_lower = [_normalize(a) for a in pred_actions] if gt_actions: matched_actions = sum( 1 for ga in gt_actions if _normalize(ga) in pred_actions_lower ) breakdown["action_coverage"] = round(0.30 * matched_actions / len(gt_actions), 4) # --- No hallucination --- if pred_entities and gt_entities: extra_keys = set(pred_entities.keys()) - set(gt_entities.keys()) if extra_keys: penalty = min(len(extra_keys) * 0.02, 0.09) breakdown["no_hallucination"] = round(max(0.0, 0.09 - penalty), 4) score = _strict_score(sum(breakdown.values())) parts = ", ".join(f"{k}={v:.2f}" for k, v in breakdown.items()) return score, breakdown, f"Task 2: {parts}" def _entity_matches(gt_val: Any, pred_val: Any) -> bool: """Flexible entity comparison — handles strings, lists, and numbers.""" if isinstance(gt_val, list) and isinstance(pred_val, list): gt_set = {_normalize(v) for v in gt_val} pred_set = {_normalize(v) for v in pred_val} return gt_set == pred_set return _normalize(gt_val) == _normalize(pred_val) # --------------------------------------------------------------------------- # Task 3 — Resolution Generation # --------------------------------------------------------------------------- def _grade_resolution(ep: Dict[str, Any]) -> Tuple[float, Dict[str, float], str]: """ Score breakdown: keyword_coverage 0.30 — fraction of required keywords found in response step_coverage 0.30 — fraction of required resolution steps matched tone_compliance 0.25 — apology / urgency / timeline adherence length_adequate 0.10 — response meets minimum length no_empty_steps 0.05 — all resolution steps are non-empty """ gt = ep["ticket_data"]["ground_truth"] history = ep.get("action_history", []) breakdown: Dict[str, float] = { "baseline": 0.01, "keyword_coverage": 0.0, "step_coverage": 0.0, "tone_compliance": 0.0, "length_adequate": 0.0, "no_empty_steps": 0.04, # assume pass unless empty steps found } respond_action = _last_action_of_type(history, "respond") if respond_action is None: breakdown["no_empty_steps"] = 0.0 return _strict_score(0.01), breakdown, "No respond action found." response_text: str = respond_action.get("response_text") or "" resolution_steps: List[str] = respond_action.get("resolution_steps") or [] response_lower = response_text.lower() # --- Keyword coverage --- required_keywords: List[str] = gt.get("required_keywords", []) if required_keywords: matched_kw = sum(1 for kw in required_keywords if kw.lower() in response_lower) breakdown["keyword_coverage"] = round(0.29 * matched_kw / len(required_keywords), 4) # --- Step coverage --- gt_steps: List[str] = gt.get("required_resolution_steps", []) if gt_steps: pred_steps_lower = [_normalize(s) for s in resolution_steps] matched_steps = sum( 1 for gs in gt_steps if _normalize(gs) in pred_steps_lower ) breakdown["step_coverage"] = round(0.30 * matched_steps / len(gt_steps), 4) # --- Tone compliance --- tone_req = gt.get("tone_requirements", {}) tone_checks = 0 tone_pass = 0 if tone_req.get("must_apologize"): tone_checks += 1 apology_words = ["apolog", "sorry", "regret", "sincerely"] if any(w in response_lower for w in apology_words): tone_pass += 1 if tone_req.get("must_acknowledge_urgency"): tone_checks += 1 urgency_words = ["urgent", "immediately", "priority", "asap", "right away", "as soon as"] if any(w in response_lower for w in urgency_words): tone_pass += 1 if tone_req.get("must_provide_timeline"): tone_checks += 1 timeline_words = ["within", "hours", "minutes", "by end of", "shortly", "today", "tomorrow", "timeline", "expect"] if any(w in response_lower for w in timeline_words): tone_pass += 1 if tone_checks > 0: breakdown["tone_compliance"] = round(0.25 * tone_pass / tone_checks, 4) else: breakdown["tone_compliance"] = 0.25 # no tone requirements = full marks # --- Length adequate --- min_len = gt.get("expected_response_length_min", 80) if len(response_text) >= min_len: breakdown["length_adequate"] = 0.10 # --- Non-empty steps --- if not resolution_steps or any(not s.strip() for s in resolution_steps): breakdown["no_empty_steps"] = 0.0 score = _strict_score(sum(breakdown.values())) parts = ", ".join(f"{k}={v:.2f}" for k, v in breakdown.items()) return score, breakdown, f"Task 3: {parts}"