Medical Diagnosis Multi-Agent System

Build an AI-powered medical diagnosis system with specialized agents for symptom analysis, medical history review, and treatment recommendations.

Overview

This example demonstrates how to build a multi-agent healthcare system that can:

• Analyze patient symptoms and medical history
• Suggest potential diagnoses based on evidence
• Recommend appropriate medical tests
• Provide treatment suggestions and referrals
• Handle emergency escalations
• Maintain HIPAA compliance and patient privacy

Perfect for: Telemedicine platforms, healthcare clinics, medical triage systems, and patient support applications.

How It Works

Workflow Steps

1. Patient Intake: Manager agent receives patient symptoms and history
2. Triage Assessment: Evaluates urgency level (emergency, urgent, routine)
3. Specialist Routing: Delegates to appropriate medical specialist agents
4. Diagnosis Analysis: Specialist agents analyze symptoms and suggest diagnoses
5. Test Recommendations: Suggests relevant medical tests if needed
6. Treatment Plan: Provides evidence-based treatment recommendations
7. Follow-up Scheduling: Arranges appropriate follow-up care

Agent Hierarchy

Install AgentRouter

# Install AgentRouter SDK
!pip install agentrouter -q

# Import required libraries
from agentrouter import ManagerAgent, WorkerAgent, tool
from agentrouter.visualization import PipelineInspector
import asyncio
import nest_asyncio
from typing import Dict, Any, List
from datetime import datetime
import json

nest_asyncio.apply()

Complete Code

# ============================================
# MEDICAL DIAGNOSIS MULTI-AGENT SYSTEM
# ============================================

# Configuration
API_KEY = "your-api-key-here" # Replace with your actual API key

# ============================================
# TOOL DEFINITIONS
# ============================================

@tool(
schema={
"type": "function",
"function": {
"name": "analyze_symptoms",
"description": "Processes patient symptoms to identify potential medical conditions with urgency assessment and diagnostic test recommendations",
"parameters": {
"type": "object",
"properties": {
"symptoms": {
"type": "array",
"items": {"type": "string"},
"description": "List of patient symptoms"
},
"duration": {
"type": "string",
"description": "Duration of symptoms (e.g., '2 days', '1 week')"
},
"severity": {
"type": "string",
"enum": ["mild", "moderate", "severe"],
"description": "Severity of symptoms"
}
},
"required": ["symptoms", "severity"]
}
}
}
)
async def analyze_symptoms(symptoms: List[str], duration: str = "unknown", severity: str = "moderate") -> Dict[str, Any]:
"""Analyze symptoms and suggest potential conditions"""

# Simulated symptom-condition mapping
condition_database = {
("fever", "cough"): {
"conditions": ["Common Cold", "Influenza", "COVID-19"],
"urgency": "moderate",
"tests_recommended": ["Rapid flu test", "COVID-19 test"]
},
("chest pain", "shortness of breath"): {
"conditions": ["Angina", "Heart Attack", "Panic Attack"],
"urgency": "high",
"tests_recommended": ["ECG", "Troponin levels", "Chest X-ray"]
},
("headache", "dizziness"): {
"conditions": ["Migraine", "Vertigo", "Dehydration"],
"urgency": "low",
"tests_recommended": ["Blood pressure check", "Neurological exam"]
},
("rash", "itching"): {
"conditions": ["Allergic Reaction", "Eczema", "Contact Dermatitis"],
"urgency": "low",
"tests_recommended": ["Allergy test", "Skin biopsy if persistent"]
}
}

# Find matching conditions
symptoms_lower = [s.lower() for s in symptoms]
for key_symptoms, data in condition_database.items():
if any(symptom in " ".join(symptoms_lower) for symptom in key_symptoms):
return {
"symptoms_analyzed": symptoms,
"duration": duration,
"severity": severity,
"potential_conditions": data["conditions"],
"urgency_level": data["urgency"],
"recommended_tests": data["tests_recommended"],
"analysis_timestamp": datetime.now().isoformat()
}

return {
"symptoms_analyzed": symptoms,
"duration": duration,
"severity": severity,
"potential_conditions": ["Further evaluation needed"],
"urgency_level": "moderate",
"recommended_tests": ["Complete blood count", "Basic metabolic panel"],
"analysis_timestamp": datetime.now().isoformat()
}

@tool(
schema={
"type": "function",
"function": {
"name": "check_medical_history",
"description": "Retrieves comprehensive patient medical records including allergies, medications, chronic conditions, and family history for informed clinical decisions",
"parameters": {
"type": "object",
"properties": {
"patient_id": {
"type": "string",
"description": "Patient identifier"
},
"condition_category": {
"type": "string",
"enum": ["cardiac", "neurological", "respiratory", "dermatological", "general"],
"description": "Category of conditions to check"
}
},
"required": ["patient_id"]
}
}
}
)
async def check_medical_history(patient_id: str, condition_category: str = "general") -> Dict[str, Any]:
"""Check patient medical history"""

# Simulated patient database
return {
"patient_id": patient_id,
"age": 45,
"gender": "Female",
"allergies": ["Penicillin", "Shellfish"],
"chronic_conditions": ["Hypertension", "Type 2 Diabetes"],
"current_medications": [
{"name": "Metformin", "dose": "500mg", "frequency": "twice daily"},
{"name": "Lisinopril", "dose": "10mg", "frequency": "once daily"}
],
"recent_visits": [
{"date": "2025-01-05", "reason": "Routine checkup", "outcome": "stable"},
{"date": "2024-12-15", "reason": "Flu symptoms", "outcome": "resolved"}
],
"immunizations": ["COVID-19", "Flu 2024", "Tetanus 2023"],
"family_history": ["Heart disease", "Diabetes"],
"category_checked": condition_category
}

@tool(
schema={
"type": "function",
"function": {
"name": "recommend_treatment",
"description": "Generates evidence-based treatment plans with medications, lifestyle modifications, and follow-up protocols tailored to patient conditions and contraindications",
"parameters": {
"type": "object",
"properties": {
"condition": {
"type": "string",
"description": "Diagnosed or suspected condition"
},
"patient_id": {
"type": "string",
"description": "Patient identifier"
},
"severity": {
"type": "string",
"enum": ["mild", "moderate", "severe"],
"description": "Condition severity"
}
},
"required": ["condition", "patient_id"]
}
}
}
)
async def recommend_treatment(condition: str, patient_id: str, severity: str = "moderate") -> Dict[str, Any]:
"""Recommend treatment based on condition"""

treatments = {
"Common Cold": {
"medications": ["Acetaminophen for fever", "Rest and fluids"],
"lifestyle": ["Get plenty of rest", "Stay hydrated", "Avoid contact with others"],
"follow_up": "If symptoms persist beyond 10 days"
},
"Hypertension": {
"medications": ["ACE inhibitors", "Beta blockers", "Diuretics"],
"lifestyle": ["Low sodium diet", "Regular exercise", "Stress management"],
"follow_up": "Monthly blood pressure monitoring"
},
"Migraine": {
"medications": ["Triptans for acute attacks", "Preventive medications if frequent"],
"lifestyle": ["Identify triggers", "Regular sleep schedule", "Stress reduction"],
"follow_up": "Neurologist referral if frequent"
}
}

treatment_plan = treatments.get(condition, {
"medications": ["Consult with physician"],
"lifestyle": ["General health maintenance"],
"follow_up": "Schedule appointment with primary care"
})

return {
"patient_id": patient_id,
"condition": condition,
"severity": severity,
"treatment_plan": treatment_plan,
"contraindications_checked": True,
"prescription_required": severity in ["moderate", "severe"],
"emergency_referral": severity == "severe",
"created_at": datetime.now().isoformat()
}

@tool(
schema={
"type": "function",
"function": {
"name": "order_lab_tests",
"description": "Initiates diagnostic laboratory test orders with priority routing, preparation instructions, and result timeline management",
"parameters": {
"type": "object",
"properties": {
"patient_id": {
"type": "string",
"description": "Patient identifier"
},
"tests": {
"type": "array",
"items": {"type": "string"},
"description": "List of tests to order"
},
"urgency": {
"type": "string",
"enum": ["routine", "urgent", "stat"],
"description": "Test urgency level"
}
},
"required": ["patient_id", "tests"]
}
}
}
)
async def order_lab_tests(patient_id: str, tests: List[str], urgency: str = "routine") -> Dict[str, Any]:
"""Order laboratory tests"""

order_id = f"LAB{datetime.now().strftime('%Y%m%d%H%M%S')}"

return {
"order_id": order_id,
"patient_id": patient_id,
"tests_ordered": tests,
"urgency": urgency,
"estimated_results": "2-4 hours" if urgency == "stat" else "24-48 hours",
"lab_location": "Main Hospital Lab - Room 201",
"preparation_required": "Fasting required for glucose tests" if any("glucose" in t.lower() for t in tests) else "None",
"order_status": "pending",
"ordered_at": datetime.now().isoformat()
}

@tool(
schema={
"type": "function",
"function": {
"name": "schedule_followup",
"description": "Coordinates specialist referrals and follow-up appointments with appropriate urgency triaging and automated scheduling",
"parameters": {
"type": "object",
"properties": {
"patient_id": {
"type": "string",
"description": "Patient identifier"
},
"specialist_type": {
"type": "string",
"description": "Type of specialist needed"
},
"urgency": {
"type": "string",
"enum": ["routine", "soon", "urgent"],
"description": "Appointment urgency"
},
"reason": {
"type": "string",
"description": "Reason for follow-up"
}
},
"required": ["patient_id", "reason"]
}
}
}
)
async def schedule_followup(patient_id: str, reason: str, specialist_type: str = "Primary Care", urgency: str = "routine") -> Dict[str, Any]:
"""Schedule follow-up appointment"""

appointment_windows = {
"urgent": "Within 24-48 hours",
"soon": "Within 1 week",
"routine": "Within 2-4 weeks"
}

return {
"patient_id": patient_id,
"appointment_id": f"APT{datetime.now().strftime('%Y%m%d%H%M%S')}",
"specialist_type": specialist_type,
"reason": reason,
"urgency": urgency,
"available_window": appointment_windows[urgency],
"location": "Medical Center - Building A",
"preparation": "Bring list of current medications and recent test results",
"status": "scheduled",
"created_at": datetime.now().isoformat()
}

# ============================================
# AGENT SYSTEM SETUP
# ============================================

class MedicalDiagnosisSystem:
"""Complete medical diagnosis multi-agent system"""

def __init__(self, api_key: str):
self.api_key = api_key
self.manager = None
self.agents = {}

def setup(self):
"""Setup the multi-agent hierarchy"""

# Create Manager Agent - Medical Director
self.manager = ManagerAgent(
name="medical_director",
api_key=self.api_key,
model="usf-mini",
backstory="""I am an experienced Medical Director with 20 years of clinical practice
and healthcare management. I oversee a team of medical professionals and ensure
accurate diagnoses and appropriate treatment plans while maintaining patient safety.""",
goal="Provide accurate medical assessments and coordinate appropriate care",
instruction="""
1. Evaluate patient symptoms and medical history comprehensively
2. Determine urgency level and appropriate specialist routing
3. Coordinate between different medical specialists
4. Ensure evidence-based diagnosis and treatment recommendations
5. Maintain patient safety and HIPAA compliance
6. Escalate emergency cases immediately
""",
knowledge_cutoff="15 January 2025",
temperature=0.3, # Lower temperature for medical accuracy
max_iterations=25
)

# Register manager tools
self.manager.register_tool(analyze_symptoms)
self.manager.register_tool(check_medical_history)

# Create Level 1 Workers

# Triage Nurse
triage_nurse = self.manager.create_worker(
name="Triage_Nurse",
role="Assesses patient acuity levels and prioritizes emergency care routing",
backstory="Certified emergency nurse with expertise in rapid patient assessment",
goal="Quickly assess patient urgency and route to appropriate care",
instruction="Evaluate symptoms for emergency indicators and prioritize care"
)
triage_nurse.register_tool(analyze_symptoms)
self.agents["triage_nurse"] = triage_nurse

# General Practitioner
general_practitioner = self.manager.create_worker(
name="General_Practitioner",
role="Diagnoses common conditions and manages primary healthcare delivery",
backstory="Board-certified family medicine physician with comprehensive training",
goal="Provide primary care diagnosis and treatment",
instruction="Handle common conditions and coordinate specialist referrals"
)
general_practitioner.register_tool(check_medical_history)
general_practitioner.register_tool(recommend_treatment)
general_practitioner.register_tool(order_lab_tests)
self.agents["general_practitioner"] = general_practitioner

# Specialist Coordinator
specialist_coordinator = self.manager.create_worker(
name="Specialist_Coordinator",
role="Routes complex cases to appropriate specialists and manages referral workflows",
backstory="Healthcare coordinator managing specialist consultations",
goal="Route patients to appropriate specialists",
instruction="Match patient conditions with specialist expertise"
)
specialist_coordinator.register_tool(schedule_followup)
self.agents["specialist_coordinator"] = specialist_coordinator

# Create Level 2 Workers (Specialists under Coordinator)

# Cardiologist
cardiologist = specialist_coordinator.create_worker(
name="Cardiologist",
role="Diagnoses and treats heart diseases, arrhythmias, and vascular conditions",
backstory="Board-certified cardiologist with expertise in heart conditions",
goal="Diagnose and treat cardiovascular conditions"
)
cardiologist.register_tool(order_lab_tests)
cardiologist.register_tool(recommend_treatment)

# Neurologist
neurologist = specialist_coordinator.create_worker(
name="Neurologist",
role="Manages brain, spinal cord, and peripheral nervous system disorders",
backstory="Neurologist specializing in brain and nervous system disorders",
goal="Diagnose and treat neurological conditions"
)
neurologist.register_tool(order_lab_tests)
neurologist.register_tool(recommend_treatment)

# Dermatologist
dermatologist = specialist_coordinator.create_worker(
name="Dermatologist",
role="Evaluates and treats skin conditions, lesions, and dermatological diseases",
backstory="Dermatologist with expertise in skin conditions",
goal="Diagnose and treat skin disorders"
)
dermatologist.register_tool(recommend_treatment)

# Create support workers under General Practitioner

# Lab Technician
lab_tech = general_practitioner.create_worker(
name="Lab_Technician",
role="Processes diagnostic tests and validates laboratory result accuracy",
backstory="Certified lab technician with expertise in diagnostic testing",
goal="Process and interpret laboratory tests"
)
lab_tech.register_tool(order_lab_tests)

# Pharmacist
pharmacist = general_practitioner.create_worker(
name="Pharmacist",
role="Verifies drug interactions and optimizes medication therapy protocols",
backstory="Licensed pharmacist with expertise in medication management",
goal="Ensure safe and effective medication therapy"
)
pharmacist.register_tool(recommend_treatment)

print("✅ Medical Diagnosis System initialized successfully!")

# Visualize hierarchy
inspector = PipelineInspector(self.manager)
print("\n📊 MEDICAL TEAM HIERARCHY:")
print(inspector.visualize(format='mermaid'))
print("\n🔧 MEDICAL TOOLS AVAILABLE:")
# Get all tools from inspector
for agent in inspector.pipeline_data['agents']:
    if agent['tools']:
        print(f"  {agent['name']}:")
        for tool_name in agent['tools']:
            print(f"    • {tool_name}")

async def diagnose_patient(self, symptoms: str, patient_id: str = "PAT001"):
"""Process patient diagnosis request"""

messages = [
{
"role": "system",
"content": f"Patient ID: {patient_id}\nDate: {datetime.now().strftime('%Y-%m-%d %H:%M')}\nMode: Medical Consultation"
},
{
"role": "user",
"content": symptoms
}
]

print(f"\n👤 Patient ({patient_id}): {symptoms}")
print("🏥 Processing with medical AI system...")

try:
response = await self.manager.execute(messages)

if response and "choices" in response:
result = response["choices"][0]["message"]["content"]
print(f"\n✅ Medical Assessment:\n{result}")
return {
"success": True,
"patient_id": patient_id,
"symptoms": symptoms,
"assessment": result,
"timestamp": datetime.now().isoformat()
}

except Exception as e:
print(f"❌ Error: {str(e)}")
return {
"success": False,
"patient_id": patient_id,
"symptoms": symptoms,
"error": str(e),
"timestamp": datetime.now().isoformat()
}

# ============================================
# EXAMPLE USAGE
# ============================================

async def run_medical_example():
"""Run a medical diagnosis scenario"""

# Initialize system
medical_system = MedicalDiagnosisSystem(API_KEY)
medical_system.setup()

# Test scenarios
test_cases = [
{
"patient_id": "PAT001",
"symptoms": "I've been having severe chest pain and shortness of breath for the past hour. The pain radiates to my left arm."
},
{
"patient_id": "PAT002",
"symptoms": "I have a persistent headache for 3 days with sensitivity to light and nausea."
}
]

print("\n" + "="*60)
print("🏥 RUNNING MEDICAL DIAGNOSIS SCENARIOS")
print("="*60)

results = []
for test_case in test_cases[:1]: # Run first case for demo
result = await medical_system.diagnose_patient(
test_case["symptoms"],
test_case["patient_id"]
)
results.append(result)
print("\n" + "-"*60)

return results

# ============================================
# MAIN EXECUTION
# ============================================

# Run the example
results = await run_medical_example()

# Display results
print("\n📋 DIAGNOSIS RESULTS:")
for result in results:
print(json.dumps(result, indent=2))

Expected Output

✅ Medical Diagnosis System initialized successfully!

📊 MEDICAL TEAM HIERARCHY:
graph TB
    Medical_Director["Medical Director"]
    Triage_Nurse["Triage Nurse"]
    General_Practitioner["General Practitioner"]
    Specialist_Coordinator["Specialist Coordinator"]
    Cardiologist["Cardiologist"]
    Neurologist["Neurologist"]
    Dermatologist["Dermatologist"]
    Lab_Technician["Lab Technician"]
    Pharmacist["Pharmacist"]
    
    Medical_Director --> Triage_Nurse
    Medical_Director --> General_Practitioner
    Medical_Director --> Specialist_Coordinator
    Specialist_Coordinator --> Cardiologist
    Specialist_Coordinator --> Neurologist
    Specialist_Coordinator --> Dermatologist
    General_Practitioner --> Lab_Technician
    General_Practitioner --> Pharmacist

🔧 MEDICAL TOOLS AVAILABLE:
  • analyze_symptoms
  • check_medical_history
  • recommend_treatment
  • order_lab_tests
  • schedule_followup

============================================================
🏥 RUNNING MEDICAL DIAGNOSIS SCENARIOS
============================================================

👤 Patient (PAT001): I've been having severe chest pain and shortness of breath for the past hour. The pain radiates to my left arm.
🏥 Processing with medical AI system...

✅ Medical Assessment:
⚠️ URGENT MEDICAL ATTENTION REQUIRED

Based on your symptoms, this appears to be a potential cardiac emergency. The combination of severe chest pain, shortness of breath, and pain radiating to your left arm are classic signs of a possible heart attack (myocardial infarction).

IMMEDIATE ACTIONS REQUIRED:
1. Call 911 or emergency services immediately
2. Chew an aspirin (325mg) if available and not allergic
3. Remain calm and sit in a comfortable position
4. Do not drive yourself to the hospital

DIAGNOSTIC TESTS ORDERED (STAT):
- ECG (Electrocardiogram)
- Troponin levels
- Complete blood count
- Chest X-ray

This is being treated as a medical emergency. The emergency department has been notified and is preparing for your arrival.

📋 DIAGNOSIS RESULTS:
{
  "success": true,
  "patient_id": "PAT001",
  "symptoms": "I've been having severe chest pain and shortness of breath for the past hour. The pain radiates to my left arm.",
  "assessment": "⚠️ URGENT MEDICAL ATTENTION REQUIRED...",
  "timestamp": "2025-01-12T10:30:45.123456"
}

Customization Options

1. Add Specialized Medical Departments

# Add Radiology Department
radiologist = specialist_coordinator.create_worker(
    name="Radiologist",
    role="Medical Imaging Specialist",
    backstory="Expert in interpreting medical imaging",
    goal="Analyze X-rays, CT scans, and MRI results"
)

2. Integrate with Electronic Health Records

@tool(schema={...})
async def fetch_ehr_records(patient_id: str, record_type: str):
    """Fetch patient records from EHR system"""
    # Integration with hospital EHR system
    return {"records": ehr_data, "last_updated": timestamp}

3. Add Telemedicine Capabilities

@tool(schema={...})
async def initiate_video_consultation(patient_id: str, doctor_id: str):
    """Start video consultation session"""
    return {"session_id": session_id, "join_url": video_url}

Production Considerations

1. HIPAA Compliance: Ensure all patient data is encrypted and access is logged
2. Medical Liability: Include disclaimers and human physician oversight
3. Emergency Protocols: Immediate escalation for life-threatening conditions
4. Integration: Connect with hospital EHR, lab systems, and pharmacy networks
5. Audit Trail: Comprehensive logging of all medical decisions
6. Failsafe Mechanisms: Always have human physician backup available
7. Regulatory Compliance: Follow local medical practice regulations