Medical Knowledge Assistant for Hospital

Scenario Overview

A hospital system needs an AI assistant to help medical staff quickly find relevant protocols, drug interactions, and treatment guidelines. The system must be able to:

• Handle complex medical terminology and abbreviations
• Provide accurate, evidence-based medical information
• Filter by patient demographics and medical conditions
• Ensure patient privacy and data protection
• Maintain compliance with medical regulations

User Journey

1. Doctor Query

User: "What's the protocol for treating sepsis in pediatric patients with diabetes?"

2. System Response Flow

Step 1: Query Understanding & Medical Expansion

# Medical query expansion with domain-specific terminology
expanded_query = medical_expander.expand(
    query="What's the protocol for treating sepsis in pediatric patients with diabetes?",
    domain="medical",
    context={
        "user_role": "physician",
        "specialty": "pediatrics",
        "patient_age_group": "pediatric",
        "condition": "sepsis",
        "comorbidity": "diabetes"
    }
)

# Results in:
# - "sepsis" → "septic shock", "SIRS", "severe sepsis", "bacteremia"
# - "pediatric" → "children", "infants", "adolescents", "age-specific"
# - "diabetes" → "diabetes mellitus", "DM", "hyperglycemia", "insulin"
# - "protocol" → "guidelines", "treatment protocol", "clinical pathway"

Step 2: Faceted Search with Medical Filters

# Multi-dimensional filtering for medical information
faceted_search = faceted_search_engine.search(
    query=expanded_query,
    filters={
        "patient_age_group": "pediatric",
        "condition": "sepsis",
        "comorbidity": "diabetes",
        "document_type": "clinical_guidelines",
        "evidence_level": "high",
        "last_updated": "2024-01-01"
    },
    facets=["patient_demographics", "medical_conditions", "treatment_protocols"]
)

Step 3: Hybrid Retrieval with Medical Knowledge

# Search medical databases and clinical guidelines
retrieval_results = hybrid_retriever.retrieve(
    query=expanded_query,
    k=25,  # More candidates for complex medical queries
    filters={
        "category": "clinical_guidelines",
        "specialty": "pediatrics",
        "condition": "sepsis",
        "evidence_level": "high"
    }
)

Step 4: Medical Reranking

# Rerank based on medical relevance and evidence quality
reranked_results = medical_reranker.rerank(
    query=expanded_query,
    documents=retrieval_results,
    context={
        "patient_age": "pediatric",
        "condition_severity": "severe",
        "comorbidities": ["diabetes"]
    },
    top_k=8
)

Step 5: Medical Response Generation

# Generate evidence-based medical response
response = medical_generator.generate(
    query=expanded_query,
    context=reranked_results,
    safety_checks=True,
    evidence_level="high"
)

3. System Response

# Pediatric Sepsis Protocol with Diabetes Considerations

## Immediate Assessment (First Hour)
1. **Vital Signs Monitoring**
   - Continuous heart rate, blood pressure, temperature
   - Blood glucose monitoring every 2 hours
   - Watch for diabetic ketoacidosis (DKA) signs

2. **Laboratory Workup**
   - Complete blood count with differential
   - Basic metabolic panel (glucose, electrolytes)
   - Blood cultures (2 sets from different sites)
   - Lactate level
   - HbA1c if not recent

## Fluid Resuscitation
- **Initial bolus**: 20ml/kg normal saline over 15-20 minutes
- **Glucose management**: Target 140-180 mg/dL
- **Insulin therapy**: Consider continuous insulin infusion if glucose >200 mg/dL
- **Monitor**: Watch for fluid overload in diabetic patients

## Antibiotic Therapy
- **Empiric coverage**: Vancomycin + Piperacillin-Tazobactam
- **Adjust for diabetes**: Consider broader coverage for diabetic foot infections
- **Timing**: Administer within 1 hour of recognition

## Vasopressor Support
- **Norepinephrine**: First-line vasopressor
- **Glucose monitoring**: Essential during vasopressor therapy
- **Insulin adjustment**: May need increased insulin requirements

## Special Considerations for Diabetic Patients
- **DKA risk**: Monitor for ketones if glucose >250 mg/dL
- **Infection source**: Higher risk of diabetic foot infections
- **Insulin resistance**: May require higher insulin doses
- **Renal function**: Monitor closely due to diabetic nephropathy risk

## Monitoring and Follow-up
- **Glucose**: Every 2 hours initially
- **Lactate**: Repeat in 2-4 hours
- **Clinical response**: Reassess every 6 hours
- **Endocrinology consult**: Consider if glucose difficult to control

**Note**: This protocol should be adapted based on local hospital guidelines and patient-specific factors. Always consult with senior medical staff for complex cases.

Implementation Code

1. Medical Knowledge Base Setup

# config/medical_config.py
from packages.rag import HybridRetriever, CrossEncoderReranker
from packages.rag.faceted_search import FacetedSearchEngine
from packages.agents import RAGAgentGraph, AgentConfig
from packages.rag.query_expansion import MedicalQueryExpander
from packages.observability import MetricsCollector, StructuredLogger

class MedicalConfig:
    def __init__(self):
        # Medical query expansion
        self.medical_expander = MedicalQueryExpander(
            domain="medical",
            medical_terminology_file="data/medical_terminology.json",
            drug_database_file="data/drug_interactions.json",
            specialty_abbreviations_file="data/medical_abbreviations.json"
        )
        
        # Faceted search for medical information
        self.faceted_search = FacetedSearchEngine(
            facets={
                "patient_demographics": ["age_group", "gender", "weight_category"],
                "medical_conditions": ["primary_diagnosis", "comorbidities", "severity"],
                "treatment_protocols": ["protocol_type", "evidence_level", "specialty"],
                "medications": ["drug_class", "interactions", "contraindications"]
            }
        )
        
        # Hybrid retrieval with medical focus
        self.hybrid_retriever = HybridRetriever(
            vector_retriever=VectorRetriever(
                model_name="text-embedding-3-large",
                vector_store=OpenSearchStore(
                    index_name="medical_knowledge_base"
                )
            ),
            bm25_retriever=BM25Retriever(
                index_path="data/medical_bm25_index"
            ),
            alpha=0.7  # Favor vector search for medical terminology
        )
        
        # Medical-specific reranking
        self.medical_reranker = MedicalReranker(
            model_name="cross-encoder/ms-marco-MiniLM-L-6-v2",
            medical_context_weight=0.8,
            evidence_level_weight=0.9
        )
        
        # Agent configuration for medical domain
        self.agent_config = AgentConfig(
            model_name="gpt-4-turbo-preview",
            temperature=0.05,  # Lower temperature for medical accuracy
            max_steps=7,
            retrieval_k=25,
            rerank_k=8,
            enable_web_search=False,  # Disable web search for medical accuracy
            enable_escalation=True,
            cost_limit=0.15
        )

2. Medical Knowledge Base Data

# data/medical_knowledge_base.json
{
  "documents": [
    {
      "id": "sepsis_pediatric_protocol_001",
      "title": "Pediatric Sepsis Treatment Protocol",
      "content": "Comprehensive protocol for treating sepsis in pediatric patients...",
      "metadata": {
        "patient_demographics": {
          "age_group": "pediatric",
          "gender": "all",
          "weight_category": "all"
        },
        "medical_conditions": {
          "primary_diagnosis": "sepsis",
          "comorbidities": ["diabetes", "immunocompromised"],
          "severity": "severe"
        },
        "treatment_protocols": {
          "protocol_type": "clinical_guidelines",
          "evidence_level": "high",
          "specialty": "pediatrics"
        },
        "medications": {
          "drug_class": ["antibiotics", "vasopressors"],
          "interactions": ["insulin", "glucose_management"],
          "contraindications": ["allergy_history"]
        },
        "last_updated": "2024-01-15",
        "source": "American Academy of Pediatrics",
        "evidence_level": "A"
      }
    }
  ]
}

3. Medical Query Expansion

# data/medical_terminology.json
{
  "sepsis": {
    "synonyms": ["septic shock", "SIRS", "severe sepsis", "bacteremia", "septicemia"],
    "medical_abbreviations": ["SIRS", "SS", "SIRS"],
    "related_conditions": ["shock", "organ_dysfunction", "infection"],
    "confidence": 0.95
  },
  "pediatric": {
    "synonyms": ["children", "infants", "adolescents", "pediatric patients"],
    "age_ranges": ["0-18 years", "neonatal", "infant", "toddler", "adolescent"],
    "confidence": 0.9
  },
  "diabetes": {
    "synonyms": ["diabetes mellitus", "DM", "hyperglycemia", "diabetic"],
    "types": ["type_1", "type_2", "gestational"],
    "complications": ["DKA", "diabetic_nephropathy", "diabetic_foot"],
    "confidence": 0.95
  }
}

4. Medical Agent Implementation

# agents/medical_agent.py
import asyncio
from typing import Dict, Any, List
from packages.agents import RAGAgentGraph
from packages.rag.faceted_search import FacetedSearchEngine
from packages.observability import MetricsCollector, StructuredLogger

class MedicalAgent:
    def __init__(self, config: MedicalConfig):
        self.config = config
        self.agent_graph = RAGAgentGraph(
            config=config.agent_config,
            tool_registry=config.tool_registry
        )
        self.metrics = config.metrics_collector
        self.logger = config.logger
    
    async def handle_medical_query(self, query: str, patient_context: Dict[str, Any]) -> Dict[str, Any]:
        """Handle medical query with full pipeline."""
        start_time = time.time()
        
        try:
            # Step 1: Medical query expansion
            expanded_query = await self._expand_medical_query(query, patient_context)
            
            # Step 2: Faceted search
            faceted_results = await self._faceted_search(expanded_query, patient_context)
            
            # Step 3: Hybrid retrieval
            retrieval_results = await self._retrieve_medical_documents(expanded_query, patient_context)
            
            # Step 4: Medical reranking
            reranked_results = await self._rerank_medical_documents(expanded_query, retrieval_results, patient_context)
            
            # Step 5: Medical response generation
            response = await self.agent_graph.ainvoke({
                "query": expanded_query,
                "retrieved_docs": retrieval_results,
                "reranked_docs": reranked_results,
                "patient_context": patient_context,
                "safety_checks": True
            })
            
            # Step 6: Medical safety validation
            validated_response = await self._validate_medical_response(response, patient_context)
            
            # Step 7: Logging and metrics
            await self._log_medical_interaction(query, response, patient_context)
            
            return validated_response
            
        except Exception as e:
            self.logger.error(f"Medical query failed: {e}")
            return await self._handle_medical_error(query, e, patient_context)
    
    async def _expand_medical_query(self, query: str, patient_context: Dict[str, Any]) -> str:
        """Expand query with medical terminology."""
        expanded = self.config.medical_expander.expand(
            query=query,
            domain="medical",
            context=patient_context
        )
        
        self.logger.info(
            "Medical query expanded",
            original_query=query,
            expanded_query=expanded,
            patient_age=patient_context.get("age_group"),
            specialty=patient_context.get("specialty")
        )
        
        return expanded
    
    async def _faceted_search(self, query: str, patient_context: Dict[str, Any]) -> List[Dict[str, Any]]:
        """Perform faceted search for medical information."""
        filters = {
            "patient_demographics": {
                "age_group": patient_context.get("age_group", "adult"),
                "gender": patient_context.get("gender", "all")
            },
            "medical_conditions": {
                "primary_diagnosis": patient_context.get("primary_diagnosis"),
                "comorbidities": patient_context.get("comorbidities", [])
            },
            "treatment_protocols": {
                "evidence_level": "high",
                "specialty": patient_context.get("specialty", "general")
            }
        }
        
        results = await self.config.faceted_search.search(
            query=query,
            filters=filters,
            facets=["patient_demographics", "medical_conditions", "treatment_protocols"]
        )
        
        return results
    
    async def _validate_medical_response(self, response: Dict[str, Any], patient_context: Dict[str, Any]) -> Dict[str, Any]:
        """Validate medical response for safety and accuracy."""
        # Check for contraindications
        contraindications = await self._check_contraindications(response, patient_context)
        
        # Check for drug interactions
        drug_interactions = await self._check_drug_interactions(response, patient_context)
        
        # Add safety warnings if needed
        if contraindications or drug_interactions:
            response["safety_warnings"] = {
                "contraindications": contraindications,
                "drug_interactions": drug_interactions
            }
        
        return response

5. Medical Safety and Compliance

# safety/medical_safety.py
from packages.agents.policies import SafetyPolicy
from packages.observability import StructuredLogger

class MedicalSafetyPolicy(SafetyPolicy):
    """Safety policy for medical domain."""
    
    def __init__(self):
        self.logger = StructuredLogger()
        self.medical_abbreviations = self._load_medical_abbreviations()
        self.drug_interactions = self._load_drug_interactions()
    
    def evaluate(self, input_data: Dict[str, Any]) -> PolicyResult:
        """Evaluate medical safety requirements."""
        query = input_data.get("text", "")
        patient_context = input_data.get("patient_context", {})
        
        # Check for PII in medical context
        pii_violations = self._check_medical_pii(query, patient_context)
        if pii_violations:
            return PolicyResult(
                action=PolicyAction.BLOCK,
                reason=f"PII detected in medical context: {pii_violations}",
                confidence=1.0
            )
        
        # Check for inappropriate medical advice
        inappropriate_advice = self._check_inappropriate_advice(query)
        if inappropriate_advice:
            return PolicyResult(
                action=PolicyAction.BLOCK,
                reason=f"Inappropriate medical advice detected: {inappropriate_advice}",
                confidence=0.9
            )
        
        # Check for drug interaction warnings
        drug_warnings = self._check_drug_warnings(query, patient_context)
        if drug_warnings:
            return PolicyResult(
                action=PolicyAction.WARN,
                reason=f"Drug interaction warnings: {drug_warnings}",
                confidence=0.8
            )
        
        return PolicyResult(
            action=PolicyAction.ALLOW,
            reason="Medical query passed safety checks",
            confidence=1.0
        )

Features Demonstrated

1. Faceted Search

• Multi-dimensional filtering by patient demographics
• Medical condition and treatment protocol filtering
• Evidence level and specialty-based filtering

2. Medical Query Expansion

• Domain-specific medical terminology expansion
• Medical abbreviation and acronym handling
• Specialty-specific terminology recognition

3. Medical Safety & Compliance

• PII detection and protection
• Drug interaction checking
• Contraindication validation
• Medical advice appropriateness checking

4. Business Intelligence

• Medical query pattern analysis
• Knowledge gap identification in medical protocols
• Quality trend monitoring for medical responses

5. Response Consistency

• Uniform medical protocol interpretation
• Consistent evidence-based recommendations
• Standardized medical terminology usage

6. Analytics & Monitoring

• Medical query performance tracking
• Safety incident monitoring
• Compliance reporting

Next Steps

1. Deploy the medical system with proper compliance controls
2. Ingest medical knowledge base with proper metadata
3. Configure medical safety policies and validation
4. Train medical staff on the new system
5. Monitor medical accuracy and safety metrics

Related Stories

• IT Support Agent - Similar structure with IT domain
• Financial Compliance Assistant - Compliance-focused implementation
• Research Lab Knowledge Management - Research-focused medical implementation

Ready to implement? Start with the medical knowledge base setup and work through each component step by step! 🏥