diff --git "a/app.py" "b/app.py"
--- "a/app.py"
+++ "b/app.py"
@@ -2,1029 +2,1458 @@ import streamlit as st
import pandas as pd
import json
import os
-from typing import Union, List, Dict, Optional, Tuple
+from typing import Union, List, Dict, Optional, Tuple, Any
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
+from dataclasses import dataclass, asdict, field
import hashlib
from collections import defaultdict
+import re
+from enum import Enum
-# Set page configuration
-st.set_page_config(
- page_title="MedAssist - AI Medical Preconsultation",
- layout="wide",
- initial_sidebar_state="expanded",
- page_icon="š„"
-)
+class UrgencyLevel(Enum):
+ LOW = "low"
+ MODERATE = "moderate"
+ HIGH = "high"
+ EMERGENCY = "emergency"
-# Enhanced CSS for medical theme
-st.markdown("""
-
-""", unsafe_allow_html=True)
+class ConversationState(Enum):
+ GREETING = "greeting"
+ SYMPTOM_GATHERING = "symptom_gathering"
+ CLARIFICATION = "clarification"
+ ASSESSMENT = "assessment"
+ RECOMMENDATION = "recommendation"
+ FOLLOW_UP = "follow_up"
@dataclass
-class ConversationEntry:
- """Data structure for storing conversation entries"""
- timestamp: str
- user_input: str
- assistant_response: str
- symptoms: List[str]
- severity_score: float
- confidence_score: float
- search_queries_used: List[str]
- user_feedback: Optional[int] = None # 1-5 rating
- was_helpful: Optional[bool] = None
+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)
@dataclass
-class AgentPerformance:
- """Track agent performance metrics"""
- agent_name: str
- total_queries: int = 0
- successful_responses: int = 0
- average_confidence: float = 0.0
- user_satisfaction: float = 0.0
- learning_rate: float = 0.01
- expertise_areas: Dict[str, float] = None
-
- def __post_init__(self):
- if self.expertise_areas is None:
- self.expertise_areas = defaultdict(float)
+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 MedicalSearchTool:
- """Enhanced medical search tool with domain-specific optimization"""
-
- 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"
- ]
-
- def search_medical_info(self, query: str, search_type: str = "symptoms") -> str:
- """Search for medical information with safety considerations"""
- try:
- # Add medical context to search
- medical_queries = {
- "symptoms": f"medical symptoms {query} causes diagnosis",
- "treatment": f"medical treatment {query} therapy options",
- "prevention": f"disease prevention {query} health tips",
- "general": f"medical information {query} health facts"
- }
-
- enhanced_query = medical_queries.get(search_type, medical_queries["general"])
-
- # Perform search with medical focus
- search_results = list(self.ddgs.text(
- enhanced_query,
- max_results=5,
- region='wt-wt',
- safesearch='on'
- ))
-
- if not search_results:
- return "No relevant medical information found. Please consult with a healthcare professional."
-
- # Filter and format results with medical authority preference
- 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" Information: {snippet}\n"
- f" Source: {url}\n"
- )
-
- return "\n".join(formatted_results)
-
- except Exception as e:
- return f"Search temporarily unavailable: {str(e)}"
+@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 GroqLLM:
- """Medical-optimized LLM client"""
+class ConversationalMedicalAgent:
+ """Enhanced conversational medical agent with specialized capabilities"""
- def __init__(self, model_name="openai/gpt-oss-20b"):
- self.client = Groq(api_key=os.environ.get("GROQ_API_KEY"))
- self.model_name = model_name
- self.medical_context = """
- You are a medical AI assistant for preconsultation guidance.
- IMPORTANT: Always remind users that this is not a substitute for professional medical advice.
- Provide helpful information while emphasizing the need for proper medical consultation.
- """
+ def __init__(self, agent_id: str, specialization: str, personality_traits: Dict[str, float] = None):
+ 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.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
+
+ def _initialize_question_bank(self) -> Dict[str, List[str]]:
+ """Initialize specialized question banks for different scenarios"""
+ 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?"
+ ],
+ '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?",
+ "How has this been affecting your daily activities?",
+ "Have you been under more stress than usual lately?",
+ "How has your sleep been?"
+ ],
+ '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?"
+ ]
+ }
- def generate_response(self, prompt: str, conversation_history: List[str] = None) -> Tuple[str, float]:
- """Generate response with confidence scoring"""
- try:
- # Build context with conversation history
- context = self.medical_context
- if conversation_history:
- context += f"\n\nConversation History:\n{chr(10).join(conversation_history[-5:])}"
-
- full_prompt = f"{context}\n\nUser Query: {prompt}\n\nPlease provide helpful medical guidance while emphasizing the importance of professional medical consultation."
-
- completion = self.client.chat.completions.create(
- model=self.model_name,
- messages=[{"role": "user", "content": full_prompt}],
- temperature=0.3, # Lower temperature for medical accuracy
- max_tokens=1500,
- stream=False
- )
-
- response = completion.choices[0].message.content if completion.choices else "Unable to generate response"
-
- # Calculate confidence score based on response characteristics
- confidence = self._calculate_confidence(response, prompt)
-
- return response, confidence
-
- except Exception as e:
- return f"LLM temporarily unavailable: {str(e)}", 0.0
+ def _initialize_conversation_flows(self) -> Dict[str, List[str]]:
+ """Initialize conversation flow patterns for different conditions"""
+ return {
+ 'respiratory': [
+ 'symptom_exploration', 'timeline_clarification', 'lifestyle_context'
+ ],
+ 'cardiovascular': [
+ 'pain_assessment', 'symptom_exploration', 'medical_history', 'lifestyle_context'
+ ],
+ 'neurological': [
+ 'symptom_exploration', 'timeline_clarification', 'pain_assessment'
+ ],
+ 'gastrointestinal': [
+ 'symptom_exploration', 'timeline_clarification', 'lifestyle_context'
+ ],
+ 'musculoskeletal': [
+ 'pain_assessment', 'symptom_exploration', 'lifestyle_context'
+ ],
+ 'mental_health': [
+ 'symptom_exploration', 'timeline_clarification', 'lifestyle_context'
+ ]
+ }
- def _calculate_confidence(self, response: str, query: str) -> float:
- """Calculate confidence score based on response quality"""
- confidence_factors = 0.0
+ 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"""
- # Check for medical disclaimers (increases confidence in safety)
- if any(phrase in response.lower() for phrase in ["consult", "doctor", "medical professional", "healthcare provider"]):
- confidence_factors += 0.3
+ # Analyze user input
+ input_analysis = self._analyze_user_input(user_input)
- # Check response length (adequate detail)
- if 200 <= len(response) <= 1000:
- confidence_factors += 0.2
+ # Update conversation context
+ self._update_conversation_context(user_input, input_analysis)
- # Check for structured information
- if any(marker in response for marker in ["1.", "ā¢", "-", "**"]):
- confidence_factors += 0.2
+ # Determine response strategy
+ response_strategy = self._determine_response_strategy(input_analysis, patient_profile)
- # Check for balanced information (not overly certain)
- if any(phrase in response.lower() for phrase in ["may", "might", "could", "possible", "typically"]):
- confidence_factors += 0.3
+ # Generate appropriate response
+ response = self._generate_contextual_response(
+ user_input, input_analysis, response_strategy, conversation_history
+ )
- return min(confidence_factors, 1.0)
-
-class EvolutionaryMedicalAgent:
- """Evolutionary agent with reinforcement learning capabilities"""
+ # Update learning from interaction
+ self._update_learning(user_input, response, input_analysis)
+
+ 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', [])
+ }
- def __init__(self, agent_id: str, specialization: str):
- self.agent_id = agent_id
- self.specialization = specialization
- self.performance = AgentPerformance(agent_name=agent_id)
- self.knowledge_base = defaultdict(float)
- self.response_patterns = {}
- self.learning_memory = []
+ 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
+ }
- def process_query(self, query: str, context: str, search_results: str) -> Tuple[str, float]:
- """Process query and adapt based on specialization"""
+ input_lower = user_input.lower()
- # Update query count
- self.performance.total_queries += 1
+ # 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'
+ ]
- # Extract key terms for learning
- key_terms = self._extract_medical_terms(query)
+ for symptom in symptom_keywords:
+ if symptom in input_lower:
+ analysis['symptoms_mentioned'].append(symptom)
- # Build specialized response based on agent's expertise
- specialized_prompt = f"""
- As a {self.specialization} specialist, analyze this medical query:
- Query: {query}
- Context: {context}
- Search Results: {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]
- Provide specialized insights based on your expertise in {self.specialization}.
- Always emphasize the need for professional medical consultation.
- """
+ # 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]
- # Simulate processing (in real implementation, this would use the LLM)
- response = f"Based on my specialization in {self.specialization}, {query.lower()} suggests several considerations. However, please consult with a healthcare professional for proper diagnosis and treatment."
+ # Severity indicators
+ severity_words = {
+ 'mild': ['slight', 'mild', 'little', 'minor'],
+ 'moderate': ['moderate', 'noticeable', 'bothersome'],
+ 'severe': ['severe', 'intense', 'unbearable', 'excruciating', 'terrible']
+ }
- confidence = 0.7 + (self.performance.average_confidence * 0.3)
+ for level, words in severity_words.items():
+ if any(word in input_lower for word in words):
+ analysis['severity_indicators'].append(level)
- # Update expertise in relevant areas
- for term in key_terms:
- self.knowledge_base[term] += 0.1
+ # Emotional state
+ worry_words = ['worried', 'concerned', 'scared', 'anxious', 'frightened']
+ calm_words = ['fine', 'okay', 'manageable', 'better']
- return response, confidence
-
- def update_from_feedback(self, query: str, response: str, feedback_score: int, was_helpful: bool):
- """Update agent based on user feedback (reinforcement learning)"""
-
- # Calculate reward signal
- reward = (feedback_score - 3) / 2 # Convert 1-5 scale to -1 to 1
- if was_helpful:
- reward += 0.2
-
- # Update performance metrics
- if feedback_score >= 3:
- self.performance.successful_responses += 1
-
- # Update satisfaction and confidence
- self.performance.user_satisfaction = (
- (self.performance.user_satisfaction * (self.performance.total_queries - 1) + feedback_score) /
- self.performance.total_queries
- )
+ 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'
- # Store learning memory
- self.learning_memory.append({
- 'query': query,
- 'response': response,
- 'reward': reward,
- 'timestamp': datetime.now().isoformat()
- })
+ # 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]
- # Adapt learning rate based on performance
- if self.performance.user_satisfaction > 4.0:
- self.performance.learning_rate *= 0.95 # Slow down learning when performing well
- elif self.performance.user_satisfaction < 3.0:
- self.performance.learning_rate *= 1.1 # Speed up learning when performing poorly
+ # Question detection
+ if '?' in user_input or user_input.lower().startswith(('what', 'how', 'when', 'where', 'why', 'should', 'can', 'is', 'are')):
+ analysis['question_type'] = 'question'
- # Update expertise areas based on feedback
- terms = self._extract_medical_terms(query)
- for term in terms:
- self.knowledge_base[term] += reward * self.performance.learning_rate
+ return analysis
- def _extract_medical_terms(self, text: str) -> List[str]:
- """Extract medical terms from text for learning"""
- medical_keywords = [
- 'pain', 'fever', 'headache', 'nausea', 'fatigue', 'cough', 'cold', 'flu',
- 'diabetes', 'hypertension', 'infection', 'allergy', 'asthma', 'arthritis',
- 'anxiety', 'depression', 'insomnia', 'migraine', 'rash', 'swelling'
- ]
-
- found_terms = []
- text_lower = text.lower()
- for term in medical_keywords:
- if term in text_lower:
- found_terms.append(term)
- return found_terms
+ 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)
- def get_expertise_summary(self) -> Dict:
- """Get summary of agent's learned expertise"""
- return {
- 'specialization': self.specialization,
- 'total_queries': self.performance.total_queries,
- 'success_rate': (self.performance.successful_responses / max(1, self.performance.total_queries)) * 100,
- 'user_satisfaction': self.performance.user_satisfaction,
- 'learning_rate': self.performance.learning_rate,
- 'top_expertise_areas': dict(sorted(self.knowledge_base.items(), key=lambda x: x[1], reverse=True)[:5])
+ 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
}
-
-class MedicalConsultationSystem:
- """Main medical consultation system with evolutionary agents"""
+
+ # 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'
+
+ return strategy
- def __init__(self):
- self.llm = GroqLLM()
- self.search_tool = MedicalSearchTool()
- self.agents = self._initialize_agents()
- self.conversation_history = []
- self.conversation_data = []
+ def _prioritize_questions(self, analysis: Dict[str, Any]) -> List[str]:
+ """Prioritize questions based on information gaps"""
+ priorities = []
- def _initialize_agents(self) -> Dict[str, EvolutionaryMedicalAgent]:
- """Initialize specialized medical agents"""
- return {
- "general_practitioner": EvolutionaryMedicalAgent("gp", "General Practice Medicine"),
- "symptom_analyzer": EvolutionaryMedicalAgent("symptom", "Symptom Analysis and Triage"),
- "wellness_advisor": EvolutionaryMedicalAgent("wellness", "Preventive Care and Wellness"),
- "mental_health": EvolutionaryMedicalAgent("mental", "Mental Health and Psychology"),
- "emergency_assessor": EvolutionaryMedicalAgent("emergency", "Emergency Assessment and Urgent Care")
+ gathered = self.conversation_context.information_gathered
+
+ # High priority: missing critical information
+ if not gathered.get('symptoms'):
+ priorities.append('symptom_exploration')
+
+ if 'pain' in analysis.get('symptoms_mentioned', []) and not analysis.get('pain_descriptors'):
+ priorities.append('pain_assessment')
+
+ if not gathered.get('timeline'):
+ priorities.append('timeline_clarification')
+
+ # Medium priority: contextual information
+ if not gathered.get('lifestyle_factors'):
+ priorities.append('lifestyle_context')
+
+ if not gathered.get('medical_history'):
+ priorities.append('medical_history')
+
+ return priorities
+
+ 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': []
}
+
+ # 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
- def process_medical_query(self, user_query: str) -> Dict:
- """Process medical query through evolutionary agent system"""
+ 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."
+ ]
+
+ # 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
+ }
- timestamp = datetime.now().isoformat()
+ # 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}. "
- # Determine which agents should handle this query
- relevant_agents = self._select_relevant_agents(user_query)
+ # Ask follow-up questions based on priority
+ question_categories = strategy.get('question_priority', ['symptom_exploration'])
- # Search for medical information
- search_results = self.search_tool.search_medical_info(user_query, "symptoms")
+ 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)
- # Build conversation context
- context = "\n".join(self.conversation_history[-3:]) if self.conversation_history else ""
+ # Add main question to response text
+ if response['questions']:
+ response['text'] += f"To better understand your situation: {response['questions'][0]}"
- # Get responses from relevant agents
- agent_responses = {}
- for agent_name in relevant_agents:
- agent = self.agents[agent_name]
- response, confidence = agent.process_query(user_query, context, search_results)
- agent_responses[agent_name] = {
- 'response': response,
- 'confidence': confidence,
- 'specialization': agent.specialization
- }
+ response['next_steps'] = ['Continue symptom assessment', 'Gather timeline information']
- # Generate main LLM response
- main_response, main_confidence = self.llm.generate_response(
- f"{user_query}\n\nRelevant Information: {search_results}",
- self.conversation_history
- )
+ return response
+
+ def _select_contextual_question(self, questions: List[str], analysis: Dict[str, Any]) -> Optional[str]:
+ """Select most appropriate question based on context"""
- # Combine responses intelligently
- final_response = self._combine_responses(main_response, agent_responses)
+ gathered = self.conversation_context.information_gathered
- # Update conversation history
- self.conversation_history.extend([
- f"User: {user_query}",
- f"Assistant: {final_response}"
- ])
+ # Filter out questions we've already asked
+ available_questions = [q for q in questions if q not in self.conversation_context.questions_asked]
- # Extract symptoms for analysis
- symptoms = self._extract_symptoms(user_query)
- severity_score = self._assess_severity(user_query, symptoms)
-
- # Store conversation data
- conversation_entry = ConversationEntry(
- timestamp=timestamp,
- user_input=user_query,
- assistant_response=final_response,
- symptoms=symptoms,
- severity_score=severity_score,
- confidence_score=main_confidence,
- search_queries_used=[user_query]
- )
+ if not available_questions:
+ return None
- self.conversation_data.append(conversation_entry)
+ # 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]
- return {
- 'response': final_response,
- 'confidence': main_confidence,
- 'severity_score': severity_score,
- 'symptoms_detected': symptoms,
- 'agents_consulted': relevant_agents,
- 'agent_responses': agent_responses,
- 'search_performed': True
+ 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'])
+
+ response['text'] = f"Based on our conversation, you're experiencing {', '.join(symptoms)} "
+
+ if severity:
+ response['text'] += f"with {severity[0]} intensity "
+
+ if timeline:
+ response['text'] += f"that started {timeline[0]}. "
+
+ # Provide preliminary assessment
+ assessment = self._generate_preliminary_assessment(symptoms, severity, timeline)
+ response['text'] += assessment
+
+ # Recommend next steps
+ next_steps = self._recommend_next_steps(symptoms, severity, analysis)
+ response['next_steps'] = next_steps
+
+ response['text'] += f"\n\nI recommend: {'. '.join(next_steps)}"
+
+ return response
- def _select_relevant_agents(self, query: str) -> List[str]:
- """Select most relevant agents for the query"""
- query_lower = query.lower()
- relevant_agents = ["general_practitioner"] # Always include GP
-
- # Mental health keywords
- mental_health_keywords = ["stress", "anxiety", "depression", "sleep", "mood", "worry", "panic", "sad"]
- if any(keyword in query_lower for keyword in mental_health_keywords):
- relevant_agents.append("mental_health")
-
- # Emergency keywords
- emergency_keywords = ["severe", "intense", "emergency", "urgent", "chest pain", "difficulty breathing", "blood"]
- if any(keyword in query_lower for keyword in emergency_keywords):
- relevant_agents.append("emergency_assessor")
-
- # Wellness keywords
- wellness_keywords = ["prevention", "healthy", "nutrition", "exercise", "lifestyle", "diet"]
- if any(keyword in query_lower for keyword in wellness_keywords):
- relevant_agents.append("wellness_advisor")
-
- # Always include symptom analyzer for health queries
- if any(keyword in query_lower for keyword in ["pain", "ache", "hurt", "symptom", "feel"]):
- relevant_agents.append("symptom_analyzer")
-
- return list(set(relevant_agents))
+ def _generate_preliminary_assessment(self, symptoms: List[str], severity: List[str], timeline: List[str]) -> str:
+ """Generate preliminary assessment based on gathered information"""
+
+ # This is a simplified assessment logic
+ # In a real system, this would be much more sophisticated
+
+ 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.",
+ }
+
+ # 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
+
+ return "Based on your symptoms, it would be good to monitor how you're feeling and consider speaking with a healthcare provider."
- def _combine_responses(self, main_response: str, agent_responses: Dict) -> str:
- """Intelligently combine responses from multiple agents"""
- if not agent_responses:
- return main_response
+ def _recommend_next_steps(self, symptoms: List[str], severity: List[str], analysis: Dict[str, Any]) -> List[str]:
+ """Recommend appropriate next steps"""
+
+ steps = []
+
+ # 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")
- combined = main_response + "\n\n**Specialist Insights:**\n"
- for agent_name, data in agent_responses.items():
- if data['confidence'] > 0.6: # Only include confident responses
- combined += f"\nā¢ **{data['specialization']}**: {data['response'][:200]}...\n"
+ # 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")
- return combined
+ return steps or ["Monitor your symptoms and consult with a healthcare provider if you have concerns"]
- def _extract_symptoms(self, query: str) -> List[str]:
- """Extract symptoms from user query"""
- common_symptoms = [
- 'fever', 'headache', 'nausea', 'pain', 'cough', 'fatigue', 'dizziness',
- 'rash', 'swelling', 'shortness of breath', 'chest pain', 'abdominal pain'
- ]
+ 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': []
+ }
+
+ # Specific urgent situations
+ urgency_markers = analysis.get('urgency_markers', [])
+
+ 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."
- query_lower = query.lower()
- detected_symptoms = [symptom for symptom in common_symptoms if symptom in query_lower]
- return detected_symptoms
+ return urgent_response
- def _assess_severity(self, query: str, symptoms: List[str]) -> float:
- """Assess severity of reported symptoms (0-10 scale)"""
- severity_score = 0.0
- query_lower = query.lower()
-
- # High severity indicators
- high_severity = ["severe", "intense", "unbearable", "emergency", "chest pain", "difficulty breathing"]
- medium_severity = ["moderate", "persistent", "recurring", "worse", "concerning"]
-
- if any(indicator in query_lower for indicator in high_severity):
- severity_score += 7.0
- elif any(indicator in query_lower for indicator in medium_severity):
- severity_score += 4.0
+ 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"""
+
+ gathered = self.conversation_context.information_gathered
+ missing_info = []
+
+ if not gathered.get('severity'):
+ missing_info.append("severity")
+ if not gathered.get('timeline'):
+ missing_info.append("when it started")
+
+ response_text = "Thank you for that information. To better help you, I'd like to understand "
+
+ if len(missing_info) == 1:
+ response_text += f"more about the {missing_info[0]}."
else:
- severity_score += 2.0
+ response_text += f"more about {' and '.join(missing_info)}."
+
+ 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 = []
+
+ if "severity" in missing_info:
+ questions.append("On a scale of 1-10, how would you rate the intensity of your symptoms?")
- # Add points for multiple symptoms
- severity_score += min(len(symptoms) * 0.5, 2.0)
+ if "when it started" in missing_info:
+ questions.append("When did you first notice these symptoms?")
- return min(severity_score, 10.0)
+ return questions
- def update_agent_performance(self, query_index: int, feedback_score: int, was_helpful: bool):
- """Update agent performance based on user feedback"""
- if query_index < len(self.conversation_data):
- entry = self.conversation_data[query_index]
- entry.user_feedback = feedback_score
- entry.was_helpful = was_helpful
-
- # Update all agents that were involved in this query
- for agent in self.agents.values():
- agent.update_from_feedback(entry.user_input, entry.assistant_response, feedback_score, was_helpful)
+ def _generate_follow_up_response(self) -> Dict[str, Any]:
+ """Generate follow-up response"""
+ 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
+ }
- def get_system_metrics(self) -> Dict:
- """Get comprehensive system performance metrics"""
- total_conversations = len(self.conversation_data)
+ 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
+ })
- if total_conversations == 0:
- return {"status": "No conversations yet"}
+ # Update knowledge base
+ for symptom in analysis.get('symptoms_mentioned', []):
+ self.knowledge_base[symptom] += 0.1
- avg_confidence = np.mean([entry.confidence_score for entry in self.conversation_data])
- avg_severity = np.mean([entry.severity_score for entry in self.conversation_data])
+ # 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 get_agent_status(self) -> Dict[str, Any]:
+ """Get current agent status and learning metrics"""
- feedback_entries = [entry for entry in self.conversation_data if entry.user_feedback is not None]
- avg_feedback = np.mean([entry.user_feedback for entry in feedback_entries]) if feedback_entries else 0
+ 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
return {
- "total_conversations": total_conversations,
- "average_confidence": avg_confidence,
- "average_severity": avg_severity,
- "average_user_feedback": avg_feedback,
- "agent_performance": {name: agent.get_expertise_summary() for name, agent in self.agents.items()}
+ '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
-# Initialize session state
-if 'medical_system' not in st.session_state:
- st.session_state.medical_system = MedicalConsultationSystem()
-if 'chat_messages' not in st.session_state:
- st.session_state.chat_messages = []
-
-medical_system = st.session_state.medical_system
-
-# Main interface
-st.markdown("""
-
-""", unsafe_allow_html=True)
-
-# Medical disclaimer
-st.markdown("""
-
-
ā ļø Important Medical Disclaimer
-
This AI system provides general health information and is NOT a substitute for professional medical advice, diagnosis, or treatment. Always consult with qualified healthcare professionals for medical concerns. In case of emergency, contact emergency services immediately.
-
', unsafe_allow_html=True)
-
- for i, message in enumerate(st.session_state.chat_messages):
- if message["role"] == "user":
- st.markdown(f'
š¤ You: {message["content"]}
', unsafe_allow_html=True)
- else:
- st.markdown(f'
š¤ MedAssist: {message["content"]}
', unsafe_allow_html=True)
-
- # Add feedback buttons for assistant messages
- col_a, col_b, col_c = st.columns([1, 1, 8])
- with col_a:
- if st.button("š", key=f"helpful_{i}"):
- medical_system.update_agent_performance(i//2, 5, True)
- st.success("Feedback recorded!")
- with col_b:
- if st.button("š", key=f"not_helpful_{i}"):
- medical_system.update_agent_performance(i//2, 2, False)
- st.info("Feedback recorded. We'll improve!")
-
- st.markdown('
', unsafe_allow_html=True)
- user_input = st.text_input("Describe your symptoms or health concerns:",
- placeholder="e.g., I've been having headaches for 3 days...",
- key="medical_input")
-
- col_send, col_clear = st.columns([1, 4])
- with col_send:
- send_message = st.button("Send š¤", type="primary")
- with col_clear:
- if st.button("Clear Chat šļø"):
- st.session_state.chat_messages = []
- st.rerun()
- st.markdown('
', unsafe_allow_html=True)
-
-with col2:
- st.markdown("### š¤ AI Agent Status")
+ @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
+ }
+ )
+
+ # 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?"
+ ]
+ })
+
+ return agent
- # Agent status display
- for agent_name, agent in medical_system.agents.items():
- expertise = agent.get_expertise_summary()
-
- st.markdown(f"""
-
-
{agent.specialization}
-
Queries: {expertise['total_queries']}
-
Success Rate: {expertise['success_rate']:.1f}%
-
Satisfaction: {expertise['user_satisfaction']:.1f}/5
-
Learning Rate: {expertise['learning_rate']:.3f}
-
-
Total Chats: {metrics['total_conversations']}
-
Avg Confidence: {metrics['average_confidence']:.2f}
-
Avg Severity: {metrics['average_severity']:.1f}/10
-
User Rating: {metrics['average_user_feedback']:.1f}/5
-
-
MedAssist v1.0 | AI-Powered Medical Preconsultation System
-
š¤ Evolutionary Learning Agents ā¢ š Real-time Medical Search ā¢ š¬ Intelligent Chat Interface
-
ā ļø This system is for informational purposes only and is not a substitute for professional medical advice
-