Spaces:
Sleeping
Sleeping
| from enum import Enum | |
| from typing import List | |
| import logging | |
| from app.schemas.text_triage_schema import ( | |
| TriageRequest, | |
| AIAnalysis, | |
| TriageResponse, | |
| UrgencyLevel, | |
| Species, | |
| GumColor, | |
| Mentation, | |
| Sex, | |
| ) | |
| from app.intelligence.triage.risk_context_builder_v2 import build_risk_context_v2 | |
| logger = logging.getLogger(__name__) | |
| class ClinicalPathway(Enum): | |
| LOCALIZED_SWELLING = "localized_swelling" | |
| GI_UPSET = "gi_upset" | |
| RESPIRATORY = "respiratory" | |
| NEUROLOGIC = "neurologic" | |
| SYSTEMIC_ILLNESS = "systemic_illness" | |
| UNKNOWN = "unknown" | |
| class TextRuleEngine: | |
| """ | |
| Deterministic, safety-first triage engine. | |
| LLMs may inform severity, but rules own urgency. | |
| """ | |
| COLOR_MAP = { | |
| UrgencyLevel.CRITICAL: "#FF0000", | |
| UrgencyLevel.URGENT: "#FFA500", | |
| UrgencyLevel.CONSULT: "#FFFF00", | |
| UrgencyLevel.MONITOR: "#00FF00", | |
| } | |
| # PATHWAY DETERMINATION | |
| def _determine_pathway(self, data: TriageRequest, ai: AIAnalysis) -> ClinicalPathway: | |
| if ai.is_open_mouth_breathing: | |
| return ClinicalPathway.RESPIRATORY | |
| if ai.mentation == Mentation.UNRESPONSIVE: | |
| return ClinicalPathway.NEUROLOGIC | |
| if ai.is_localized_swelling: | |
| return ClinicalPathway.LOCALIZED_SWELLING | |
| if ( | |
| (data.fluids and (data.fluids.is_vomiting or data.fluids.is_diarrhea)) | |
| or ai.vomit_frequency > 0 | |
| ): | |
| return ClinicalPathway.GI_UPSET | |
| if data.vitals and data.vitals.gum_color in { | |
| GumColor.PALE, GumColor.BLUE, GumColor.YELLOW | |
| }: | |
| return ClinicalPathway.SYSTEMIC_ILLNESS | |
| return ClinicalPathway.UNKNOWN | |
| # CORE EVALUATION (STEP 3) | |
| async def evaluate_risk(self, data: TriageRequest, ai: AIAnalysis) -> TriageResponse: | |
| decision_trace: List[str] = [] | |
| reasons: List[str] = [] | |
| # 🔥 STEP 3A — LLM SAFETY OVERRIDE (FIRST) | |
| if ai.gi_severity == "SEVERE" or ai.red_flags: | |
| decision_trace.append("OVERRIDE:LLM_SEVERE_RISK") | |
| logger.info("TRIAGE_TRACE | " + " | ".join(decision_trace)) | |
| return await self._response( | |
| urgency=UrgencyLevel.URGENT, | |
| reasons=[ | |
| "Concerning symptoms reported", | |
| *ai.red_flags | |
| ], | |
| species=data.species, | |
| pathway=ClinicalPathway.GI_UPSET, | |
| reasoning=None | |
| ) | |
| # Default baseline | |
| urgency = UrgencyLevel.MONITOR | |
| pathway = self._determine_pathway(data, ai) | |
| decision_trace.append(f"PATHWAY:{pathway.value}") | |
| # HARD EMERGENCY STOPS | |
| if ( | |
| data.species == Species.CAT | |
| and data.signalment.sex == Sex.MALE | |
| and ai.is_straining_to_urinate | |
| ): | |
| decision_trace.append("RULE:MALE_CAT_URINARY_BLOCK") | |
| logger.info("TRIAGE_TRACE | " + " | ".join(decision_trace)) | |
| return await self._critical( | |
| "🚨 Emergency: Possible Urinary Blockage", | |
| "A male cat straining to urinate may have a dangerous blockage", | |
| [ | |
| "Go to an emergency veterinary clinic immediately", | |
| "Do not wait for symptoms to improve" | |
| ] | |
| ) | |
| if data.species == Species.CAT and ai.is_open_mouth_breathing: | |
| decision_trace.append("RULE:CAT_RESP_DISTRESS") | |
| logger.info("TRIAGE_TRACE | " + " | ".join(decision_trace)) | |
| return await self._critical( | |
| "🚨 Emergency: Breathing Difficulty", | |
| "Open-mouth breathing in cats is always abnormal", | |
| [ | |
| "Seek emergency veterinary care immediately", | |
| "Keep your cat calm during transport" | |
| ] | |
| ) | |
| # PATHWAY RULES | |
| if pathway == ClinicalPathway.GI_UPSET: | |
| reasons.append("Digestive upset reported") | |
| if data.fluids and data.fluids.vomit_count >= 6: | |
| urgency = UrgencyLevel.CRITICAL | |
| elif data.fluids and data.fluids.is_vomiting and data.fluids.is_diarrhea: | |
| urgency = UrgencyLevel.URGENT | |
| else: | |
| urgency = UrgencyLevel.CONSULT | |
| elif pathway == ClinicalPathway.LOCALIZED_SWELLING: | |
| reasons.append("Localized swelling observed") | |
| urgency = ( | |
| UrgencyLevel.CONSULT | |
| if ai.mentation == Mentation.BRIGHT | |
| else UrgencyLevel.URGENT | |
| ) | |
| elif pathway == ClinicalPathway.SYSTEMIC_ILLNESS: | |
| reasons.append("Concerning systemic signs detected") | |
| urgency = UrgencyLevel.CRITICAL | |
| # MODIFIERS | |
| if ai.pain_signs_detected: | |
| reasons.append("Signs of discomfort noted") | |
| urgency = max(urgency, UrgencyLevel.CONSULT) | |
| if ai.mentation == Mentation.LETHARGIC: | |
| reasons.append("Reduced energy or responsiveness") | |
| urgency = max(urgency, UrgencyLevel.CONSULT) | |
| if not reasons: | |
| reasons.append("Assessment based on limited concerning signs") | |
| logger.info("TRIAGE_TRACE | " + " | ".join(decision_trace)) | |
| response = await self._response(urgency, reasons, data.species, pathway) | |
| # STEP 4 — CONFIDENCE SMOOTHING | |
| if ai.confidence < 0.4: | |
| response.action_steps.insert( | |
| 0, | |
| "More details could help assess this more accurately" | |
| ) | |
| response.action_steps.extend([ | |
| "Has this happened more than once today?", | |
| "Is there any blood, collapse, or severe pain?", | |
| "Has appetite or energy changed noticeably?" | |
| ]) | |
| return response | |
| async def evaluate_with_reasoning( | |
| self, | |
| data: TriageRequest, | |
| reasoning | |
| ) -> TriageResponse: | |
| """ | |
| LLM-informed, rule-owned triage. | |
| """ | |
| reasons: List[str] = [] | |
| pathway = ClinicalPathway.UNKNOWN | |
| urgency = UrgencyLevel.MONITOR | |
| # HARD CLINICAL OVERRIDES FIRST | |
| if ( | |
| data.species == Species.CAT | |
| and data.signalment.sex == Sex.MALE | |
| and any("litter" in rf.lower() or "urinate" in rf.lower() for rf in reasoning.key_findings) | |
| ): | |
| return await self._critical( | |
| "🚨 Emergency: Possible Urinary Blockage", | |
| "A male cat repeatedly straining to urinate may have a dangerous blockage", | |
| [ | |
| "Go to an emergency veterinary clinic immediately", | |
| "Do not wait for symptoms to improve" | |
| ] | |
| ) | |
| # ----------------------------- | |
| # INTERPRET LLM SIGNALS (NOT AUTHORITY) | |
| # ----------------------------- | |
| reasons.extend(reasoning.key_findings) | |
| if reasoning.red_flags: | |
| urgency = UrgencyLevel.URGENT | |
| reasons.extend(reasoning.red_flags) | |
| # Collapse, blood, severe weakness → never MONITOR | |
| if any( | |
| kw in " ".join(reasoning.red_flags).lower() | |
| for kw in ["collapse", "blood", "dark vomit", "seizure"] | |
| ): | |
| urgency = UrgencyLevel.URGENT | |
| # MODERATE ≠ consult by default | |
| if reasoning.risk_level == "MODERATE" and urgency == UrgencyLevel.MONITOR: | |
| urgency = UrgencyLevel.CONSULT | |
| # LOW confidence → do NOT escalate, ask questions | |
| followups = [] | |
| if reasoning.confidence < 0.5 and reasoning.missing_information: | |
| followups = reasoning.missing_information[:3] | |
| if not reasons: | |
| reasons.append("Assessment based on limited concerning signs") | |
| response = await self._response( | |
| urgency=urgency, | |
| reasons=reasons, | |
| species=data.species, | |
| pathway=pathway | |
| ) | |
| if followups: | |
| response.action_steps.insert( | |
| 0, | |
| "More details could help assess this more accurately" | |
| ) | |
| for q in followups: | |
| response.action_steps.insert(1, q) | |
| return response | |
| # RESPONSE BUILDERS | |
| async def _critical(self, headline: str, reason: str, steps: List[str], reasoning= None) -> TriageResponse: | |
| steps.append( | |
| "If your pet worsens during travel, go to the nearest emergency clinic" | |
| ) | |
| return TriageResponse( | |
| urgency=UrgencyLevel.CRITICAL, | |
| color_hex=self.COLOR_MAP[UrgencyLevel.CRITICAL], | |
| headline=headline, | |
| primary_reason=reason, | |
| action_steps=steps, | |
| risk_context=await build_risk_context_v2( | |
| reasoning=reasoning, | |
| urgency=UrgencyLevel.CRITICAL | |
| ), | |
| ) | |
| async def _response( | |
| self, | |
| urgency, | |
| reasons, | |
| species, | |
| pathway, | |
| reasoning=None | |
| ) -> TriageResponse: | |
| headlines = { | |
| UrgencyLevel.CRITICAL: "🚨 Emergency: Immediate Veterinary Care Needed", | |
| UrgencyLevel.URGENT: "⚠️ Urgent: Veterinary Attention Recommended Soon", | |
| UrgencyLevel.CONSULT: "🔶 Veterinary Visit Recommended", | |
| UrgencyLevel.MONITOR: "✅ Monitor and Support at Home", | |
| } | |
| steps: List[str] = [] | |
| if urgency == UrgencyLevel.CRITICAL: | |
| steps.append("Go to an emergency veterinary hospital right away") | |
| elif urgency == UrgencyLevel.URGENT: | |
| steps.append("Contact your veterinarian as soon as possible") | |
| elif urgency == UrgencyLevel.CONSULT: | |
| steps.append("Plan a veterinary visit within the next 24–48 hours") | |
| else: | |
| steps.append("Continue monitoring your pet at home") | |
| if pathway == ClinicalPathway.GI_UPSET: | |
| steps.append("Offer small amounts of water frequently") | |
| steps.append("Pause food for 6–12 hours unless advised otherwise") | |
| if urgency in {UrgencyLevel.MONITOR, UrgencyLevel.CONSULT}: | |
| steps.append( | |
| "If symptoms worsen or you feel unsure, seek veterinary advice" | |
| ) | |
| return TriageResponse( | |
| urgency=urgency, | |
| color_hex=self.COLOR_MAP[urgency], | |
| headline=headlines[urgency], | |
| primary_reason="; ".join(reasons), | |
| action_steps=steps, | |
| risk_context=await build_risk_context_v2( | |
| reasoning=reasoning, # safe fallback | |
| urgency=urgency | |
| ), | |
| ) | |
| _engine_instance = None | |
| def get_rule_engine() -> TextRuleEngine: | |
| global _engine_instance | |
| if _engine_instance is None: | |
| _engine_instance = TextRuleEngine() | |
| return _engine_instance | |