diff --git "a/app.py" "b/app.py" --- "a/app.py" +++ "b/app.py" @@ -2,1458 +2,1053 @@ import streamlit as st import pandas as pd import json import os -from typing import Union, List, Dict, Optional, Tuple, Any +from typing import Union, List, Dict, Optional, Tuple from groq import Groq from duckduckgo_search import DDGS from datetime import datetime, timedelta import time import numpy as np import pickle -from dataclasses import dataclass, asdict, field +from dataclasses import dataclass, asdict import hashlib from collections import defaultdict import re -from enum import Enum -class UrgencyLevel(Enum): - LOW = "low" - MODERATE = "moderate" - HIGH = "high" - EMERGENCY = "emergency" +# Set page configuration +st.set_page_config( + page_title="MedChat - AI Doctor Consultation", + layout="wide", + initial_sidebar_state="expanded", + page_icon="👩‍⚕️" +) -class ConversationState(Enum): - GREETING = "greeting" - SYMPTOM_GATHERING = "symptom_gathering" - CLARIFICATION = "clarification" - ASSESSMENT = "assessment" - RECOMMENDATION = "recommendation" - FOLLOW_UP = "follow_up" - -@dataclass -class PatientProfile: - """Enhanced patient profile for personalized care""" - age_range: Optional[str] = None - gender: Optional[str] = None - chronic_conditions: List[str] = field(default_factory=list) - medications: List[str] = field(default_factory=list) - allergies: List[str] = field(default_factory=list) - lifestyle_factors: Dict[str, Any] = field(default_factory=dict) - previous_symptoms: List[str] = field(default_factory=list) - risk_factors: List[str] = field(default_factory=list) +# Enhanced CSS for conversational medical theme +st.markdown(""" + +""", unsafe_allow_html=True) @dataclass -class SymptomAnalysis: - """Comprehensive symptom analysis structure""" - primary_symptoms: List[str] - secondary_symptoms: List[str] - duration: Optional[str] - severity: float # 1-10 scale - progression: str # improving, worsening, stable - triggers: List[str] - relieving_factors: List[str] - associated_symptoms: List[str] - red_flags: List[str] # Warning signs - urgency_level: UrgencyLevel +class ConversationState: + """Track the current state of the medical conversation""" + phase: str = "greeting" # greeting, history_taking, symptom_exploration, assessment, recommendations, follow_up + patient_concerns: List[str] = None + symptoms_discussed: Dict[str, dict] = None + medical_history: Dict[str, any] = None + current_focus: str = "" + questions_asked: List[str] = None + patient_responses: Dict[str, str] = None + conversation_depth: int = 0 + urgency_level: str = "routine" # routine, urgent, emergency + + def __post_init__(self): + if self.patient_concerns is None: + self.patient_concerns = [] + if self.symptoms_discussed is None: + self.symptoms_discussed = {} + if self.medical_history is None: + self.medical_history = {} + if self.questions_asked is None: + self.questions_asked = [] + if self.patient_responses is None: + self.patient_responses = {} @dataclass -class ConversationContext: - """Track conversation flow and context""" - state: ConversationState - questions_asked: List[str] - information_gathered: Dict[str, Any] - clarifications_needed: List[str] - conversation_depth: int - patient_engagement: float # 0-1 scale - conversation_quality: float # 0-1 scale +class ConversationEntry: + """Enhanced conversation entry with conversational context""" + timestamp: str + user_input: str + assistant_response: str + conversation_phase: str + symptoms: List[str] + severity_score: float + confidence_score: float + search_queries_used: List[str] + follow_up_questions: List[str] + agent_insights: Dict[str, str] + user_feedback: Optional[int] = None + was_helpful: Optional[bool] = None class ConversationalMedicalAgent: - """Enhanced conversational medical agent with specialized capabilities""" + """Enhanced medical agent with conversational capabilities""" - def __init__(self, agent_id: str, specialization: str, personality_traits: Dict[str, float] = None): + def __init__(self, agent_id: str, specialization: str, personality: str = "professional"): self.agent_id = agent_id self.specialization = specialization - self.personality_traits = personality_traits or { - 'empathy': 0.8, - 'directness': 0.6, - 'technical_detail': 0.5, - 'reassurance': 0.7, - 'proactive_questioning': 0.8 - } - - # Conversation management - self.conversation_context = ConversationContext( - state=ConversationState.GREETING, - questions_asked=[], - information_gathered={}, - clarifications_needed=[], - conversation_depth=0, - patient_engagement=1.0, - conversation_quality=1.0 - ) - - # Knowledge and learning - self.knowledge_base = defaultdict(float) - self.conversation_patterns = defaultdict(list) - self.success_patterns = defaultdict(float) - - # Specialized question banks + self.personality = personality + self.conversation_patterns = self._load_conversation_patterns() self.question_bank = self._initialize_question_bank() - self.conversation_flows = self._initialize_conversation_flows() - - # Performance tracking - self.total_conversations = 0 - self.successful_assessments = 0 - self.patient_satisfaction_scores = [] - self.learning_rate = 0.02 + self.response_templates = self._load_response_templates() - def _initialize_question_bank(self) -> Dict[str, List[str]]: - """Initialize specialized question banks for different scenarios""" + def _load_conversation_patterns(self) -> Dict[str, List[str]]: + """Load conversational patterns for natural dialogue""" return { - 'symptom_exploration': [ - "Can you describe when you first noticed this symptom?", - "How would you rate the severity on a scale of 1-10?", - "Does anything make it better or worse?", - "Have you noticed any patterns or triggers?", - "Are there any other symptoms you've experienced alongside this?" + "greeting": [ + "Hello! I'm here to help with your health concerns. What brings you in today?", + "Good to see you today. How are you feeling, and what can I help you with?", + "Welcome! I'd like to understand what's been concerning you about your health lately." ], - 'pain_assessment': [ - "Where exactly do you feel the pain?", - "How would you describe the pain - sharp, dull, throbbing, burning?", - "Does the pain radiate or spread to other areas?", - "What were you doing when the pain started?", - "Have you taken anything for the pain?" - ], - 'timeline_clarification': [ - "When did this first start?", - "Has it been getting better, worse, or staying the same?", - "Have you had this problem before?", - "How long have you been dealing with this?" - ], - 'lifestyle_context': [ - "Have you made any recent changes to your routine?", + "follow_up": [ + "Can you tell me more about that?", + "When did you first notice this?", "How has this been affecting your daily activities?", - "Have you been under more stress than usual lately?", - "How has your sleep been?" + "Have you noticed any patterns with this symptom?" ], - 'medical_history': [ - "Do you have any ongoing medical conditions?", - "Are you currently taking any medications?", - "Do you have any known allergies?", - "Has anyone in your family had similar issues?" + "empathy": [ + "I can understand how concerning that must be for you.", + "That does sound uncomfortable. Let's explore this further.", + "Thank you for sharing that with me. It helps me understand better." + ], + "clarification": [ + "Just to make sure I understand correctly...", + "Let me clarify what you're experiencing...", + "I want to make sure I have the full picture..." ] } - def _initialize_conversation_flows(self) -> Dict[str, List[str]]: - """Initialize conversation flow patterns for different conditions""" + def _initialize_question_bank(self) -> Dict[str, List[str]]: + """Initialize comprehensive question bank for different scenarios""" return { - 'respiratory': [ - 'symptom_exploration', 'timeline_clarification', 'lifestyle_context' - ], - 'cardiovascular': [ - 'pain_assessment', 'symptom_exploration', 'medical_history', 'lifestyle_context' - ], - 'neurological': [ - 'symptom_exploration', 'timeline_clarification', 'pain_assessment' + "symptom_exploration": [ + "On a scale of 1-10, how would you rate the intensity?", + "Does anything make it better or worse?", + "How long have you been experiencing this?", + "Does it happen at specific times of day?", + "Have you tried anything to relieve it?" ], - 'gastrointestinal': [ - 'symptom_exploration', 'timeline_clarification', 'lifestyle_context' + "medical_history": [ + "Do you have any chronic medical conditions?", + "Are you currently taking any medications?", + "Any family history of similar issues?", + "Have you had any surgeries or hospitalizations?", + "Any known allergies to medications?" ], - 'musculoskeletal': [ - 'pain_assessment', 'symptom_exploration', 'lifestyle_context' + "lifestyle": [ + "How would you describe your stress levels lately?", + "How is your sleep quality?", + "Any recent changes in diet or exercise?", + "Do you smoke or drink alcohol?", + "Any recent travel or exposure to illness?" ], - 'mental_health': [ - 'symptom_exploration', 'timeline_clarification', 'lifestyle_context' + "associated_symptoms": [ + "Are you experiencing any other symptoms alongside this?", + "Any fever, nausea, or dizziness?", + "How is your appetite?", + "Any changes in bowel movements or urination?", + "Any skin changes or rashes?" ] } - def generate_conversational_response(self, user_input: str, patient_profile: PatientProfile, - conversation_history: List[str] = None) -> Dict[str, Any]: - """Generate conversational response with context awareness""" - - # Analyze user input - input_analysis = self._analyze_user_input(user_input) - - # Update conversation context - self._update_conversation_context(user_input, input_analysis) - - # Determine response strategy - response_strategy = self._determine_response_strategy(input_analysis, patient_profile) - - # Generate appropriate response - response = self._generate_contextual_response( - user_input, input_analysis, response_strategy, conversation_history - ) - - # Update learning from interaction - self._update_learning(user_input, response, input_analysis) - + def _load_response_templates(self) -> Dict[str, str]: + """Load response templates for different conversation phases""" return { - 'response': response['text'], - 'questions_to_ask': response.get('questions', []), - 'urgency_assessment': response.get('urgency', UrgencyLevel.LOW), - 'conversation_state': self.conversation_context.state, - 'information_gathered': dict(self.conversation_context.information_gathered), - 'next_steps': response.get('next_steps', []), - 'confidence_score': response.get('confidence', 0.7), - 'empathy_indicators': response.get('empathy_cues', []) + "symptom_acknowledgment": "I see that you're experiencing {symptom}. That can certainly be {emotion_word}. Let me ask you a few questions to better understand what might be going on.", + "information_gathering": "Based on what you've told me about {symptom}, I'd like to gather some more information. {question}", + "assessment": "From what you've described - {symptom_summary} - there are a few possibilities we should consider. {assessment_details}", + "reassurance": "I want to reassure you that {concern} is {reassurance_level}. However, it's important that we {recommended_action}.", + "urgent_referral": "Based on your symptoms, especially {urgent_symptoms}, I recommend that you {urgent_action} as this may require immediate attention." } - def _analyze_user_input(self, user_input: str) -> Dict[str, Any]: - """Comprehensive analysis of user input""" - analysis = { - 'symptoms_mentioned': [], - 'pain_descriptors': [], - 'temporal_indicators': [], - 'severity_indicators': [], - 'emotional_state': 'neutral', - 'question_type': 'statement', - 'medical_terminology_used': False, - 'urgency_markers': [], - 'information_completeness': 0.5 - } - - input_lower = user_input.lower() + def generate_conversational_response(self, query: str, conversation_state: ConversationState, + search_results: str = "") -> Tuple[str, List[str], Dict[str, any]]: + """Generate contextual, conversational response""" - # Extract symptoms - symptom_keywords = [ - 'pain', 'ache', 'hurt', 'sore', 'fever', 'headache', 'nausea', 'dizzy', - 'tired', 'fatigue', 'cough', 'sneeze', 'rash', 'swelling', 'numbness', - 'tingling', 'burning', 'stiffness', 'weakness', 'shortness of breath' - ] + # Determine conversation phase and adjust response + phase_response = self._determine_response_strategy(query, conversation_state) - for symptom in symptom_keywords: - if symptom in input_lower: - analysis['symptoms_mentioned'].append(symptom) + # Generate main response based on specialization and conversation context + main_response = self._craft_specialized_response(query, conversation_state, search_results) - # Pain descriptors - pain_words = ['sharp', 'dull', 'throbbing', 'burning', 'stabbing', 'cramping', 'aching'] - analysis['pain_descriptors'] = [word for word in pain_words if word in input_lower] + # Generate follow-up questions + follow_up_questions = self._generate_follow_up_questions(query, conversation_state) - # Temporal indicators - time_words = ['yesterday', 'today', 'week', 'month', 'sudden', 'gradual', 'chronic'] - analysis['temporal_indicators'] = [word for word in time_words if word in input_lower] - - # Severity indicators - severity_words = { - 'mild': ['slight', 'mild', 'little', 'minor'], - 'moderate': ['moderate', 'noticeable', 'bothersome'], - 'severe': ['severe', 'intense', 'unbearable', 'excruciating', 'terrible'] + # Provide agent insights + insights = { + "specialization_note": f"From a {self.specialization} perspective, {self._get_specialist_insight(query)}", + "conversation_guidance": self._suggest_conversation_direction(conversation_state), + "urgency_assessment": self._assess_urgency(query, conversation_state) } - for level, words in severity_words.items(): - if any(word in input_lower for word in words): - analysis['severity_indicators'].append(level) - - # Emotional state - worry_words = ['worried', 'concerned', 'scared', 'anxious', 'frightened'] - calm_words = ['fine', 'okay', 'manageable', 'better'] - - if any(word in input_lower for word in worry_words): - analysis['emotional_state'] = 'anxious' - elif any(word in input_lower for word in calm_words): - analysis['emotional_state'] = 'calm' - - # Urgency markers - urgent_words = ['emergency', 'urgent', 'severe', 'can\'t breathe', 'chest pain', 'blood'] - analysis['urgency_markers'] = [word for word in urgent_words if word in input_lower] - - # Question detection - if '?' in user_input or user_input.lower().startswith(('what', 'how', 'when', 'where', 'why', 'should', 'can', 'is', 'are')): - analysis['question_type'] = 'question' - - return analysis - - def _update_conversation_context(self, user_input: str, analysis: Dict[str, Any]): - """Update conversation context based on user input""" - self.conversation_context.conversation_depth += 1 - - # Update information gathered - if analysis['symptoms_mentioned']: - if 'symptoms' not in self.conversation_context.information_gathered: - self.conversation_context.information_gathered['symptoms'] = [] - self.conversation_context.information_gathered['symptoms'].extend(analysis['symptoms_mentioned']) - - if analysis['temporal_indicators']: - self.conversation_context.information_gathered['timeline'] = analysis['temporal_indicators'] - - if analysis['severity_indicators']: - self.conversation_context.information_gathered['severity'] = analysis['severity_indicators'] - - # Update conversation state based on depth and information - if self.conversation_context.conversation_depth == 1: - self.conversation_context.state = ConversationState.SYMPTOM_GATHERING - elif len(self.conversation_context.information_gathered) < 3: - self.conversation_context.state = ConversationState.CLARIFICATION - elif self.conversation_context.conversation_depth > 3: - self.conversation_context.state = ConversationState.ASSESSMENT - - # Update patient engagement based on response completeness - response_length = len(user_input.split()) - if response_length < 5: - self.conversation_context.patient_engagement *= 0.9 - elif response_length > 20: - self.conversation_context.patient_engagement = min(1.0, self.conversation_context.patient_engagement * 1.1) + return main_response, follow_up_questions, insights - def _determine_response_strategy(self, analysis: Dict[str, Any], patient_profile: PatientProfile) -> Dict[str, Any]: - """Determine appropriate response strategy based on context""" - strategy = { - 'primary_goal': 'gather_information', - 'tone': 'empathetic', - 'detail_level': 'moderate', - 'question_priority': [], - 'urgency_response': False - } + def _determine_response_strategy(self, query: str, state: ConversationState) -> str: + """Determine the appropriate response strategy based on conversation state""" + query_lower = query.lower() - # Adjust based on emotional state - if analysis['emotional_state'] == 'anxious': - strategy['tone'] = 'reassuring' - strategy['primary_goal'] = 'provide_comfort' - - # Adjust based on urgency markers - if analysis['urgency_markers']: - strategy['urgency_response'] = True - strategy['primary_goal'] = 'emergency_assessment' - strategy['tone'] = 'direct' - - # Adjust based on conversation depth - if self.conversation_context.conversation_depth < 3: - strategy['primary_goal'] = 'gather_information' - strategy['question_priority'] = self._prioritize_questions(analysis) - else: - strategy['primary_goal'] = 'provide_assessment' - strategy['detail_level'] = 'high' + # Check for emergency indicators + emergency_keywords = ["chest pain", "can't breathe", "severe", "emergency", "blood", "unconscious"] + if any(keyword in query_lower for keyword in emergency_keywords): + state.urgency_level = "emergency" + return "emergency_response" - return strategy + # Check for urgent indicators + urgent_keywords = ["intense", "sudden", "worsening", "spreading", "difficulty"] + if any(keyword in query_lower for keyword in urgent_keywords): + state.urgency_level = "urgent" + return "urgent_response" + + # Determine conversation phase + if state.phase == "greeting": + return "initial_response" + elif state.phase == "history_taking": + return "information_gathering" + elif state.phase == "symptom_exploration": + return "detailed_exploration" + else: + return "general_response" - def _prioritize_questions(self, analysis: Dict[str, Any]) -> List[str]: - """Prioritize questions based on information gaps""" - priorities = [] - - gathered = self.conversation_context.information_gathered + def _craft_specialized_response(self, query: str, state: ConversationState, search_results: str) -> str: + """Craft a specialized response based on agent expertise""" - # High priority: missing critical information - if not gathered.get('symptoms'): - priorities.append('symptom_exploration') + # Base response structure + response_parts = [] - if 'pain' in analysis.get('symptoms_mentioned', []) and not analysis.get('pain_descriptors'): - priorities.append('pain_assessment') + # Acknowledgment and empathy + if state.conversation_depth == 0: # First interaction + response_parts.append(self._get_greeting_response(query)) + else: + response_parts.append(self._get_acknowledgment(query, state)) - if not gathered.get('timeline'): - priorities.append('timeline_clarification') + # Specialized analysis + specialist_analysis = self._provide_specialist_analysis(query, search_results) + if specialist_analysis: + response_parts.append(specialist_analysis) - # Medium priority: contextual information - if not gathered.get('lifestyle_factors'): - priorities.append('lifestyle_context') + # Guidance and recommendations + guidance = self._provide_guidance(query, state) + if guidance: + response_parts.append(guidance) - if not gathered.get('medical_history'): - priorities.append('medical_history') + # Medical disclaimer + response_parts.append("Please remember that this guidance is for informational purposes. For a proper diagnosis and treatment plan, it's important to consult with a healthcare provider.") - return priorities + return "\n\n".join(response_parts) - def _generate_contextual_response(self, user_input: str, analysis: Dict[str, Any], - strategy: Dict[str, Any], history: List[str] = None) -> Dict[str, Any]: - """Generate contextual response based on strategy""" - - response = { - 'text': '', - 'questions': [], - 'urgency': UrgencyLevel.LOW, - 'next_steps': [], - 'confidence': 0.7, - 'empathy_cues': [] + def _generate_follow_up_questions(self, query: str, state: ConversationState) -> List[str]: + """Generate contextually relevant follow-up questions""" + questions = [] + query_lower = query.lower() + + # Symptom-specific questions + if "pain" in query_lower: + questions.extend([ + "Where exactly is the pain located?", + "How would you describe the pain - sharp, dull, throbbing?", + "What were you doing when the pain started?" + ]) + + if "headache" in query_lower: + questions.extend([ + "Where on your head do you feel the pain?", + "How often do you get headaches like this?", + "Any visual changes or sensitivity to light?" + ]) + + # General health questions based on conversation phase + if state.phase == "greeting": + questions.extend([ + "How long have you been experiencing this?", + "Have you noticed any other symptoms?", + "Is this your first time having this issue?" + ]) + + # Limit to most relevant questions + return questions[:3] + + def _get_specialist_insight(self, query: str) -> str: + """Provide specialist insight based on agent's expertise""" + insights = { + "General Practice Medicine": "this requires a comprehensive evaluation of your overall health and symptoms", + "Symptom Analysis and Triage": "we need to prioritize your symptoms and determine the appropriate level of care", + "Preventive Care and Wellness": "there may be lifestyle factors we can address to prevent this from recurring", + "Mental Health and Psychology": "stress and mental health can significantly impact physical symptoms", + "Emergency Assessment and Urgent Care": "we need to quickly assess if this requires immediate medical attention" } - - # Handle emergency situations first - if strategy['urgency_response']: - response.update(self._handle_urgent_situation(analysis)) - return response - - # Generate empathetic opening based on emotional state - empathy_opening = self._generate_empathy_opening(analysis, strategy) - - # Generate main response based on conversation state - if self.conversation_context.state == ConversationState.GREETING: - main_response = self._generate_greeting_response() - elif self.conversation_context.state == ConversationState.SYMPTOM_GATHERING: - main_response = self._generate_information_gathering_response(analysis, strategy) - elif self.conversation_context.state == ConversationState.CLARIFICATION: - main_response = self._generate_clarification_response(analysis, strategy) - elif self.conversation_context.state == ConversationState.ASSESSMENT: - main_response = self._generate_assessment_response(analysis) - else: - main_response = self._generate_follow_up_response() - - # Combine response parts - response['text'] = f"{empathy_opening} {main_response['text']}" - response['questions'] = main_response.get('questions', []) - response['next_steps'] = main_response.get('next_steps', []) - response['confidence'] = main_response.get('confidence', 0.7) - - # Add medical disclaimer - response['text'] += "\n\n*Please remember that this is preliminary guidance and should not replace professional medical consultation.*" - - return response + return insights.get(self.specialization, "this warrants careful evaluation") + +class ConversationalMedicalSystem: + """Enhanced medical system with natural conversation flow""" - def _generate_empathy_opening(self, analysis: Dict[str, Any], strategy: Dict[str, Any]) -> str: - """Generate empathetic opening based on patient's emotional state""" - - if analysis['emotional_state'] == 'anxious': - openings = [ - "I understand this must be concerning for you.", - "I can sense you're worried about this - that's completely natural.", - "Thank you for sharing this with me. I'm here to help." - ] - elif analysis['symptoms_mentioned']: - openings = [ - "I hear that you're experiencing some symptoms.", - "Thank you for describing what you're going through.", - "Let me help you understand what might be happening." - ] - else: - openings = [ - "I'm listening and want to help.", - "Tell me more about what's bothering you.", - "I appreciate you taking the time to share this." - ] + def __init__(self): + self.llm = GroqLLM() + self.search_tool = MedicalSearchTool() + self.agents = self._initialize_conversational_agents() + self.conversation_state = ConversationState() + self.conversation_history = [] + self.conversation_data = [] - # Adjust based on personality traits - empathy_level = self.personality_traits.get('empathy', 0.5) - if empathy_level > 0.8: - return openings[0] + " I want to make sure we address your concerns thoroughly." - elif empathy_level > 0.5: - return openings[0] - else: - return "Let's work through this together." - - def _generate_information_gathering_response(self, analysis: Dict[str, Any], strategy: Dict[str, Any]) -> Dict[str, Any]: - """Generate response focused on gathering information""" - - response = { - 'text': '', - 'questions': [], - 'next_steps': [], - 'confidence': 0.6 + def _initialize_conversational_agents(self) -> Dict[str, ConversationalMedicalAgent]: + """Initialize conversational medical agents""" + return { + "primary_physician": ConversationalMedicalAgent("primary", "General Practice Medicine", "warm"), + "symptom_specialist": ConversationalMedicalAgent("symptom", "Symptom Analysis and Triage", "analytical"), + "wellness_coach": ConversationalMedicalAgent("wellness", "Preventive Care and Wellness", "encouraging"), + "mental_health_counselor": ConversationalMedicalAgent("mental", "Mental Health and Psychology", "empathetic"), + "emergency_consultant": ConversationalMedicalAgent("emergency", "Emergency Assessment and Urgent Care", "direct") } - - # Acknowledge what they've shared - if analysis['symptoms_mentioned']: - symptoms_text = ', '.join(analysis['symptoms_mentioned']) - response['text'] = f"I understand you're experiencing {symptoms_text}. " - - # Ask follow-up questions based on priority - question_categories = strategy.get('question_priority', ['symptom_exploration']) - - for category in question_categories[:2]: # Limit to 2 categories per response - if category in self.question_bank: - questions = self.question_bank[category] - # Select appropriate question based on what we know - selected_question = self._select_contextual_question(questions, analysis) - if selected_question: - response['questions'].append(selected_question) - - # Add main question to response text - if response['questions']: - response['text'] += f"To better understand your situation: {response['questions'][0]}" - - response['next_steps'] = ['Continue symptom assessment', 'Gather timeline information'] - - return response - def _select_contextual_question(self, questions: List[str], analysis: Dict[str, Any]) -> Optional[str]: - """Select most appropriate question based on context""" - - gathered = self.conversation_context.information_gathered + def process_conversational_query(self, user_input: str) -> Dict: + """Process query with full conversational context""" - # Filter out questions we've already asked - available_questions = [q for q in questions if q not in self.conversation_context.questions_asked] + timestamp = datetime.now().isoformat() + self.conversation_state.conversation_depth += 1 - if not available_questions: - return None + # Update conversation phase based on input and history + self._update_conversation_phase(user_input) - # Select based on what information we're missing - if 'severity' not in gathered and any('severity' in q.lower() or 'scale' in q.lower() for q in available_questions): - selected = next((q for q in available_questions if 'severity' in q.lower() or 'scale' in q.lower()), None) - elif 'timeline' not in gathered and any('when' in q.lower() or 'start' in q.lower() for q in available_questions): - selected = next((q for q in available_questions if 'when' in q.lower() or 'start' in q.lower()), None) - else: - selected = available_questions[0] - - if selected: - self.conversation_context.questions_asked.append(selected) - - return selected - - def _generate_assessment_response(self, analysis: Dict[str, Any]) -> Dict[str, Any]: - """Generate assessment and recommendations""" - - response = { - 'text': '', - 'next_steps': [], - 'confidence': 0.8 - } - - # Summarize what we've learned - gathered = self.conversation_context.information_gathered - symptoms = gathered.get('symptoms', []) - severity = gathered.get('severity', ['mild']) - timeline = gathered.get('timeline', ['recent']) + # Extract and update patient information + self._extract_patient_information(user_input) - response['text'] = f"Based on our conversation, you're experiencing {', '.join(symptoms)} " + # Select primary agent for this interaction + primary_agent = self._select_primary_agent(user_input) - if severity: - response['text'] += f"with {severity[0]} intensity " + # Get supporting agents + supporting_agents = self._select_supporting_agents(user_input) - if timeline: - response['text'] += f"that started {timeline[0]}. " + # Perform medical search if needed + search_results = "" + if self._should_search(user_input): + search_results = self.search_tool.search_medical_info(user_input, self._determine_search_type()) - # Provide preliminary assessment - assessment = self._generate_preliminary_assessment(symptoms, severity, timeline) - response['text'] += assessment + # Generate conversational response from primary agent + main_response, follow_up_questions, primary_insights = primary_agent.generate_conversational_response( + user_input, self.conversation_state, search_results + ) - # Recommend next steps - next_steps = self._recommend_next_steps(symptoms, severity, analysis) - response['next_steps'] = next_steps + # Get insights from supporting agents + supporting_insights = {} + for agent_name in supporting_agents: + agent = self.agents[agent_name] + _, _, insights = agent.generate_conversational_response(user_input, self.conversation_state, search_results) + supporting_insights[agent_name] = insights + + # Generate LLM-enhanced response for natural conversation + enhanced_response = self._enhance_with_llm(user_input, main_response, self.conversation_state) + + # Update conversation tracking + self._update_conversation_tracking(user_input, enhanced_response) + + # Create conversation entry + entry = ConversationEntry( + timestamp=timestamp, + user_input=user_input, + assistant_response=enhanced_response, + conversation_phase=self.conversation_state.phase, + symptoms=list(self.conversation_state.symptoms_discussed.keys()), + severity_score=self._calculate_severity(), + confidence_score=0.8, # Placeholder + search_queries_used=[user_input] if search_results else [], + follow_up_questions=follow_up_questions, + agent_insights={primary_agent.agent_id: primary_insights, **supporting_insights} + ) - response['text'] += f"\n\nI recommend: {'. '.join(next_steps)}" + self.conversation_data.append(entry) - return response + return { + 'response': enhanced_response, + 'follow_up_questions': follow_up_questions, + 'conversation_phase': self.conversation_state.phase, + 'urgency_level': self.conversation_state.urgency_level, + 'primary_agent': primary_agent.agent_id, + 'supporting_agents': supporting_agents, + 'agent_insights': {primary_agent.agent_id: primary_insights, **supporting_insights}, + 'patient_summary': self._get_patient_summary(), + 'search_performed': bool(search_results) + } - def _generate_preliminary_assessment(self, symptoms: List[str], severity: List[str], timeline: List[str]) -> str: - """Generate preliminary assessment based on gathered information""" + def _update_conversation_phase(self, user_input: str): + """Update conversation phase based on user input and history""" + input_lower = user_input.lower() - # This is a simplified assessment logic - # In a real system, this would be much more sophisticated + if self.conversation_state.conversation_depth == 1: + self.conversation_state.phase = "greeting" + elif any(word in input_lower for word in ["history", "medical", "medication", "allergy"]): + self.conversation_state.phase = "history_taking" + elif any(word in input_lower for word in ["pain", "symptom", "feel", "hurt"]): + self.conversation_state.phase = "symptom_exploration" + elif any(word in input_lower for word in ["what should", "recommend", "treatment", "help"]): + self.conversation_state.phase = "recommendations" + elif self.conversation_state.conversation_depth > 3: + self.conversation_state.phase = "follow_up" + + def _extract_patient_information(self, user_input: str): + """Extract and store patient information from conversation""" + input_lower = user_input.lower() - assessments = { - ('headache', 'mild'): "This could be related to stress, dehydration, or tension.", - ('headache', 'severe'): "Severe headaches warrant medical attention, especially if they're unusual for you.", - ('fever', 'any'): "Fever often indicates your body is fighting an infection.", - ('pain', 'severe'): "Severe pain should be evaluated by a healthcare professional.", + # Extract symptoms + symptom_patterns = { + 'headache': ['headache', 'head pain', 'migraine'], + 'fever': ['fever', 'temperature', 'hot', 'chills'], + 'nausea': ['nausea', 'sick', 'throw up', 'vomit'], + 'fatigue': ['tired', 'exhausted', 'fatigue', 'weak'], + 'cough': ['cough', 'coughing'], + 'pain': ['pain', 'hurt', 'ache', 'sore'] } - # Simple pattern matching - for symptom in symptoms: - severity_level = severity[0] if severity else 'any' - - for (assess_symptom, assess_severity), assessment in assessments.items(): - if symptom == assess_symptom and (assess_severity == 'any' or assess_severity == severity_level): - return assessment + for symptom, patterns in symptom_patterns.items(): + if any(pattern in input_lower for pattern in patterns): + if symptom not in self.conversation_state.symptoms_discussed: + self.conversation_state.symptoms_discussed[symptom] = { + 'first_mentioned': datetime.now().isoformat(), + 'details': [] + } + self.conversation_state.symptoms_discussed[symptom]['details'].append(user_input) + + # Extract duration information + duration_patterns = [ + r'(\d+)\s+(day|days|week|weeks|month|months)', + r'(yesterday|today|last night)', + r'(few days|several days|about a week)' + ] - return "Based on your symptoms, it would be good to monitor how you're feeling and consider speaking with a healthcare provider." + for pattern in duration_patterns: + matches = re.search(pattern, input_lower) + if matches: + self.conversation_state.medical_history['duration'] = matches.group(0) + break - def _recommend_next_steps(self, symptoms: List[str], severity: List[str], analysis: Dict[str, Any]) -> List[str]: - """Recommend appropriate next steps""" - - steps = [] + def _select_primary_agent(self, user_input: str) -> ConversationalMedicalAgent: + """Select the primary agent to handle this conversation turn""" + input_lower = user_input.lower() - # Check for urgent indicators - if analysis.get('urgency_markers') or (severity and 'severe' in severity): - steps.append("Seek immediate medical attention") - return steps - - # General recommendations based on symptoms - if 'fever' in symptoms: - steps.append("Monitor your temperature regularly") - steps.append("Stay hydrated and rest") - - if 'pain' in symptoms: - steps.append("Consider over-the-counter pain relief if appropriate") - steps.append("Apply ice or heat as feels comfortable") - - if 'headache' in symptoms: - steps.append("Ensure you're staying hydrated") - steps.append("Try to rest in a quiet, dark room") - - # Always recommend professional consultation - if severity and 'severe' not in severity: - steps.append("Schedule an appointment with your healthcare provider if symptoms persist or worsen") - else: - steps.append("Consult with a healthcare professional for proper diagnosis and treatment") + # Emergency situations + emergency_keywords = ["chest pain", "can't breathe", "emergency", "severe", "blood"] + if any(keyword in input_lower for keyword in emergency_keywords): + return self.agents["emergency_consultant"] - return steps or ["Monitor your symptoms and consult with a healthcare provider if you have concerns"] - - def _handle_urgent_situation(self, analysis: Dict[str, Any]) -> Dict[str, Any]: - """Handle urgent medical situations""" - - urgent_response = { - 'text': "⚠️ Based on what you've described, this may require immediate medical attention. ", - 'urgency': UrgencyLevel.EMERGENCY, - 'next_steps': [ - "Call emergency services (911) immediately", - "Do not drive yourself to the hospital", - "If possible, have someone stay with you" - ], - 'confidence': 0.9, - 'questions': [] - } + # Mental health focus + mental_keywords = ["stress", "anxiety", "depression", "worried", "scared", "panic"] + if any(keyword in input_lower for keyword in mental_keywords): + return self.agents["mental_health_counselor"] - # Specific urgent situations - urgency_markers = analysis.get('urgency_markers', []) + # Wellness and prevention + wellness_keywords = ["prevent", "healthy", "lifestyle", "diet", "exercise"] + if any(keyword in input_lower for keyword in wellness_keywords): + return self.agents["wellness_coach"] - if 'chest pain' in urgency_markers: - urgent_response['text'] += "Chest pain can be a sign of a serious condition and needs immediate evaluation." - elif 'can\'t breathe' in urgency_markers: - urgent_response['text'] += "Difficulty breathing requires immediate medical care." - elif 'blood' in urgency_markers: - urgent_response['text'] += "Significant bleeding or blood in unexpected places needs urgent attention." - else: - urgent_response['text'] += "Please seek immediate medical care." + # Symptom analysis + symptom_keywords = ["symptom", "pain", "hurt", "feel", "experience"] + if any(keyword in input_lower for keyword in symptom_keywords): + return self.agents["symptom_specialist"] - return urgent_response + # Default to primary physician + return self.agents["primary_physician"] - def _generate_greeting_response(self) -> Dict[str, Any]: - """Generate initial greeting response""" - return { - 'text': f"Hello! I'm your {self.specialization} AI assistant. I'm here to help you understand your health concerns and guide you toward appropriate care. What's bringing you here today?", - 'questions': ["What symptoms or health concerns would you like to discuss?"], - 'next_steps': ['Begin symptom assessment'], - 'confidence': 0.8 - } - - def _generate_clarification_response(self, analysis: Dict[str, Any], strategy: Dict[str, Any]) -> Dict[str, Any]: - """Generate response seeking clarification""" + def _enhance_with_llm(self, user_input: str, agent_response: str, state: ConversationState) -> str: + """Enhance response with LLM for natural conversation""" - gathered = self.conversation_context.information_gathered - missing_info = [] + conversation_context = "\n".join(self.conversation_history[-6:]) # Last 3 exchanges - if not gathered.get('severity'): - missing_info.append("severity") - if not gathered.get('timeline'): - missing_info.append("when it started") + prompt = f""" + You are having a natural, empathetic medical conversation with a patient. - response_text = "Thank you for that information. To better help you, I'd like to understand " + Conversation context: {conversation_context} - if len(missing_info) == 1: - response_text += f"more about the {missing_info[0]}." - else: - response_text += f"more about {' and '.join(missing_info)}." + Patient just said: "{user_input}" - return { - 'text': response_text, - 'questions': self._generate_clarification_questions(missing_info), - 'next_steps': ['Complete symptom assessment'], - 'confidence': 0.7 - } - - def _generate_clarification_questions(self, missing_info: List[str]) -> List[str]: - """Generate specific clarification questions""" - questions = [] + Current conversation phase: {state.phase} + Patient's concerns so far: {', '.join(state.patient_concerns)} + Symptoms discussed: {', '.join(state.symptoms_discussed.keys())} + + Agent suggested response: {agent_response} - if "severity" in missing_info: - questions.append("On a scale of 1-10, how would you rate the intensity of your symptoms?") + Please provide a natural, conversational response that: + 1. Shows empathy and understanding + 2. Builds on the conversation naturally + 3. Asks relevant follow-up questions + 4. Maintains a professional but warm tone + 5. Always emphasizes the need for professional medical care - if "when it started" in missing_info: - questions.append("When did you first notice these symptoms?") + Keep the response conversational and avoid being too clinical or robotic. + """ - return questions + try: + enhanced_response, _ = self.llm.generate_response(prompt, self.conversation_history) + return enhanced_response + except Exception as e: + return agent_response # Fallback to agent response - def _generate_follow_up_response(self) -> Dict[str, Any]: - """Generate follow-up response""" + def _get_patient_summary(self) -> Dict: + """Generate a summary of the patient's current situation""" return { - 'text': "Is there anything else about your symptoms or health concerns that you'd like to discuss?", - 'questions': ["Any other symptoms or concerns you'd like to address?"], - 'next_steps': ['Provide additional support', 'Encourage professional consultation'], - 'confidence': 0.6 + 'chief_concerns': self.conversation_state.patient_concerns, + 'symptoms_discussed': list(self.conversation_state.symptoms_discussed.keys()), + 'conversation_depth': self.conversation_state.conversation_depth, + 'current_phase': self.conversation_state.phase, + 'urgency_level': self.conversation_state.urgency_level, + 'questions_asked': len(self.conversation_state.questions_asked) } + +# Continue with the rest of the implementation... +# [The rest of the classes like GroqLLM, MedicalSearchTool would remain the same but with enhanced conversational capabilities] + +class GroqLLM: + """Medical-optimized LLM client with conversational enhancement""" - def _update_learning(self, user_input: str, response: Dict[str, Any], analysis: Dict[str, Any]): - """Update agent learning based on interaction""" - - # Track conversation patterns - pattern_key = f"{self.conversation_context.state.value}_{analysis['question_type']}" - self.conversation_patterns[pattern_key].append({ - 'input_analysis': analysis, - 'response_strategy': response, - 'conversation_depth': self.conversation_context.conversation_depth - }) - - # Update knowledge base - for symptom in analysis.get('symptoms_mentioned', []): - self.knowledge_base[symptom] += 0.1 - - # Update success patterns (simplified - in real system would be based on outcomes) - if response.get('confidence', 0) > 0.8: - self.success_patterns[pattern_key] += 0.1 + def __init__(self, model_name="llama-3.1-70b-versatile"): + self.client = Groq(api_key=os.environ.get("GROQ_API_KEY")) + self.model_name = model_name + self.medical_context = """ + You are an empathetic AI medical assistant having a natural conversation with a patient. + Your responses should be: + - Conversational and warm, not clinical or robotic + - Empathetic to the patient's concerns + - Informative but accessible + - Always emphasizing the importance of professional medical care + + Remember: This is a conversation, not a medical exam. Show genuine care and understanding. + """ + + def generate_response(self, prompt: str, conversation_history: List[str] = None) -> Tuple[str, float]: + """Generate conversational response""" + try: + context = self.medical_context + if conversation_history: + recent_history = conversation_history[-6:] # Last 3 exchanges + context += f"\n\nRecent conversation:\n{chr(10).join(recent_history)}" + + full_prompt = f"{context}\n\n{prompt}" + + completion = self.client.chat.completions.create( + model=self.model_name, + messages=[{"role": "user", "content": full_prompt}], + temperature=0.7, # Higher temperature for more natural conversation + max_tokens=1000, + stream=False + ) + + response = completion.choices[0].message.content if completion.choices else "I'm here to help. Could you tell me more about what's concerning you?" + confidence = self._calculate_conversational_confidence(response, prompt) + + return response, confidence + + except Exception as e: + return "I'm here to listen and help. Could you share more about what's been concerning you?", 0.5 - def get_agent_status(self) -> Dict[str, Any]: - """Get current agent status and learning metrics""" + def _calculate_conversational_confidence(self, response: str, query: str) -> float: + """Calculate confidence for conversational responses""" + confidence = 0.5 # Base confidence - total_patterns = sum(len(patterns) for patterns in self.conversation_patterns.values()) - avg_success = np.mean(list(self.success_patterns.values())) if self.success_patterns else 0 + # Check for empathetic language + empathy_words = ["understand", "concerning", "help", "support", "care", "here for you"] + if any(word in response.lower() for word in empathy_words): + confidence += 0.2 - return { - 'specialization': self.specialization, - 'total_conversations': self.total_conversations, - 'conversation_patterns_learned': total_patterns, - 'average_success_rate': avg_success, - 'current_conversation_depth': self.conversation_context.conversation_depth, - 'information_completeness': len(self.conversation_context.information_gathered) / 5.0, - 'patient_engagement': self.conversation_context.patient_engagement, - 'top_knowledge_areas': dict(sorted(self.knowledge_base.items(), key=lambda x: x[1], reverse=True)[:5]), - 'personality_traits': self.personality_traits, - 'conversation_state': self.conversation_context.state.value - } - - def reset_conversation(self): - """Reset conversation context for new patient interaction""" - self.conversation_context = ConversationContext( - state=ConversationState.GREETING, - questions_asked=[], - information_gathered={}, - clarifications_needed=[], - conversation_depth=0, - patient_engagement=1.0, - conversation_quality=1.0 - ) - self.total_conversations += 1 - -class SpecializedMedicalAgents: - """Factory for creating specialized medical agents with unique capabilities""" - - @staticmethod - def create_triage_agent() -> ConversationalMedicalAgent: - """Create emergency triage specialist""" - return ConversationalMedicalAgent( - agent_id="triage_specialist", - specialization="Emergency Triage and Urgent Care Assessment", - personality_traits={ - 'empathy': 0.7, - 'directness': 0.9, - 'technical_detail': 0.8, - 'reassurance': 0.6, - 'proactive_questioning': 0.9 - } - ) - - @staticmethod - def create_mental_health_agent() -> ConversationalMedicalAgent: - """Create mental health specialist""" - agent = ConversationalMedicalAgent( - agent_id="mental_health", - specialization="Mental Health and Psychological Wellbeing", - personality_traits={ - 'empathy': 0.9, - 'directness': 0.5, - 'technical_detail': 0.4, - 'reassurance': 0.9, - 'proactive_questioning': 0.8 - } - ) + # Check for medical disclaimers + if any(phrase in response.lower() for phrase in ["healthcare provider", "medical professional", "doctor", "consult"]): + confidence += 0.2 - # Add specialized mental health questions - agent.question_bank.update({ - 'mood_assessment': [ - "How has your mood been over the past few weeks?", - "Have you noticed changes in your energy levels?", - "How would you describe your sleep patterns lately?", - "Have you been feeling hopeless or overwhelmed?" - ], - 'anxiety_screening': [ - "Do you find yourself worrying more than usual?", - "Have you experienced any panic attacks or intense anxiety?", - "Are there specific situations that trigger your anxiety?", - "How has anxiety been affecting your daily life?" - ], - 'depression_screening': [ - "Have you lost interest in activities you usually enjoy?", - "How has your appetite been lately?", - "Do you find it hard to concentrate on tasks?", - "Have you had thoughts of hurting yourself?" - ] - }) + # Check for conversational elements + if any(phrase in response for phrase in ["?", "tell me", "share", "describe"]): + confidence += 0.1 - return agent - - @staticmethod - def create_pediatric_agent() -> ConversationalMedicalAgent: - """Create pediatric specialist for children's health""" - return ConversationalMedicalAgent( - agent_id="pediatric_care", - specialization="Pediatric Care and Child Health", - personality_traits={ - 'empathy': 0.9, - 'directness': 0.4, - 'technical_detail': 0.3, - 'reassurance': 0.9, - 'proactive_questioning': 0.8 - } - ) + return min(confidence, 1.0) + +class MedicalSearchTool: + """Medical search with conversational context""" - @staticmethod - def create_chronic_care_agent() -> ConversationalMedicalAgent: - """Create chronic disease management specialist""" - return ConversationalMedicalAgent( - agent_id="chronic_care", - specialization="Chronic Disease Management and Long-term Care", - personality_traits={ - 'empathy': 0.8, - 'directness': 0.7, - 'technical_detail': 0.8, - 'reassurance': 0.7, - 'proactive_questioning': 0.6 - } - ) + def __init__(self): + self.ddgs = DDGS() + self.medical_sources = [ + "mayoclinic.org", "webmd.com", "healthline.com", "medlineplus.gov", + "nih.gov", "who.int", "cdc.gov", "ncbi.nlm.nih.gov" + ] - @staticmethod - def create_womens_health_agent() -> ConversationalMedicalAgent: - """Create women's health specialist""" - return ConversationalMedicalAgent( - agent_id="womens_health", - specialization="Women's Health and Reproductive Care", - personality_traits={ - 'empathy': 0.8, - 'directness': 0.6, - 'technical_detail': 0.7, - 'reassurance': 0.8, - 'proactive_questioning': 0.7 + def search_medical_info(self, query: str, search_type: str = "symptoms") -> str: + """Search with conversational context awareness""" + try: + medical_queries = { + "symptoms": f"symptoms causes {query} patient concerns", + "treatment": f"treatment options {query} patient care", + "prevention": f"prevention {query} patient education", + "general": f"patient information {query} healthcare" } - ) + + enhanced_query = medical_queries.get(search_type, medical_queries["general"]) + + search_results = list(self.ddgs.text( + enhanced_query, + max_results=5, + region='wt-wt', + safesearch='on' + )) + + if not search_results: + return "I'll help guide our conversation, though I'd recommend discussing this with your healthcare provider for specific medical advice." + + formatted_results = [] + for idx, result in enumerate(search_results, 1): + title = result.get('title', 'No title') + snippet = result.get('body', 'No description') + url = result.get('href', 'No URL') + + # Prioritize trusted medical sources + source_trust = "⭐" if any(source in url for source in self.medical_sources) else "" + + formatted_results.append( + f"{idx}. {source_trust} {title}\n" + f" {snippet}\n" + f" Source: {url}\n" + ) + + return "\n".join(formatted_results) + + except Exception as e: + return f"I'm having trouble accessing medical resources right now, but I can still help guide our conversation based on general medical knowledge." -class ConversationalMedicalSystem: - """Enhanced conversational medical consultation system""" +# Initialize session state for conversational system +if 'conversational_system' not in st.session_state: + st.session_state.conversational_system = ConversationalMedicalSystem() +if 'chat_messages' not in st.session_state: + st.session_state.chat_messages = [] +if 'conversation_context' not in st.session_state: + st.session_state.conversation_context = {} +if 'typing' not in st.session_state: + st.session_state.typing = False + +medical_system = st.session_state.conversational_system + +# Main interface +st.markdown(""" +
+

👩‍⚕️ MedChat - Your AI Medical Companion

+

Natural, empathetic conversations about your health concerns

+
+""", unsafe_allow_html=True) + +# Medical disclaimer +st.markdown(""" +
+

⚠️ Important Medical Disclaimer

+

I'm here to have a supportive conversation about your health concerns and provide general information. However, I cannot diagnose or treat medical conditions. For proper medical care, please consult with qualified healthcare professionals. In emergencies, contact emergency services immediately.

+
+""", unsafe_allow_html=True) + +# Main chat layout +col_main, col_sidebar = st.columns([4, 1]) + +with col_main: + st.markdown("### 💬 Let's Talk About Your Health") - def __init__(self): - self.agents = self._initialize_agents() - self.current_agent = None - self.patient_profile = PatientProfile() - self.conversation_history = [] - self.session_data = { - 'start_time': datetime.now(), - 'interactions': [], - 'agent_switches': [], - 'final_assessment': None - } - - # Enhanced search and LLM integration - self.search_tool = MedicalSearchTool() - self.llm = GroqLLM() - - # Learning and analytics - self.system_metrics = { - 'total_sessions': 0, - 'successful_assessments': 0, - 'agent_performance': defaultdict(list), - 'common_symptoms': defaultdict(int), - 'user_satisfaction': [] + # Display conversation phase + if medical_system.conversation_state.phase: + phase_display = { + "greeting": "👋 Getting to know you", + "history_taking": "📋 Understanding your background", + "symptom_exploration": "🔍 Exploring your symptoms", + "assessment": "🧠 Assessing the situation", + "recommendations": "💡 Discussing next steps", + "follow_up": "🔄 Following up on your concerns" } + current_phase = phase_display.get(medical_system.conversation_state.phase, "💬 In conversation") + st.markdown(f'
{current_phase}
', unsafe_allow_html=True) - def _initialize_agents(self) -> Dict[str, ConversationalMedicalAgent]: - """Initialize all specialized agents""" - return { - 'triage': SpecializedMedicalAgents.create_triage_agent(), - 'primary_care': SpecializedMedicalAgents.create_primary_care_agent(), - 'mental_health': SpecializedMedicalAgents.create_mental_health_agent(), - 'pediatric': SpecializedMedicalAgents.create_pediatric_agent(), - 'chronic_care': SpecializedMedicalAgents.create_chronic_care_agent(), - 'womens_health': SpecializedMedicalAgents.create_womens_health_agent() - } + # Chat container + chat_container = st.container() + with chat_container: + st.markdown('
', unsafe_allow_html=True) + + # Display conversation history + for i, message in enumerate(st.session_state.chat_messages): + if message["role"] == "user": + st.markdown(f'
{message["content"]}
', unsafe_allow_html=True) + else: + st.markdown(f'
{message["content"]}
', unsafe_allow_html=True) + + # Show agent insights if available + if "insights" in message: + for agent, insight in message["insights"].items(): + if insight.get("specialization_note"): + st.markdown(f'
Specialist Note: {insight["specialization_note"]}
', unsafe_allow_html=True) + + # Show follow-up questions + if "follow_up_questions" in message and message["follow_up_questions"]: + st.markdown('
', unsafe_allow_html=True) + st.markdown("I'd like to know more about:") + for question in message["follow_up_questions"][:3]: # Show max 3 + if st.button(question, key=f"followup_{i}_{hash(question)}", help="Click to discuss this"): + st.session_state.chat_messages.append({"role": "user", "content": question}) + st.rerun() + st.markdown('
', unsafe_allow_html=True) + + # Typing indicator + if st.session_state.typing: + st.markdown('
Dr. AI is thinking... 💭
', unsafe_allow_html=True) + + st.markdown('
', unsafe_allow_html=True) - def start_conversation(self, initial_message: str = None) -> Dict[str, Any]: - """Start a new conversation session""" - - # Reset all agents for new conversation - for agent in self.agents.values(): - agent.reset_conversation() - - # Select initial agent based on first message if provided - if initial_message: - self.current_agent = self._select_appropriate_agent(initial_message) - else: - self.current_agent = self.agents['primary_care'] # Default to primary care - - self.session_data = { - 'start_time': datetime.now(), - 'interactions': [], - 'agent_switches': [], - 'final_assessment': None - } - - # Generate greeting - if initial_message: - return self.process_message(initial_message) - else: - greeting_response = self.current_agent.generate_conversational_response( - "", self.patient_profile, self.conversation_history - ) - return { - 'response': greeting_response['response'], - 'agent_used': self.current_agent.agent_id, - 'conversation_state': greeting_response['conversation_state'].value, - 'next_questions': greeting_response.get('questions_to_ask', []), - 'urgency_level': greeting_response['urgency_assessment'].value, - 'confidence': greeting_response['confidence_score'] - } + # Chat input area + st.markdown("---") - def process_message(self, user_message: str, context: Dict[str, Any] = None) -> Dict[str, Any]: - """Process user message through conversational system""" - - # Record interaction - interaction_start = datetime.now() - - # Check if we need to switch agents - agent_switch_needed = self._evaluate_agent_switch_need(user_message) - if agent_switch_needed: - old_agent = self.current_agent.agent_id - self.current_agent = self._select_appropriate_agent(user_message) - self.session_data['agent_switches'].append({ - 'timestamp': interaction_start.isoformat(), - 'from_agent': old_agent, - 'to_agent': self.current_agent.agent_id, - 'reason': agent_switch_needed - }) - - # Get search results if needed - search_results = "" - if self._should_search(user_message): - search_results = self.search_tool.search_medical_info(user_message) - - # Generate agent response - agent_response = self.current_agent.generate_conversational_response( - user_message, self.patient_profile, self.conversation_history - ) - - # Enhance response with LLM if needed - enhanced_response = self._enhance_with_llm(user_message, agent_response, search_results) - - # Update conversation history - self.conversation_history.extend([ - f"User: {user_message}", - f"Assistant ({self.current_agent.specialization}): {enhanced_response['response']}" - ]) - - # Update patient profile based on conversation - self._update_patient_profile(user_message, agent_response) - - # Record interaction metrics - interaction_data = { - 'timestamp': interaction_start.isoformat(), - 'user_input': user_message, - 'agent_used': self.current_agent.agent_id, - 'response_generated': enhanced_response['response'], - 'confidence_score': enhanced_response['confidence_score'], - 'urgency_level': enhanced_response['urgency_assessment'].value, - 'questions_asked': enhanced_response.get('questions_to_ask', []), - 'information_gathered': enhanced_response['information_gathered'], - 'processing_time': (datetime.now() - interaction_start).total_seconds() - } - - self.session_data['interactions'].append(interaction_data) - - # Update system metrics - self._update_system_metrics(interaction_data) - - return { - 'response': enhanced_response['response'], - 'agent_used': self.current_agent.agent_id, - 'agent_specialization': self.current_agent.specialization, - 'conversation_state': enhanced_response['conversation_state'].value, - 'next_questions': enhanced_response.get('questions_to_ask', []), - 'urgency_level': enhanced_response['urgency_assessment'].value, - 'confidence': enhanced_response['confidence_score'], - 'information_gathered': enhanced_response['information_gathered'], - 'recommended_actions': enhanced_response.get('next_steps', []), - 'search_performed': bool(search_results), - 'agent_switched': len(self.session_data['agent_switches']) > 0, - 'session_metrics': self._get_session_metrics() - } + # Conversation starters for new users + if len(st.session_state.chat_messages) == 0: + st.markdown("**How can I help you today?** You can start by:") + + starter_col1, starter_col2, starter_col3 = st.columns(3) + + with starter_col1: + if st.button("💬 Tell me about symptoms I'm having", key="starter_symptoms"): + starter_text = "I've been experiencing some symptoms and I'd like to talk about them." + st.session_state.chat_messages.append({"role": "user", "content": starter_text}) + st.rerun() + + with starter_col2: + if st.button("🤔 Ask about a health concern", key="starter_concern"): + starter_text = "I have a health concern I'd like to discuss." + st.session_state.chat_messages.append({"role": "user", "content": starter_text}) + st.rerun() + + with starter_col3: + if st.button("🩺 Get a health checkup conversation", key="starter_checkup"): + starter_text = "I'd like to have a general health discussion." + st.session_state.chat_messages.append({"role": "user", "content": starter_text}) + st.rerun() - def _select_appropriate_agent(self, message: str) -> ConversationalMedicalAgent: - """Select most appropriate agent based on message content""" - - message_lower = message.lower() - - # Emergency/urgent indicators - use triage agent - urgent_keywords = ['emergency', 'urgent', 'severe pain', 'chest pain', 'can\'t breathe', - 'blood', 'unconscious', 'severe', 'intense'] - if any(keyword in message_lower for keyword in urgent_keywords): - return self.agents['triage'] - - # Mental health indicators - mental_health_keywords = ['depressed', 'anxiety', 'panic', 'suicide', 'worried', 'stress', - 'mood', 'sleep problems', 'can\'t concentrate', 'hopeless'] - if any(keyword in message_lower for keyword in mental_health_keywords): - return self.agents['mental_health'] - - # Pediatric indicators - pediatric_keywords = ['child', 'baby', 'toddler', 'kid', 'infant', 'my son', 'my daughter', - 'year old', 'months old'] - if any(keyword in message_lower for keyword in pediatric_keywords): - return self.agents['pediatric'] - - # Women's health indicators - womens_health_keywords = ['pregnancy', 'period', 'menstrual', 'breast', 'gynecologic', - 'prenatal', 'postpartum', 'contraception'] - if any(keyword in message_lower for keyword in womens_health_keywords): - return self.agents['womens_health'] - - # Chronic care indicators - chronic_keywords = ['diabetes', 'hypertension', 'arthritis', 'chronic', 'ongoing', - 'long-term', 'medication management'] - if any(keyword in message_lower for keyword in chronic_keywords): - return self.agents['chronic_care'] - - # Default to primary care - return self.agents['primary_care'] + # Main input area + user_input = st.text_area( + "What would you like to talk about?", + placeholder="For example: 'I've been having headaches for the past few days...' or 'I'm worried about some symptoms I'm having...'", + height=100, + key="main_input" + ) - def _evaluate_agent_switch_need(self, message: str) -> Optional[str]: - """Evaluate if we need to switch to a different agent""" - - if not self.current_agent: - return "no_current_agent" - - optimal_agent = self._select_appropriate_agent(message) - - if optimal_agent.agent_id != self.current_agent.agent_id: - return f"better_specialization_{optimal_agent.agent_id}" - - # Check if current agent is struggling (low confidence, repeated questions) - if len(self.session_data['interactions']) > 3: - recent_interactions = self.session_data['interactions'][-3:] - avg_confidence = np.mean([i['confidence_score'] for i in recent_interactions]) - - if avg_confidence < 0.5: - return "low_confidence_pattern" - - return None + # Input controls + input_col1, input_col2, input_col3, input_col4 = st.columns([2, 1, 1, 2]) - def _should_search(self, message: str) -> bool: - """Determine if we should perform medical search""" - - search_indicators = [ - 'what is', 'tell me about', 'information about', 'research', 'studies', - 'treatment for', 'cause of', 'symptoms of' - ] - - return any(indicator in message.lower() for indicator in search_indicators) + with input_col1: + send_message = st.button("💬 Send Message", type="primary") - def _enhance_with_llm(self, user_message: str, agent_response: Dict[str, Any], - search_results: str) -> Dict[str, Any]: - """Enhance agent response with LLM-generated content""" - - if not search_results and agent_response['confidence_score'] > 0.8: - return agent_response # Agent response is already good - - # Prepare context for LLM - context = f""" - Agent Specialization: {self.current_agent.specialization} - Agent Response: {agent_response['response']} - User Message: {user_message} - Search Results: {search_results} - Conversation History: {' | '.join(self.conversation_history[-4:])} - - Please enhance the agent response while maintaining the conversational tone and medical accuracy. - Focus on being helpful, empathetic, and providing clear guidance. - """ - - try: - llm_response, llm_confidence = self.llm.generate_response(context, self.conversation_history) - - # Combine agent and LLM responses intelligently - if llm_confidence > agent_response['confidence_score']: - enhanced_response = llm_response - final_confidence = llm_confidence - else: - enhanced_response = agent_response['response'] - final_confidence = agent_response['confidence_score'] - - # Update the response - agent_response['response'] = enhanced_response - agent_response['confidence_score'] = final_confidence - - except Exception as e: - # Fall back to agent response if LLM fails - pass - - return agent_response + with input_col2: + if st.button("🔄 New Topic"): + # Save current conversation but start new topic + medical_system.conversation_state.phase = "greeting" + medical_system.conversation_state.conversation_depth = 0 + st.success("Ready to discuss a new topic!") - def _update_patient_profile(self, user_message: str, agent_response: Dict[str, Any]): - """Update patient profile based on conversation""" - - message_lower = user_message.lower() - - # Extract age indicators - age_patterns = [ - (r'\b(\d+)\s*year', lambda m: f"{m.group(1)} years"), - (r'\bchild\b', lambda m: "child"), - (r'\binfant\b|\bbaby\b', lambda m: "infant"), - (r'\belderly\b|\bsenior\b', lambda m: "elderly") - ] - - for pattern, extractor in age_patterns: - match = re.search(pattern, message_lower) - if match and not self.patient_profile.age_range: - self.patient_profile.age_range = extractor(match) - break - - # Extract chronic conditions - chronic_conditions = ['diabetes', 'hypertension', 'arthritis', 'asthma', 'depression'] - for condition in chronic_conditions: - if condition in message_lower and condition not in self.patient_profile.chronic_conditions: - self.patient_profile.chronic_conditions.append(condition) - - # Extract symptoms for tracking - info_gathered = agent_response.get('information_gathered', {}) - if 'symptoms' in info_gathered: - for symptom in info_gathered['symptoms']: - if symptom not in self.patient_profile.previous_symptoms: - self.patient_profile.previous_symptoms.append(symptom) + with input_col3: + if st.button("🗑️ Clear Chat"): + st.session_state.chat_messages = [] + st.session_state.conversational_system = ConversationalMedicalSystem() + st.rerun() + +with col_sidebar: + st.markdown("### 🤖 Conversation Status") - def _update_system_metrics(self, interaction_data: Dict[str, Any]): - """Update system-wide metrics""" - - agent_id = interaction_data['agent_used'] - self.system_metrics['agent_performance'][agent_id].append({ - 'confidence': interaction_data['confidence_score'], - 'processing_time': interaction_data['processing_time'], - 'urgency_handled': interaction_data['urgency_level'] - }) - - # Track common symptoms - for symptom in interaction_data.get('information_gathered', {}).get('symptoms', []): - self.system_metrics['common_symptoms'][symptom] += 1 + # Current conversation summary + if medical_system.conversation_state.conversation_depth > 0: + summary = medical_system._get_patient_summary() + + st.markdown(f""" +
+

📊 Our Conversation

+

Phase: {summary['current_phase'].replace('_', ' ').title()}

+

Messages: {summary['conversation_depth']}

+

Topics Discussed: {len(summary['symptoms_discussed'])}

+

Priority: {summary['urgency_level'].title()}

+
+ """, unsafe_allow_html=True) + + if summary['symptoms_discussed']: + st.markdown("**Symptoms We've Discussed:**") + for symptom in summary['symptoms_discussed']: + st.markdown(f"• {symptom.title()}") - def _get_session_metrics(self) -> Dict[str, Any]: - """Get current session metrics""" - - if not self.session_data['interactions']: - return {} - - interactions = self.session_data['interactions'] - - return { - 'total_interactions': len(interactions), - 'session_duration': (datetime.now() - self.session_data['start_time']).total_seconds(), - 'average_confidence': np.mean([i['confidence_score'] for i in interactions]), - 'agents_used': list(set(i['agent_used'] for i in interactions)), - 'agent_switches': len(self.session_data['agent_switches']), - 'urgent_situations_detected': sum(1 for i in interactions if i['urgency_level'] != 'low'), - 'information_completeness': len(self.patient_profile.previous_symptoms) / 5.0, - 'average_processing_time': np.mean([i['processing_time'] for i in interactions]) - } + # Quick response suggestions + st.markdown("### 💡 Quick Responses") - def get_comprehensive_assessment(self) -> Dict[str, Any]: - """Generate comprehensive assessment of the session""" - - if not self.session_data['interactions']: - return {'status': 'no_interactions'} - - # Compile all information gathered - all_symptoms = [] - all_urgency_levels = [] - all_confidence_scores = [] - - for interaction in self.session_data['interactions']: - symptoms = interaction.get('information_gathered', {}).get('symptoms', []) - all_symptoms.extend(symptoms) - all_urgency_levels.append(interaction['urgency_level']) - all_confidence_scores.append(interaction['confidence_score']) - - # Determine overall assessment - unique_symptoms = list(set(all_symptoms)) - max_urgency = max(all_urgency_levels) if all_urgency_levels else 'low' - avg_confidence = np.mean(all_confidence_scores) if all_confidence_scores else 0.5 - - # Generate recommendations - recommendations = self._generate_final_recommendations( - unique_symptoms, max_urgency, avg_confidence - ) - - assessment = { - 'session_summary': { - 'total_interactions': len(self.session_data['interactions']), - 'duration_minutes': (datetime.now() - self.session_data['start_time']).total_seconds() / 60, - 'agents_consulted': list(set(i['agent_used'] for i in self.session_data['interactions'])), - 'agent_switches': len(self.session_data['agent_switches']) - }, - 'clinical_summary': { - 'symptoms_identified': unique_symptoms, - 'highest_urgency_level': max_urgency, - 'overall_confidence': avg_confidence, - 'patient_profile': asdict(self.patient_profile) - }, - 'recommendations': recommendations, - 'next_steps': self._generate_next_steps(max_urgency, unique_symptoms), - 'follow_up_guidance': self._generate_follow_up_guidance(), - 'system_metrics': self._get_session_metrics() - } - - self.session_data['final_assessment'] = assessment - return assessment + quick_responses = [ + "Yes, that's right", + "No, that's not quite right", + "Can you explain more?", + "I'm not sure about that", + "That makes sense", + "What should I do next?", + "How serious is this?", + "When should I see a doctor?" + ] - def _generate_final_recommendations(self, symptoms: List[str], urgency: str, confidence: float) -> List[str]: - """Generate final recommendations based on session""" - - recommendations = [] - - if urgency == 'emergency': - recommendations.append("Seek immediate emergency medical attention") - recommendations.append("Call 911 or go to the nearest emergency room") - elif urgency == 'high': - recommendations.append("Schedule urgent appointment with healthcare provider") - recommendations.append("Consider urgent care if primary care unavailable") - elif urgency == 'moderate': - recommendations.append("Schedule appointment with primary care provider within 1-2 weeks") - recommendations.append("Monitor symptoms and seek care if worsening") - else: - recommendations.append("Consider routine appointment if symptoms persist") - recommendations.append("Continue self-monitoring and basic self-care") - - if confidence < 0.6: - recommendations.append("This assessment has lower confidence - professional evaluation recommended") - - if len(symptoms) > 3: - recommendations.append("Multiple symptoms warrant comprehensive medical evaluation") - - return recommendations + for response in quick_responses[:4]: # Show only first 4 + if st.button(response, key=f"quick_{response}", help="Quick response"): + st.session_state.chat_messages.append({"role": "user", "content": response}) + st.rerun() - def _generate_next_steps(self, urgency: str, symptoms: List[str]) -> List[str]: - """Generate specific next steps""" - - steps = [] - - if urgency in ['emergency', 'high']: - steps.append("Do not delay seeking medical care") - steps.append("Prepare list of all symptoms and medications") - else: - steps.append("Keep symptom diary until medical appointment") - steps.append("Note any changes or new symptoms") - - if 'pain' in symptoms: - steps.append("Track pain levels, location, and triggers") - - if 'fever' in symptoms: - steps.append("Monitor temperature regularly") - - steps.append("Bring this conversation summary to your healthcare provider") - - return steps + # Agent status + st.markdown("### 🏥 Medical Team Status") - def _generate_follow_up_guidance(self) -> Dict[str, Any]: - """Generate follow-up guidance""" - - return { - 'when_to_seek_immediate_care': [ - "Symptoms suddenly worsen significantly", - "New severe symptoms develop", - "Difficulty breathing or chest pain", - "Signs of serious allergic reaction" - ], - 'monitoring_guidelines': [ - "Keep detailed symptom diary", - "Note medication effects", - "Track symptom patterns", - "Monitor vital signs if appropriate" - ], - 'lifestyle_recommendations': [ - "Maintain adequate hydration", - "Get sufficient rest", - "Follow medication schedules", - "Avoid known triggers if identified" - ], - 'communication_tips': [ - "Prepare questions for healthcare provider", - "Bring medication list", - "Share this AI consultation summary", - "Be specific about symptom timing and severity" - ] - } + active_agents = ["primary_physician", "symptom_specialist", "wellness_coach"] - def export_session_data(self) -> str: - """Export complete session data as JSON""" - - export_data = { - 'session_metadata': { - 'start_time': self.session_data['start_time'].isoformat(), - 'end_time': datetime.now().isoformat(), - 'total_interactions': len(self.session_data['interactions']), - 'system_version': '1.0' - }, - 'patient_profile': asdict(self.patient_profile), - 'conversation_history': self.conversation_history, - 'interactions': self.session_data['interactions'], - 'agent_switches': self.session_data['agent_switches'], - 'final_assessment': self.session_data.get('final_assessment'), - 'session_metrics': self._get_session_metrics() - } - - return json.dumps(export_data, indent=2, default=str) + for agent_id in active_agents: + agent = medical_system.agents[agent_id] + status_color = "🟢" if agent_id == "primary_physician" else "🟡" + st.markdown(f"{status_color} **{agent.specialization}**") + st.markdown(f"*{agent.personality.title()} approach*") + st.markdown("---") + + # Emergency reminder + st.markdown(""" +
+

🚨 Emergency?

+

If you're having a medical emergency, please call emergency services immediately rather than using this chat.

+
+ """, unsafe_allow_html=True) -# Integration helper for Streamlit -class StreamlitMedicalInterface: - """Streamlit interface integration for conversational medical system""" +# Process user input +if send_message and user_input: + # Add user message + st.session_state.chat_messages.append({"role": "user", "content": user_input}) - def __init__(self): - if 'medical_system' not in st.session_state: - st.session_state.medical_system = ConversationalMedicalSystem() - if 'conversation_started' not in st.session_state: - st.session_state.conversation_started = False - if 'chat_messages' not in st.session_state: - st.session_state.chat_messages = [] + # Show typing indicator + st.session_state.typing = True + st.rerun() - def render_interface(self): - """Render the complete Streamlit interface""" - - st.markdown("## Conversational Medical Preconsultation") - - # Medical disclaimer - st.warning("This AI system provides preliminary health guidance and is NOT a substitute for professional medical advice. Always consult qualified healthcare professionals for medical concerns.") - - # Agent status sidebar - self._render_agent_status() + # Process the conversational query + with st.spinner("Having a thoughtful conversation..."): + result = medical_system.process_conversational_query(user_input) + + # Create assistant message with all context + assistant_message = { + "role": "assistant", + "content": result['response'], + "follow_up_questions": result.get('follow_up_questions', []), + "insights": result.get('agent_insights', {}), + "phase": result.get('conversation_phase', ''), + "urgency": result.get('urgency_level', 'routine') + } - # Main chat interface - self._render_chat_interface() + st.session_state.chat_messages.append(assistant_message) - # Session analytics - self._render_session_analytics() + # Update conversation context + st.session_state.conversation_context = { + 'phase': result['conversation_phase'], + 'urgency': result['urgency_level'], + 'primary_agent': result['primary_agent'], + 'patient_summary': result['patient_summary'] + } - def _render_agent_status(self): - """Render agent status in sidebar""" - - with st.sidebar: - st.markdown("### AI Medical Agents") - - medical_system = st.session_state.medical_system - - for agent_id, agent in medical_system.agents.items(): - status = agent.get_agent_status() - - with st.expander(f"{status['specialization']}", expanded=False): - st.metric("Conversations", status['total_conversations']) - st.metric("Engagement", f"{status['patient_engagement']:.1%}") - st.metric("Info Complete", f"{status['information_completeness']:.1%}") - - if agent == medical_system.current_agent: - st.success("Currently Active") - st.write(f"State: {status['conversation_state']}") - - # System metrics - st.markdown("### Session Overview") - session_metrics = medical_system._get_session_metrics() - - if session_metrics: - st.metric("Total Interactions", session_metrics.get('total_interactions', 0)) - st.metric("Avg Confidence", f"{session_metrics.get('average_confidence', 0):.2f}") - st.metric("Agent Switches", session_metrics.get('agent_switches', 0)) + # Remove typing indicator + st.session_state.typing = False + st.rerun() + +# Auto-follow-up system +if len(st.session_state.chat_messages) > 0: + last_message = st.session_state.chat_messages[-1] - def _render_chat_interface(self): - """Render main chat interface""" - - medical_system = st.session_state.medical_system + if (last_message["role"] == "assistant" and + medical_system.conversation_state.conversation_depth > 2 and + medical_system.conversation_state.phase not in ["recommendations", "follow_up"]): - # Display chat messages - for message in st.session_state.chat_messages: - with st.chat_message(message["role"]): - st.write(message["content"]) - - if message["role"] == "assistant" and "metadata" in message: - metadata = message["metadata"] - - col1, col2, col3 = st.columns(3) - with col1: - st.caption(f"Agent: {metadata.get('agent_specialization', 'Unknown')}") - with col2: - st.caption(f"Confidence: {metadata.get('confidence', 0):.2f}") - with col3: - urgency = metadata.get('urgency_level', 'low') - color = {'low': 'normal', 'moderate': 'orange', 'high': 'red', 'emergency': 'red'}[urgency] - st.caption(f"Urgency: {urgency}") - - # Chat input - if prompt := st.chat_input("Describe your symptoms or health concerns..."): - - # Add user message - st.session_state.chat_messages.append({"role": "user", "content": prompt}) - - with st.chat_message("user"): - st.write(prompt) - - # Process with medical system - with st.spinner("AI medical agents are analyzing..."): - - if not st.session_state.conversation_started: - response_data = medical_system.start_conversation(prompt) - st.session_state.conversation_started = True - else: - response_data = medical_system.process_message(prompt) - - # Add assistant response - assistant_message = { - "role": "assistant", - "content": response_data['response'], - "metadata": response_data - } - - st.session_state.chat_messages.append(assistant_message) - - with st.chat_message("assistant"): - st.write(response_data['response']) - - # Show metadata - col1, col2, col3 = st.columns(3) - with col1: - st.caption(f"Agent: {response_data.get('agent_specialization', 'Unknown')}") - with col2: - st.caption(f"Confidence: {response_data.get('confidence', 0):.2f}") - with col3: - urgency = response_data.get('urgency_level', 'low') - st.caption(f"Urgency: {urgency}") - - # Show recommended actions if any - if response_data.get('recommended_actions'): - st.markdown("**Recommended Actions:**") - for action in response_data['recommended_actions']: - st.write(f"• {action}") - - st.rerun() + # Suggest natural follow-up after a few exchanges + if len(st.session_state.chat_messages) % 6 == 0: # Every 3 exchanges + st.info("💭 Would you like me to summarize what we've discussed so far, or is there something specific you'd like to explore further?") + +# Conversation analytics (simplified) +if st.sidebar.button("📊 View Conversation Insights"): + st.sidebar.markdown("### 📈 Conversation Analytics") - def _render_session_analytics(self): - """Render session analytics""" + if medical_system.conversation_data: + # Simple metrics + total_exchanges = len(medical_system.conversation_data) + avg_response_length = np.mean([len(entry.assistant_response) for entry in medical_system.conversation_data]) + phases_covered = set([entry.conversation_phase for entry in medical_system.conversation_data]) + + st.sidebar.metric("Total Exchanges", total_exchanges) + st.sidebar.metric("Avg Response Length", f"{avg_response_length:.0f} chars") + st.sidebar.metric("Conversation Phases", len(phases_covered)) + + # Conversation flow + st.sidebar.markdown("**Conversation Flow:**") + for entry in medical_system.conversation_data[-5:]: # Last 5 + st.sidebar.markdown(f"• {entry.conversation_phase.replace('_', ' ').title()}") + +# Export conversation +if st.sidebar.button("💾 Save Our Conversation"): + if st.session_state.chat_messages: + conversation_export = { + 'timestamp': datetime.now().isoformat(), + 'conversation_summary': st.session_state.conversation_context, + 'messages': st.session_state.chat_messages, + 'patient_summary': medical_system._get_patient_summary(), + 'total_exchanges': medical_system.conversation_state.conversation_depth + } - if st.button("Generate Session Assessment"): - medical_system = st.session_state.medical_system - assessment = medical_system.get_comprehensive_assessment() - - if assessment.get('status') != 'no_interactions': - st.markdown("### Session Assessment") - - # Clinical summary - clinical = assessment['clinical_summary'] - col1, col2 = st.columns(2) - - with col1: - st.markdown("**Symptoms Identified:**") - for symptom in clinical['symptoms_identified']: - st.write(f"• {symptom}") - - with col2: - st.metric("Urgency Level", clinical['highest_urgency_level']) - st.metric("Overall Confidence", f"{clinical['overall_confidence']:.2f}") - - # Recommendations - st.markdown("**Recommendations:**") - for rec in assessment['recommendations']: - st.write(f"• {rec}") - - # Next steps - st.markdown("**Next Steps:**") - for step in assessment['next_steps']: - st.write(f"• {step}") - - # Export option - if st.button("Export Session Data"): - export_data = medical_system.export_session_data() - st.download_button( - "Download Session Report", - data=export_data, - file_name=f"medical_consultation_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json", - mime="application/json" - ) + st.sidebar.download_button( + label="📄 Download Conversation", + data=json.dumps(conversation_export, indent=2), + file_name=f"medical_conversation_{datetime.now().strftime('%Y%m%d_%H%M')}.json", + mime="application/json" + ) + st.sidebar.success("Conversation ready for download!") -# Usage example and main execution -if __name__ == "__main__": - - # Initialize Streamlit interface - interface = StreamlitMedicalInterface() - interface.render_interface() - \ No newline at end of file +# Health reminders +st.markdown("---") +st.markdown("### 🌟 Gentle Health Reminders") + +reminders_col1, reminders_col2 = st.columns(2) + +with reminders_col1: + st.markdown(""" + **💧 Daily Wellness:** + • Stay hydrated throughout the day + • Take breaks from screens regularly + • Practice deep breathing + • Get some sunlight when possible + """) + +with reminders_col2: + st.markdown(""" + **🤝 Remember:** + • This chat supplements, doesn't replace medical care + • Trust your instincts about your body + • Don't hesitate to seek professional help + • Your health concerns are always valid + """) + +# Footer +st.markdown(""" +
+

MedChat - Conversational Medical AI

+

Providing empathetic, intelligent health conversations • Always encouraging professional medical care

+

This AI system is designed to have supportive conversations about health topics and encourage appropriate medical care.

+
+""", unsafe_allow_html=True) \ No newline at end of file