Healthcare & Medical AI Implementation

Implementation Example

# Clinical Decision Support System with LLMs
import asyncio
from typing import Dict, List, Optional, Any
from dataclasses import dataclass
from datetime import datetime
import numpy as np
from enum import Enum

class ClinicalPriority(Enum):
    """Clinical priority levels"""
    EMERGENCY = "emergency"
    URGENT = "urgent"
    ROUTINE = "routine"
    PREVENTIVE = "preventive"

@dataclass
class PatientData:
    """Patient data structure"""
    patient_id: str
    age: int
    gender: str
    chief_complaint: str
    symptoms: List[str]
    vital_signs: Dict[str, float]
    medical_history: List[str]
    current_medications: List[str]
    lab_results: Optional[Dict[str, Any]] = None
    imaging_results: Optional[Dict[str, Any]] = None

@dataclass
class ClinicalRecommendation:
    """Clinical recommendation from AI"""
    diagnosis_suggestions: List[Dict[str, float]]  # diagnosis: confidence
    recommended_tests: List[str]
    treatment_options: List[Dict[str, Any]]
    risk_factors: List[str]
    priority_level: ClinicalPriority
    evidence_references: List[str]
    confidence_score: float
    reasoning: str

class HealthcareLLMSystem:
    """Healthcare-specific LLM implementation with safety controls"""
    
    def __init__(self, api_key: str, model: str = "gpt-4"):
        self.api_key = api_key
        self.model = model
        self.safety_checks = HealthcareSafetySystem()
        self.audit_logger = ClinicalAuditLogger()
        self.knowledge_base = MedicalKnowledgeBase()
        
    async def analyze_patient(
        self, 
        patient_data: PatientData,
        include_differential: bool = True,
        max_recommendations: int = 5
    ) -> ClinicalRecommendation:
        """Analyze patient data and provide clinical recommendations"""
        
        # Step 1: Validate and anonymize patient data
        anonymized_data = self._anonymize_patient_data(patient_data)
        
        # Step 2: Extract clinical features
        clinical_features = self._extract_clinical_features(patient_data)
        
        # Step 3: Generate clinical context
        context = self._generate_clinical_context(
            patient_data, 
            clinical_features
        )
        
        # Step 4: Query LLM with safety controls
        llm_response = await self._query_llm_safely(context)
        
        # Step 5: Validate and structure response
        recommendation = self._parse_clinical_response(
            llm_response,
            patient_data
        )
        
        # Step 6: Apply medical knowledge validation
        validated_recommendation = await self.knowledge_base.validate(
            recommendation
        )
        
        # Step 7: Risk assessment
        risk_assessment = self._assess_clinical_risks(
            patient_data,
            validated_recommendation
        )
        validated_recommendation.risk_factors = risk_assessment
        
        # Step 8: Audit logging
        await self.audit_logger.log_clinical_decision(
            patient_id=patient_data.patient_id,
            recommendation=validated_recommendation,
            timestamp=datetime.utcnow()
        )
        
        return validated_recommendation
    
    def _anonymize_patient_data(
        self, 
        patient_data: PatientData
    ) -> Dict[str, Any]:
        """Anonymize patient data for HIPAA compliance"""
        return {
            "age": patient_data.age,
            "gender": patient_data.gender,
            "clinical_data": {
                "chief_complaint": patient_data.chief_complaint,
                "symptoms": patient_data.symptoms,
                "vital_signs": patient_data.vital_signs,
                "medical_history": patient_data.medical_history,
                "current_medications": patient_data.current_medications,
                "lab_results": patient_data.lab_results,
                "imaging_results": patient_data.imaging_results
            }
        }
    
    def _extract_clinical_features(
        self, 
        patient_data: PatientData
    ) -> Dict[str, Any]:
        """Extract relevant clinical features"""
        features = {
            "symptom_cluster": self._cluster_symptoms(patient_data.symptoms),
            "vital_abnormalities": self._detect_vital_abnormalities(
                patient_data.vital_signs
            ),
            "medication_interactions": self._check_medication_interactions(
                patient_data.current_medications
            ),
            "risk_profile": self._calculate_risk_profile(patient_data),
        }
        
        if patient_data.lab_results:
            features["lab_abnormalities"] = self._analyze_lab_results(
                patient_data.lab_results
            )
        
        return features
    
    def _generate_clinical_context(
        self,
        patient_data: PatientData,
        clinical_features: Dict[str, Any]
    ) -> str:
        """Generate structured clinical context for LLM"""
        context = f"""
Clinical Decision Support Request

Patient Profile:
- Age: {patient_data.age}, Gender: {patient_data.gender}
- Chief Complaint: {patient_data.chief_complaint}

Current Symptoms:
{self._format_symptoms(patient_data.symptoms)}

Vital Signs:
{self._format_vital_signs(patient_data.vital_signs)}

Medical History:
{self._format_medical_history(patient_data.medical_history)}

Current Medications:
{self._format_medications(patient_data.current_medications)}

Clinical Analysis Required:
1. Differential diagnosis with confidence scores
2. Recommended diagnostic tests
3. Treatment options with evidence basis
4. Risk factors and contraindications
5. Clinical priority assessment

Please provide evidence-based recommendations following current clinical guidelines.
Include references to support recommendations.
"""
        
        if patient_data.lab_results:
            context += f"\n\nLab Results:\n{self._format_lab_results(patient_data.lab_results)}"
        
        return context
    
    async def _query_llm_safely(self, context: str) -> str:
        """Query LLM with safety checks"""
        # Add safety prompt
        safety_prompt = """
IMPORTANT: You are providing clinical decision support. 
- Base recommendations on evidence-based medicine
- Include confidence levels for all suggestions
- Highlight any critical or emergency conditions
- Note when human physician review is essential
- Never make definitive diagnoses
- Always recommend appropriate follow-up
"""
        
        full_prompt = safety_prompt + "\n\n" + context
        
        # Implement actual LLM call here
        # This is a placeholder for the actual implementation
        response = await self._call_llm_api(full_prompt)
        
        # Validate response for safety
        if not self.safety_checks.validate_clinical_response(response):
            raise ValueError("Clinical response failed safety validation")
        
        return response
    
    def _parse_clinical_response(
        self,
        llm_response: str,
        patient_data: PatientData
    ) -> ClinicalRecommendation:
        """Parse and structure LLM response"""
        # This would parse the actual LLM response
        # Simplified example:
        return ClinicalRecommendation(
            diagnosis_suggestions=[
                {"Acute Bronchitis": 0.75},
                {"Pneumonia": 0.45},
                {"COVID-19": 0.35}
            ],
            recommended_tests=[
                "Chest X-ray",
                "Complete Blood Count (CBC)",
                "COVID-19 PCR test"
            ],
            treatment_options=[
                {
                    "name": "Supportive care",
                    "description": "Rest, fluids, antipyretics",
                    "evidence_level": "High"
                },
                {
                    "name": "Antibiotic therapy",
                    "description": "If bacterial infection confirmed",
                    "evidence_level": "Conditional"
                }
            ],
            risk_factors=[
                "Age-related complications",
                "Potential drug interactions"
            ],
            priority_level=self._determine_priority(patient_data),
            evidence_references=[
                "UpToDate: Acute Bronchitis in Adults",
                "CDC Guidelines for Respiratory Infections"
            ],
            confidence_score=0.82,
            reasoning="Based on symptom presentation and vital signs..."
        )
    
    def _determine_priority(
        self, 
        patient_data: PatientData
    ) -> ClinicalPriority:
        """Determine clinical priority level"""
        # Check for emergency indicators
        if self._has_emergency_indicators(patient_data):
            return ClinicalPriority.EMERGENCY
        
        # Check for urgent indicators
        if self._has_urgent_indicators(patient_data):
            return ClinicalPriority.URGENT
        
        # Default to routine
        return ClinicalPriority.ROUTINE
    
    def _has_emergency_indicators(
        self, 
        patient_data: PatientData
    ) -> bool:
        """Check for emergency medical conditions"""
        emergency_vitals = {
            "heart_rate": lambda x: x > 150 or x < 40,
            "blood_pressure_systolic": lambda x: x > 180 or x < 90,
            "oxygen_saturation": lambda x: x < 90,
            "temperature": lambda x: x > 40 or x < 35
        }
        
        for vital, check in emergency_vitals.items():
            if vital in patient_data.vital_signs:
                if check(patient_data.vital_signs[vital]):
                    return True
        
        # Check for emergency symptoms
        emergency_symptoms = [
            "chest pain", "difficulty breathing", "unconscious",
            "severe bleeding", "stroke symptoms"
        ]
        
        for symptom in patient_data.symptoms:
            if any(es in symptom.lower() for es in emergency_symptoms):
                return True
        
        return False
    
    def _assess_clinical_risks(
        self,
        patient_data: PatientData,
        recommendation: ClinicalRecommendation
    ) -> List[str]:
        """Assess clinical risks based on patient data"""
        risks = []
        
        # Age-related risks
        if patient_data.age > 65:
            risks.append("Elderly patient - consider adjusted dosing")
        elif patient_data.age < 18:
            risks.append("Pediatric patient - verify pediatric protocols")
        
        # Medication interaction risks
        interaction_risks = self._check_medication_interactions(
            patient_data.current_medications
        )
        risks.extend(interaction_risks)
        
        # Comorbidity risks
        comorbidity_risks = self._assess_comorbidity_risks(
            patient_data.medical_history
        )
        risks.extend(comorbidity_risks)
        
        return risks
    
    def _check_medication_interactions(
        self, 
        medications: List[str]
    ) -> List[str]:
        """Check for potential medication interactions"""
        # Simplified example - would integrate with drug interaction database
        interactions = []
        
        # Check for common dangerous combinations
        if "Warfarin" in medications and "Aspirin" in medications:
            interactions.append("Warning: Warfarin + Aspirin increases bleeding risk")
        
        return interactions
    
    def _assess_comorbidity_risks(
        self, 
        medical_history: List[str]
    ) -> List[str]:
        """Assess risks based on comorbidities"""
        risks = []
        
        # Diabetes-related risks
        if any("diabetes" in condition.lower() for condition in medical_history):
            risks.append("Diabetic patient - monitor glucose levels")
        
        # Cardiovascular risks
        if any("heart" in condition.lower() for condition in medical_history):
            risks.append("Cardiac history - monitor cardiovascular parameters")
        
        return risks

class HealthcareSafetySystem:
    """Safety validation for healthcare AI responses"""
    
    def validate_clinical_response(self, response: str) -> bool:
        """Validate clinical response for safety"""
        # Check for dangerous recommendations
        dangerous_patterns = [
            "stop all medications",
            "ignore symptoms",
            "delay emergency care",
            "self-diagnose",
            "replace physician"
        ]
        
        response_lower = response.lower()
        for pattern in dangerous_patterns:
            if pattern in response_lower:
                return False
        
        # Ensure appropriate disclaimers
        required_disclaimers = [
            "consult", "physician", "medical professional"
        ]
        
        has_disclaimer = any(
            disclaimer in response_lower 
            for disclaimer in required_disclaimers
        )
        
        return has_disclaimer

class ClinicalAuditLogger:
    """HIPAA-compliant audit logging"""
    
    async def log_clinical_decision(
        self,
        patient_id: str,
        recommendation: ClinicalRecommendation,
        timestamp: datetime
    ):
        """Log clinical decision for audit trail"""
        audit_entry = {
            "timestamp": timestamp.isoformat(),
            "patient_id_hash": self._hash_patient_id(patient_id),
            "action": "clinical_decision_support",
            "priority": recommendation.priority_level.value,
            "confidence": recommendation.confidence_score,
            "recommendations_count": len(recommendation.diagnosis_suggestions),
            "tests_recommended": len(recommendation.recommended_tests),
            "evidence_based": len(recommendation.evidence_references) > 0
        }
        
        # Log to secure audit system
        await self._write_audit_log(audit_entry)
    
    def _hash_patient_id(self, patient_id: str) -> str:
        """Hash patient ID for privacy"""
        import hashlib
        return hashlib.sha256(patient_id.encode()).hexdigest()
    
    async def _write_audit_log(self, entry: Dict[str, Any]):
        """Write to audit log system"""
        # Implementation would write to secure, immutable audit log
        pass

class MedicalKnowledgeBase:
    """Medical knowledge validation system"""
    
    async def validate(
        self, 
        recommendation: ClinicalRecommendation
    ) -> ClinicalRecommendation:
        """Validate recommendations against medical knowledge"""
        # Validate diagnoses against ICD-10
        validated_diagnoses = []
        for diagnosis in recommendation.diagnosis_suggestions:
            if self._validate_diagnosis(list(diagnosis.keys())[0]):
                validated_diagnoses.append(diagnosis)
        
        recommendation.diagnosis_suggestions = validated_diagnoses
        
        # Validate recommended tests
        validated_tests = []
        for test in recommendation.recommended_tests:
            if self._validate_test(test):
                validated_tests.append(test)
        
        recommendation.recommended_tests = validated_tests
        
        return recommendation
    
    def _validate_diagnosis(self, diagnosis: str) -> bool:
        """Validate diagnosis against medical databases"""
        # Would check against ICD-10, SNOMED CT, etc.
        return True  # Simplified
    
    def _validate_test(self, test: str) -> bool:
        """Validate test against standard medical tests"""
        # Would check against LOINC, CPT codes, etc.
        return True  # Simplified

# Example usage
async def main():
    # Initialize healthcare AI system
    healthcare_ai = HealthcareLLMSystem(
        api_key="your-api-key",
        model="gpt-4"
    )
    
    # Example patient data
    patient = PatientData(
        patient_id="PAT-12345",
        age=45,
        gender="Male",
        chief_complaint="Persistent cough and fever",
        symptoms=[
            "Cough for 5 days",
            "Fever up to 38.5°C",
            "Mild chest discomfort",
            "Fatigue"
        ],
        vital_signs={
            "heart_rate": 88,
            "blood_pressure_systolic": 125,
            "blood_pressure_diastolic": 80,
            "temperature": 38.2,
            "oxygen_saturation": 96,
            "respiratory_rate": 20
        },
        medical_history=[
            "Hypertension",
            "Type 2 Diabetes"
        ],
        current_medications=[
            "Metformin 1000mg daily",
            "Lisinopril 10mg daily"
        ],
        lab_results={
            "white_blood_cell_count": 11.5,
            "c_reactive_protein": 15.2
        }
    )
    
    # Get clinical recommendations
    recommendation = await healthcare_ai.analyze_patient(patient)
    
    # Display recommendations
    print(f"Priority Level: {recommendation.priority_level.value}")
    print(f"Confidence Score: {recommendation.confidence_score}")
    print("\nDifferential Diagnoses:")
    for diagnosis in recommendation.diagnosis_suggestions:
        for name, confidence in diagnosis.items():
            print(f"  - {name}: {confidence*100:.1f}% confidence")
    
    print("\nRecommended Tests:")
    for test in recommendation.recommended_tests:
        print(f"  - {test}")
    
    print("\nRisk Factors:")
    for risk in recommendation.risk_factors:
        print(f"  - {risk}")

if __name__ == "__main__":
    asyncio.run(main())