""" graders/base_grader.py — Deterministic, rule-based grader Handles: step detection, edge cases, emotion mismatch, wrong assumptions """ from __future__ import annotations import re from typing import Dict, Any, List from models import StepName, Task # ── Keyword banks ────────────────────────────────────────────────────────────── EMPATHY_KEYWORDS = [ "sorry", "apologize", "apology", "understand", "frustrat", "hear you", "feel", "concern", "inconvenien", "deeply sorry", "sincerely", "personally sorry", "personally and deeply sorry", "i take full", "full responsibility", ] COLLECT_INFO_KEYWORDS = [ "order number", "order id", "account number", "could you provide", "can you share", "what is your", "please confirm", "transaction id", "booking reference", "invoice", "date of purchase", "when did", "which product", "what happened", "could you tell me", "may i have", "can i have your", "full name", "email address", "phone number", # Additional strong collect-info signals "could you please provide", "please provide", "provide me with", "associated with your account", "your account", "look into this", "look into further", "assist you further", "help me locate", "need a few details", "need some information", "need your", "retrieve your", "verify your", "pull up your", "access your", "your email", "your order", "your reference", "your details", "please share your", "share your", "to look into", "to assist you", "to help you", "can you provide", "can you confirm", "can you give", "i would need", "i will need", "so i can", "in order to", ] INVESTIGATE_KEYWORDS = [ "checking", "looking into", "pulling up", "reviewing", "investigating", "let me check", "i can see", "our records show", "according to our system", "i have found", "i found", "found that", "it appears", "it seems", "the issue", "root cause", ] RESOLUTION_KEYWORDS = [ "refund", "replacement", "credit", "we will", "i will", "here is what", "the solution", "fix this", "resolved", "compensation", "escalate", "expedite", "waive", "reimburse", "send a new", "process a", "arrange for", "free of charge", ] # De-escalation signals required for hard tasks DEESCALATION_KEYWORDS = [ "i completely understand", "that is unacceptable", "i take full responsibility", "this should not have happened", "you have every right", "personally ensure", "highest priority", "immediate attention", "personally and deeply sorry", "sincerely apologize", "i hear your frustration", "deeply sorry for the experience", ] # Wrong-assumption triggers (generic bad guesses) WRONG_ASSUMPTION_PHRASES = [ "you probably forgot", "you must have", "it is your fault", "you should have", "clearly you", "obviously you", ] # Generic / non-answer phrases GENERIC_FILLERS = [ "ok", "okay", "done", "sure", "got it", "understood", "no problem", "alright", "i see", "noted", ] class BaseGrader: """ Deterministic grader. Returns a dict with: correct, detected_action, wrong_assumption, skipped_step, notes """ def grade( self, response: str, expected_step: StepName, task: Task, ) -> Dict[str, Any]: r = response.lower().strip() result: Dict[str, Any] = { "correct": False, "detected_action": "unknown", "wrong_assumption": False, "skipped_step": False, "notes": [], } # ── Edge-case guard ──────────────────────────────────────────────────── if self._is_irrelevant(r): result["detected_action"] = "irrelevant" result["notes"].append("Response is irrelevant / generic filler.") return result # ── Detect what the agent actually did ───────────────────────────────── detected = self._detect_action(r, task) result["detected_action"] = detected # ── Correctness check ────────────────────────────────────────────────── result["correct"] = (detected == expected_step.value) # ── Skip-step check ──────────────────────────────────────────────────── # e.g. agent jumps straight to RESOLUTION while expected EMPATHY step_rank = { StepName.EMPATHY: 0, StepName.COLLECT_INFO: 1, StepName.INVESTIGATE: 2, StepName.RESOLUTION: 3, } detected_rank = self._action_to_rank(detected) expected_rank = step_rank[expected_step] if detected_rank > expected_rank + 0: result["skipped_step"] = True result["notes"].append( f"Skipped step: expected '{expected_step.value}', " f"agent jumped to '{detected}'." ) # ── Wrong-assumption check ───────────────────────────────────────────── if any(phrase in r for phrase in WRONG_ASSUMPTION_PHRASES): result["wrong_assumption"] = True result["notes"].append("Agent made a wrong assumption about the customer.") # ── Hard-task: de-escalation required ───────────────────────────────── if task.escalation_risk and expected_step == StepName.EMPATHY: if not any(kw in r for kw in DEESCALATION_KEYWORDS): result["correct"] = False result["notes"].append( "Hard task requires de-escalation language; none detected." ) return result # ── Helpers ──────────────────────────────────────────────────────────────── def _detect_action(self, r: str, task: Task) -> str: """Return the name of the step the response most resembles.""" scores = { "empathy": self._score_keywords(r, EMPATHY_KEYWORDS), "collect_info": self._score_keywords(r, COLLECT_INFO_KEYWORDS), "investigate": self._score_keywords(r, INVESTIGATE_KEYWORDS), "resolution": self._score_keywords(r, RESOLUTION_KEYWORDS), } best = max(scores, key=scores.get) if scores[best] == 0: return "unknown" return best def _score_keywords(self, text: str, keywords: List[str]) -> int: return sum(1 for kw in keywords if kw in text) def _is_irrelevant(self, r: str) -> bool: # Very short if len(r.split()) <= 3: return True # Pure filler if r in GENERIC_FILLERS: return True # No customer-support vocabulary at all all_keywords = ( EMPATHY_KEYWORDS + COLLECT_INFO_KEYWORDS + INVESTIGATE_KEYWORDS + RESOLUTION_KEYWORDS ) return not any(kw in r for kw in all_keywords) def _action_to_rank(self, action: str) -> int: mapping = { "empathy": 0, "collect_info": 1, "investigate": 2, "resolution": 3, "unknown": -1, "irrelevant": -1, } return mapping.get(action, -1) # ── Hard-task specific grader ────────────────────────────────────────────────── class HardTaskGrader(BaseGrader): """ Extended grader for the hard escalation scenario. Adds: emotional control score, conflict handling, professional tone. """ PROFESSIONAL_PHRASES = [ "i assure you", "rest assured", "our team will", "i will personally", "i will escalate", "within 24 hours", "we take this very seriously", "top priority", ] UNPROFESSIONAL_PHRASES = [ "calm down", "you need to", "stop complaining", "it is not our fault", "nothing we can do", "policy says", ] def grade( self, response: str, expected_step: StepName, task: Task, ) -> Dict[str, Any]: result = super().grade(response, expected_step, task) r = response.lower() notes = result["notes"] # Emotional-control bonus / penalty if any(kw in r for kw in self.PROFESSIONAL_PHRASES): notes.append("✅ Professional / reassuring language detected.") if any(kw in r for kw in self.UNPROFESSIONAL_PHRASES): result["correct"] = False result["wrong_assumption"] = True notes.append("❌ Unprofessional language detected (e.g. 'calm down').") # Resolution accuracy: must offer concrete action on resolution step if expected_step == StepName.RESOLUTION: concrete = [ "refund", "replacement", "credit", "waive", "free", "expedite", "send", "process", "arrange", ] if not any(c in r for c in concrete): result["correct"] = False notes.append( "❌ Resolution step requires a concrete action (refund/replacement/credit)." ) result["notes"] = notes return result