""" MedGuard - Healthcare Multi-Agent Medication Safety System Hugging Face Space Entry Point MCP 1st Birthday Hackathon Submission This is a self-contained demo that mirrors the FULL production system: - 5 Specialized AI Agents (Drug Interaction, Personalization, Guideline, Cost, Explanation) - LangGraph-style orchestration with conditional routing - BaseAgent pattern with LLM/MCP integration - Pharmacogenomics (CYP enzyme analysis) - Beers Criteria for elderly patients - Polypharmacy detection (5+/10+ medications) - ML-based novel interaction prediction - PubMed literature enhancement - Comprehensive drug interaction database - Clinical decision support with severity-based prioritization Architecture mirrors production: - src/agents/base_agent.py -> BaseAgent abstract class - src/agents/drug_interaction_agent_enhanced.py -> DrugInteractionAgentEnhanced - src/agents/personalization_agent.py -> PersonalizationAgent - src/agents/guideline_compliance_agent.py -> GuidelineComplianceAgent - src/agents/cost_optimization_agent.py -> CostOptimizationAgent - src/agents/explanation_agent.py -> ExplanationAgent - src/orchestration/coordinator_enhanced.py -> MedicationSafetyOrchestrator """ import gradio as gr import os from typing import Dict, Any, List, Optional, Tuple from dataclasses import dataclass, field from enum import Enum from datetime import datetime from abc import ABC, abstractmethod import json import random # Try to import LangChain for LLM support (optional for demo) try: from langchain_google_genai import ChatGoogleGenerativeAI from langchain_core.messages import HumanMessage, SystemMessage HAS_LLM = bool(os.environ.get("GOOGLE_API_KEY")) except ImportError: HAS_LLM = False # ============================================================================= # MODELS (from src/models/) # ============================================================================= class InteractionSeverity(str, Enum): MINOR = "minor" MODERATE = "moderate" MAJOR = "major" CONTRAINDICATED = "contraindicated" class EvidenceLevel(str, Enum): DEFINITIVE = "definitive" PROBABLE = "probable" POSSIBLE = "possible" THEORETICAL = "theoretical" @dataclass class GeneticMarker: """Pharmacogenomic marker from src/models/patient.py""" gene: str variant: str phenotype: str # e.g., "poor_metabolizer", "normal", "ultrarapid" implications: List[str] = field(default_factory=list) @dataclass class LabResult: """Lab result from src/models/patient.py""" test_name: str value: float unit: str reference_range: str test_date: Optional[datetime] = None is_abnormal: bool = False @dataclass class Medication: name: str rxnorm_code: str dosage: str frequency: str route: str = "oral" active: bool = True drug_class: str = "" # Added for therapeutic class analysis @dataclass class Comorbidity: condition: str icd10_code: str active: bool = True @dataclass class Allergy: allergen: str severity: str reaction: str @dataclass class PatientProfile: patient_id: str name: str age: int sex: str weight_kg: float height_cm: float medications: List[Medication] = field(default_factory=list) comorbidities: List[Comorbidity] = field(default_factory=list) allergies: List[Allergy] = field(default_factory=list) egfr: Optional[float] = None liver_function: str = "normal" genetic_markers: List[GeneticMarker] = field(default_factory=list) lab_results: List[LabResult] = field(default_factory=list) @property def bmi(self) -> float: """Calculate BMI""" if self.height_cm and self.weight_kg: height_m = self.height_cm / 100 return round(self.weight_kg / (height_m ** 2), 1) return 0.0 @property def has_renal_impairment(self) -> bool: """Check if patient has renal impairment (eGFR < 60)""" return self.egfr is not None and self.egfr < 60 @property def medication_count(self) -> int: """Count of active medications""" return len([m for m in self.medications if m.active]) @dataclass class DrugInteraction: drug1_name: str drug2_name: str severity: InteractionSeverity evidence_level: EvidenceLevel mechanism: str clinical_effect: str management_strategy: str confidence_score: float = 0.85 literature_refs: List[str] = field(default_factory=list) # Added for PubMed citations # ============================================================================= # MCP TOOL SIMULATION (from src/mcp/healthcare_mcp_server.py) # ============================================================================= class MCPToolSimulator: """ Simulates the 10 MCP tools from the production healthcare_mcp_server.py This demonstrates how MCP clients (like Claude Desktop) interact with our server. In production, these are exposed via the `mcp` SDK with stdio transport. Tools: 1. analyze_medication_safety - Full 5-agent analysis 2. check_drug_interactions - DDI check only 3. get_personalized_dosing - Patient-specific dosing 4. check_guideline_compliance - Clinical guidelines 5. optimize_medication_costs - Cost savings 6. get_patient_profile - Patient data 7. search_clinical_guidelines - BioBERT vector search 8. explain_medication_decision - Patient-friendly explanation 9. search_pubmed_literature - MCP Search integration 10. search_fda_safety_alerts - FDA safety data """ def __init__(self): self.tool_registry = { "analyze_medication_safety": self._analyze_medication_safety, "check_drug_interactions": self._check_drug_interactions, "get_personalized_dosing": self._get_personalized_dosing, "check_guideline_compliance": self._check_guideline_compliance, "optimize_medication_costs": self._optimize_medication_costs, "get_patient_profile": self._get_patient_profile, "search_clinical_guidelines": self._search_clinical_guidelines, "explain_medication_decision": self._explain_medication_decision, "search_pubmed_literature": self._search_pubmed_literature, "search_fda_safety_alerts": self._search_fda_safety_alerts, } # Simulated PubMed data for common queries self.pubmed_data = { "warfarin aspirin": [ { "pmid": "27432982", "title": "Dual Antiplatelet Therapy with Aspirin and Clopidogrel: A Systematic Review", "authors": "Kumbhani DJ, et al.", "journal": "JAMA Cardiology", "year": 2020, "key_finding": "Combined therapy significantly increases major bleeding risk (HR 1.73)" }, { "pmid": "29562146", "title": "Warfarin plus Aspirin: Bleeding Risk Meta-Analysis", "authors": "Garcia DA, et al.", "journal": "NEJM", "year": 2021, "key_finding": "Major bleeding increased from 1.2% to 2.8% with combination" } ], "serotonin syndrome ssri tramadol": [ { "pmid": "31678302", "title": "Serotonin Syndrome: Recognition and Management", "authors": "Boyer EW, Shannon M", "journal": "NEJM", "year": 2019, "key_finding": "SSRI + tramadol combination associated with 4x increased risk" } ], "statin myopathy cyp3a4": [ { "pmid": "28954892", "title": "Statin-Associated Muscle Symptoms: CYP3A4 Interactions", "authors": "Stroes ES, et al.", "journal": "European Heart Journal", "year": 2022, "key_finding": "CYP3A4 inhibitors increase simvastatin AUC by 10-15 fold" } ] } # Simulated FDA safety alerts self.fda_alerts = { "warfarin": [ { "alert_id": "FDA-2020-1234", "date": "2020-08-15", "type": "safety_alert", "title": "Risk of Major Bleeding with Warfarin", "description": "FDA warns of increased bleeding risk with NSAIDs or antiplatelet agents" } ], "simvastatin": [ { "alert_id": "FDA-2011-0140", "date": "2011-06-08", "type": "safety_alert", "title": "Simvastatin 80mg Dose Restriction", "description": "FDA restricts 80mg dose due to myopathy risk, especially with CYP3A4 inhibitors" } ], "metformin": [ { "alert_id": "FDA-2016-0832", "date": "2016-04-08", "type": "safety_alert", "title": "Metformin Use in Renal Impairment", "description": "Updated labeling: can use in mild-moderate renal impairment, contraindicated if eGFR <30" } ] } async def call_tool(self, tool_name: str, arguments: Dict) -> Dict[str, Any]: """ Simulate calling an MCP tool. This mirrors how Claude Desktop calls our MCP server via: @server.call_tool() async def call_tool(name: str, arguments: dict) -> list[TextContent]: """ if tool_name not in self.tool_registry: return {"error": f"Unknown tool: {tool_name}", "success": False} handler = self.tool_registry[tool_name] return await handler(arguments) def list_tools(self) -> List[Dict]: """ Return list of available tools (mirrors @server.list_tools()). """ return [ { "name": "analyze_medication_safety", "description": "Comprehensive medication safety analysis using 5 AI agents", "parameters": ["patient_id", "query", "include_cost_analysis"] }, { "name": "check_drug_interactions", "description": "Fast drug-drug interaction check using Neo4j knowledge graph", "parameters": ["medications", "patient_allergies"] }, { "name": "get_personalized_dosing", "description": "Calculate personalized dosing based on patient factors", "parameters": ["patient_id", "medication_name", "indication"] }, { "name": "check_guideline_compliance", "description": "Verify medication regimen against clinical practice guidelines", "parameters": ["patient_id", "condition", "proposed_treatment"] }, { "name": "optimize_medication_costs", "description": "Find cost-effective alternatives maintaining clinical efficacy", "parameters": ["current_medications", "insurance_type"] }, { "name": "get_patient_profile", "description": "Retrieve complete patient profile with medications and allergies", "parameters": ["patient_id", "include_history"] }, { "name": "search_clinical_guidelines", "description": "Semantic search through clinical guidelines using BioBERT", "parameters": ["query", "limit"] }, { "name": "explain_medication_decision", "description": "Generate patient-friendly explanation of recommendations", "parameters": ["analysis_summary", "education_level", "language"] }, { "name": "search_pubmed_literature", "description": "Search PubMed for clinical evidence via MCP Search protocol", "parameters": ["query", "study_types", "years", "max_results"] }, { "name": "search_fda_safety_alerts", "description": "Search FDA safety communications and recalls via MCP Search", "parameters": ["drug_name", "alert_types", "years"] } ] async def _analyze_medication_safety(self, args: Dict) -> Dict: """Tool 1: Full multi-agent analysis (calls orchestrator).""" patient_id = args.get("patient_id", "P001") patient = DEMO_PATIENTS.get(patient_id) if not patient: return {"error": f"Patient not found: {patient_id}", "success": False} # This would call the full orchestrator in production return { "success": True, "tool": "analyze_medication_safety", "message": f"Full 5-agent analysis queued for patient {patient_id}", "agents_to_run": [ "DrugInteractionAgentEnhanced", "PersonalizationAgent", "GuidelineComplianceAgent", "CostOptimizationAgent", "ExplanationAgent" ] } async def _check_drug_interactions(self, args: Dict) -> Dict: """Tool 2: Quick DDI check.""" medications = args.get("medications", []) # Simulate Neo4j query response interactions_found = [] med_names = [m.get("name", "").lower() for m in medications] # Check common dangerous pairs if "warfarin" in med_names and "aspirin" in med_names: interactions_found.append({ "drug1": "Warfarin", "drug2": "Aspirin", "severity": "MAJOR", "mechanism": "Additive anticoagulant effects" }) if "sertraline" in med_names and "tramadol" in med_names: interactions_found.append({ "drug1": "Sertraline", "drug2": "Tramadol", "severity": "MAJOR", "mechanism": "Serotonin syndrome risk" }) return { "success": True, "tool": "check_drug_interactions", "medications_analyzed": len(medications), "interactions_found": len(interactions_found), "interactions": interactions_found, "database": "Neo4j knowledge graph" } async def _get_personalized_dosing(self, args: Dict) -> Dict: """Tool 3: Personalized dosing calculation.""" patient_id = args.get("patient_id") medication = args.get("medication_name") patient = DEMO_PATIENTS.get(patient_id) if not patient: return {"error": f"Patient not found: {patient_id}", "success": False} adjustments = [] if patient.egfr and patient.egfr < 60: adjustments.append(f"Renal adjustment needed (eGFR: {patient.egfr})") if patient.age >= 65: adjustments.append("Geriatric dosing considerations") if patient.genetic_markers: for marker in patient.genetic_markers: adjustments.append(f"Pharmacogenomic: {marker.gene} {marker.phenotype}") return { "success": True, "tool": "get_personalized_dosing", "patient_id": patient_id, "medication": medication, "adjustments_needed": adjustments, "factors_analyzed": ["age", "renal_function", "hepatic_function", "pharmacogenomics"] } async def _check_guideline_compliance(self, args: Dict) -> Dict: """Tool 4: Clinical guideline compliance check.""" condition = args.get("condition", "") proposed = args.get("proposed_treatment", []) guideline_sources = { "atrial fibrillation": "AHA/ACC/HRS 2023 AFib Guidelines", "heart failure": "ACC/AHA 2022 Heart Failure Guidelines", "diabetes": "ADA 2024 Standards of Care", "hypertension": "ACC/AHA 2017 Hypertension Guidelines" } source = next((v for k, v in guideline_sources.items() if k in condition.lower()), "Clinical Guidelines") return { "success": True, "tool": "check_guideline_compliance", "condition": condition, "proposed_treatment": proposed, "guideline_source": source, "compliance_status": "Under review", "vector_search": "BioBERT embeddings via Qdrant" } async def _optimize_medication_costs(self, args: Dict) -> Dict: """Tool 5: Cost optimization suggestions.""" medications = args.get("current_medications", []) insurance = args.get("insurance_type", "commercial") savings = [] for med in medications: med_lower = med.lower() if isinstance(med, str) else "" if "lipitor" in med_lower: savings.append({"current": "Lipitor", "generic": "Atorvastatin", "savings": 285}) if "crestor" in med_lower: savings.append({"current": "Crestor", "generic": "Rosuvastatin", "savings": 260}) return { "success": True, "tool": "optimize_medication_costs", "medications_analyzed": len(medications), "insurance_type": insurance, "generic_opportunities": savings, "total_potential_savings": sum(s["savings"] for s in savings) } async def _get_patient_profile(self, args: Dict) -> Dict: """Tool 6: Retrieve patient profile.""" patient_id = args.get("patient_id") patient = DEMO_PATIENTS.get(patient_id) if not patient: return {"error": f"Patient not found: {patient_id}", "success": False} return { "success": True, "tool": "get_patient_profile", "patient_id": patient.patient_id, "demographics": { "name": patient.name, "age": patient.age, "sex": patient.sex, "bmi": patient.bmi }, "medications_count": len(patient.medications), "allergies_count": len(patient.allergies), "comorbidities_count": len(patient.comorbidities), "has_genetic_data": len(patient.genetic_markers) > 0 } async def _search_clinical_guidelines(self, args: Dict) -> Dict: """Tool 7: Semantic search through guidelines.""" query = args.get("query", "") limit = args.get("limit", 5) return { "success": True, "tool": "search_clinical_guidelines", "query": query, "search_method": "BioBERT semantic embeddings", "database": "Qdrant vector store", "results_count": min(limit, 3), # Simulated "results": [ {"title": f"Guideline for {query}", "relevance": 0.95}, {"title": f"Best practices: {query}", "relevance": 0.87} ] } async def _explain_medication_decision(self, args: Dict) -> Dict: """Tool 8: Generate patient-friendly explanation.""" summary = args.get("analysis_summary", "") level = args.get("education_level", "high_school") return { "success": True, "tool": "explain_medication_decision", "original_length": len(summary), "reading_level": level, "explanation": "Your medications have been carefully reviewed by our AI system. We checked for drug interactions, made sure the doses are right for you, and verified everything follows medical guidelines.", "key_points": [ "All medications analyzed for safety", "Your specific health factors considered", "Recommendations follow evidence-based guidelines" ] } async def _search_pubmed_literature(self, args: Dict) -> Dict: """Tool 9: Search PubMed via MCP Search protocol.""" query = args.get("query", "") max_results = args.get("max_results", 10) # Find matching simulated results articles = [] for key, data in self.pubmed_data.items(): if any(term in query.lower() for term in key.split()): articles.extend(data) return { "success": True, "tool": "search_pubmed_literature", "query": query, "protocol": "MCP Search", "results_count": len(articles[:max_results]), "articles": articles[:max_results], "search_url": f"https://pubmed.ncbi.nlm.nih.gov/?term={query.replace(' ', '+')}" } async def _search_fda_safety_alerts(self, args: Dict) -> Dict: """Tool 10: Search FDA safety alerts via MCP Search.""" drug_name = args.get("drug_name", "") alerts = self.fda_alerts.get(drug_name.lower(), []) return { "success": True, "tool": "search_fda_safety_alerts", "drug": drug_name, "protocol": "MCP Search", "alerts_count": len(alerts), "alerts": alerts, "search_url": f"https://www.fda.gov/drugs/drug-safety-and-availability/search?keys={drug_name}" } # Global MCP tool simulator instance mcp_tools = MCPToolSimulator() # ============================================================================= # BASE AGENT (from src/agents/base_agent.py) # ============================================================================= class BaseAgent(ABC): """ Abstract base class for all healthcare agents. Mirrors src/agents/base_agent.py with LLM/MCP integration patterns. """ def __init__(self, llm=None, agent_name: str = "BaseAgent"): self.llm = llm self.agent_name = agent_name self.execution_trace: List[Dict] = [] @abstractmethod async def execute(self, patient: PatientProfile, **kwargs) -> Dict[str, Any]: """Execute agent analysis. Must be implemented by subclasses.""" pass def _log_execution(self, action: str, details: Dict): """Log execution step for audit trail.""" self.execution_trace.append({ "timestamp": datetime.now().isoformat(), "agent": self.agent_name, "action": action, "details": details }) async def _call_llm(self, prompt: str, system_prompt: str = "") -> str: """Call LLM for analysis. Falls back to rule-based if unavailable.""" if self.llm and HAS_LLM: try: messages = [] if system_prompt: messages.append(SystemMessage(content=system_prompt)) messages.append(HumanMessage(content=prompt)) response = await self.llm.ainvoke(messages) return response.content except Exception as e: self._log_execution("llm_error", {"error": str(e)}) return "" return "" async def execute_with_error_handling(self, patient: PatientProfile, **kwargs) -> Dict[str, Any]: """Execute with comprehensive error handling.""" try: self._log_execution("start", {"patient_id": patient.patient_id}) result = await self.execute(patient, **kwargs) self._log_execution("complete", {"success": True}) return result except Exception as e: self._log_execution("error", {"error": str(e)}) return { "agent": self.agent_name, "error": str(e), "success": False } # ============================================================================= # AGENT IMPLEMENTATIONS (from src/agents/) # ============================================================================= class DrugInteractionAgentEnhanced(BaseAgent): """ Agent 1: Enhanced Drug Interaction Detection Mirrors src/agents/drug_interaction_agent_enhanced.py Features: - Known interaction database (Neo4j in production) - Metabolic pathway conflict detection (CYP enzymes) - ML-based novel interaction prediction - Literature enhancement from PubMed """ def __init__(self, llm=None): super().__init__(llm, "DrugInteractionAgentEnhanced") # Comprehensive drug interaction database (expanded from production) self.interaction_db = { # Anticoagulant interactions ("warfarin", "aspirin"): DrugInteraction( drug1_name="Warfarin", drug2_name="Aspirin", severity=InteractionSeverity.MAJOR, evidence_level=EvidenceLevel.DEFINITIVE, mechanism="Pharmacodynamic: Additive anticoagulant/antiplatelet effects", clinical_effect="Significantly increased bleeding risk, including GI hemorrhage and intracranial bleeding", management_strategy="If combination necessary: use low-dose aspirin (81mg), monitor INR closely, add PPI for gastroprotection", literature_refs=["PMID:27432982", "PMID:29562146"] ), ("warfarin", "ibuprofen"): DrugInteraction( drug1_name="Warfarin", drug2_name="Ibuprofen", severity=InteractionSeverity.MAJOR, evidence_level=EvidenceLevel.DEFINITIVE, mechanism="NSAIDs inhibit platelet function and can cause GI erosions; also CYP2C9 interaction", clinical_effect="Increased bleeding risk, GI hemorrhage, INR elevation", management_strategy="Avoid combination. If necessary, use lowest NSAID dose for shortest duration with PPI" ), ("warfarin", "amiodarone"): DrugInteraction( drug1_name="Warfarin", drug2_name="Amiodarone", severity=InteractionSeverity.MAJOR, evidence_level=EvidenceLevel.DEFINITIVE, mechanism="CYP2C9 and CYP3A4 inhibition by amiodarone increases warfarin levels", clinical_effect="30-50% increase in INR, bleeding risk", management_strategy="Reduce warfarin dose by 30-50% when starting amiodarone. Weekly INR for first month" ), ("warfarin", "metformin"): DrugInteraction( drug1_name="Warfarin", drug2_name="Metformin", severity=InteractionSeverity.MINOR, evidence_level=EvidenceLevel.POSSIBLE, mechanism="Theoretical protein binding displacement", clinical_effect="Minimal clinical significance", management_strategy="No dose adjustment needed, routine INR monitoring sufficient" ), # Serotonin syndrome risks ("sertraline", "tramadol"): DrugInteraction( drug1_name="Sertraline", drug2_name="Tramadol", severity=InteractionSeverity.MAJOR, evidence_level=EvidenceLevel.DEFINITIVE, mechanism="Pharmacodynamic: Both increase serotonin levels via different mechanisms", clinical_effect="Risk of serotonin syndrome (hyperthermia, agitation, tremor, hyperreflexia)", management_strategy="Avoid combination if possible. If necessary, use lowest doses and monitor for serotonin syndrome symptoms", literature_refs=["PMID:31678302"] ), ("sertraline", "sumatriptan"): DrugInteraction( drug1_name="Sertraline", drug2_name="Sumatriptan", severity=InteractionSeverity.MODERATE, evidence_level=EvidenceLevel.PROBABLE, mechanism="Both drugs increase serotonin levels", clinical_effect="Theoretical risk of serotonin syndrome, though rare in practice", management_strategy="Monitor for serotonin syndrome symptoms. Use lowest effective triptan dose" ), ("sertraline", "zolpidem"): DrugInteraction( drug1_name="Sertraline", drug2_name="Zolpidem", severity=InteractionSeverity.MODERATE, evidence_level=EvidenceLevel.PROBABLE, mechanism="Additive CNS depression", clinical_effect="Enhanced sedation, impaired psychomotor function, increased fall risk", management_strategy="Use lowest effective doses, warn about morning drowsiness, fall precautions" ), # ACE inhibitor interactions ("lisinopril", "potassium"): DrugInteraction( drug1_name="Lisinopril", drug2_name="Potassium Chloride", severity=InteractionSeverity.MAJOR, evidence_level=EvidenceLevel.DEFINITIVE, mechanism="ACE inhibitors reduce aldosterone, decreasing potassium excretion", clinical_effect="Risk of hyperkalemia, potentially life-threatening cardiac arrhythmias", management_strategy="Monitor potassium levels closely (within 1 week of starting), avoid if K+ >5.0" ), ("lisinopril", "spironolactone"): DrugInteraction( drug1_name="Lisinopril", drug2_name="Spironolactone", severity=InteractionSeverity.MAJOR, evidence_level=EvidenceLevel.DEFINITIVE, mechanism="Both drugs increase serum potassium through different mechanisms", clinical_effect="High risk of hyperkalemia", management_strategy="If used together, start low dose, monitor K+ weekly initially" ), # Cardiovascular combinations ("furosemide", "carvedilol"): DrugInteraction( drug1_name="Furosemide", drug2_name="Carvedilol", severity=InteractionSeverity.MODERATE, evidence_level=EvidenceLevel.PROBABLE, mechanism="Additive hypotensive effects; diuretic-induced hypovolemia enhances beta-blocker effect", clinical_effect="Postural hypotension, dizziness, syncope risk", management_strategy="Monitor BP, especially on initiation. Rise slowly from sitting/lying position" ), ("metoprolol", "albuterol"): DrugInteraction( drug1_name="Metoprolol", drug2_name="Albuterol", severity=InteractionSeverity.MODERATE, evidence_level=EvidenceLevel.DEFINITIVE, mechanism="Pharmacodynamic antagonism: Beta-blocker opposes beta-agonist bronchodilation", clinical_effect="Reduced bronchodilator efficacy, potential bronchospasm in susceptible patients", management_strategy="Use cardioselective beta-blocker at lowest effective dose. Consider alternative antihypertensive" ), ("metoprolol", "diltiazem"): DrugInteraction( drug1_name="Metoprolol", drug2_name="Diltiazem", severity=InteractionSeverity.MAJOR, evidence_level=EvidenceLevel.DEFINITIVE, mechanism="Both drugs slow AV conduction; diltiazem inhibits CYP2D6 increasing metoprolol levels", clinical_effect="Risk of severe bradycardia, heart block, hypotension", management_strategy="Avoid combination if possible. If needed, use lowest doses with close monitoring" ), # Statin interactions ("simvastatin", "amlodipine"): DrugInteraction( drug1_name="Simvastatin", drug2_name="Amlodipine", severity=InteractionSeverity.MAJOR, evidence_level=EvidenceLevel.DEFINITIVE, mechanism="CYP3A4 inhibition by amlodipine increases simvastatin exposure", clinical_effect="Increased risk of myopathy and rhabdomyolysis", management_strategy="Limit simvastatin to 20mg/day with amlodipine. Consider switching to pravastatin or rosuvastatin" ), ("simvastatin", "clarithromycin"): DrugInteraction( drug1_name="Simvastatin", drug2_name="Clarithromycin", severity=InteractionSeverity.CONTRAINDICATED, evidence_level=EvidenceLevel.DEFINITIVE, mechanism="Strong CYP3A4 inhibition dramatically increases statin levels", clinical_effect="High risk of rhabdomyolysis, acute kidney injury", management_strategy="CONTRAINDICATED. Hold simvastatin during clarithromycin course" ), ("atorvastatin", "grapefruit"): DrugInteraction( drug1_name="Atorvastatin", drug2_name="Grapefruit", severity=InteractionSeverity.MODERATE, evidence_level=EvidenceLevel.DEFINITIVE, mechanism="Grapefruit inhibits intestinal CYP3A4, increasing statin absorption", clinical_effect="Increased statin exposure and myopathy risk", management_strategy="Limit grapefruit intake or switch to pravastatin/rosuvastatin" ), # Digoxin interactions ("digoxin", "amiodarone"): DrugInteraction( drug1_name="Digoxin", drug2_name="Amiodarone", severity=InteractionSeverity.MAJOR, evidence_level=EvidenceLevel.DEFINITIVE, mechanism="Amiodarone inhibits P-glycoprotein, increasing digoxin levels", clinical_effect="70-100% increase in digoxin levels, toxicity risk", management_strategy="Reduce digoxin dose by 50% when adding amiodarone. Monitor levels" ), ("digoxin", "verapamil"): DrugInteraction( drug1_name="Digoxin", drug2_name="Verapamil", severity=InteractionSeverity.MAJOR, evidence_level=EvidenceLevel.DEFINITIVE, mechanism="P-glycoprotein inhibition; additive AV nodal suppression", clinical_effect="Increased digoxin levels, bradycardia, heart block", management_strategy="Reduce digoxin dose by 25-50%. Monitor for bradycardia" ), # Diabetes drug interactions ("metformin", "contrast"): DrugInteraction( drug1_name="Metformin", drug2_name="IV Contrast", severity=InteractionSeverity.MAJOR, evidence_level=EvidenceLevel.DEFINITIVE, mechanism="Contrast-induced nephropathy impairs metformin excretion", clinical_effect="Risk of lactic acidosis if renal function declines", management_strategy="Hold metformin 48h before and after IV contrast. Check renal function before restarting" ), ("glipizide", "fluconazole"): DrugInteraction( drug1_name="Glipizide", drug2_name="Fluconazole", severity=InteractionSeverity.MAJOR, evidence_level=EvidenceLevel.DEFINITIVE, mechanism="CYP2C9 inhibition increases sulfonylurea levels", clinical_effect="Severe hypoglycemia risk", management_strategy="Reduce glipizide dose by 50%. Monitor blood glucose closely" ), # Opioid interactions ("oxycodone", "benzodiazepine"): DrugInteraction( drug1_name="Oxycodone", drug2_name="Benzodiazepines", severity=InteractionSeverity.MAJOR, evidence_level=EvidenceLevel.DEFINITIVE, mechanism="Additive CNS and respiratory depression", clinical_effect="Increased risk of sedation, respiratory depression, overdose death", management_strategy="Avoid combination. FDA black box warning. If necessary, use lowest doses" ), } # Metabolic pathway database for CYP conflict detection self.cyp_substrates = { "CYP2D6": ["metoprolol", "carvedilol", "codeine", "tramadol", "fluoxetine", "paroxetine"], "CYP3A4": ["simvastatin", "atorvastatin", "amlodipine", "diltiazem", "midazolam", "fentanyl"], "CYP2C9": ["warfarin", "phenytoin", "glipizide", "losartan", "celecoxib"], "CYP2C19": ["omeprazole", "clopidogrel", "diazepam", "citalopram", "escitalopram"], "CYP1A2": ["theophylline", "caffeine", "clozapine", "olanzapine"] } self.cyp_inhibitors = { "CYP2D6": ["fluoxetine", "paroxetine", "bupropion", "quinidine"], "CYP3A4": ["clarithromycin", "ketoconazole", "itraconazole", "ritonavir", "grapefruit", "diltiazem", "verapamil", "amiodarone"], "CYP2C9": ["fluconazole", "amiodarone", "metronidazole"], "CYP2C19": ["omeprazole", "fluconazole", "fluvoxamine"], "CYP1A2": ["ciprofloxacin", "fluvoxamine"] } async def execute(self, patient: PatientProfile, **kwargs) -> Dict[str, Any]: """Execute comprehensive drug interaction analysis.""" self._log_execution("starting_analysis", {"med_count": patient.medication_count}) # Step 1: Check known interactions known_interactions = await self._check_known_interactions(patient) # Step 2: Check metabolic conflicts metabolic_conflicts = await self._check_metabolic_interactions(patient) # Step 3: Predict novel interactions (ML/LLM in production) novel_predictions = await self._predict_novel_interactions(patient) # Step 4: Enhance with literature (PubMed in production) literature_enhanced = await self._enhance_with_literature(known_interactions) # Combine and categorize all_interactions = known_interactions + metabolic_conflicts + novel_predictions critical_interactions = [i for i in all_interactions if i.get("severity") in ["major", "contraindicated"]] return { "agent": self.agent_name, "interactions": all_interactions, "critical_interactions": critical_interactions, "metabolic_conflicts": metabolic_conflicts, "total_found": len(all_interactions), "critical_count": len(critical_interactions), "literature_enhanced": literature_enhanced, "confidence": 0.92 } async def _check_known_interactions(self, patient: PatientProfile) -> List[Dict]: """Check against known interaction database.""" interactions = [] meds = [m for m in patient.medications if m.active] for i, med1 in enumerate(meds): for med2 in meds[i+1:]: key1 = (med1.name.lower(), med2.name.lower()) key2 = (med2.name.lower(), med1.name.lower()) interaction = self.interaction_db.get(key1) or self.interaction_db.get(key2) if interaction: interactions.append({ "drug1": interaction.drug1_name, "drug2": interaction.drug2_name, "severity": interaction.severity.value, "evidence_level": interaction.evidence_level.value, "mechanism": interaction.mechanism, "clinical_effect": interaction.clinical_effect, "management": interaction.management_strategy, "confidence": interaction.confidence_score, "literature_refs": interaction.literature_refs, "source": "knowledge_base" }) return interactions async def _check_metabolic_interactions(self, patient: PatientProfile) -> List[Dict]: """Detect CYP enzyme-mediated drug interactions.""" conflicts = [] med_names = [m.name.lower() for m in patient.medications if m.active] for enzyme, substrates in self.cyp_substrates.items(): # Find substrates patient is taking patient_substrates = [med for med in med_names if any(sub in med for sub in substrates)] # Find inhibitors patient is taking inhibitors = self.cyp_inhibitors.get(enzyme, []) patient_inhibitors = [med for med in med_names if any(inh in med for inh in inhibitors)] # Report conflicts for substrate in patient_substrates: for inhibitor in patient_inhibitors: if substrate != inhibitor: # Don't report self-interaction conflicts.append({ "drug1": substrate.title(), "drug2": inhibitor.title(), "severity": "moderate", "evidence_level": "probable", "mechanism": f"CYP {enzyme} inhibition by {inhibitor.title()} may increase {substrate.title()} levels", "clinical_effect": f"Potential for increased {substrate.title()} exposure and toxicity", "management": f"Monitor for {substrate.title()} toxicity, consider dose reduction", "confidence": 0.75, "source": "metabolic_analysis" }) return conflicts async def _predict_novel_interactions(self, patient: PatientProfile) -> List[Dict]: """Use ML/LLM to predict potential novel interactions not in database.""" # In production, this calls LLM with structured output # For demo, we simulate ML confidence scoring if len(patient.medications) < 5: return [] # Simulate novel interaction detection for polypharmacy return [{ "drug1": "Multi-drug Combination", "drug2": f"{patient.medication_count} medications", "severity": "moderate", "evidence_level": "theoretical", "mechanism": "Complex polypharmacy interaction network", "clinical_effect": "Cumulative CNS depression, metabolic burden, and fall risk", "management": "Comprehensive medication review recommended", "confidence": 0.65, "source": "ml_prediction" }] async def _enhance_with_literature(self, interactions: List[Dict]) -> Dict: """Enhance findings with PubMed literature (simulated).""" # In production, this calls PubMed MCP Search enhanced_count = sum(1 for i in interactions if i.get("literature_refs")) return { "total_citations": enhanced_count * 2, # Simulated "recent_publications": enhanced_count, "meta_analyses_available": enhanced_count > 0 } class PersonalizationAgent(BaseAgent): """ Agent 2: Patient-Specific Personalization Mirrors src/agents/personalization_agent.py Features: - Pharmacogenomics analysis (CYP2D6, CYP2C9, CYP2C19, CYP3A4) - Renal dose adjustments based on eGFR - Hepatic dose adjustments - Beers Criteria for elderly (complete list) - Age-specific considerations (pediatric, geriatric) - Pregnancy/lactation (if applicable) """ def __init__(self, llm=None): super().__init__(llm, "PersonalizationAgent") # Comprehensive renal adjustment database self.renal_adjustments = { "metformin": {"egfr_threshold": 30, "action": "contraindicated", "severity": "critical"}, "gabapentin": {"egfr_threshold": 60, "action": "dose_reduce_50%", "severity": "high"}, "pregabalin": {"egfr_threshold": 60, "action": "dose_reduce_50%", "severity": "high"}, "digoxin": {"egfr_threshold": 50, "action": "dose_reduce_25%", "severity": "high"}, "enoxaparin": {"egfr_threshold": 30, "action": "dose_reduce_50%", "severity": "critical"}, "dabigatran": {"egfr_threshold": 30, "action": "contraindicated", "severity": "critical"}, "rivaroxaban": {"egfr_threshold": 15, "action": "dose_reduce", "severity": "high"}, "apixaban": {"egfr_threshold": 25, "action": "dose_reduce", "severity": "high"}, "methotrexate": {"egfr_threshold": 60, "action": "dose_reduce", "severity": "high"}, "lithium": {"egfr_threshold": 60, "action": "dose_reduce_monitor", "severity": "high"}, "allopurinol": {"egfr_threshold": 60, "action": "dose_reduce", "severity": "moderate"}, "colchicine": {"egfr_threshold": 30, "action": "dose_reduce_50%", "severity": "high"}, "morphine": {"egfr_threshold": 50, "action": "extended_interval", "severity": "high"}, "codeine": {"egfr_threshold": 50, "action": "avoid", "severity": "high"}, } # Hepatic adjustment database self.hepatic_adjustments = { "acetaminophen": {"max_dose": "2g/day", "severity": "high"}, "methotrexate": {"action": "contraindicated", "severity": "critical"}, "simvastatin": {"action": "dose_reduce", "severity": "high"}, "atorvastatin": {"action": "dose_reduce", "severity": "moderate"}, "warfarin": {"action": "dose_reduce_monitor", "severity": "high"}, "tramadol": {"action": "extended_interval", "severity": "moderate"}, } # Complete Beers Criteria list (2023 AGS update) self.beers_criteria = { # First-generation antihistamines "diphenhydramine": {"reason": "Highly anticholinergic, sedating", "alternative": "Loratadine, cetirizine"}, "hydroxyzine": {"reason": "Anticholinergic, sedating", "alternative": "Non-sedating antihistamines"}, "chlorpheniramine": {"reason": "Anticholinergic", "alternative": "Second-gen antihistamines"}, "promethazine": {"reason": "Highly anticholinergic, sedating", "alternative": "Ondansetron for nausea"}, # Benzodiazepines "diazepam": {"reason": "Long half-life, increased fall risk, cognitive impairment", "alternative": "Non-benzo sleep aids if needed"}, "alprazolam": {"reason": "Fall risk, cognitive impairment", "alternative": "SSRIs for anxiety"}, "lorazepam": {"reason": "Fall risk, cognitive impairment", "alternative": "Consider trazodone for insomnia"}, "clonazepam": {"reason": "Long-acting, fall risk", "alternative": "SSRIs/SNRIs"}, "temazepam": {"reason": "Fall risk", "alternative": "Sleep hygiene, melatonin"}, # Non-benzo hypnotics "zolpidem": {"reason": "ER visits, motor vehicle accidents, falls, fractures", "alternative": "Sleep hygiene, low-dose trazodone"}, "eszopiclone": {"reason": "Similar concerns to zolpidem", "alternative": "Melatonin, CBT-I"}, # Anticholinergics "amitriptyline": {"reason": "Highly anticholinergic, sedating", "alternative": "Nortriptyline if TCA needed"}, "imipramine": {"reason": "Anticholinergic", "alternative": "SSRIs/SNRIs"}, "doxepin_high": {"reason": "Anticholinergic at doses >6mg", "alternative": "Doxepin 3-6mg if needed"}, "oxybutynin": {"reason": "Anticholinergic, cognitive effects", "alternative": "Mirabegron"}, "tolterodine": {"reason": "Anticholinergic", "alternative": "Mirabegron, behavioral therapy"}, # NSAIDs "indomethacin": {"reason": "Highest GI risk, CNS effects", "alternative": "Acetaminophen, topical NSAIDs"}, "ketorolac": {"reason": "High GI bleeding risk", "alternative": "Acetaminophen"}, # Muscle relaxants "cyclobenzaprine": {"reason": "Anticholinergic, sedating, limited efficacy", "alternative": "Physical therapy"}, "methocarbamol": {"reason": "Sedating, fall risk", "alternative": "Physical therapy"}, "carisoprodol": {"reason": "Metabolized to meprobamate, addiction risk", "alternative": "Physical therapy"}, # Antipsychotics "haloperidol": {"reason": "Increased mortality in dementia", "alternative": "Avoid if possible"}, "quetiapine": {"reason": "Metabolic effects, sedation, fall risk", "alternative": "Minimize use"}, # Other "meperidine": {"reason": "Neurotoxic metabolite, seizure risk", "alternative": "Other opioids if needed"}, "nitrofurantoin": {"reason": "Pulmonary toxicity, ineffective if CrCl<30", "alternative": "Other antibiotics based on culture"}, } # Pharmacogenomics database (CYP enzyme phenotypes) self.pharmacogenomics = { "CYP2D6": { "poor_metabolizer": { "affected_drugs": ["codeine", "tramadol", "oxycodone", "metoprolol", "carvedilol"], "effect": "reduced_efficacy_or_toxicity", "recommendations": { "codeine": "AVOID - minimal to no efficacy, use alternative analgesic", "tramadol": "AVOID - reduced efficacy, consider alternative", "metoprolol": "Reduce dose 50-75%, monitor for bradycardia", "carvedilol": "Reduce dose, monitor closely" } }, "ultrarapid_metabolizer": { "affected_drugs": ["codeine", "tramadol"], "effect": "increased_toxicity", "recommendations": { "codeine": "AVOID - risk of fatal respiratory depression", "tramadol": "AVOID - increased seizure and respiratory risk" } } }, "CYP2C9": { "poor_metabolizer": { "affected_drugs": ["warfarin", "phenytoin", "celecoxib"], "effect": "increased_drug_levels", "recommendations": { "warfarin": "Reduce initial dose 50%, more frequent INR monitoring", "phenytoin": "Reduce dose, monitor levels closely" } } }, "CYP2C19": { "poor_metabolizer": { "affected_drugs": ["clopidogrel", "omeprazole", "citalopram"], "effect": "variable_by_drug", "recommendations": { "clopidogrel": "Consider prasugrel or ticagrelor instead", "omeprazole": "Standard dosing okay (actually more effective)", "citalopram": "Reduce dose, max 20mg" } }, "ultrarapid_metabolizer": { "affected_drugs": ["omeprazole", "pantoprazole"], "effect": "reduced_efficacy", "recommendations": { "omeprazole": "May need higher doses or switch to rabeprazole" } } } } async def execute(self, patient: PatientProfile, **kwargs) -> Dict[str, Any]: """Execute comprehensive personalization analysis.""" self._log_execution("starting_personalization", { "age": patient.age, "egfr": patient.egfr, "liver_function": patient.liver_function }) findings = { "agent": self.agent_name, "renal_adjustments": [], "hepatic_adjustments": [], "age_concerns": [], "pharmacogenomics": [], "polypharmacy_alert": None, "risk_score": 0.0 } # 1. Renal adjustments findings["renal_adjustments"] = await self._check_renal_adjustments(patient) # 2. Hepatic adjustments findings["hepatic_adjustments"] = await self._check_hepatic_adjustments(patient) # 3. Age-related concerns (Beers Criteria for elderly) findings["age_concerns"] = await self._analyze_age_factors(patient) # 4. Pharmacogenomics findings["pharmacogenomics"] = await self._analyze_pharmacogenetics(patient) # 5. Polypharmacy assessment findings["polypharmacy_alert"] = await self._check_polypharmacy(patient) # Calculate risk score critical_count = len([a for a in findings["renal_adjustments"] if a.get("severity") == "critical"]) high_count = len([a for a in findings["renal_adjustments"] if a.get("severity") == "high"]) high_count += len([a for a in findings["age_concerns"] if a.get("severity") == "high"]) findings["risk_score"] = min(1.0, critical_count * 0.3 + high_count * 0.15) return findings async def _check_renal_adjustments(self, patient: PatientProfile) -> List[Dict]: """Check for medications needing renal adjustment.""" adjustments = [] if not patient.egfr: return adjustments for med in patient.medications: if not med.active: continue med_lower = med.name.lower() for drug, criteria in self.renal_adjustments.items(): if drug in med_lower and patient.egfr < criteria["egfr_threshold"]: adjustments.append({ "medication": med.name, "egfr": patient.egfr, "threshold": criteria["egfr_threshold"], "severity": criteria["severity"], "action": criteria["action"], "recommendation": f"{criteria['action'].replace('_', ' ').title()} - eGFR {patient.egfr} below threshold {criteria['egfr_threshold']}" }) return adjustments async def _check_hepatic_adjustments(self, patient: PatientProfile) -> List[Dict]: """Check for medications needing hepatic adjustment.""" adjustments = [] if patient.liver_function not in ["mild_impairment", "moderate_impairment", "severe_impairment"]: return adjustments for med in patient.medications: if not med.active: continue med_lower = med.name.lower() for drug, criteria in self.hepatic_adjustments.items(): if drug in med_lower: adjustments.append({ "medication": med.name, "liver_function": patient.liver_function, "severity": criteria["severity"], "recommendation": criteria.get("action", criteria.get("max_dose", "Review needed")) }) return adjustments async def _analyze_age_factors(self, patient: PatientProfile) -> List[Dict]: """Analyze age-related medication concerns.""" concerns = [] # Elderly patients (Beers Criteria) if patient.age >= 65: concerns.append({ "concern": "Elderly patient (≥65 years)", "severity": "info", "recommendations": [ "Review for fall risk with sedating medications", "Consider renal function decline with age", "Check for anticholinergic burden", "Evaluate for deprescribing opportunities" ] }) # Check Beers Criteria medications for med in patient.medications: if not med.active: continue med_lower = med.name.lower() for drug, criteria in self.beers_criteria.items(): if drug in med_lower: concerns.append({ "medication": med.name, "concern": f"Potentially Inappropriate in Elderly (Beers Criteria)", "reason": criteria["reason"], "alternative": criteria["alternative"], "severity": "high", "source": "AGS Beers Criteria 2023" }) # Very elderly (85+) if patient.age >= 85: concerns.append({ "concern": "Very elderly patient (≥85 years)", "severity": "high", "recommendations": [ "Prioritize functional status over disease targets", "Consider prognosis in treatment decisions", "Aggressive deprescribing review", "Enhanced fall prevention" ] }) return concerns async def _analyze_pharmacogenetics(self, patient: PatientProfile) -> List[Dict]: """Analyze pharmacogenomic implications.""" findings = [] if not patient.genetic_markers: return findings for marker in patient.genetic_markers: gene = marker.gene phenotype = marker.phenotype if gene in self.pharmacogenomics: gene_data = self.pharmacogenomics[gene].get(phenotype, {}) for med in patient.medications: if not med.active: continue med_lower = med.name.lower() for affected_drug in gene_data.get("affected_drugs", []): if affected_drug in med_lower: recommendation = gene_data.get("recommendations", {}).get(affected_drug, "Review needed") findings.append({ "medication": med.name, "gene": gene, "variant": marker.variant, "phenotype": phenotype, "effect": gene_data.get("effect", "unknown"), "recommendation": recommendation, "severity": "high" if "AVOID" in recommendation.upper() else "moderate" }) return findings async def _check_polypharmacy(self, patient: PatientProfile) -> Optional[Dict]: """Assess polypharmacy risk.""" active_meds = [m for m in patient.medications if m.active] count = len(active_meds) if count >= 10: return { "level": "severe", "medication_count": count, "severity": "high", "recommendation": "Comprehensive medication review REQUIRED. Consider deprescribing.", "risks": [ "Significantly increased drug interaction risk", "Medication non-adherence likely", "Increased fall risk", "Cognitive impairment risk", "Hospitalization risk increased 3-fold" ] } elif count >= 5: return { "level": "moderate", "medication_count": count, "severity": "moderate", "recommendation": "Medication review recommended. Evaluate each medication's necessity.", "risks": [ "Increased drug interaction probability", "Potential adherence challenges", "Consider simplifying regimen" ] } return None class GuidelineComplianceAgent(BaseAgent): """ Agent 3: Clinical Guideline Compliance Mirrors src/agents/guideline_compliance_agent.py Features: - Evidence-based guideline checking (AHA/ACC, ADA, ESC) - Condition-specific therapy recommendations - Polypharmacy detection (5+, 10+ thresholds) - Compliance scoring - Vector store search for guidelines (simulated) """ def __init__(self, llm=None): super().__init__(llm, "GuidelineComplianceAgent") # Comprehensive clinical guidelines database self.guidelines = { "atrial fibrillation": { "required_classes": ["anticoagulant"], "first_line": ["apixaban", "rivaroxaban", "dabigatran", "edoxaban"], "alternative": ["warfarin"], "contraindicated_with": ["mechanical_heart_valve"], "source": "AHA/ACC/HRS 2023 AFib Guidelines", "key_recommendations": [ "CHA2DS2-VASc scoring for stroke risk", "DOACs preferred over warfarin unless mechanical valve or moderate-severe mitral stenosis", "Rate vs rhythm control based on symptoms" ] }, "heart failure": { "required_classes": ["ace_inhibitor_or_arb_or_arni", "beta_blocker", "diuretic", "sglt2i"], "first_line": ["sacubitril-valsartan", "lisinopril", "losartan", "carvedilol", "metoprolol succinate", "bisoprolol", "furosemide", "empagliflozin", "dapagliflozin"], "mra_if_severe": ["spironolactone", "eplerenone"], "source": "ACC/AHA 2022 Heart Failure Guidelines", "key_recommendations": [ "GDMT optimization with 4 pillars: ACEI/ARB/ARNI, BB, MRA, SGLT2i", "SGLT2i now class I recommendation regardless of diabetes", "Target doses of evidence-based therapies" ] }, "type 2 diabetes": { "required_classes": ["antidiabetic"], "first_line": ["metformin"], "with_cv_disease": ["empagliflozin", "dapagliflozin", "semaglutide", "liraglutide"], "with_ckd": ["empagliflozin", "dapagliflozin", "finerenone"], "source": "ADA 2024 Standards of Care", "key_recommendations": [ "A1c target typically <7% (individualized)", "SGLT2i or GLP-1 RA for patients with ASCVD, HF, or CKD", "Weight management integral to treatment" ] }, "hypertension": { "required_classes": ["antihypertensive"], "first_line": ["lisinopril", "amlodipine", "losartan", "hydrochlorothiazide", "chlorthalidone"], "with_diabetes": ["lisinopril", "losartan"], "with_ckd": ["lisinopril", "losartan"], "source": "ACC/AHA 2017 Hypertension Guidelines", "key_recommendations": [ "BP target <130/80 for high-risk patients", "ACEi/ARB preferred with diabetes or CKD", "Two-drug combination often needed" ] }, "coronary artery disease": { "required_classes": ["antiplatelet", "statin", "beta_blocker_if_prior_mi"], "antiplatelet": ["aspirin", "clopidogrel", "ticagrelor", "prasugrel"], "statins": ["atorvastatin", "rosuvastatin"], "source": "ACC/AHA CAD Guidelines 2023", "key_recommendations": [ "High-intensity statin therapy", "Aspirin for secondary prevention", "DAPT post-PCI per guidelines" ] }, "chronic kidney disease": { "required_classes": ["ace_inhibitor_or_arb", "sglt2i_if_appropriate"], "first_line": ["lisinopril", "losartan", "dapagliflozin", "empagliflozin"], "avoid": ["nsaids", "high_dose_vitamin_c"], "source": "KDIGO 2024 CKD Guidelines", "key_recommendations": [ "ACEi/ARB for albuminuric CKD", "SGLT2i for eGFR ≥20 with albuminuria", "BP target <120 systolic if tolerated" ] } } # Drug class mappings for compliance checking self.drug_classes = { "anticoagulant": ["warfarin", "apixaban", "rivaroxaban", "dabigatran", "edoxaban", "enoxaparin", "heparin"], "antiplatelet": ["aspirin", "clopidogrel", "ticagrelor", "prasugrel"], "ace_inhibitor": ["lisinopril", "enalapril", "ramipril", "benazepril", "captopril"], "arb": ["losartan", "valsartan", "irbesartan", "olmesartan", "candesartan"], "beta_blocker": ["metoprolol", "carvedilol", "bisoprolol", "atenolol", "propranolol"], "statin": ["atorvastatin", "rosuvastatin", "simvastatin", "pravastatin", "lovastatin"], "sglt2i": ["empagliflozin", "dapagliflozin", "canagliflozin", "ertugliflozin"], "glp1_ra": ["semaglutide", "liraglutide", "dulaglutide", "exenatide"], "diuretic": ["furosemide", "hydrochlorothiazide", "chlorthalidone", "bumetanide", "torsemide"], "mra": ["spironolactone", "eplerenone"], "antidiabetic": ["metformin", "glipizide", "glyburide", "pioglitazone", "sitagliptin", "empagliflozin", "semaglutide"] } async def execute(self, patient: PatientProfile, **kwargs) -> Dict[str, Any]: """Execute guideline compliance analysis.""" self._log_execution("checking_guidelines", { "conditions": [c.condition for c in patient.comorbidities] }) findings = { "agent": self.agent_name, "compliant_therapies": [], "missing_therapies": [], "guideline_recommendations": [], "compliance_score": 1.0, "polypharmacy_check": None } # Get patient's medication classes patient_med_classes = await self._map_medications_to_classes(patient) # Check each condition against guidelines for comorbidity in patient.comorbidities: if not comorbidity.active: continue condition_findings = await self._check_condition_guidelines( patient, comorbidity.condition, patient_med_classes ) findings["compliant_therapies"].extend(condition_findings.get("compliant", [])) findings["missing_therapies"].extend(condition_findings.get("missing", [])) findings["guideline_recommendations"].extend(condition_findings.get("recommendations", [])) # Polypharmacy check findings["polypharmacy_check"] = await self._check_polypharmacy(patient) # Calculate compliance score total_conditions = len([c for c in patient.comorbidities if c.active]) compliant_count = len(findings["compliant_therapies"]) if total_conditions > 0: findings["compliance_score"] = min(1.0, compliant_count / max(total_conditions, 1)) return findings async def _map_medications_to_classes(self, patient: PatientProfile) -> Dict[str, List[str]]: """Map patient's medications to therapeutic classes.""" patient_classes = {} for med in patient.medications: if not med.active: continue med_lower = med.name.lower() for drug_class, drugs in self.drug_classes.items(): if any(drug in med_lower for drug in drugs): if drug_class not in patient_classes: patient_classes[drug_class] = [] patient_classes[drug_class].append(med.name) return patient_classes async def _check_condition_guidelines( self, patient: PatientProfile, condition: str, patient_classes: Dict ) -> Dict: """Check if patient's regimen complies with guidelines for condition.""" result = {"compliant": [], "missing": [], "recommendations": []} condition_lower = condition.lower() for guideline_condition, guideline in self.guidelines.items(): if guideline_condition in condition_lower or condition_lower in guideline_condition: # Check for required therapy classes all_meds = guideline.get("first_line", []) + guideline.get("alternative", []) med_names_lower = [m.name.lower() for m in patient.medications if m.active] has_therapy = any( any(drug in med for drug in all_meds) for med in med_names_lower ) if has_therapy: result["compliant"].append({ "condition": condition, "status": "compliant", "source": guideline["source"] }) else: result["missing"].append({ "condition": condition, "recommended_drugs": guideline.get("first_line", [])[:3], "source": guideline["source"], "severity": "moderate" }) # Add key recommendations for rec in guideline.get("key_recommendations", [])[:2]: result["recommendations"].append({ "condition": condition, "recommendation": rec, "source": guideline["source"] }) break # Found matching guideline return result async def _check_polypharmacy(self, patient: PatientProfile) -> Dict: """Check for polypharmacy concerns.""" active_count = len([m for m in patient.medications if m.active]) if active_count >= 10: return { "status": "severe_polypharmacy", "count": active_count, "severity": "high", "recommendation": "Deprescribing review strongly recommended" } elif active_count >= 5: return { "status": "polypharmacy", "count": active_count, "severity": "moderate", "recommendation": "Consider medication review for optimization" } return { "status": "acceptable", "count": active_count, "severity": "low" } class CostOptimizationAgent(BaseAgent): """ Agent 4: Cost Optimization Mirrors src/agents/cost_optimization_agent.py Features: - Brand to generic substitution - Therapeutic alternatives (same class, lower cost) - Formulary optimization - Estimated cost calculations """ def __init__(self, llm=None): super().__init__(llm, "CostOptimizationAgent") # Comprehensive brand to generic mappings with pricing self.generic_alternatives = { "lipitor": {"generic": "Atorvastatin", "brand_monthly": 300, "generic_monthly": 15, "savings": 285}, "crestor": {"generic": "Rosuvastatin", "brand_monthly": 280, "generic_monthly": 20, "savings": 260}, "zocor": {"generic": "Simvastatin", "brand_monthly": 200, "generic_monthly": 10, "savings": 190}, "plavix": {"generic": "Clopidogrel", "brand_monthly": 230, "generic_monthly": 12, "savings": 218}, "nexium": {"generic": "Esomeprazole", "brand_monthly": 250, "generic_monthly": 20, "savings": 230}, "prilosec": {"generic": "Omeprazole", "brand_monthly": 180, "generic_monthly": 8, "savings": 172}, "synthroid": {"generic": "Levothyroxine", "brand_monthly": 80, "generic_monthly": 10, "savings": 70}, "glucophage": {"generic": "Metformin", "brand_monthly": 120, "generic_monthly": 8, "savings": 112}, "zoloft": {"generic": "Sertraline", "brand_monthly": 200, "generic_monthly": 15, "savings": 185}, "prozac": {"generic": "Fluoxetine", "brand_monthly": 180, "generic_monthly": 10, "savings": 170}, "ambien": {"generic": "Zolpidem", "brand_monthly": 150, "generic_monthly": 15, "savings": 135}, "norvasc": {"generic": "Amlodipine", "brand_monthly": 160, "generic_monthly": 12, "savings": 148}, "prinivil": {"generic": "Lisinopril", "brand_monthly": 140, "generic_monthly": 8, "savings": 132}, "zestril": {"generic": "Lisinopril", "brand_monthly": 140, "generic_monthly": 8, "savings": 132}, "lopressor": {"generic": "Metoprolol Tartrate", "brand_monthly": 120, "generic_monthly": 10, "savings": 110}, "toprol_xl": {"generic": "Metoprolol Succinate", "brand_monthly": 180, "generic_monthly": 25, "savings": 155}, "lasix": {"generic": "Furosemide", "brand_monthly": 100, "generic_monthly": 8, "savings": 92}, "coumadin": {"generic": "Warfarin", "brand_monthly": 90, "generic_monthly": 10, "savings": 80}, "xanax": {"generic": "Alprazolam", "brand_monthly": 180, "generic_monthly": 15, "savings": 165}, "valium": {"generic": "Diazepam", "brand_monthly": 150, "generic_monthly": 10, "savings": 140}, "klonopin": {"generic": "Clonazepam", "brand_monthly": 170, "generic_monthly": 12, "savings": 158}, } # Therapeutic class alternatives (equivalent efficacy, lower cost) self.therapeutic_alternatives = { # Statins (atorvastatin/rosuvastatin often preferred for efficacy/cost) "crestor": { "alternative": "Atorvastatin 40-80mg", "reason": "Similar LDL reduction, lower cost", "class": "statin" }, "livalo": { "alternative": "Atorvastatin or Rosuvastatin", "reason": "Better studied, lower cost", "class": "statin" }, # DOACs (consider warfarin in some contexts) "eliquis": { "alternative": "Warfarin (if monitoring feasible)", "reason": "Significantly lower cost, reversal agent available", "class": "anticoagulant", "caution": "Requires INR monitoring" }, "xarelto": { "alternative": "Warfarin or Eliquis (if cost similar)", "reason": "Cost consideration, twice-daily Eliquis may be preferable", "class": "anticoagulant" }, # PPIs (omeprazole typically first-line) "protonix": { "alternative": "Omeprazole", "reason": "Equivalent efficacy, significantly lower cost", "class": "ppi" }, "dexilant": { "alternative": "Omeprazole or Esomeprazole", "reason": "Equivalent for most indications, much lower cost", "class": "ppi" }, # Antihypertensives "benicar": { "alternative": "Losartan", "reason": "Similar efficacy, lower cost, better studied", "class": "arb" }, # Sleep aids "lunesta": { "alternative": "Zolpidem or sleep hygiene", "reason": "Lower cost, similar efficacy", "class": "sleep_aid" }, } # Insurance formulary tiers (simplified) self.formulary_info = { "medicare": { "preferred_generics": ["metformin", "lisinopril", "amlodipine", "atorvastatin", "omeprazole"], "high_cost_brands": ["eliquis", "jardiance", "ozempic", "humira"] }, "commercial": { "preferred_generics": ["atorvastatin", "metoprolol", "lisinopril", "sertraline"], "step_therapy_required": ["ozempic", "trulicity", "humira"] } } async def execute(self, patient: PatientProfile, **kwargs) -> Dict[str, Any]: """Execute cost optimization analysis.""" self._log_execution("analyzing_costs", {"med_count": patient.medication_count}) findings = { "agent": self.agent_name, "generic_opportunities": [], "therapeutic_alternatives": [], "formulary_recommendations": [], "potential_monthly_savings": 0.0, "potential_annual_savings": 0.0 } # 1. Check for generic alternatives findings["generic_opportunities"] = await self._check_generic_availability(patient) # 2. Check for therapeutic alternatives findings["therapeutic_alternatives"] = await self._find_therapeutic_alternatives(patient) # 3. Calculate total savings generic_savings = sum(opp["monthly_savings"] for opp in findings["generic_opportunities"]) findings["potential_monthly_savings"] = generic_savings findings["potential_annual_savings"] = generic_savings * 12 return findings async def _check_generic_availability(self, patient: PatientProfile) -> List[Dict]: """Check for available generic substitutions.""" opportunities = [] for med in patient.medications: if not med.active: continue med_lower = med.name.lower().replace(" ", "_").replace("-", "_") for brand, info in self.generic_alternatives.items(): if brand in med_lower or med_lower in brand: opportunities.append({ "current_medication": med.name, "generic_alternative": info["generic"], "brand_monthly_cost": info["brand_monthly"], "generic_monthly_cost": info["generic_monthly"], "monthly_savings": info["savings"], "annual_savings": info["savings"] * 12, "recommendation": f"Switch to {info['generic']} to save ${info['savings']}/month" }) return opportunities async def _find_therapeutic_alternatives(self, patient: PatientProfile) -> List[Dict]: """Find cost-effective therapeutic alternatives.""" alternatives = [] for med in patient.medications: if not med.active: continue med_lower = med.name.lower() for brand, info in self.therapeutic_alternatives.items(): if brand in med_lower: alternatives.append({ "current_medication": med.name, "alternative": info["alternative"], "reason": info["reason"], "therapeutic_class": info["class"], "caution": info.get("caution", None) }) return alternatives class ExplanationAgent(BaseAgent): """ Agent 5: Clinical Summary and Explanation Generation Mirrors src/agents/explanation_agent.py Features: - Executive summary for clinicians - Patient-friendly explanations (adjustable reading level) - Prioritized recommendations - Safety score calculation - Actionable next steps """ def __init__(self, llm=None): super().__init__(llm, "ExplanationAgent") async def execute(self, patient: PatientProfile, **kwargs) -> Dict[str, Any]: """Generate comprehensive explanation from all agent findings.""" interaction_findings = kwargs.get("interaction_findings", {}) personalization_findings = kwargs.get("personalization_findings", {}) guideline_findings = kwargs.get("guideline_findings", {}) cost_findings = kwargs.get("cost_findings", {}) return await self.generate( patient, interaction_findings, personalization_findings, guideline_findings, cost_findings ) async def generate( self, patient: PatientProfile, interaction_findings: Dict, personalization_findings: Dict, guideline_findings: Dict, cost_findings: Dict ) -> Dict[str, Any]: """Synthesize all agent findings into actionable recommendations.""" self._log_execution("generating_summary", {"patient_id": patient.patient_id}) # Collect and prioritize recommendations recommendations = [] # Priority 1: Critical drug interactions for interaction in interaction_findings.get("critical_interactions", []): recommendations.append({ "priority": 1, "category": "CRITICAL DRUG INTERACTION", "title": f"{interaction['drug1']} + {interaction['drug2']}", "severity": "critical", "description": interaction["clinical_effect"], "action": interaction["management"], "evidence": interaction.get("evidence_level", "definitive"), "literature": interaction.get("literature_refs", []) }) # Priority 1: Critical renal contraindications for adj in personalization_findings.get("renal_adjustments", []): if adj.get("severity") == "critical": recommendations.append({ "priority": 1, "category": "RENAL CONTRAINDICATION", "title": f"{adj['medication']} - eGFR {adj['egfr']}", "severity": "critical", "description": f"eGFR {adj['egfr']} below threshold {adj['threshold']}", "action": adj["recommendation"] }) # Priority 1: Pharmacogenomic alerts for pgx in personalization_findings.get("pharmacogenomics", []): if "AVOID" in pgx.get("recommendation", "").upper(): recommendations.append({ "priority": 1, "category": "PHARMACOGENOMIC ALERT", "title": f"{pgx['medication']} - {pgx['gene']} {pgx['phenotype']}", "severity": "critical", "description": f"Patient is {pgx['phenotype']} for {pgx['gene']}", "action": pgx["recommendation"] }) # Priority 2: Major (non-critical) interactions for interaction in interaction_findings.get("interactions", []): if interaction["severity"] == "major": # Check if already in critical is_critical = any( rec["title"] == f"{interaction['drug1']} + {interaction['drug2']}" for rec in recommendations if rec["priority"] == 1 ) if not is_critical: recommendations.append({ "priority": 2, "category": "Major Drug Interaction", "title": f"{interaction['drug1']} + {interaction['drug2']}", "severity": "high", "description": interaction["clinical_effect"], "action": interaction["management"] }) # Priority 2: Beers Criteria medications (elderly) for concern in personalization_findings.get("age_concerns", []): if concern.get("severity") == "high" and "medication" in concern: recommendations.append({ "priority": 2, "category": "Beers Criteria", "title": f"{concern['medication']} - Inappropriate in Elderly", "severity": "high", "description": concern.get("reason", "Potentially inappropriate"), "action": f"Consider: {concern.get('alternative', 'safer alternative')}" }) # Priority 2: High renal adjustments (non-critical) for adj in personalization_findings.get("renal_adjustments", []): if adj.get("severity") == "high": recommendations.append({ "priority": 2, "category": "Renal Dose Adjustment", "title": f"{adj['medication']} - Dose Adjustment Needed", "severity": "high", "description": f"eGFR {adj['egfr']} requires adjustment", "action": adj["recommendation"] }) # Priority 2: Missing guideline-recommended therapies for missing in guideline_findings.get("missing_therapies", []): recommendations.append({ "priority": 2, "category": "Guideline Gap", "title": f"Consider therapy for {missing['condition']}", "severity": "moderate", "description": f"Recommended: {', '.join(missing.get('recommended_drugs', [])[:3])}", "action": f"Per {missing.get('source', 'clinical guidelines')}" }) # Priority 2: Severe polypharmacy polypharm = personalization_findings.get("polypharmacy_alert") if polypharm and polypharm.get("level") == "severe": recommendations.append({ "priority": 2, "category": "Polypharmacy Alert", "title": f"{polypharm['medication_count']} Active Medications", "severity": "high", "description": "Significantly increased risk of adverse events", "action": polypharm["recommendation"] }) # Priority 3: Moderate interactions for interaction in interaction_findings.get("interactions", []): if interaction["severity"] == "moderate": recommendations.append({ "priority": 3, "category": "Moderate Interaction", "title": f"{interaction['drug1']} + {interaction['drug2']}", "severity": "moderate", "description": interaction["clinical_effect"], "action": interaction["management"] }) # Priority 3: Cost optimization opportunities for opp in cost_findings.get("generic_opportunities", []): recommendations.append({ "priority": 3, "category": "Cost Savings Opportunity", "title": f"Switch to {opp['generic_alternative']}", "severity": "low", "description": f"Save ${opp['monthly_savings']}/month (${opp['annual_savings']}/year)", "action": f"Consider generic substitution for {opp['current_medication']}" }) # Sort by priority recommendations.sort(key=lambda x: (x["priority"], -1 if x["severity"] == "critical" else 0)) # Calculate safety score critical_count = len([r for r in recommendations if r["priority"] == 1]) high_count = len([r for r in recommendations if r["priority"] == 2 and r["severity"] == "high"]) moderate_count = len([r for r in recommendations if r["priority"] == 2 and r["severity"] == "moderate"]) safety_score = 100 - (critical_count * 25) - (high_count * 10) - (moderate_count * 3) safety_score = max(0, min(100, safety_score)) # Determine if human review required requires_human_review = ( critical_count > 0 or high_count >= 2 or any(adj.get("severity") == "critical" for adj in personalization_findings.get("renal_adjustments", [])) ) # Generate executive summary executive_summary = await self._generate_executive_summary( patient, recommendations, safety_score, requires_human_review ) # Generate patient-friendly summary patient_summary = await self._generate_patient_summary( patient, recommendations, safety_score ) return { "agent": self.agent_name, "recommendations": recommendations, "safety_score": safety_score, "requires_human_review": requires_human_review, "executive_summary": executive_summary, "patient_summary": patient_summary, "summary_stats": { "total_recommendations": len(recommendations), "critical_items": critical_count, "high_priority_items": high_count, "moderate_items": moderate_count, "cost_savings": cost_findings.get("potential_monthly_savings", 0) } } async def _generate_executive_summary( self, patient: PatientProfile, recommendations: List[Dict], safety_score: int, requires_human_review: bool ) -> str: """Generate executive summary for clinicians.""" critical_recs = [r for r in recommendations if r["priority"] == 1] high_recs = [r for r in recommendations if r["priority"] == 2] summary_parts = [ f"**Patient:** {patient.name} ({patient.age}y {patient.sex})", f"**Medications:** {patient.medication_count} active", f"**Safety Score:** {safety_score}/100" ] if requires_human_review: summary_parts.append("\n**⚠️ REQUIRES CLINICAL REVIEW**") if critical_recs: summary_parts.append(f"\n**Critical Issues ({len(critical_recs)}):**") for rec in critical_recs[:3]: summary_parts.append(f" • {rec['title']}: {rec['action'][:50]}...") if high_recs: summary_parts.append(f"\n**High Priority ({len(high_recs)}):**") for rec in high_recs[:3]: summary_parts.append(f" • {rec['title']}") return "\n".join(summary_parts) async def _generate_patient_summary( self, patient: PatientProfile, recommendations: List[Dict], safety_score: int ) -> str: """Generate patient-friendly summary.""" if safety_score >= 80: status = "Your medications have been reviewed and appear to be safe for you." elif safety_score >= 60: status = "Some concerns were found that your doctor should review with you." else: status = "Important concerns were found. Please discuss with your healthcare provider soon." critical_count = len([r for r in recommendations if r["priority"] == 1]) summary = f"{status}\n\n" if critical_count > 0: summary += f"**{critical_count} important issue(s)** need your doctor's attention.\n" summary += "\nYour care team will discuss any needed changes with you." return summary # ============================================================================= # ORCHESTRATOR (from src/orchestration/coordinator_enhanced.py) # ============================================================================= class MedicationSafetyOrchestrator: """ LangGraph-Style Medication Safety Coordinator Mirrors src/orchestration/coordinator_enhanced.py Features: - StateGraph-style workflow with conditional routing - Severity-based routing (critical -> human_review, parallel_analysis, low_risk) - 5 specialized agent nodes - Execution tracing and audit logging - Consensus building across agents Production uses LangGraph StateGraph with: - add_node() for each agent - add_conditional_edges() for routing based on severity - Parallel execution for independent agents """ def __init__(self): # Initialize all agents (mirrors production agent initialization) self.drug_agent = DrugInteractionAgentEnhanced() self.personalization_agent = PersonalizationAgent() self.guideline_agent = GuidelineComplianceAgent() self.cost_agent = CostOptimizationAgent() self.explanation_agent = ExplanationAgent() # Workflow state self.execution_trace = [] async def analyze(self, patient: PatientProfile) -> Dict[str, Any]: """ Run full multi-agent analysis pipeline with conditional routing. Mirrors LangGraph workflow: 1. START -> interaction_check 2. interaction_check -> conditional_route 3. conditional_route -> - "critical" -> human_review_node - "parallel_analysis" -> [personalization, guideline, cost] in parallel - "low_risk" -> standard_analysis 4. All paths -> explanation 5. explanation -> END """ results = { "patient_id": patient.patient_id, "patient_name": patient.name, "timestamp": datetime.now().isoformat(), "agents_run": [], "execution_path": [], "workflow_type": "langgraph_conditional_routing" } # =================================================================== # Node 1: Drug Interaction Analysis (Entry Point) # =================================================================== results["execution_path"].append("START -> interaction_check") interaction_results = await self.drug_agent.execute_with_error_handling(patient) results["drug_interactions"] = interaction_results results["agents_run"].append("DrugInteractionAgentEnhanced") # =================================================================== # Conditional Routing Based on Severity # =================================================================== route = self._route_after_interaction_check(interaction_results) results["execution_path"].append(f"interaction_check -> {route}") results["routing_decision"] = route # =================================================================== # Node 2: Personalization (runs in all paths) # =================================================================== results["execution_path"].append("PersonalizationAgent") personalization_results = await self.personalization_agent.execute_with_error_handling(patient) results["personalization"] = personalization_results results["agents_run"].append("PersonalizationAgent") # =================================================================== # Node 3: Guideline Compliance (conditional - if comorbidities exist) # =================================================================== if patient.comorbidities and route != "low_risk": results["execution_path"].append("GuidelineComplianceAgent") guideline_results = await self.guideline_agent.execute_with_error_handling(patient) results["guidelines"] = guideline_results results["agents_run"].append("GuidelineComplianceAgent") else: guideline_results = { "compliant_therapies": [], "missing_therapies": [], "compliance_score": 1.0, "skipped": route == "low_risk" or not patient.comorbidities } results["guidelines"] = guideline_results results["execution_path"].append("GuidelineComplianceAgent (skipped - " + ("low_risk" if route == "low_risk" else "no comorbidities") + ")") # =================================================================== # Node 4: Cost Optimization (runs in parallel_analysis and standard paths) # =================================================================== if route != "critical": # Cost not priority in critical path results["execution_path"].append("CostOptimizationAgent") cost_results = await self.cost_agent.execute_with_error_handling(patient) results["cost_optimization"] = cost_results results["agents_run"].append("CostOptimizationAgent") else: cost_results = { "generic_opportunities": [], "therapeutic_alternatives": [], "potential_monthly_savings": 0, "skipped": "critical_path" } results["cost_optimization"] = cost_results results["execution_path"].append("CostOptimizationAgent (skipped - critical path)") # =================================================================== # Special Node: Human Review Flag (for critical findings) # =================================================================== if route == "critical": results["execution_path"].append("human_review_node (FLAGGED)") results["human_review_required"] = True results["human_review_reasons"] = self._get_critical_reasons( interaction_results, personalization_results ) else: results["human_review_required"] = False # =================================================================== # Node 5: Explanation Agent (Final Synthesis) # =================================================================== if route == "critical": results["execution_path"].append("ExplanationAgent (DETAILED - Critical findings)") else: results["execution_path"].append("ExplanationAgent (Standard)") explanation_results = await self.explanation_agent.generate( patient, interaction_results, personalization_results, guideline_results, cost_results ) results["explanation"] = explanation_results results["agents_run"].append("ExplanationAgent") results["execution_path"].append("END") # =================================================================== # Build Consensus # =================================================================== results["consensus"] = self._build_consensus( interaction_results, personalization_results, guideline_results, explanation_results ) return results def _route_after_interaction_check(self, interaction_results: Dict) -> str: """ Conditional routing logic (mirrors LangGraph conditional_edges). Returns: "critical" - Route to human review, skip cost optimization "parallel_analysis" - Run personalization, guideline, cost in parallel "low_risk" - Streamlined analysis path """ critical_count = interaction_results.get("critical_count", 0) total_interactions = interaction_results.get("total_found", 0) # Check for contraindicated combinations has_contraindicated = any( i.get("severity") == "contraindicated" for i in interaction_results.get("interactions", []) ) if has_contraindicated or critical_count >= 2: return "critical" elif critical_count >= 1 or total_interactions >= 3: return "parallel_analysis" else: return "low_risk" def _get_critical_reasons( self, interaction_results: Dict, personalization_results: Dict ) -> List[str]: """Get reasons for critical routing.""" reasons = [] for interaction in interaction_results.get("critical_interactions", []): reasons.append( f"Critical interaction: {interaction['drug1']} + {interaction['drug2']}" ) for adj in personalization_results.get("renal_adjustments", []): if adj.get("severity") == "critical": reasons.append( f"Renal contraindication: {adj['medication']} with eGFR {adj['egfr']}" ) return reasons def _build_consensus( self, interaction: Dict, personalization: Dict, guideline: Dict, explanation: Dict ) -> Dict: """Build consensus across all agent findings.""" return { "safety_score": explanation.get("safety_score", 0), "requires_human_review": explanation.get("requires_human_review", False), "total_concerns": explanation.get("summary_stats", {}).get("total_recommendations", 0), "critical_concerns": explanation.get("summary_stats", {}).get("critical_items", 0), "high_priority_concerns": explanation.get("summary_stats", {}).get("high_priority_items", 0), "guideline_compliance": guideline.get("compliance_score", 1.0) if not guideline.get("skipped") else "N/A", "cost_savings_available": explanation.get("summary_stats", {}).get("cost_savings", 0), "agent_agreement": self._calculate_agent_agreement(interaction, personalization, guideline) } def _calculate_agent_agreement( self, interaction: Dict, personalization: Dict, guideline: Dict ) -> float: """Calculate how much agents agree on risk level.""" # Simple heuristic: if multiple agents flag issues, agreement is high flags = 0 if interaction.get("critical_count", 0) > 0: flags += 1 if personalization.get("risk_score", 0) > 0.3: flags += 1 if guideline.get("compliance_score", 1) < 0.8: flags += 1 # Agreement is high if either all flag or none flag if flags == 0 or flags == 3: return 0.95 elif flags == 1: return 0.75 else: return 0.85 # ============================================================================= # DEMO DATA (simulating src/repositories/ - Enhanced with full production data) # ============================================================================= DEMO_PATIENTS = { "P001": PatientProfile( patient_id="P001", name="John Smith", age=67, sex="male", weight_kg=82, height_cm=175, egfr=58, # Mild-moderate CKD liver_function="normal", medications=[ Medication("Warfarin", "11289", "5mg", "daily", drug_class="anticoagulant"), Medication("Metformin", "6809", "1000mg", "twice daily", drug_class="antidiabetic"), Medication("Lisinopril", "29046", "20mg", "daily", drug_class="ace_inhibitor"), Medication("Aspirin", "1191", "81mg", "daily", drug_class="antiplatelet"), ], comorbidities=[ Comorbidity("Hypertension", "I10"), Comorbidity("Type 2 Diabetes", "E11.9"), Comorbidity("Atrial Fibrillation", "I48.91"), ], allergies=[ Allergy("Penicillin", "severe", "anaphylaxis"), Allergy("Sulfa drugs", "moderate", "rash"), ], genetic_markers=[ GeneticMarker("CYP2C9", "*1/*3", "intermediate_metabolizer", ["Warfarin: May require lower doses"]), ], lab_results=[ LabResult("INR", 2.8, "ratio", "2.0-3.0"), LabResult("Serum Creatinine", 1.4, "mg/dL", "0.7-1.3", is_abnormal=True), LabResult("HbA1c", 7.2, "%", "<7.0%", is_abnormal=True), ] ), "P002": PatientProfile( patient_id="P002", name="Maria Garcia", age=45, sex="female", weight_kg=68, height_cm=163, egfr=95, liver_function="normal", medications=[ Medication("Sertraline", "36437", "100mg", "daily", drug_class="ssri"), Medication("Tramadol", "10689", "50mg", "as needed", drug_class="opioid"), Medication("Zolpidem", "39993", "10mg", "at bedtime", drug_class="hypnotic"), ], comorbidities=[ Comorbidity("Major Depressive Disorder", "F33.1"), Comorbidity("Chronic Pain Syndrome", "G89.29"), Comorbidity("Insomnia", "G47.00"), ], allergies=[], genetic_markers=[ GeneticMarker("CYP2D6", "*1/*4", "intermediate_metabolizer", ["Tramadol: Reduced efficacy possible"]), ], lab_results=[] ), "P003": PatientProfile( patient_id="P003", name="Robert Chen", age=72, sex="male", weight_kg=75, height_cm=170, egfr=38, # Stage 3b CKD liver_function="mild_impairment", medications=[ Medication("Furosemide", "4603", "40mg", "daily", drug_class="diuretic"), Medication("Carvedilol", "20352", "25mg", "twice daily", drug_class="beta_blocker"), Medication("Potassium Chloride", "6252", "20mEq", "daily", drug_class="electrolyte"), Medication("Lisinopril", "29046", "10mg", "daily", drug_class="ace_inhibitor"), Medication("Albuterol", "435", "2 puffs", "as needed", drug_class="bronchodilator"), Medication("Metoprolol", "6918", "25mg", "twice daily", drug_class="beta_blocker"), Medication("Digoxin", "3407", "0.125mg", "daily", drug_class="cardiac_glycoside"), Medication("Spironolactone", "9997", "25mg", "daily", drug_class="mra"), ], comorbidities=[ Comorbidity("Heart Failure", "I50.9"), Comorbidity("COPD", "J44.9"), Comorbidity("Chronic Kidney Disease Stage 3b", "N18.4"), Comorbidity("Atrial Fibrillation", "I48.91"), ], allergies=[ Allergy("Iodine contrast", "severe", "anaphylactoid reaction"), ], genetic_markers=[], lab_results=[ LabResult("Potassium", 5.2, "mEq/L", "3.5-5.0", is_abnormal=True), LabResult("Digoxin Level", 1.8, "ng/mL", "0.8-2.0"), LabResult("BNP", 450, "pg/mL", "<100", is_abnormal=True), ] ), "P004": PatientProfile( patient_id="P004", name="Sarah Johnson", age=55, sex="female", weight_kg=90, height_cm=165, egfr=72, liver_function="normal", medications=[ Medication("Simvastatin", "36567", "40mg", "daily", drug_class="statin"), Medication("Amlodipine", "17767", "10mg", "daily", drug_class="ccb"), Medication("Metoprolol", "6918", "50mg", "twice daily", drug_class="beta_blocker"), Medication("Omeprazole", "7646", "20mg", "daily", drug_class="ppi"), ], comorbidities=[ Comorbidity("Hypertension", "I10"), Comorbidity("Hyperlipidemia", "E78.5"), Comorbidity("GERD", "K21.0"), ], allergies=[], genetic_markers=[], lab_results=[ LabResult("LDL Cholesterol", 95, "mg/dL", "<100"), LabResult("Total Cholesterol", 185, "mg/dL", "<200"), ] ), "P005": PatientProfile( patient_id="P005", name="James Wilson", age=78, sex="male", weight_kg=70, height_cm=168, egfr=45, # Stage 3a CKD liver_function="normal", medications=[ Medication("Codeine/Acetaminophen", "2670", "30/300mg", "every 6 hours", drug_class="opioid"), Medication("Diazepam", "3322", "5mg", "at bedtime", drug_class="benzodiazepine"), Medication("Diphenhydramine", "3498", "25mg", "at bedtime", drug_class="antihistamine"), Medication("Gabapentin", "25480", "600mg", "three times daily", drug_class="anticonvulsant"), Medication("Oxybutynin", "7628", "5mg", "twice daily", drug_class="anticholinergic"), Medication("Amitriptyline", "704", "25mg", "at bedtime", drug_class="tca"), ], comorbidities=[ Comorbidity("Chronic Pain", "G89.29"), Comorbidity("Insomnia", "G47.00"), Comorbidity("Overactive Bladder", "N32.81"), Comorbidity("Peripheral Neuropathy", "G62.9"), ], allergies=[], genetic_markers=[ GeneticMarker("CYP2D6", "*4/*4", "poor_metabolizer", ["Codeine: AVOID - no efficacy", "Tramadol: AVOID"]), ], lab_results=[] ), } # ============================================================================= # GRADIO UI # ============================================================================= orchestrator = MedicationSafetyOrchestrator() def format_results(results: Dict[str, Any]) -> str: """Format analysis results as markdown.""" output = [] # Header output.append(f"# 🏥 MedGuard Analysis Report\n") output.append(f"**Patient:** {results['patient_name']} (ID: {results['patient_id']})") output.append(f"**Generated:** {results['timestamp'][:19]}") output.append(f"**Workflow:** {results.get('workflow_type', 'langgraph_conditional_routing')}\n") # Consensus Summary consensus = results["consensus"] safety_score = consensus["safety_score"] if safety_score >= 80: score_emoji = "✅" score_label = "LOW RISK" elif safety_score >= 60: score_emoji = "⚠️" score_label = "MODERATE RISK" elif safety_score >= 40: score_emoji = "🔶" score_label = "HIGH RISK" else: score_emoji = "🔴" score_label = "CRITICAL RISK" output.append(f"## {score_emoji} Safety Score: {safety_score}/100 ({score_label})\n") if consensus.get("requires_human_review") or results.get("human_review_required"): output.append("> ⚠️ **REQUIRES IMMEDIATE CLINICAL REVIEW**\n") if results.get("human_review_reasons"): output.append("**Reasons:**") for reason in results["human_review_reasons"]: output.append(f" - {reason}") output.append("") # Agent Execution Path (LangGraph style) - Text format for compatibility output.append("### 🔄 LangGraph Execution Path\n") output.append("```") for i, step in enumerate(results["execution_path"]): prefix = " " * (1 if "->" in step or "→" in step else 0) # Clean step for display clean_step = step.replace("→", "->") if i == 0: output.append(f"[1] {clean_step}") else: output.append(f"[{i+1}] {clean_step}") output.append("```") output.append(f"\n**Routing Decision:** `{results.get('routing_decision', 'parallel_analysis')}`\n") # Executive Summary (from ExplanationAgent) if results.get("explanation", {}).get("executive_summary"): output.append("---\n## 📋 Executive Summary\n") output.append(results["explanation"]["executive_summary"]) output.append("") # Recommendations recommendations = results.get("explanation", {}).get("recommendations", []) if recommendations: output.append("---\n## 🎯 Prioritized Recommendations\n") for rec in recommendations: priority_emoji = {"critical": "🔴", "high": "🟠", "moderate": "🟡", "low": "🟢"}.get(rec.get("severity"), "ℹ️") priority_label = {1: "CRITICAL", 2: "HIGH", 3: "MODERATE"}.get(rec["priority"], "INFO") output.append(f"### {priority_emoji} {rec['category']}: {rec['title']}") output.append(f"**Priority:** {priority_label} | **Severity:** {rec.get('severity', 'unknown').upper()}") output.append(f"**Issue:** {rec['description']}") output.append(f"**Action:** {rec['action']}") if rec.get("literature"): output.append(f"**Evidence:** {', '.join(rec['literature'])}") output.append("") # Drug Interactions Detail interactions = results.get("drug_interactions", {}).get("interactions", []) if interactions: output.append("---\n## 💊 Drug Interaction Details\n") for interaction in interactions: severity_emoji = {"contraindicated": "⛔", "major": "🔴", "moderate": "🟠", "minor": "🟡"}.get(interaction.get("severity"), "") output.append(f"### {severity_emoji} {interaction['drug1']} + {interaction['drug2']}") output.append(f"- **Severity:** {interaction.get('severity', 'unknown').upper()}") output.append(f"- **Evidence Level:** {interaction.get('evidence_level', 'unknown')}") output.append(f"- **Mechanism:** {interaction.get('mechanism', 'N/A')}") output.append(f"- **Clinical Effect:** {interaction.get('clinical_effect', 'N/A')}") output.append(f"- **Management:** {interaction.get('management', 'N/A')}") if interaction.get("source") == "metabolic_analysis": output.append(f"- **Source:** CYP Enzyme Analysis") output.append("") # Personalization Findings personalization = results.get("personalization", {}) has_personalization = ( personalization.get("renal_adjustments") or personalization.get("age_concerns") or personalization.get("pharmacogenomics") or personalization.get("polypharmacy_alert") ) if has_personalization: output.append("---\n## 👤 Patient-Specific Factors\n") # Renal adjustments for adj in personalization.get("renal_adjustments", []): severity_emoji = "🔴" if adj.get("severity") == "critical" else "🟠" output.append(f"### {severity_emoji} Renal Adjustment: {adj['medication']}") output.append(f"- **Patient eGFR:** {adj['egfr']} mL/min/1.73m²") output.append(f"- **Threshold:** {adj['threshold']} mL/min/1.73m²") output.append(f"- **Action:** {adj['recommendation']}\n") # Pharmacogenomics for pgx in personalization.get("pharmacogenomics", []): output.append(f"### 🧬 Pharmacogenomic Alert: {pgx['medication']}") output.append(f"- **Gene:** {pgx['gene']} ({pgx['variant']})") output.append(f"- **Phenotype:** {pgx['phenotype']}") output.append(f"- **Effect:** {pgx.get('effect', 'variable')}") output.append(f"- **Recommendation:** {pgx['recommendation']}\n") # Age concerns (Beers Criteria) for concern in personalization.get("age_concerns", []): if concern.get("medication"): output.append(f"### 👴 Beers Criteria: {concern['medication']}") output.append(f"- **Concern:** {concern.get('concern', 'Potentially inappropriate')}") output.append(f"- **Reason:** {concern.get('reason', 'N/A')}") output.append(f"- **Alternative:** {concern.get('alternative', 'See guidelines')}\n") elif concern.get("recommendations"): output.append(f"### 👴 {concern.get('concern', 'Age-Related Considerations')}") for rec in concern.get("recommendations", []): output.append(f"- {rec}") output.append("") # Polypharmacy polypharm = personalization.get("polypharmacy_alert") if polypharm and polypharm.get("level") != "acceptable": output.append(f"### 💊 Polypharmacy Alert") output.append(f"- **Medication Count:** {polypharm['medication_count']}") output.append(f"- **Level:** {polypharm['level'].title()}") output.append(f"- **Recommendation:** {polypharm['recommendation']}") if polypharm.get("risks"): output.append("- **Risks:**") for risk in polypharm["risks"][:3]: output.append(f" - {risk}") output.append("") # Guideline Compliance guidelines = results.get("guidelines", {}) if not guidelines.get("skipped"): output.append("---\n## 📚 Guideline Compliance\n") compliance_score = guidelines.get("compliance_score", 1.0) output.append(f"**Compliance Score:** {compliance_score:.0%}\n") for therapy in guidelines.get("compliant_therapies", []): output.append(f"✅ **{therapy['condition']}** - Compliant per {therapy.get('source', 'guidelines')}") for missing in guidelines.get("missing_therapies", []): output.append(f"⚠️ **{missing['condition']}** - Consider adding guideline-recommended therapy") output.append(f" Recommended: {', '.join(missing.get('recommended_drugs', []))}") output.append(f" Source: {missing.get('source', 'Clinical guidelines')}") for rec in guidelines.get("guideline_recommendations", [])[:2]: output.append(f"\n💡 **{rec['condition']}:** {rec['recommendation']}") output.append("") # Cost Optimization cost = results.get("cost_optimization", {}) if cost.get("potential_monthly_savings", 0) > 0 or cost.get("generic_opportunities"): output.append("---\n## 💰 Cost Optimization Opportunities\n") if cost.get("potential_monthly_savings", 0) > 0: output.append(f"**Potential Monthly Savings:** ${cost['potential_monthly_savings']:.2f}") output.append(f"**Potential Annual Savings:** ${cost.get('potential_annual_savings', 0):.2f}\n") for opp in cost.get("generic_opportunities", []): output.append(f"💊 **{opp['current_medication']}** → **{opp['generic_alternative']}**") output.append(f" Save ${opp['monthly_savings']}/month (${opp.get('annual_savings', 0)}/year)") for alt in cost.get("therapeutic_alternatives", [])[:2]: output.append(f"\n🔄 **Therapeutic Alternative:** {alt['current_medication']}") output.append(f" Consider: {alt['alternative']}") output.append(f" Reason: {alt['reason']}") output.append("") # Consensus Summary output.append("---\n## 📊 Consensus Summary\n") output.append(f"| Metric | Value |") output.append(f"|--------|-------|") output.append(f"| Safety Score | {consensus['safety_score']}/100 |") output.append(f"| Human Review Required | {'Yes ⚠️' if consensus.get('requires_human_review') else 'No ✓'} |") output.append(f"| Critical Concerns | {consensus.get('critical_concerns', 0)} |") output.append(f"| High Priority Concerns | {consensus.get('high_priority_concerns', 0)} |") output.append(f"| Total Recommendations | {consensus.get('total_concerns', 0)} |") if consensus.get('guideline_compliance') != 'N/A': output.append(f"| Guideline Compliance | {consensus.get('guideline_compliance', 1.0):.0%} |") output.append(f"| Agent Agreement | {consensus.get('agent_agreement', 0.85):.0%} |") output.append(f"| Cost Savings Available | ${consensus.get('cost_savings_available', 0):.0f}/month |") # Footer output.append("\n---") output.append("*Generated by MedGuard Multi-Agent System*") output.append("*MCP 1st Birthday Hackathon Submission*") output.append(f"*Agents: {len(results.get('agents_run', []))} | Workflow: LangGraph Conditional Routing*") return "\n".join(output) async def analyze_patient(patient_id: str) -> str: """Run analysis on selected patient.""" patient = DEMO_PATIENTS.get(patient_id) if not patient: return "Patient not found" results = await orchestrator.analyze(patient) return format_results(results) def run_analysis(patient_id: str) -> str: """Synchronous wrapper for Gradio.""" import asyncio return asyncio.run(analyze_patient(patient_id)) async def call_mcp_tool(tool_name: str, patient_id: str, extra_params: str) -> str: """Call an MCP tool and format the response.""" import json # Build arguments based on tool args = {"patient_id": patient_id} # Parse extra params if provided if extra_params.strip(): try: extra = json.loads(extra_params) args.update(extra) except json.JSONDecodeError: # Treat as simple query string args["query"] = extra_params # Add tool-specific defaults if tool_name == "check_drug_interactions": patient = DEMO_PATIENTS.get(patient_id) if patient: args["medications"] = [{"name": m.name} for m in patient.medications] if tool_name == "optimize_medication_costs": patient = DEMO_PATIENTS.get(patient_id) if patient: args["current_medications"] = [m.name for m in patient.medications] if tool_name == "search_pubmed_literature": args["query"] = extra_params or "warfarin aspirin bleeding" if tool_name == "search_fda_safety_alerts": args["drug_name"] = extra_params or "warfarin" if tool_name == "search_clinical_guidelines": args["query"] = extra_params or "atrial fibrillation anticoagulation" # Call the tool result = await mcp_tools.call_tool(tool_name, args) # Format output as markdown output = [f"# 🔧 MCP Tool: `{tool_name}`\n"] output.append(f"**Patient:** {patient_id}") output.append(f"**Timestamp:** {datetime.now().isoformat()[:19]}\n") output.append("## Request") output.append("```json") output.append(json.dumps(args, indent=2, default=str)) output.append("```\n") output.append("## Response") output.append("```json") output.append(json.dumps(result, indent=2, default=str)) output.append("```\n") if result.get("success"): output.append("✅ **Tool executed successfully**") else: output.append(f"❌ **Error:** {result.get('error', 'Unknown error')}") return "\n".join(output) def run_mcp_tool(tool_name: str, patient_id: str, extra_params: str) -> str: """Synchronous wrapper for MCP tool calls.""" import asyncio return asyncio.run(call_mcp_tool(tool_name, patient_id, extra_params)) def get_mcp_tools_list() -> str: """Return formatted list of MCP tools.""" tools = mcp_tools.list_tools() output = ["# 🔧 MCP Server Tools\n"] output.append("These tools are exposed via the MCP protocol for Claude Desktop and other MCP clients.\n") output.append("| # | Tool Name | Description |") output.append("|---|-----------|-------------|") for i, tool in enumerate(tools, 1): output.append(f"| {i} | `{tool['name']}` | {tool['description'][:60]}... |") output.append("\n---\n") output.append("### MCP Protocol Details\n") output.append("- **Transport:** stdio (for Claude Desktop) or SSE (for web clients)") output.append("- **SDK:** Python `mcp` package") output.append("- **Resources:** 3 (guidelines, drug database, safety alerts)") output.append("- **Server Name:** `healthcare-multi-agent-system`") return "\n".join(output) # Build Gradio Interface with gr.Blocks(title="MedGuard - AI Medication Safety") as app: gr.Markdown(""" # 🏥 MedGuard ## AI-Powered Medication Safety Analysis **Multi-Agent System for Drug Interaction Detection and Clinical Decision Support** *MCP 1st Birthday Hackathon Submission - Track 1: Building MCP* """) with gr.Tabs(): # ===================================================================== # TAB 1: MULTI-AGENT ANALYSIS # ===================================================================== with gr.TabItem("🔬 Multi-Agent Analysis", id="analysis"): gr.Markdown(""" This demo showcases a **production-ready** healthcare AI system with: - **5 Specialized AI Agents** coordinated via LangGraph-style orchestration - **Conditional routing** based on severity (critical → human review path) - **Evidence-based drug interaction detection** (25+ known interactions) - **Pharmacogenomics analysis** (CYP2D6, CYP2C9, CYP2C19, CYP3A4) """) with gr.Row(): with gr.Column(scale=1): gr.Markdown("### Select Patient") patient_dropdown = gr.Dropdown( choices=[ ("P001 - John Smith (67y, AFib/DM/HTN, CKD Stage 3, CYP2C9 variant)", "P001"), ("P002 - Maria Garcia (45y, Depression/Pain/Insomnia, Serotonin risk)", "P002"), ("P003 - Robert Chen (72y, HF/COPD/CKD 3b, 8 meds, hyperkalemia)", "P003"), ("P004 - Sarah Johnson (55y, HTN/HLD/GERD, Statin interaction)", "P004"), ("P005 - James Wilson (78y, Polypharmacy, Beers Criteria, CYP2D6 PM)", "P005"), ], label="Patient", value="P001" ) analyze_btn = gr.Button("🔬 Run Multi-Agent Analysis", variant="primary", size="lg") gr.Markdown(""" ### 🤖 Agent Architecture | Agent | Role | |-------|------| | **DrugInteractionAgent** | DDI detection, CYP conflicts, ML prediction | | **PersonalizationAgent** | Renal/hepatic, pharmacogenomics, Beers | | **GuidelineAgent** | Clinical guideline compliance | | **CostAgent** | Generic substitution, formulary | | **ExplanationAgent** | Synthesis and prioritization | ### 🔄 LangGraph Orchestration ``` START → interaction_check → [route] ├─ "critical" → human_review → explanation ├─ "parallel" → [personalization, guideline, cost] → explanation └─ "low_risk" → streamlined_analysis → explanation → END ``` """) with gr.Column(scale=2): analysis_output = gr.Markdown( value=""" ## 👋 Welcome to MedGuard Select a patient and click **Run Multi-Agent Analysis** to begin. ### Demo Patients Each patient demonstrates different aspects of the system: | Patient | Key Features | |---------|--------------| | **P001 - John Smith** | Warfarin + Aspirin (major DDI), CKD, CYP2C9 variant | | **P002 - Maria Garcia** | Sertraline + Tramadol (serotonin syndrome risk) | | **P003 - Robert Chen** | 8 medications, hyperkalemia risk, HF + COPD | | **P004 - Sarah Johnson** | Simvastatin + Amlodipine (CYP3A4 interaction) | | **P005 - James Wilson** | 6 Beers Criteria meds, CYP2D6 poor metabolizer | --- ### What the System Analyzes 1. **Drug-Drug Interactions** - Known interactions + CYP metabolic conflicts 2. **Patient Factors** - Renal function, age, pharmacogenomics 3. **Guideline Compliance** - AHA/ACC, ADA, ESC evidence-based standards 4. **Cost Optimization** - Generic alternatives and formulary savings 5. **Prioritized Recommendations** - Critical → High → Moderate severity """, label="Analysis Results" ) # Event handler for analysis tab analyze_btn.click( fn=run_analysis, inputs=[patient_dropdown], outputs=[analysis_output] ) # ===================================================================== # TAB 2: MCP TOOLS DEMO # ===================================================================== with gr.TabItem("🔧 MCP Tools Demo", id="mcp_tools"): gr.Markdown(""" ## MCP Server Tools This tab demonstrates the **10 MCP tools** exposed by the healthcare server. These tools are callable via the Model Context Protocol by Claude Desktop and other MCP clients. **MCP Server Features:** - **Transport:** stdio (Claude Desktop) or SSE (web clients) - **SDK:** Python `mcp` package - **Resources:** 3 (clinical guidelines, drug database, FDA alerts) """) with gr.Row(): with gr.Column(scale=1): mcp_tools_list_btn = gr.Button("📋 List All MCP Tools", variant="secondary") gr.Markdown("### Call Individual Tool") tool_dropdown = gr.Dropdown( choices=[ ("analyze_medication_safety", "analyze_medication_safety"), ("check_drug_interactions", "check_drug_interactions"), ("get_personalized_dosing", "get_personalized_dosing"), ("check_guideline_compliance", "check_guideline_compliance"), ("optimize_medication_costs", "optimize_medication_costs"), ("get_patient_profile", "get_patient_profile"), ("search_clinical_guidelines", "search_clinical_guidelines"), ("explain_medication_decision", "explain_medication_decision"), ("search_pubmed_literature", "search_pubmed_literature"), ("search_fda_safety_alerts", "search_fda_safety_alerts"), ], label="Select MCP Tool", value="check_drug_interactions" ) tool_patient_dropdown = gr.Dropdown( choices=[ ("P001 - John Smith", "P001"), ("P002 - Maria Garcia", "P002"), ("P003 - Robert Chen", "P003"), ("P004 - Sarah Johnson", "P004"), ("P005 - James Wilson", "P005"), ], label="Patient Context", value="P001" ) extra_params = gr.Textbox( label="Extra Parameters (optional)", placeholder="JSON or query string (e.g., 'warfarin bleeding' for PubMed)", lines=2 ) call_tool_btn = gr.Button("🚀 Call MCP Tool", variant="primary") gr.Markdown(""" ### Tool Descriptions | Tool | Purpose | |------|---------| | `analyze_medication_safety` | Full 5-agent pipeline | | `check_drug_interactions` | DDI detection only | | `get_personalized_dosing` | Patient-specific adjustments | | `check_guideline_compliance` | Evidence-based compliance | | `optimize_medication_costs` | Generic alternatives | | `get_patient_profile` | Demographics & history | | `search_clinical_guidelines` | BioBERT vector search | | `explain_medication_decision` | Patient-friendly explanation | | `search_pubmed_literature` | **MCP Search integration** | | `search_fda_safety_alerts` | **MCP Search integration** | """) with gr.Column(scale=2): mcp_output = gr.Markdown( value=""" ## 🔧 MCP Tools Demo This tab allows you to **call individual MCP tools** to see how they work. ### How MCP Integration Works 1. **Claude Desktop** connects to the MCP server via stdio 2. **Tools are discoverable** via the `tools/list` protocol method 3. **Claude can call tools** using `tools/call` with JSON arguments 4. **Responses are structured** for easy parsing by LLMs ### Example Claude Desktop Interaction ``` User: Check drug interactions for patient P001 Claude: I'll use the check_drug_interactions MCP tool. [Calls tool with patient_id="P001"] Result: Found 2 interactions: - Warfarin + Aspirin: Major bleeding risk - Warfarin + CYP2C9 variant: Dose adjustment needed ``` ### Try It! 1. Select a tool from the dropdown 2. Choose a patient context 3. Click **Call MCP Tool** to see the raw response """, label="MCP Tool Output" ) # Event handlers for MCP tools tab mcp_tools_list_btn.click( fn=get_mcp_tools_list, inputs=[], outputs=[mcp_output] ) call_tool_btn.click( fn=run_mcp_tool, inputs=[tool_dropdown, tool_patient_dropdown, extra_params], outputs=[mcp_output] ) # ===================================================================== # TAB 3: ARCHITECTURE # ===================================================================== with gr.TabItem("🏗️ Architecture", id="architecture"): gr.Markdown(""" ## Production Architecture ### 🏗️ Technology Stack | Component | Technology | |-----------|------------| | **MCP Server** | Python `mcp` SDK with stdio transport, 10 tools + 3 resources | | **Orchestration** | LangGraph StateGraph with conditional severity routing | | **LLM** | Google Gemini 1.5 Flash (primary) / Claude (fallback) | | **Knowledge Graph** | Neo4j for drug interaction network | | **Vector Search** | Qdrant + BioBERT embeddings | | **API** | FastAPI with HIPAA-compliant audit logging | | **Frontend** | Gradio (demo) + React Admin (production) | | **Databases** | PostgreSQL + Redis for session management | --- ### 🔄 LangGraph Workflow ``` ┌────────────────────────────────────────────────────────────────┐ │ MedicationSafetyOrchestrator │ │ │ │ START │ │ │ │ │ ▼ │ │ ┌─────────────────────────┐ │ │ │ DrugInteractionAgent │◄── Entry point │ │ │ (CYP analysis, DDI, ML) │ │ │ └───────────┬─────────────┘ │ │ │ │ │ ▼ │ │ ┌─────────────────────────┐ │ │ │ CONDITIONAL ROUTER │ │ │ │ Based on severity │ │ │ └───────────┬─────────────┘ │ │ │ │ │ ┌──────────┼──────────┐ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ "critical" "parallel" "low_risk" │ │ │ │ │ │ │ ▼ │ │ │ │ ┌────────┐ │ │ │ │ │ Human │ │ │ │ │ │ Review │ │ │ │ │ │ Flag │ │ │ │ │ └───┬────┘ │ │ │ │ │ ┌────┴────┐ │ │ │ │ │ Run in │ │ │ │ │ │ Parallel│ │ │ │ │ └────┬────┘ │ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ ┌─────────────────────────┐ │ │ │ PersonalizationAgent │ │ │ │ GuidelineComplianceAgent│ │ │ │ CostOptimizationAgent │ │ │ └───────────┬─────────────┘ │ │ │ │ │ ▼ │ │ ┌─────────────────────────┐ │ │ │ ExplanationAgent │◄── Final synthesis │ │ │ (Prioritized recs) │ │ │ └───────────┬─────────────┘ │ │ │ │ │ ▼ │ │ END │ └────────────────────────────────────────────────────────────────┘ ``` --- ### 🎯 MCP Server Tools | # | Tool Name | Description | |---|-----------|-------------| | 1 | `analyze_medication_safety` | Full multi-agent analysis pipeline | | 2 | `check_drug_interactions` | Drug-drug interaction detection | | 3 | `get_personalized_dosing` | Patient-specific dose adjustments | | 4 | `check_guideline_compliance` | Clinical guideline compliance check | | 5 | `optimize_medication_costs` | Cost optimization with generics | | 6 | `get_patient_profile` | Patient demographics and history | | 7 | `search_clinical_guidelines` | Vector search clinical guidelines | | 8 | `explain_medication_decision` | Patient-friendly explanation | | 9 | `search_pubmed_literature` | PubMed literature search (MCP Search) | | 10 | `search_fda_safety_alerts` | FDA safety alert search (MCP Search) | --- ### 📁 MCP Resources | Resource URI | Description | |--------------|-------------| | `guidelines://clinical-practice` | Clinical practice guidelines database | | `database://drug-interactions` | Drug interaction knowledge base | | `alerts://fda-safety` | FDA safety communications | --- ### 🎯 Hackathon Tracks - **Track 1 (Building MCP):** ✅ Production-ready MCP server with 10 tools + 3 resources - **Track 2 (Consumer Use):** ✅ Claude Desktop integration working **GitHub:** [mcp1stbirthday_hack](https://github.com/SmartGridsML/mcp1stbirthday_hack) """) gr.Markdown(""" --- *Generated by MedGuard Multi-Agent System | MCP 1st Birthday Hackathon Submission* """) # Launch if __name__ == "__main__": app.launch()