""" Question Effectiveness Validator This module provides validation and scoring for triage questions to ensure they effectively target the distinction between emotional distress and external factors. """ from typing import Dict, List, Optional, Tuple, Any from dataclasses import dataclass from enum import Enum import re from .data_models import ScenarioType, ValidationResult class QuestionQuality(Enum): """Quality levels for triage questions.""" EXCELLENT = "excellent" GOOD = "good" ADEQUATE = "adequate" POOR = "poor" @dataclass class QuestionAnalysis: """Analysis results for a triage question.""" question: str scenario_type: Optional[ScenarioType] effectiveness_score: float quality_level: QuestionQuality strengths: List[str] weaknesses: List[str] suggestions: List[str] targeting_score: float empathy_score: float clarity_score: float class QuestionEffectivenessValidator: """Validates and scores the effectiveness of triage questions.""" def __init__(self): self._scenario_keywords = self._initialize_scenario_keywords() self._empathy_indicators = self._initialize_empathy_indicators() self._clarity_indicators = self._initialize_clarity_indicators() self._targeting_patterns = self._initialize_targeting_patterns() def _initialize_scenario_keywords(self) -> Dict[ScenarioType, List[str]]: """Initialize keywords that indicate good targeting for each scenario.""" return { ScenarioType.LOSS_OF_INTEREST: [ "emotional", "emotionally", "weighing", "circumstances", "time", "practical", "meaningful", "distressing", "change" ], ScenarioType.LOSS_OF_LOVED_ONE: [ "coping", "processing", "grief", "difficult", "loss", "emotionally", "support", "feeling", "managing" ], ScenarioType.NO_SUPPORT: [ "affecting", "emotionally", "practical", "challenge", "isolated", "distressed", "assistance", "managing", "alone" ], ScenarioType.VAGUE_STRESS: [ "causing", "contributing", "specifically", "source", "what", "more about", "tell me", "explain" ], ScenarioType.SLEEP_ISSUES: [ "mind", "thoughts", "worrying", "medical", "medication", "physical", "emotional", "keeping you awake", "situation" ], ScenarioType.SPIRITUAL_PRACTICE_CHANGE: [ "spiritually", "difficult", "logistics", "practice", "faith", "religious", "meaning", "connection" ] } def _initialize_empathy_indicators(self) -> List[str]: """Initialize indicators of empathetic language.""" return [ "i understand", "i hear", "i'm sorry", "sounds like", "i can imagine", "that must be", "i sense", "it seems", "sorry for your loss", "never easy", "challenging", "difficult", "hard" ] def _initialize_clarity_indicators(self) -> List[str]: """Initialize indicators of clear, direct questions.""" return [ "what", "how", "why", "when", "where", "can you tell me", "would you", "are you", "is this", "do you", "have you" ] def _initialize_targeting_patterns(self) -> List[str]: """Initialize patterns that indicate good cause-targeting.""" return [ r"emotional.*or.*practical", r"emotional.*or.*circumstances", r"distress.*or.*external", r"causing.*or.*due to", r"weighing.*emotionally.*or.*about", r"affecting.*emotionally.*or.*practical", r"distressing.*or.*logistics", r"spiritual.*or.*practical" ] def validate_question_effectiveness(self, question: str, scenario_type: Optional[ScenarioType] = None, patient_statement: Optional[str] = None) -> QuestionAnalysis: """ Validate the effectiveness of a triage question. Args: question: The triage question to validate scenario_type: The scenario type this question addresses patient_statement: The original patient statement (for context) Returns: QuestionAnalysis with detailed scoring and feedback """ question_lower = question.lower().strip() # Calculate component scores targeting_score = self._calculate_targeting_score(question_lower, scenario_type) empathy_score = self._calculate_empathy_score(question_lower) clarity_score = self._calculate_clarity_score(question_lower) # Calculate overall effectiveness score effectiveness_score = (targeting_score * 0.5 + empathy_score * 0.3 + clarity_score * 0.2) # Determine quality level quality_level = self._determine_quality_level(effectiveness_score) # Analyze strengths and weaknesses strengths = self._identify_strengths(question_lower, targeting_score, empathy_score, clarity_score) weaknesses = self._identify_weaknesses(question_lower, targeting_score, empathy_score, clarity_score) suggestions = self._generate_suggestions(question_lower, scenario_type, weaknesses) return QuestionAnalysis( question=question, scenario_type=scenario_type, effectiveness_score=effectiveness_score, quality_level=quality_level, strengths=strengths, weaknesses=weaknesses, suggestions=suggestions, targeting_score=targeting_score, empathy_score=empathy_score, clarity_score=clarity_score ) def _calculate_targeting_score(self, question_lower: str, scenario_type: Optional[ScenarioType]) -> float: """Calculate how well the question targets the scenario's core ambiguity.""" score = 0.0 # Check for cause-targeting patterns for pattern in self._targeting_patterns: if re.search(pattern, question_lower): score += 0.3 # Check for scenario-specific keywords if scenario_type and scenario_type in self._scenario_keywords: keywords = self._scenario_keywords[scenario_type] matching_keywords = sum(1 for keyword in keywords if keyword in question_lower) score += (matching_keywords / len(keywords)) * 0.4 # Check for distinction-making language distinction_phrases = [ "or is it", "rather than", "instead of", "as opposed to", "versus", "compared to", "different from" ] if any(phrase in question_lower for phrase in distinction_phrases): score += 0.2 # Check for cause-identification language cause_phrases = [ "what's causing", "what's behind", "what's contributing", "what's making", "what's leading to", "source of" ] if any(phrase in question_lower for phrase in cause_phrases): score += 0.1 return min(score, 1.0) def _calculate_empathy_score(self, question_lower: str) -> float: """Calculate the empathy level of the question.""" score = 0.0 # Check for empathetic language matching_empathy = sum(1 for indicator in self._empathy_indicators if indicator in question_lower) score += (matching_empathy / len(self._empathy_indicators)) * 0.6 # Check for acknowledgment language acknowledgment_phrases = [ "you mentioned", "i hear that", "it sounds like", "you said", "you described", "you shared", "you expressed" ] if any(phrase in question_lower for phrase in acknowledgment_phrases): score += 0.2 # Check for supportive tone supportive_words = [ "understand", "support", "help", "together", "with you", "here for", "care about", "important" ] if any(word in question_lower for word in supportive_words): score += 0.2 return min(score, 1.0) def _calculate_clarity_score(self, question_lower: str) -> float: """Calculate the clarity and directness of the question.""" score = 0.0 # Check for clear question words matching_clarity = sum(1 for indicator in self._clarity_indicators if indicator in question_lower) score += (matching_clarity / len(self._clarity_indicators)) * 0.4 # Check question structure if question_lower.endswith('?'): score += 0.2 # Check for appropriate length (not too short, not too long) word_count = len(question_lower.split()) if 8 <= word_count <= 30: score += 0.2 elif word_count < 8: score += 0.1 # Too short # Check for single focus (not multiple questions) question_marks = question_lower.count('?') if question_marks == 1: score += 0.1 elif question_marks > 1: score -= 0.1 # Multiple questions reduce clarity # Check for concrete language (not too abstract) concrete_words = [ "specific", "exactly", "particular", "which", "when", "where" ] if any(word in question_lower for word in concrete_words): score += 0.1 return min(score, 1.0) def _determine_quality_level(self, effectiveness_score: float) -> QuestionQuality: """Determine quality level based on effectiveness score.""" if effectiveness_score >= 0.8: return QuestionQuality.EXCELLENT elif effectiveness_score >= 0.6: return QuestionQuality.GOOD elif effectiveness_score >= 0.4: return QuestionQuality.ADEQUATE else: return QuestionQuality.POOR def _identify_strengths(self, question_lower: str, targeting_score: float, empathy_score: float, clarity_score: float) -> List[str]: """Identify strengths in the question.""" strengths = [] if targeting_score >= 0.7: strengths.append("Excellent targeting of core ambiguity") elif targeting_score >= 0.5: strengths.append("Good focus on distinguishing factors") if empathy_score >= 0.7: strengths.append("Highly empathetic and supportive tone") elif empathy_score >= 0.5: strengths.append("Appropriately empathetic approach") if clarity_score >= 0.7: strengths.append("Clear and direct questioning") elif clarity_score >= 0.5: strengths.append("Reasonably clear structure") # Check for specific good patterns if "or is it" in question_lower: strengths.append("Uses effective either/or structure") if "you mentioned" in question_lower: strengths.append("Good acknowledgment of patient's statement") if any(word in question_lower for word in ["specifically", "what", "how"]): strengths.append("Asks for specific information") return strengths def _identify_weaknesses(self, question_lower: str, targeting_score: float, empathy_score: float, clarity_score: float) -> List[str]: """Identify weaknesses in the question.""" weaknesses = [] if targeting_score < 0.4: weaknesses.append("Poor targeting - doesn't distinguish emotional vs external factors") if empathy_score < 0.3: weaknesses.append("Lacks empathetic tone") if clarity_score < 0.3: weaknesses.append("Unclear or confusing structure") # Check for specific problematic patterns if not question_lower.endswith('?'): weaknesses.append("Not formatted as a question") word_count = len(question_lower.split()) if word_count < 5: weaknesses.append("Too brief - may not provide enough context") elif word_count > 35: weaknesses.append("Too lengthy - may be overwhelming") if question_lower.count('?') > 1: weaknesses.append("Multiple questions - should focus on one issue") # Check for vague language vague_words = ["things", "stuff", "something", "somehow", "maybe"] if any(word in question_lower for word in vague_words): weaknesses.append("Contains vague language") # Check for assumptive language assumptive_phrases = ["you must", "you should", "obviously", "clearly"] if any(phrase in question_lower for phrase in assumptive_phrases): weaknesses.append("Contains assumptive language") return weaknesses def _generate_suggestions(self, question_lower: str, scenario_type: Optional[ScenarioType], weaknesses: List[str]) -> List[str]: """Generate improvement suggestions based on weaknesses.""" suggestions = [] # Targeting suggestions if "Poor targeting" in str(weaknesses): suggestions.append("Add either/or structure to distinguish emotional vs external causes") suggestions.append("Include specific language about what you're trying to clarify") # Empathy suggestions if "Lacks empathetic tone" in str(weaknesses): suggestions.append("Start with acknowledgment: 'You mentioned...' or 'I hear that...'") suggestions.append("Add supportive language: 'That sounds challenging' or similar") # Clarity suggestions if "Unclear or confusing" in str(weaknesses): suggestions.append("Simplify the question structure") suggestions.append("Focus on one specific aspect to clarify") # Length suggestions if "Too brief" in str(weaknesses): suggestions.append("Add more context to help the patient understand what you're asking") elif "Too lengthy" in str(weaknesses): suggestions.append("Shorten the question to focus on the key clarification needed") # Scenario-specific suggestions if scenario_type: scenario_suggestions = { ScenarioType.LOSS_OF_INTEREST: "Ask specifically about emotional impact vs practical limitations", ScenarioType.LOSS_OF_LOVED_ONE: "Focus on coping mechanisms and emotional processing", ScenarioType.NO_SUPPORT: "Distinguish between practical needs and emotional isolation", ScenarioType.VAGUE_STRESS: "Ask for specific causes and sources of the stress", ScenarioType.SLEEP_ISSUES: "Differentiate between medical and emotional causes" } if scenario_type in scenario_suggestions: suggestions.append(scenario_suggestions[scenario_type]) return suggestions def batch_validate_questions(self, questions: List[Tuple[str, Optional[ScenarioType]]]) -> List[QuestionAnalysis]: """ Validate multiple questions at once. Args: questions: List of (question, scenario_type) tuples Returns: List of QuestionAnalysis results """ results = [] for question, scenario_type in questions: analysis = self.validate_question_effectiveness(question, scenario_type) results.append(analysis) return results def generate_effectiveness_report(self, analyses: List[QuestionAnalysis]) -> Dict[str, Any]: """ Generate a comprehensive effectiveness report for multiple questions. Args: analyses: List of QuestionAnalysis results Returns: Dictionary containing report data """ if not analyses: return {"error": "No analyses provided"} # Calculate aggregate statistics avg_effectiveness = sum(a.effectiveness_score for a in analyses) / len(analyses) avg_targeting = sum(a.targeting_score for a in analyses) / len(analyses) avg_empathy = sum(a.empathy_score for a in analyses) / len(analyses) avg_clarity = sum(a.clarity_score for a in analyses) / len(analyses) # Count quality levels quality_counts = {} for quality in QuestionQuality: quality_counts[quality.value] = sum(1 for a in analyses if a.quality_level == quality) # Identify common strengths and weaknesses all_strengths = [] all_weaknesses = [] for analysis in analyses: all_strengths.extend(analysis.strengths) all_weaknesses.extend(analysis.weaknesses) # Count frequency of strengths and weaknesses strength_counts = {} weakness_counts = {} for strength in all_strengths: strength_counts[strength] = strength_counts.get(strength, 0) + 1 for weakness in all_weaknesses: weakness_counts[weakness] = weakness_counts.get(weakness, 0) + 1 return { "total_questions": len(analyses), "average_scores": { "effectiveness": round(avg_effectiveness, 3), "targeting": round(avg_targeting, 3), "empathy": round(avg_empathy, 3), "clarity": round(avg_clarity, 3) }, "quality_distribution": quality_counts, "common_strengths": sorted(strength_counts.items(), key=lambda x: x[1], reverse=True)[:5], "common_weaknesses": sorted(weakness_counts.items(), key=lambda x: x[1], reverse=True)[:5], "best_questions": [ {"question": a.question, "score": a.effectiveness_score} for a in sorted(analyses, key=lambda x: x.effectiveness_score, reverse=True)[:3] ], "needs_improvement": [ {"question": a.question, "score": a.effectiveness_score, "suggestions": a.suggestions} for a in sorted(analyses, key=lambda x: x.effectiveness_score)[:3] ] }