Agent Configuration

┌─────────────────────────────────────────┐
│            Incoming Call                │
└──────────────┬──────────────────────────┘
               │
      ┌────────┴────────┐
      │                 │
┌─────▼────┐      ┌────▼─────┐
│  Audio   │      │   NLP    │
│  Track   │      │  Track   │
└─────┬────┘      └────┬─────┘
      │                │
      │  ┌─────────┐   │
      ├─►│ STT     │◄──┤
      │  └─────────┘   │
      │                │
      │  ┌─────────┐   │
      ├─►│ Distress│   │
      │  └─────────┘   │
      │                │
      │  ┌─────────┐   │
      └─►│ Emotion │◄──┤
         └────┬────┘   │
              │        │
         ┌────▼────────▼────┐
         │   Service        │
         │   Classify       │
         └────┬─────────────┘
              │
         ┌────▼────┐
         │ Summary │
         └────┬────┘
              │
         ┌────▼────┐
         │ Ranking │
         └─────────┘

class STTClient:
    def __init__(self, on_partial=None, on_final=None):
        self.on_partial = on_partial  # Live transcript updates
        self.on_final = on_final      # Final sentence completions
        self.call_id = None
    
    async def start(self, sample_rate: int = 8000):
        # Connect to Deepgram WebSocket
        await self.deepgram_ws.connect(
            model="nova-2",
            language="en-US",
            sample_rate=sample_rate,
            encoding="linear16"
        )
    
    async def feed(self, pcm_data: bytes):
        # Stream audio chunks
        await self.deepgram_ws.send(pcm_data)
    
    async def finalize(self) -> dict:
        # Close connection, return final transcript
        await self.deepgram_ws.close()
        return {"transcript": self.final_text}

def _on_partial(text: str, call_id: str):
    LIVE_SIGNALS[call_id]["transcript_live"] = text
    print(f"[live] {text}")

def _on_final(text: str, call_id: str):
    LIVE_SIGNALS[call_id]["transcript"] = text

async def _analyze_emotion_heuristic(transcript: str, distress: float) -> dict:
    txt = transcript.lower()
    
    # Base label from acoustic distress
    if distress >= 0.7:
        label = "HIGHLY_DISTRESSED"
    elif distress >= 0.3:
        label = "DISTRESSED"
    elif distress >= 0.15:
        label = "TENSE"
    else:
        label = "CALM"
    
    # Override with life-threatening keywords
    life_threatening = [
        "shot", "stabbed", "can't breathe", "not breathing",
        "overdose", "heart attack", "unconscious"
    ]
    if any(k in txt for k in life_threatening):
        label = "HIGHLY_DISTRESSED"
        intensity = max(distress, 0.8)
    
    # Negative keywords increase intensity
    neg_keywords = ["angry", "scared", "emergency", "hurt", "panic"]
    if any(k in txt for k in neg_keywords):
        sentiment = "negative"
        intensity = max(intensity, 0.5)
    
    return {
        "label": label,
        "intensity": intensity,
        "sentiment": sentiment,
        "source": "heuristic"
    }

async def _analyze_emotion_deepgram(transcript: str, distress: float) -> dict:
    resp = await httpx.post(
        "https://api.deepgram.com/v1/read?sentiment=true&language=en",
        headers={"Authorization": f"Token {DEEPGRAM_API_KEY}"},
        json={"text": transcript}
    )
    
    sentiment = resp.json()["results"]["sentiments"]["average"]["sentiment"]
    confidence = resp.json()["results"]["sentiments"]["average"]["sentiment_score"]
    
    # Fusion logic: combine sentiment + distress + crisis keywords
    if sentiment == "negative" and has_crisis_keywords(transcript):
        label = "HIGHLY_DISTRESSED"
        intensity = max(distress, 0.7)
    elif sentiment == "positive":
        label = "RELIEVED" if distress < 0.4 else "CALM"
    else:
        label = label_from_distress(distress)
    
    return {"label": label, "sentiment": sentiment, "intensity": intensity}

async def _analyze_emotion_openai(transcript: str, distress: float) -> dict:
    resp = await openai_client.chat.completions.create(
        model="gpt-4o-mini",
        response_format={"type": "json_object"},
        messages=[
            {"role": "system", "content": "Classify emotion as JSON..."},
            {"role": "user", "content": f"Transcript: {transcript}\nDistress: {distress}"}
        ]
    )
    
    result = json.loads(resp.choices[0].message.content)
    return {
        "label": result["label"],
        "sentiment": result["sentiment"],
        "intensity": result["intensity"]
    }

{
  "label": "HIGHLY_DISTRESSED",
  "intensity": 0.87,
  "sentiment": "negative",
  "distress_input": 0.82,
  "has_transcript": true
}

ServiceCategory = Literal["EMS", "FIRE", "POLICE", "OTHER"]

def classify_service_and_tags(transcript: str, distress: float) -> dict:
    text = normalize_transcript(transcript)  # Fix ASR errors
    tags = set()
    cat_scores = {"EMS": 0.0, "FIRE": 0.0, "POLICE": 0.0, "OTHER": 0.0}
    
    # Gunshot detection
    if has_any(["shot", "shooting", "gunshot", "active shooter"]):
        if not has_any(["basketball", "hoops"]):  # Avoid false positives
            tags.add("ACTIVE_SHOOTER")
            tags.add("TRAUMA")
            cat_scores["EMS"] += 0.9
            cat_scores["POLICE"] += 0.9
    
    # Breathing emergency
    if has_any(["can't breathe", "not breathing", "turning blue"]):
        if not is_negated("breathe"):
            tags.add("NOT_BREATHING")
            cat_scores["EMS"] += 1.0
    
    # Fire detection
    if has_word("fire") and not has_any(["fireplace", "campfire"]):
        tags.add("FIRE")
        cat_scores["FIRE"] += 1.0
    
    # Select category with highest score
    category = max(cat_scores, key=cat_scores.get)
    confidence = calculate_confidence(cat_scores, tags, distress)
    
    return {"category": category, "confidence": confidence, "tags": sorted(tags)}

# Example: Add opioid-specific detection
opioid_keywords = [
    "fentanyl", "heroin", "oxycodone", "narcan"
]

if has_word_any(opioid_keywords):
    add_tag("OPIOID_OVERDOSE")
    add_tag("OVERDOSE")  # Parent tag
    bump("EMS", 0.95, urgent=True)

def is_negated(phrase: str) -> bool:
    # "not bleeding" → True
    # "bleeding" → False
    match = re.search(re.escape(phrase), text)
    if not match:
        return False
    
    context = text[max(0, match.start() - 20):match.start()]
    negations = [r"\bno\b", r"\bnot\b", r"\bwithout\b", r"\bcan't\b"]
    return any(re.search(neg, context) for neg in negations)

def _normalize_transcript(text: str) -> str:
    text = re.sub(r"\bam balance\b", "ambulance", text)
    text = re.sub(r"\bkeep myself\b", "kill myself", text)  # ASR error
    text = re.sub(r"\band my life\b", "end my life", text)
    return text

async def generate_summary(transcript: str, category: str, tags: list) -> str:
    prompt = (
        "You are an emergency dispatcher assistant. "
        "Summarize the caller's situation in 1–2 clear, factual sentences. "
        "Avoid speculation. Include critical details. "
        f"Category: {category}. Tags: {', '.join(tags)}.\n\n"
        f"Transcript:\n{transcript}"
    )
    
    resp = await openai_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=80,
        temperature=0
    )
    
    return resp.choices[0].message.content.strip()

def heuristic_summary(transcript: str) -> str:
    # Extract first sentence
    text = re.sub(r"\s+", " ", transcript.strip())
    sentences = re.split(r"[.?!]", text)
    first = sentences[0].strip()
    
    # Cap at 200 chars
    return first[:200] + "..." if len(first) > 200 else first

def compute_risk_level(distress_score: float, emotion: dict, tags: list) -> dict:
    # Base score from audio distress
    score = max(0.0, min(1.0, distress_score))
    
    # Amplify with emotion intensity
    emo_intensity = emotion.get("intensity", 0.0)
    score = max(score, emo_intensity)
    
    # Boost for distress-related emotions
    if emotion.get("label") in ("DISTRESSED", "TENSE"):
        score = max(score, 0.5)
    
    # CRITICAL OVERRIDE: Life-threatening tags
    critical_tags = {
        "ACTIVE_SHOOTER", "CARDIAC_ARREST", "NOT_BREATHING",
        "UNCONSCIOUS", "MAJOR_BLEEDING", "OVERDOSE"
    }
    if set(tags) & critical_tags:
        score = max(score, 0.9)  # Force critical
        level = "CRITICAL"
    
    # ELEVATED OVERRIDE: Serious emergencies
    elevated_tags = {
        "ASSAULT", "FIRE", "STROKE", "SEIZURE",
        "BREATHING_DIFFICULTY", "VEHICLE_ACCIDENT"
    }
    elif set(tags) & elevated_tags:
        score = max(score, 0.6)
        level = "ELEVATED" if score < 0.7 else "CRITICAL"
    
    # Standard bucketing
    else:
        if score >= 0.7:
            level = "CRITICAL"
        elif score >= 0.4:
            level = "ELEVATED"
        elif score >= 0.15:
            level = "NORMAL"
        else:
            level = "LOW"
    
    return {"level": level, "score": round(score, 3)}

# Make system more sensitive (lower thresholds)
if score >= 0.5:  # Was 0.7
    level = "CRITICAL"
elif score >= 0.3:  # Was 0.4
    level = "ELEVATED"

# Add custom overrides
if "DOMESTIC_VIOLENCE" in tags and distress_score >= 0.5:
    score = max(score, 0.75)  # Boost domestic violence priority

def build_ranking(inputs: RankingInputs) -> dict:
    # Weight by risk level
    weights = {
        "CRITICAL": 1000,
        "ELEVATED": 500,
        "NORMAL": 100,
        "LOW": 10
    }
    
    base_weight = weights[inputs.risk_level]
    
    # Amplify by risk score
    score_mult = inputs.risk_score  # 0-1
    final_weight = int(base_weight * (1 + score_mult))
    
    return {
        "priority": get_priority_number(inputs.risk_level),
        "weight": final_weight,
        "score": inputs.risk_score,
        "created_at": inputs.created_at or now_iso()
    }

def sort_key(item):
    return (
        -item["ranking"]["weight"],  # Highest weight first
        -item["risk"]["score"],      # Then highest score
        item["ranking"]["created_at"] # Then oldest call
    )

# tests/test_emotion.py
import pytest
from app.agents.emotion import analyze_emotion

@pytest.mark.asyncio
async def test_life_threatening_override():
    result = await analyze_emotion(
        transcript="Someone's been shot",
        distress=0.2  # Low acoustic distress
    )
    assert result["label"] == "HIGHLY_DISTRESSED"
    assert result["intensity"] >= 0.8

# tests/test_service_classify.py
from app.agents.service_classify import classify_service_and_tags

def test_active_shooter_detection():
    result = classify_service_and_tags(
        transcript="There's an active shooter at the mall",
        distress=0.9
    )
    assert result["category"] == "EMS"
    assert "ACTIVE_SHOOTER" in result["tags"]
    assert "POLICE" in result["tags"]

pytest tests/

# Replay a recorded call through the full pipeline
python scripts/replay_call.py data/calls/1234567890.wav

[live-partial] call=428c31b6... "Help there's"
[live-partial] call=428c31b6... "Help there's a fire"
[live-final] call=428c31b6... "Help there's a fire in my kitchen"
[emotion] label=DISTRESSED, sentiment=negative, intensity=0.72
[service] category=FIRE, tags=['FIRE'], confidence=0.88
[risk] level=CRITICAL, score=0.86

# Switch to Deepgram Base (faster but less accurate)
model="base"

# Run emotion + service classification concurrently
emotion_task = asyncio.create_task(analyze_emotion(transcript, distress))
service_task = asyncio.create_task(classify_service_and_tags(transcript, distress))

emotion, service = await asyncio.gather(emotion_task, service_task)