The Service Classification Agent analyzes emergency call transcripts to determine which service(s) to dispatch. It uses sophisticated keyword detection with negation handling, ASR error correction, and multi-service detection.
result = classify_service_and_tags( transcript="There's a fire in the kitchen, smoke everywhere", distress=0.9)print(result)# {# "category": "FIRE",# "confidence": 1.0,# "tags": ["FIRE", "SMOKE"]# }
def _normalize_transcript(text: str) -> str: """ Normalize common ASR errors and variations in emergency transcripts. """ # Common phonetic substitutions text = re.sub(r"\bcant\b", "can't", text) text = re.sub(r"\bgonna\b", "going to", text) # Emergency-specific ASR errors text = re.sub(r"\bam balance\b", "ambulance", text) text = re.sub(r"\bambulence\b", "ambulance", text) text = re.sub(r"\bkeep myself\b", "kill myself", text) # Critical! return text
See line 1195-1235.Common ASR Errors:
“am balance” → “ambulance”
“keep myself” → “kill myself” (critical for suicide detection)
# "shot" can mean basketball or gunshotif has_word_any(gunshot_keywords) and not is_negated("shot"): # Context: if "shoot" appears with basketball/sports, reduce confidence if not has_any(["basketball", "hoops", "game", "sport", "court"]): add_tag("ACTIVE_SHOOTER") bump("EMS", 0.9, urgent=True) bump("POLICE", 0.9, urgent=True)
See line 130-139.More examples:
“cutting” + “kitchen” → Not a stabbing
“burning pain” without “fire” → Not a fire emergency
# Track if multiple services clearly neededservice_needed: Set[str] = set()def bump(cat: str, amount: float, urgent: bool = False): """Add score and track if service is clearly needed""" cat_scores[cat] += amount if urgent or amount >= 0.7: service_needed.add(cat)# Active shooter needs both POLICE and EMSif "ACTIVE_SHOOTER" in tags: service_needed.update(["POLICE", "EMS"])# Explosion needs FIRE, EMS, and POLICEif "EXPLOSION" in tags: service_needed.update(["FIRE", "EMS", "POLICE"])
suicide_phrases = [ "kill myself", "keep myself", # ASR error: commonly mishears "kill" as "keep" "want to die", "wanna die", "suicidal", "suicide", "end my life", "and my life", # ASR error: "end" → "and" "can't go on", "better off dead"]
See line 915-944.
The classifier includes ASR error variants like “keep myself” (mishears “kill myself”) - critical for suicide detection!
import pytestfrom app.agents.service_classify import classify_service_and_tagsdef test_classify_medical(): result = classify_service_and_tags( "Someone's having a heart attack", distress=0.8 ) assert result['category'] == 'EMS' assert 'CARDIAC_EVENT' in result['tags'] assert result['confidence'] > 0.7def test_classify_fire(): result = classify_service_and_tags( "House is on fire, smoke everywhere", distress=0.9 ) assert result['category'] == 'FIRE' assert 'FIRE' in result['tags'] assert 'SMOKE' in result['tags']def test_negation(): result = classify_service_and_tags( "No gun, just an argument", distress=0.3 ) # Should NOT detect weapon assert 'WEAPON_INVOLVED' not in result['tags']def test_multi_service(): result = classify_service_and_tags( "Active shooter, multiple people down not breathing", distress=1.0 ) # Should detect need for both POLICE and EMS assert result['category'] in ['EMS', 'POLICE'] assert 'ACTIVE_SHOOTER' in result['tags'] assert 'NOT_BREATHING' in result['tags']
def test_empty_transcript(): result = classify_service_and_tags("", distress=0.0) assert result['category'] == 'OTHER' assert result['confidence'] == 0.0 assert result['tags'] == []def test_context_false_positive(): # "shot" in basketball context result = classify_service_and_tags( "He shot the ball at the hoop and missed", distress=0.1 ) assert 'ACTIVE_SHOOTER' not in result['tags']