Spaces:
Sleeping
Sleeping
| """ | |
| 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" | |
| 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] | |
| ] | |
| } |