Healthcare Diagnosis

Medical triage with symptom analysis, lab processing, and treatment recommendations

Multi-agent healthcare diagnosis: symptom analysis → lab processing → history review → diagnosis → treatment. Includes SQLite patient records, structured output, and API deployment.

Setup

# Create environment
python3 -m venv venv && source venv/bin/activate

# Install packages
pip install praisonaiagents praisonai

# Set API key
export OPENAI_API_KEY="your-key"

Create Sample Data

# Create patient records
cat > patients.json << 'EOF'
[
  {"id": "PAT001", "name": "John Doe", "age": 45, "symptoms": ["fever", "cough", "fatigue"], "history": ["diabetes"], "allergies": ["penicillin"]},
  {"id": "PAT002", "name": "Jane Smith", "age": 32, "symptoms": ["headache", "nausea"], "history": [], "allergies": []},
  {"id": "PAT003", "name": "Bob Wilson", "age": 58, "symptoms": ["chest_pain", "shortness_of_breath"], "history": ["hypertension", "heart_disease"], "allergies": ["aspirin"]}
]
EOF

# Create lab results
cat > labs.json << 'EOF'
{
  "PAT001": {"wbc": 12000, "crp": 45, "glucose": 180, "status": "abnormal"},
  "PAT002": {"wbc": 7500, "crp": 5, "glucose": 95, "status": "normal"},
  "PAT003": {"wbc": 9000, "crp": 25, "troponin": 0.8, "status": "critical"}
}
EOF

Run: Python Code

from praisonaiagents import Agent, AgentTeam, Task, tool
from pydantic import BaseModel
from typing import List
import json

# Structured output
class DiagnosisReport(BaseModel):
    patient_id: str
    primary_diagnosis: str
    confidence: float
    severity: str  # critical, high, medium, low
    treatments: List[str]
    contraindications: List[str]
    follow_up: str

# Database persistence is configured via memory={} parameter

# Tools
@tool
def get_patient(patient_id: str) -> str:
    """Get patient information."""
    with open("patients.json") as f:
        patients = json.load(f)
    for p in patients:
        if p["id"] == patient_id:
            return json.dumps(p)
    return json.dumps({"error": "Patient not found"})

@tool
def get_lab_results(patient_id: str) -> str:
    """Get lab results for patient."""
    with open("labs.json") as f:
        labs = json.load(f)
    return json.dumps(labs.get(patient_id, {"status": "pending"}))

@tool
def check_drug_interactions(medications: str, allergies: str) -> str:
    """Check for drug interactions and allergies."""
    meds = medications.split(",")
    allergy_list = allergies.split(",")
    
    interactions = []
    for med in meds:
        if med.strip().lower() in [a.strip().lower() for a in allergy_list]:
            interactions.append(f"ALLERGY: {med}")
    
    return json.dumps({"safe": len(interactions) == 0, "warnings": interactions})

# Agents
triage = Agent(
    name="TriageNurse",
    instructions="Get patient info and assess urgency. Use get_patient tool.",
    tools=[get_patient],
    memory={
        "db": "sqlite:///healthcare.db",
        "session_id": "healthcare-diagnosis"
    }
)

lab_analyst = Agent(
    name="LabAnalyst",
    instructions="Analyze lab results. Use get_lab_results tool.",
    tools=[get_lab_results]
)

physician = Agent(
    name="Physician",
    instructions="""Generate diagnosis based on symptoms, history, and labs.
    Consider: symptom patterns, lab abnormalities, patient history.
    Check drug interactions before recommending treatment.""",
    tools=[check_drug_interactions]
)

# Tasks
triage_task = Task(
    description="Triage patient: {patient_id}",
    agent=triage,
    expected_output="Patient info with symptom severity"
)

lab_task = Task(
    description="Analyze lab results for patient",
    agent=lab_analyst,
    expected_output="Lab interpretation with abnormalities"
)

diagnosis_task = Task(
    description="Generate diagnosis and treatment plan",
    agent=physician,
    expected_output="Structured diagnosis report",
    output_pydantic=DiagnosisReport
)

# Run
agents = AgentTeam(agents=[triage, lab_analyst, physician], tasks=[triage_task, lab_task, diagnosis_task])
result = agents.start(patient_id="PAT003")
print(result)

Run: CLI

# Single patient
praisonai "Diagnose patient with fever, cough, and fatigue" --verbose

# With persistence
praisonai "Analyze symptoms: chest pain, shortness of breath" --memory --user-id doctor1

# Interactive consultation
praisonai chat --memory --user-id medical_staff

Run: agents.yaml

Create agents.yaml:

framework: praisonai
topic: "medical diagnosis system"
roles:
  triage:
    role: Triage Nurse
    goal: Assess patient urgency
    backstory: Expert at emergency triage
    tasks:
      assess:
        description: |
          Assess patient:
          - Symptom severity
          - Vital signs
          - Chief complaint
          - Urgency level (1-5)
        expected_output: Triage assessment
        
  analyst:
    role: Lab Analyst
    goal: Interpret lab results
    backstory: Expert at clinical laboratory analysis
    tasks:
      analyze:
        description: |
          Analyze labs:
          - Blood counts
          - Inflammatory markers
          - Organ function tests
          - Flag abnormalities
        expected_output: Lab interpretation
        
  physician:
    role: Attending Physician
    goal: Diagnose and treat
    backstory: Board-certified physician
    tasks:
      diagnose:
        description: |
          Generate diagnosis:
          - Primary diagnosis
          - Differential diagnoses
          - Treatment plan
          - Contraindications
          - Follow-up schedule
        expected_output: Complete diagnosis report

Run:

praisonai agents.yaml --verbose

Monitor & Verify

# View patient history
praisonai --history 10 --user-id doctor1

# Check metrics
praisonai --metrics

# Export records
praisonai --save patient_records

Serve API

from praisonaiagents import Agent, tool
import json

@tool
def quick_triage(symptoms: str, age: int) -> str:
    """Quick symptom triage."""
    symptom_list = symptoms.lower().split(",")
    
    critical = ["chest_pain", "shortness_of_breath", "stroke_symptoms"]
    high = ["fever", "severe_pain", "bleeding"]
    
    severity = "low"
    for s in symptom_list:
        s = s.strip()
        if any(c in s for c in critical):
            severity = "critical"
            break
        if any(h in s for h in high):
            severity = "high"
    
    if age > 65 and severity != "low":
        severity = "critical" if severity == "high" else "high"
    
    return json.dumps({"severity": severity, "symptoms": symptom_list})

agent = Agent(
    name="TriageAPI",
    instructions="Perform quick symptom triage.",
    tools=[quick_triage]
)

agent.launch(path="/triage", port=8000)

Test:

curl -X POST http://localhost:8000/triage \
  -H "Content-Type: application/json" \
  -d '{"message": "Triage: chest pain, shortness of breath, age 58"}'

Cleanup

rm -f healthcare.db patients.json labs.json
deactivate

Features Demonstrated

Feature	Implementation
Multi-agent	Triage → Lab → Physician
Structured Output	Pydantic DiagnosisReport
Patient Records	JSON-based data
Drug Interactions	Safety checking tool
DB Persistence	SQLite via db()
CLI	praisonai chat for consult
YAML Config	3-agent medical pipeline
API Endpoint	agent.launch()