""" Data models for the prompt management system. """ from dataclasses import dataclass, field from datetime import datetime from typing import List, Dict, Optional, Any from enum import Enum class IndicatorCategory(Enum): """Categories for spiritual distress indicators.""" EMOTIONAL = "emotional" SPIRITUAL = "spiritual" SOCIAL = "social" EXISTENTIAL = "existential" PHYSICAL = "physical" class ScenarioType(Enum): """Types of YELLOW scenarios for targeted questioning.""" LOSS_OF_INTEREST = "loss_of_interest" LOSS_OF_LOVED_ONE = "loss_of_loved_one" NO_SUPPORT = "no_support" VAGUE_STRESS = "vague_stress" SLEEP_ISSUES = "sleep_issues" SPIRITUAL_PRACTICE_CHANGE = "spiritual_practice_change" @dataclass class Indicator: """Represents a spiritual distress indicator.""" name: str category: IndicatorCategory definition: str examples: List[str] severity_weight: float context_requirements: List[str] = field(default_factory=list) def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for serialization.""" return { 'name': self.name, 'category': self.category.value, 'definition': self.definition, 'examples': self.examples, 'severity_weight': self.severity_weight, 'context_requirements': self.context_requirements } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'Indicator': """Create from dictionary.""" return cls( name=data['name'], category=IndicatorCategory(data['category']), definition=data['definition'], examples=data['examples'], severity_weight=data['severity_weight'], context_requirements=data.get('context_requirements', []) ) @dataclass class Rule: """Represents a classification rule.""" rule_id: str description: str condition: str action: str priority: int examples: List[str] = field(default_factory=list) def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for serialization.""" return { 'rule_id': self.rule_id, 'description': self.description, 'condition': self.condition, 'action': self.action, 'priority': self.priority, 'examples': self.examples } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'Rule': """Create from dictionary.""" return cls( rule_id=data['rule_id'], description=data['description'], condition=data['condition'], action=data['action'], priority=data['priority'], examples=data.get('examples', []) ) @dataclass class Template: """Represents a reusable prompt template.""" template_id: str name: str content: str variables: List[str] category: str def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for serialization.""" return { 'template_id': self.template_id, 'name': self.name, 'content': self.content, 'variables': self.variables, 'category': self.category } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'Template': """Create from dictionary.""" return cls( template_id=data['template_id'], name=data['name'], content=data['content'], variables=data['variables'], category=data['category'] ) @dataclass class QuestionPattern: """Represents a question pattern for YELLOW scenarios.""" pattern_id: str scenario_type: ScenarioType template: str target_clarification: str examples: List[str] = field(default_factory=list) def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for serialization.""" return { 'pattern_id': self.pattern_id, 'scenario_type': self.scenario_type.value, 'template': self.template, 'target_clarification': self.target_clarification, 'examples': self.examples } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'QuestionPattern': """Create from dictionary.""" return cls( pattern_id=data['pattern_id'], scenario_type=ScenarioType(data['scenario_type']), template=data['template'], target_clarification=data['target_clarification'], examples=data.get('examples', []) ) @dataclass class YellowScenario: """Represents a YELLOW scenario for targeted questioning.""" scenario_type: ScenarioType patient_statement: str context_clues: List[str] target_clarification: str question_patterns: List[QuestionPattern] def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for serialization.""" return { 'scenario_type': self.scenario_type.value, 'patient_statement': self.patient_statement, 'context_clues': self.context_clues, 'target_clarification': self.target_clarification, 'question_patterns': [p.to_dict() for p in self.question_patterns] } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'YellowScenario': """Create from dictionary.""" return cls( scenario_type=ScenarioType(data['scenario_type']), patient_statement=data['patient_statement'], context_clues=data['context_clues'], target_clarification=data['target_clarification'], question_patterns=[QuestionPattern.from_dict(p) for p in data['question_patterns']] ) @dataclass class PromptConfig: """Configuration for a specific AI agent prompt.""" agent_type: str base_prompt: str shared_indicators: List[Indicator] shared_rules: List[Rule] templates: List[Template] version: str last_updated: datetime session_override: Optional[str] = None def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for serialization.""" return { 'agent_type': self.agent_type, 'base_prompt': self.base_prompt, 'shared_indicators': [i.to_dict() for i in self.shared_indicators], 'shared_rules': [r.to_dict() for r in self.shared_rules], 'templates': [t.to_dict() for t in self.templates], 'version': self.version, 'last_updated': self.last_updated.isoformat(), 'session_override': self.session_override } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'PromptConfig': """Create from dictionary.""" return cls( agent_type=data['agent_type'], base_prompt=data['base_prompt'], shared_indicators=[Indicator.from_dict(i) for i in data['shared_indicators']], shared_rules=[Rule.from_dict(r) for r in data['shared_rules']], templates=[Template.from_dict(t) for t in data['templates']], version=data['version'], last_updated=datetime.fromisoformat(data['last_updated']), session_override=data.get('session_override') ) @dataclass class ValidationResult: """Result of prompt validation.""" is_valid: bool errors: List[str] = field(default_factory=list) warnings: List[str] = field(default_factory=list) def add_error(self, error: str): """Add an error to the result.""" self.errors.append(error) self.is_valid = False def add_warning(self, warning: str): """Add a warning to the result.""" self.warnings.append(warning) @dataclass class Message: """Represents a single message in conversation history.""" content: str classification: str timestamp: datetime confidence: float = 0.0 def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for serialization.""" return { 'content': self.content, 'classification': self.classification, 'timestamp': self.timestamp.isoformat(), 'confidence': self.confidence } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'Message': """Create from dictionary.""" return cls( content=data['content'], classification=data['classification'], timestamp=datetime.fromisoformat(data['timestamp']), confidence=data.get('confidence', 0.0) ) @dataclass class Classification: """Represents a classification result with context.""" category: str confidence: float reasoning: str indicators_found: List[str] = None context_factors: List[str] = None def __post_init__(self): if self.indicators_found is None: self.indicators_found = [] if self.context_factors is None: self.context_factors = [] def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for serialization.""" return { 'category': self.category, 'confidence': self.confidence, 'reasoning': self.reasoning, 'indicators_found': self.indicators_found, 'context_factors': self.context_factors } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'Classification': """Create from dictionary.""" return cls( category=data['category'], confidence=data['confidence'], reasoning=data['reasoning'], indicators_found=data.get('indicators_found', []), context_factors=data.get('context_factors', []) ) @dataclass class ConversationHistory: """Represents conversation history for context-aware classification.""" messages: List[Message] distress_indicators_found: List[str] context_flags: List[str] medical_context: Dict[str, Any] = None def __post_init__(self): if self.medical_context is None: self.medical_context = {'conditions': [], 'medications': []} def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for serialization.""" return { 'messages': [msg.to_dict() for msg in self.messages], 'distress_indicators_found': self.distress_indicators_found, 'context_flags': self.context_flags, 'medical_context': self.medical_context } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'ConversationHistory': """Create from dictionary.""" return cls( messages=[Message.from_dict(msg) for msg in data['messages']], distress_indicators_found=data['distress_indicators_found'], context_flags=data['context_flags'], medical_context=data.get('medical_context', {'conditions': [], 'medications': []}) ) class ErrorType(Enum): """Types of classification errors for structured feedback.""" WRONG_CLASSIFICATION = "wrong_classification" SEVERITY_MISJUDGMENT = "severity_misjudgment" MISSED_INDICATORS = "missed_indicators" FALSE_POSITIVE = "false_positive" CONTEXT_MISUNDERSTANDING = "context_misunderstanding" LANGUAGE_INTERPRETATION = "language_interpretation" class ErrorSubcategory(Enum): """Subcategories for classification errors.""" # Wrong Classification subcategories GREEN_TO_YELLOW = "green_to_yellow" GREEN_TO_RED = "green_to_red" YELLOW_TO_GREEN = "yellow_to_green" YELLOW_TO_RED = "yellow_to_red" RED_TO_GREEN = "red_to_green" RED_TO_YELLOW = "red_to_yellow" # Severity Misjudgment subcategories UNDERESTIMATED_DISTRESS = "underestimated_distress" OVERESTIMATED_DISTRESS = "overestimated_distress" # Missed Indicators subcategories EMOTIONAL_INDICATORS = "emotional_indicators" SPIRITUAL_INDICATORS = "spiritual_indicators" SOCIAL_INDICATORS = "social_indicators" # False Positive subcategories MISINTERPRETED_STATEMENT = "misinterpreted_statement" CULTURAL_MISUNDERSTANDING = "cultural_misunderstanding" # Context Misunderstanding subcategories IGNORED_HISTORY = "ignored_history" MISSED_DEFENSIVE_RESPONSE = "missed_defensive_response" # Language Interpretation subcategories LITERAL_INTERPRETATION = "literal_interpretation" MISSED_SUBTEXT = "missed_subtext" class QuestionIssueType(Enum): """Types of issues with triage questions.""" INAPPROPRIATE_QUESTION = "inappropriate_question" INSENSITIVE_LANGUAGE = "insensitive_language" WRONG_SCENARIO_TARGETING = "wrong_scenario_targeting" UNCLEAR_QUESTION = "unclear_question" LEADING_QUESTION = "leading_question" class ReferralProblemType(Enum): """Types of problems with referral generation.""" INCOMPLETE_SUMMARY = "incomplete_summary" MISSING_CONTACT_INFO = "missing_contact_info" INCORRECT_URGENCY = "incorrect_urgency" POOR_CONTEXT_DESCRIPTION = "poor_context_description" @dataclass class ClassificationError: """Represents a classification error for structured feedback.""" error_id: str error_type: ErrorType subcategory: ErrorSubcategory expected_category: str # GREEN, YELLOW, RED actual_category: str # GREEN, YELLOW, RED message_content: str reviewer_comments: str confidence_level: float # 0.0 to 1.0 timestamp: datetime session_id: Optional[str] = None additional_context: Dict[str, Any] = field(default_factory=dict) def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for serialization.""" return { 'error_id': self.error_id, 'error_type': self.error_type.value, 'subcategory': self.subcategory.value, 'expected_category': self.expected_category, 'actual_category': self.actual_category, 'message_content': self.message_content, 'reviewer_comments': self.reviewer_comments, 'confidence_level': self.confidence_level, 'timestamp': self.timestamp.isoformat(), 'session_id': self.session_id, 'additional_context': self.additional_context } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'ClassificationError': """Create from dictionary.""" return cls( error_id=data['error_id'], error_type=ErrorType(data['error_type']), subcategory=ErrorSubcategory(data['subcategory']), expected_category=data['expected_category'], actual_category=data['actual_category'], message_content=data['message_content'], reviewer_comments=data['reviewer_comments'], confidence_level=data['confidence_level'], timestamp=datetime.fromisoformat(data['timestamp']), session_id=data.get('session_id'), additional_context=data.get('additional_context', {}) ) @dataclass class QuestionIssue: """Represents an issue with triage question generation.""" issue_id: str issue_type: QuestionIssueType question_content: str scenario_type: ScenarioType reviewer_comments: str severity: str # low, medium, high timestamp: datetime session_id: Optional[str] = None suggested_improvement: Optional[str] = None def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for serialization.""" return { 'issue_id': self.issue_id, 'issue_type': self.issue_type.value, 'question_content': self.question_content, 'scenario_type': self.scenario_type.value, 'reviewer_comments': self.reviewer_comments, 'severity': self.severity, 'timestamp': self.timestamp.isoformat(), 'session_id': self.session_id, 'suggested_improvement': self.suggested_improvement } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'QuestionIssue': """Create from dictionary.""" return cls( issue_id=data['issue_id'], issue_type=QuestionIssueType(data['issue_type']), question_content=data['question_content'], scenario_type=ScenarioType(data['scenario_type']), reviewer_comments=data['reviewer_comments'], severity=data['severity'], timestamp=datetime.fromisoformat(data['timestamp']), session_id=data.get('session_id'), suggested_improvement=data.get('suggested_improvement') ) @dataclass class ReferralProblem: """Represents a problem with referral generation.""" problem_id: str problem_type: ReferralProblemType referral_content: str reviewer_comments: str severity: str # low, medium, high timestamp: datetime session_id: Optional[str] = None missing_fields: List[str] = field(default_factory=list) def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for serialization.""" return { 'problem_id': self.problem_id, 'problem_type': self.problem_type.value, 'referral_content': self.referral_content, 'reviewer_comments': self.reviewer_comments, 'severity': self.severity, 'timestamp': self.timestamp.isoformat(), 'session_id': self.session_id, 'missing_fields': self.missing_fields } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'ReferralProblem': """Create from dictionary.""" return cls( problem_id=data['problem_id'], problem_type=ReferralProblemType(data['problem_type']), referral_content=data['referral_content'], reviewer_comments=data['reviewer_comments'], severity=data['severity'], timestamp=datetime.fromisoformat(data['timestamp']), session_id=data.get('session_id'), missing_fields=data.get('missing_fields', []) ) @dataclass class ErrorPattern: """Represents a pattern identified in classification errors.""" pattern_id: str pattern_type: str description: str frequency: int affected_scenarios: List[ScenarioType] suggested_improvements: List[str] confidence_score: float def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for serialization.""" return { 'pattern_id': self.pattern_id, 'pattern_type': self.pattern_type, 'description': self.description, 'frequency': self.frequency, 'affected_scenarios': [s.value for s in self.affected_scenarios], 'suggested_improvements': self.suggested_improvements, 'confidence_score': self.confidence_score } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'ErrorPattern': """Create from dictionary.""" return cls( pattern_id=data['pattern_id'], pattern_type=data['pattern_type'], description=data['description'], frequency=data['frequency'], affected_scenarios=[ScenarioType(s) for s in data['affected_scenarios']], suggested_improvements=data['suggested_improvements'], confidence_score=data['confidence_score'] )