File size: 4,553 Bytes
ee1b868 a33aad5 ee1b868 a33aad5 ee1b868 | 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 | from __future__ import annotations
from typing import Optional
from .models import CountyRisk, EpidemiologicalContext, PatientFacts
class EpidemiologicalContextEngine:
"""Computes context-sensitive risk modifiers for adaptive questioning."""
def __init__(self, context: EpidemiologicalContext) -> None:
self.context = context
def proximity_risk_score(self, facts: PatientFacts) -> float:
score = 0.0
district = (facts.location or "").strip().lower()
active = {item.lower() for item in self.context.active_outbreak_districts}
neighboring = {item.lower() for item in self.context.neighboring_outbreak_districts}
if district and district in active:
score += 3.0
elif district and district in neighboring:
score += 1.5
if self.context.community_transmission:
score += 1.0
if self.context.recent_confirmed_cases > 0:
score += min(2.0, self.context.recent_confirmed_cases * 0.25)
if self.context.cross_border_alerts:
score += 0.5
if self.context.health_facility_alerts:
score += 0.5
return score
def get_kenya_county_risk(self, county_name: Optional[str]) -> Optional[CountyRisk]:
"""Get risk intelligence for a Kenya county using case-insensitive matching."""
if not county_name or not self.context.county_risks:
return None
lookup = county_name.strip().lower()
for name, risk in self.context.county_risks.items():
if name.lower() == lookup:
return risk
return None
def get_question_priority_hints(self, facts: PatientFacts) -> dict:
"""Generate county-informed hints that guide question priority for the LLM."""
hints = {
"county_name": facts.location,
"county_risk": None,
"is_border_county": False,
"corridor_flags": [],
"relevant_poes": [],
"high_risk_profiles": [],
"key_risk_factors": [],
"question_priority_signal": "standard",
"rationale": "",
}
county_risk = self.get_kenya_county_risk(facts.location)
if not county_risk:
return hints
hints["county_risk"] = county_risk.risk_tier
hints["is_border_county"] = county_risk.is_border_county
hints["corridor_flags"] = county_risk.corridor_flags
hints["relevant_poes"] = county_risk.relevant_poes
hints["high_risk_profiles"] = county_risk.high_risk_profiles
hints["key_risk_factors"] = county_risk.key_risk_factors
if county_risk.risk_tier in ("very_high", "high"):
if county_risk.is_border_county:
hints["question_priority_signal"] = "high_exposure_priority"
hints["rationale"] = (
f"Border county {county_risk.county_name} with {county_risk.risk_tier} risk tier; "
"prioritize cross-border exposure and travel history questions."
)
elif "healthcare_workers" in county_risk.high_risk_profiles:
hints["question_priority_signal"] = "healthcare_amplification_priority"
hints["rationale"] = (
f"County {county_risk.county_name} hosts major health facilities with potential "
"amplification risk; prioritize healthcare exposure and IPC-related questions."
)
else:
hints["question_priority_signal"] = "high_exposure_priority"
hints["rationale"] = (
f"County {county_risk.county_name} is {county_risk.risk_tier} risk; prioritize travel, "
"contact, and route-based exposure questions."
)
return hints
def context_summary(self) -> str:
active = ", ".join(self.context.active_outbreak_districts) or "None"
neighboring = ", ".join(self.context.neighboring_outbreak_districts) or "None"
alerts = ", ".join(self.context.cross_border_alerts) or "None"
return (
f"District in focus: {self.context.district}\n"
f"Active outbreak districts: {active}\n"
f"Neighboring outbreak districts: {neighboring}\n"
f"Cross-border alerts: {alerts}\n"
f"Recent confirmed cases: {self.context.recent_confirmed_cases}\n"
f"Community transmission: {'Yes' if self.context.community_transmission else 'No'}"
)
|