""" Advanced multi-step grader for customer support email workflow. Handles incremental rewards, strategy scoring, and memory utilization. """ from models import EmailAction, ActionType, StrategyType, WorkflowStep, RewardWeights from typing import Tuple, Dict, Any, Optional # Deterministic strategy mapping: (category, sentiment, priority, has_vip_history) -> expected_strategy EXPECTED_STRATEGY_MAP = { # Billing issues ("billing", "angry", "high", True): "escalate_to_human", # VIP angry about billing ("billing", "angry", "high", False): "offer_refund", # Angry about billing ("billing", "negative", "high", True): "escalate_to_human", # VIP negative about billing ("billing", "negative", "high", False): "offer_refund", # Negative about billing ("billing", "neutral", "high", True): "escalate_to_human", # VIP urgent billing ("billing", "neutral", "high", False): "auto_resolve", # Standard billing issue ("billing", "neutral", "medium", True): "escalate_to_human", # VIP billing ("billing", "neutral", "medium", False): "auto_resolve", # Standard billing ("billing", "positive", "any", True): "auto_resolve", # VIP positive feedback ("billing", "positive", "any", False): "auto_resolve", # Positive billing feedback # Technical issues ("tech", "angry", "high", True): "escalate_to_human", # VIP angry about tech ("tech", "angry", "high", False): "escalate_to_human", # Angry about tech ("tech", "negative", "high", True): "escalate_to_human", # VIP negative about tech ("tech", "negative", "high", False): "request_more_info", # Need more tech details ("tech", "neutral", "high", True): "escalate_to_human", # VIP urgent tech ("tech", "neutral", "high", False): "request_more_info", # Urgent tech issue ("tech", "neutral", "medium", True): "escalate_to_human", # VIP tech issue ("tech", "neutral", "medium", False): "auto_resolve", # Standard tech issue ("tech", "positive", "any", True): "auto_resolve", # VIP positive tech feedback ("tech", "positive", "any", False): "auto_resolve", # Positive tech feedback # Complaints ("complaint", "angry", "high", True): "escalate_to_human", # VIP angry complaint ("complaint", "angry", "high", False): "escalate_to_human", # Angry complaint ("complaint", "negative", "high", True): "escalate_to_human", # VIP negative complaint ("complaint", "negative", "high", False): "escalate_to_human", # Negative complaint ("complaint", "neutral", "high", True): "escalate_to_human", # VIP urgent complaint ("complaint", "neutral", "high", False): "offer_refund", # Neutral complaint ("complaint", "neutral", "medium", True): "escalate_to_human", # VIP complaint ("complaint", "neutral", "medium", False): "request_more_info", # Standard complaint ("complaint", "positive", "any", True): "auto_resolve", # VIP positive feedback ("complaint", "positive", "any", False): "auto_resolve", # Positive feedback # Spam ("spam", "any", "any", True): "auto_resolve", # VIP unsubscribe (rare) ("spam", "any", "any", False): "auto_resolve", # Standard unsubscribe } def get_expected_strategy(category: str, sentiment: str, priority: str, customer_history: str) -> str: """ Get the deterministically expected strategy based on category, sentiment, priority, and VIP status. Args: category: Email category sentiment: Customer sentiment priority: Priority level customer_history: Customer history Returns: Expected strategy string """ has_vip = any(keyword in customer_history.lower() for keyword in ["vip", "enterprise", "high-value"]) # Try exact match first key = (category, sentiment, priority, has_vip) if key in EXPECTED_STRATEGY_MAP: return EXPECTED_STRATEGY_MAP[key] # Try with "any" wildcards for wildcard_key in [ (category, sentiment, priority, "any"), (category, sentiment, "any", has_vip), (category, "any", priority, has_vip), (category, sentiment, "any", "any"), (category, "any", priority, "any"), (category, "any", "any", has_vip), ("any", sentiment, priority, has_vip), (category, "any", "any", "any"), ("any", sentiment, "any", "any"), ("any", "any", priority, "any"), ("any", "any", "any", has_vip), ("any", "any", "any", "any") ]: if wildcard_key in EXPECTED_STRATEGY_MAP: return EXPECTED_STRATEGY_MAP[wildcard_key] # Default fallback return "auto_resolve" def grade_category(predicted: str, ground_truth: str) -> float: """ Grade a category prediction. Args: predicted: Predicted category string ground_truth: Ground truth category string Returns: 1.0 if prediction matches ground truth, else 0.0 """ return 1.0 if predicted.lower().strip() == ground_truth.lower().strip() else 0.0 def grade_priority(predicted: str, ground_truth: str) -> float: """ Grade a priority prediction. Args: predicted: Predicted priority string ground_truth: Ground truth priority string Returns: 1.0 if prediction matches ground truth, else 0.0 """ return 1.0 if predicted.lower().strip() == ground_truth.lower().strip() else 0.0 def grade_action(email_task: Dict[str, Any], action: EmailAction) -> Tuple[float, Dict[str, Any]]: """ Grade a complete EmailAction for a single-step episode. Args: email_task: Task metadata containing label and history action: Agent action containing category, priority, and response Returns: Tuple of (total_reward, breakdown) """ ground_truth = email_task.get("label", {}) category = ground_truth.get("category", "") priority = ground_truth.get("priority", "") customer_history = email_task.get("history", "") category_score = grade_category(action.category, category) priority_score = grade_priority(action.priority, priority) response_score, response_breakdown = grade_response_quality( action, category, customer_history, "auto_resolve" ) total_reward = ( 0.4 * category_score + 0.3 * priority_score + 0.3 * response_score ) breakdown = { "category_score": category_score, "priority_score": priority_score, "response_score": response_score, **response_breakdown } return min(max(total_reward, 0.0), 1.0), breakdown def analyze_customer_sentiment(email_body: str, subject: str) -> str: """ Analyze customer sentiment from email content. Returns: "positive", "neutral", "negative", "angry" """ text = (subject + " " + email_body).lower() # Angry indicators angry_words = ["frustrated", "angry", "furious", "terrible", "worst", "horrible", "unacceptable", "disgusted", "outraged", "infuriated", "damn", "hell"] if any(word in text for word in angry_words): return "angry" # Negative indicators negative_words = ["disappointed", "unhappy", "upset", "annoyed", "irritated", "concerned", "worried", "problem", "issue", "complaint"] if any(word in text for word in negative_words): return "negative" # Positive indicators positive_words = ["thank", "appreciate", "great", "excellent", "wonderful", "pleased", "happy", "satisfied", "good", "love"] if any(word in text for word in positive_words): return "positive" return "neutral" def extract_urgency_indicators(email_body: str, subject: str) -> list: """ Extract urgency indicators from email content. """ text = (subject + " " + email_body).lower() indicators = [] urgency_keywords = [ "urgent", "immediately", "asap", "right now", "emergency", "critical", "blocking", "stuck", "can't", "unable", "broken", "refund", "compensation", "deadline", "today", "now", "quickly", "fast", "rush" ] for keyword in urgency_keywords: if keyword in text: indicators.append(keyword) return indicators def grade_classification(action: EmailAction, ground_truth: str) -> Tuple[float, Dict[str, Any]]: """ Grade classification step. Args: action: Agent's classification action ground_truth: Correct category Returns: Tuple of (score, breakdown_dict) """ if action.action_type != ActionType.CLASSIFY: return 0.0, {"error": "Wrong action type for classification step"} predicted = action.content score = 1.0 if predicted.lower().strip() == ground_truth.lower().strip() else 0.0 return score, { "predicted_category": predicted, "ground_truth_category": ground_truth, "correct": score == 1.0 } def grade_prioritization(action: EmailAction, ground_truth: str, urgency_indicators: list) -> Tuple[float, Dict[str, Any]]: """ Grade prioritization step. Args: action: Agent's prioritization action ground_truth: Correct priority urgency_indicators: Urgency keywords from email Returns: Tuple of (score, breakdown_dict) """ if action.action_type != ActionType.PRIORITIZE: return 0.0, {"error": "Wrong action type for prioritization step"} predicted = action.content correct = predicted.lower().strip() == ground_truth.lower().strip() # Bonus for correctly identifying urgency urgency_bonus = 0.2 if len(urgency_indicators) > 0 and ground_truth == "high" and correct else 0.0 score = 1.0 if correct else 0.0 score = min(1.0, score + urgency_bonus) return score, { "predicted_priority": predicted, "ground_truth_priority": ground_truth, "correct": correct, "urgency_bonus": urgency_bonus, "urgency_indicators": urgency_indicators } def grade_strategy_decision(action: EmailAction, category: str, sentiment: str, customer_history: str, priority: str) -> Tuple[float, Dict[str, Any]]: """ Grade strategy decision with deterministic mapping. Args: action: Agent's strategy action category: Email category sentiment: Customer sentiment customer_history: Customer history priority: Priority level Returns: Tuple of (score, breakdown_dict) """ if action.action_type != ActionType.DECIDE_STRATEGY: return 0.0, {"error": "Wrong action type for strategy step"} chosen_strategy = action.content expected_strategy = get_expected_strategy(category, sentiment, priority, customer_history) # Perfect match gets full score if chosen_strategy == expected_strategy: score = 1.0 correct = True else: # Partial credit for reasonable alternatives score = 0.3 # Base partial credit correct = False # Bonus for choosing escalate_to_human when expected is offer_refund (conservative approach) if expected_strategy == "offer_refund" and chosen_strategy == "escalate_to_human": score = 0.7 # Bonus for choosing offer_refund when expected is auto_resolve (generous approach) elif expected_strategy == "auto_resolve" and chosen_strategy == "offer_refund": score = 0.6 # Penalty for completely wrong strategies (e.g., auto_resolve for angry complaints) elif expected_strategy in ["escalate_to_human", "offer_refund"] and chosen_strategy == "auto_resolve": score = 0.1 return score, { "strategy": chosen_strategy, "expected_strategy": expected_strategy, "correct": correct, "category": category, "sentiment": sentiment, "priority": priority, "has_vip": any(keyword in customer_history.lower() for keyword in ["vip", "enterprise", "high-value"]) } def grade_response_quality( action: EmailAction, category: str, customer_history: str, strategy: str, state: Dict[str, Any] = None ) -> Tuple[float, Dict[str, Any]]: """ Grade response quality with advanced semantic analysis. Args: action: Agent's response action category: Email category customer_history: Customer history strategy: Chosen strategy Returns: Tuple of (score, breakdown_dict) """ if action.action_type != ActionType.RESPOND: return 0.0, {"error": "Wrong action type for response step"} response = action.content response_lower = response.lower() if not response or len(response.strip()) == 0: return 0.0, {"error": "Empty response"} word_count = len(response.split()) # Length scoring (40% weight) if word_count < 20: length_score = min(word_count / 20.0, 1.0) * 0.5 elif word_count > 150: length_score = 1.0 - min((word_count - 150) / 50.0, 0.3) else: length_score = 1.0 # Politeness scoring (30% weight) politeness_markers = [ "sorry", "apologize", "apologies", "please", "help", "grateful", "appreciate", "thank", "understand", "assist", "support", "immediate", "priority", "resolve", "solution", "fix", "happy to help", "pleased to assist", "certainly", "absolutely" ] politeness_score = 1.0 if any(marker in response_lower for marker in politeness_markers) else 0.5 # Category relevance scoring (20% weight) relevance_score = 0.5 # Base score if category == "billing": billing_keywords = ["refund", "charge", "payment", "invoice", "billing", "credit", "fee"] if any(kw in response_lower for kw in billing_keywords): relevance_score = 1.0 elif strategy == "offer_refund" and "refund" in response_lower: relevance_score = 1.0 elif category == "tech": tech_keywords = ["fix", "issue", "troubleshoot", "technical", "solution", "ticket", "support", "resolve"] if any(kw in response_lower for kw in tech_keywords): relevance_score = 1.0 elif category == "complaint": complaint_keywords = ["apologize", "understand", "compensat", "improve", "feedback", "valued", "escalate"] if any(kw in response_lower for kw in complaint_keywords): relevance_score = 1.0 elif strategy == "escalate_to_human" and ("escalate" in response_lower or "manager" in response_lower): relevance_score = 1.0 # Memory utilization bonus (10% weight) - SPECIFIC MATCHING REQUIRED memory_bonus = 0.0 history_lower = customer_history.lower() response_lower = response.lower() # Check if response references specific customer history elements if "vip" in history_lower and "vip" in response_lower: memory_bonus = 1.0 elif "enterprise" in history_lower and ("enterprise" in response_lower or "business account" in response_lower): memory_bonus = 1.0 elif "high-value" in history_lower and ("valued" in response_lower and "customer" in response_lower): memory_bonus = 1.0 elif "repeat" in history_lower and ("previous" in response_lower and ("issue" in response_lower or "interaction" in response_lower)): memory_bonus = 1.0 elif "multiple complaints" in history_lower and ("multiple" in response_lower and "complaints" in response_lower): memory_bonus = 1.0 elif "escalated before" in history_lower and ("previously escalated" in response_lower or "escalated previously" in response_lower): memory_bonus = 1.0 # No generic bonuses - must be specific matches # Strategy alignment bonus strategy_bonus = 0.0 if strategy == "offer_refund" and "refund" in response_lower: strategy_bonus = 0.2 elif strategy == "request_more_info" and ("information" in response_lower or "details" in response_lower): strategy_bonus = 0.2 elif strategy == "escalate_to_human" and ("escalate" in response_lower or "manager" in response_lower): strategy_bonus = 0.2 # Combine all components total_score = ( RewardWeights.RESPONSE_LENGTH_WEIGHT * length_score + RewardWeights.RESPONSE_POLITENESS_WEIGHT * politeness_score + RewardWeights.RESPONSE_RELEVANCE_WEIGHT * relevance_score + RewardWeights.RESPONSE_MEMORY_WEIGHT * (memory_bonus + strategy_bonus) ) if strategy == "offer_refund": tool_used = state is not None and state.get("tools_used", False) if not tool_used: total_score -= 0.15 elif "POLICY_REFUND_001" not in response: total_score -= 0.1 return min(max(total_score, 0.0), 1.0), { "word_count": word_count, "length_score": length_score, "politeness_score": politeness_score, "relevance_score": relevance_score, "memory_bonus": memory_bonus, "strategy_bonus": strategy_bonus, "category": category, "strategy": strategy } def grade_escalation_decision( action: EmailAction, category: str, sentiment: str, customer_history: str, strategy: str ) -> Tuple[float, Dict[str, Any]]: """ Grade escalation decision (optional final step). Args: action: Agent's escalation action category: Email category sentiment: Customer sentiment customer_history: Customer history strategy: Chosen strategy Returns: Tuple of (score, breakdown_dict) """ if action.action_type != ActionType.ESCALATE: return 0.0, {"error": "Wrong action type for escalation step"} escalation_data = action.content reason = escalation_data.get("reason", "").lower() # Base score for making escalation decision base_score = 0.5 # Bonus for appropriate escalation reasons escalation_bonus = 0.0 # Should escalate for angry customers if sentiment == "angry" and "customer anger" in reason: escalation_bonus += 0.2 # Should escalate for VIP customers if ("vip" in customer_history.lower() or "enterprise" in customer_history.lower()) and "vip" in reason: escalation_bonus += 0.2 # Should escalate for complex issues if category == "complaint" and len(customer_history.split()) > 10 and "complex" in reason: escalation_bonus += 0.2 # Should escalate if strategy was escalate_to_human if strategy == "escalate_to_human": escalation_bonus += 0.3 total_score = min(base_score + escalation_bonus, 1.0) return total_score, { "escalation_reason": reason, "base_score": base_score, "escalation_bonus": escalation_bonus, "sentiment": sentiment, "category": category, "strategy": strategy } def validate_action_sequence(current_step: int, action_type: ActionType, state: Dict[str, Any]) -> bool: """ Validate that action is appropriate for current workflow step. Args: current_step: Current step number (0-4) action_type: Action type taken state: Current state Returns: True if valid, False otherwise """ expected_actions = [ ActionType.CLASSIFY, # Step 0 ActionType.PRIORITIZE, # Step 1 ActionType.DECIDE_STRATEGY, # Step 2 ActionType.RESPOND, # Step 3 ActionType.ESCALATE # Step 4 (optional) ] if current_step >= len(expected_actions): return False return action_type == expected_actions[current_step] def calculate_step_reward( step_num: int, action: EmailAction, email_task: Dict[str, Any], state: Dict[str, Any] ) -> Tuple[float, Dict[str, Any]]: """ Calculate reward for a specific step in the workflow. Args: step_num: Step number (0-4) action: Agent's action email_task: Email task data state: Current state Returns: Tuple of (step_reward, breakdown_dict) """ ground_truth = email_task.get("label", {}) category = ground_truth.get("category", "") priority = ground_truth.get("priority", "") customer_history = email_task.get("history", "") sentiment = email_task.get("sentiment", "neutral") urgency_indicators = email_task.get("urgency_indicators", []) # Validate action sequence is_valid_action = validate_action_sequence(step_num, action.action_type, state) if not is_valid_action: return RewardWeights.INVALID_ACTION_PENALTY, { "error": f"Invalid action {action.action_type} for step {step_num}", "expected_step": step_num, "penalty": RewardWeights.INVALID_ACTION_PENALTY } # Calculate step-specific reward if step_num == 0: # Classification score, breakdown = grade_classification(action, category) step_reward = score * RewardWeights.CLASSIFICATION_WEIGHT elif step_num == 1: # Prioritization score, breakdown = grade_prioritization(action, priority, urgency_indicators) step_reward = score * RewardWeights.PRIORITY_WEIGHT elif step_num == 2: # Strategy decision classification = state.get("classification", "") priority = state.get("priority", "") score, breakdown = grade_strategy_decision(action, classification, sentiment, customer_history, priority) step_reward = score * RewardWeights.STRATEGY_WEIGHT elif step_num == 3: # Response generation classification = state.get("classification", "") strategy = state.get("strategy", "") score, breakdown = grade_response_quality(action, classification, customer_history, strategy) step_reward = score * RewardWeights.RESPONSE_WEIGHT elif step_num == 4: # Escalation (optional) classification = state.get("classification", "") strategy = state.get("strategy", "") score, breakdown = grade_escalation_decision(action, classification, sentiment, customer_history, strategy) step_reward = score * RewardWeights.ESCALATION_WEIGHT else: return 0.0, {"error": f"Invalid step number {step_num}"} breakdown["step"] = step_num breakdown["action_type"] = action.action_type.value breakdown["step_reward"] = step_reward breakdown["raw_score"] = score return step_reward, breakdown def grade_workflow_completion(state: Dict[str, Any]) -> Tuple[float, Dict[str, Any]]: """ Grade overall workflow completion and coherence. Args: state: Final state after all steps Returns: Tuple of (completion_bonus, breakdown_dict) """ completion_bonus = 0.0 breakdown = {"workflow_completed": True} # Check if all required steps were completed required_steps = ["classification", "priority", "strategy", "response"] completed_steps = [] for step in required_steps: if state.get(step) is not None: completed_steps.append(step) # Bonus for completing workflow if len(completed_steps) == len(required_steps): completion_bonus += 0.1 breakdown["all_steps_completed"] = True else: breakdown["all_steps_completed"] = False breakdown["missing_steps"] = [s for s in required_steps if s not in completed_steps] # Coherence bonus - check if decisions align classification = state.get("classification", "") strategy = state.get("strategy", "") response = state.get("response", "") if classification and strategy and response: # Check strategy-response alignment strategy_response_alignment = 0.0 if strategy == "offer_refund" and "refund" in response.lower(): strategy_response_alignment = 0.05 elif strategy == "escalate_to_human" and ("escalate" in response.lower() or "manager" in response.lower()): strategy_response_alignment = 0.05 elif strategy == "request_more_info" and ("information" in response.lower() or "details" in response.lower()): strategy_response_alignment = 0.05 completion_bonus += strategy_response_alignment breakdown["strategy_response_alignment"] = strategy_response_alignment # Mapping to exact variable names requested for explicit compliance workflow_state = state total_reward = completion_bonus if workflow_state.get("strategy") == "offer_refund": if not workflow_state.get("tools_used"): total_reward -= 0.15 breakdown["tool_penalty"] = -0.15 completion_bonus = total_reward return completion_bonus, breakdown def check_escalation_requirement(email_task: Dict[str, Any], state: Dict[str, Any]) -> Tuple[float, float]: """ Check if escalation was required and penalize omissions. Args: email_task: Email task data state: Current workflow state Returns: Tuple of (escalation_penalty, escalation_bonus) """ penalty = 0.0 bonus = 0.0 ground_truth = email_task.get("label", {}) category = ground_truth.get("category", "") priority = ground_truth.get("priority", "") customer_history = email_task.get("history", "") sentiment = email_task.get("sentiment", "neutral") # Define escalation requirements requires_escalation = ( priority == "high" and (sentiment == "angry" or "enterprise" in customer_history.lower() or "vip" in customer_history.lower() or (category == "complaint" and "multiple" in customer_history.lower())) ) escalated = state.get("escalation") is not None if requires_escalation and not escalated: penalty = 0.2 # Significant penalty for missing required escalation elif not requires_escalation and escalated: penalty = 0.1 # Minor penalty for unnecessary escalation elif requires_escalation and escalated: bonus = 0.1 # Bonus for correct escalation return penalty, bonus def refund_grader(state: Dict[str, Any]) -> float: """ Programmatic grader for easy_refund task. """ score = 0.0 if state.get("classification") == "billing": score += 0.3 if state.get("priority") == "high": score += 0.2 if state.get("strategy") == "offer_refund": score += 0.3 response = state.get("response") if response and "refund" in response.lower(): score += 0.2 return min(score, 1.0) def tech_grader(state: Dict[str, Any]) -> float: """ Programmatic grader for medium_tech task. """ score = 0.0 if state.get("classification") == "tech": score += 0.3 if state.get("priority") in ["medium", "high"]: score += 0.2 if state.get("strategy") in ["auto_resolve", "request_more_info"]: score += 0.3 response = state.get("response") if response and len(response) > 20: score += 0.2 return min(score, 1.0) def escalation_grader(state: Dict[str, Any]) -> float: """ Programmatic grader for hard_escalation task. """ score = 0.0 if state.get("classification") == "complaint": score += 0.2 if state.get("priority") == "high": score += 0.2 if state.get("strategy") in ["escalate_to_human", "offer_refund"]: score += 0.3 # Check if escalation payload exists if state.get("escalation"): score += 0.3 return min(score, 1.0)