File size: 5,626 Bytes
793d027 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 | """RAG tools for querying clinical guidelines via ChromaDB."""
from src.db.vector_store import search_guidelines, search_mic_reference
def search_clinical_guidelines(
query: str,
pathogen_filter: str = None,
n_results: int = 5
) -> list[dict]:
"""
Semantic search over IDSA clinical guidelines.
Args:
query: Natural language query about treatment
pathogen_filter: Optional pathogen type filter
Options: 'ESBL-E', 'CRE', 'CRAB', 'DTR-PA', 'S.maltophilia', 'AmpC-E', 'General'
n_results: Number of results to return
Returns:
List of relevant guideline excerpts with metadata
Used by: Agent 1 (Empirical), Agent 4 (Justification)
"""
results = search_guidelines(query, n_results, pathogen_filter)
# Format for agent consumption
formatted = []
for r in results:
formatted.append({
"content": r.get("content", ""),
"pathogen_type": r.get("metadata", {}).get("pathogen_type", "General"),
"source": r.get("metadata", {}).get("source", "IDSA Guidelines"),
"relevance_score": 1 - r.get("distance", 1) # Convert distance to similarity
})
return formatted
def search_mic_reference_docs(query: str, n_results: int = 3) -> list[dict]:
"""
Search MIC breakpoint reference documentation.
Args:
query: Query about MIC interpretation or breakpoints
n_results: Number of results to return
Returns:
List of relevant reference excerpts
"""
results = search_mic_reference(query, n_results)
formatted = []
for r in results:
formatted.append({
"content": r.get("content", ""),
"source": r.get("metadata", {}).get("source", "EUCAST Breakpoints"),
"relevance_score": 1 - r.get("distance", 1)
})
return formatted
def get_treatment_recommendation(
pathogen: str,
infection_site: str = None,
patient_factors: list[str] = None
) -> dict:
"""
Get treatment recommendation by searching guidelines.
Args:
pathogen: Identified or suspected pathogen
infection_site: Location of infection (e.g., "urinary", "respiratory")
patient_factors: List of patient factors (e.g., ["renal impairment", "pregnancy"])
Returns:
Treatment recommendation with guideline citations
"""
# Build comprehensive query
query_parts = [f"treatment for {pathogen} infection"]
if infection_site:
query_parts.append(f"in {infection_site}")
if patient_factors:
query_parts.append(f"considering {', '.join(patient_factors)}")
query = " ".join(query_parts)
# Search guidelines
results = search_clinical_guidelines(query, n_results=5)
# Try to determine pathogen category
pathogen_category = None
pathogen_lower = pathogen.lower()
pathogen_mapping = {
"ESBL-E": ["esbl", "extended-spectrum", "e. coli", "klebsiella"],
"CRE": ["carbapenem-resistant", "cre", "carbapenemase"],
"CRAB": ["acinetobacter", "crab"],
"DTR-PA": ["pseudomonas", "dtr"],
"S.maltophilia": ["stenotrophomonas", "maltophilia"],
}
for category, keywords in pathogen_mapping.items():
for keyword in keywords:
if keyword in pathogen_lower:
pathogen_category = category
break
# Search with pathogen filter if category identified
if pathogen_category:
filtered_results = search_clinical_guidelines(
query, pathogen_filter=pathogen_category, n_results=3
)
if filtered_results:
results = filtered_results + results[:2] # Combine results
return {
"query": query,
"pathogen_category": pathogen_category or "General",
"recommendations": results[:5],
"note": "These recommendations are from IDSA 2024 guidelines. Always verify with current institutional protocols."
}
def explain_mic_interpretation(
pathogen: str,
antibiotic: str,
mic_value: float
) -> dict:
"""
Get detailed explanation for MIC interpretation from reference docs.
Args:
pathogen: Pathogen name
antibiotic: Antibiotic name
mic_value: The MIC value to interpret
Returns:
Detailed explanation with reference citations
"""
query = f"MIC breakpoint interpretation for {antibiotic} against {pathogen}"
results = search_mic_reference_docs(query, n_results=3)
return {
"query": query,
"mic_value": mic_value,
"reference_excerpts": results,
"note": "Refer to current EUCAST v16.0 breakpoint tables for official interpretation."
}
def get_empirical_therapy_guidance(
infection_type: str,
risk_factors: list[str] = None
) -> dict:
"""
Get empirical therapy guidance for an infection type.
Args:
infection_type: Type of infection (e.g., "UTI", "pneumonia", "sepsis")
risk_factors: List of risk factors (e.g., ["prior MRSA", "recent antibiotics"])
Returns:
Empirical therapy recommendations
"""
query_parts = [f"empirical therapy for {infection_type}"]
if risk_factors:
query_parts.append(f"with risk factors: {', '.join(risk_factors)}")
query = " ".join(query_parts)
results = search_clinical_guidelines(query, n_results=5)
return {
"infection_type": infection_type,
"risk_factors": risk_factors or [],
"recommendations": results,
"note": "Empirical therapy should be de-escalated based on culture results."
}
|