Summary Generation Agent

async def generate_summary(
    transcript: str, 
    category: str, 
    tags: list[str]
) -> str:
    """
    Generate a concise dispatcher-friendly summary.
    Uses OpenAI API if available, falls back to heuristic summary otherwise.
    
    Args:
        transcript: Full text from speech-to-text
        category: Service category (EMS, FIRE, POLICE, OTHER)
        tags: Semantic tags like ["CARDIAC_EVENT", "TRAUMA"]
        
    Returns:
        A 1-2 sentence summary of the emergency situation
    """

from app.agents.summary import generate_summary

summary = await generate_summary(
    transcript="My dad is clutching his chest and saying he can't breathe. He's 65 and has a history of heart problems.",
    category="EMS",
    tags=["CARDIAC_EVENT", "BREATHING_DIFFICULTY"]
)

print(summary)
# "Caller reports 65-year-old father experiencing chest pain and difficulty breathing with cardiac history."

summary = await generate_summary(
    transcript="There's smoke coming from the apartment above mine. I can smell it and hear the alarm going off. I don't know if anyone's up there.",
    category="FIRE",
    tags=["SMOKE", "FIRE"]
)

print(summary)
# "Caller reports smoke and fire alarm from upstairs apartment with unknown occupancy."

summary = await generate_summary(
    transcript="Someone's breaking into my neighbor's house. I can see them through the window. They just broke the glass on the back door.",
    category="POLICE",
    tags=["BURGLARY"]
)

print(summary)
# "Caller reports burglary in progress at neighbor's residence with suspect visible breaking glass."

from app.agents.service_classify import classify_service_and_tags
from app.agents.summary import generate_summary

async def analyze_call(transcript: str, distress: float):
    # Get service classification first
    service = classify_service_and_tags(transcript, distress)
    
    # Generate summary using classification results
    summary = await generate_summary(
        transcript=transcript,
        category=service['category'],
        tags=service['tags']
    )
    
    return {
        "service": service['category'],
        "tags": service['tags'],
        "summary": summary
    }

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 _client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": prompt}],
    max_tokens=80,      # Keep summaries concise
    temperature=0,      # Deterministic output
)

content = resp.choices[0].message.content
return content.strip()

# Default model
model="gpt-4o-mini"  # Fast and cost-effective

# Alternative models
# model="gpt-4o"      # More accurate but slower/expensive
# model="gpt-5-nano"  # Fastest, lower quality

def heuristic_summary(transcript: str) -> str:
    """
    Safe fallback when no OpenAI API key is available.
    Extracts a short clean first-sentence summary.
    """
    if not transcript:
        return "No transcript available."
    
    text = transcript.strip()
    
    # normalize whitespace
    text = re.sub(r"\s+", " ", text)
    
    # find first sentence
    sentences = re.split(r"[.?!]", text)
    first_sentence = sentences[0].strip()
    
    if not first_sentence:
        return text[:120] + "..."
    
    # cap length
    if len(first_sentence) > 200:
        return first_sentence[:200] + "..."
    
    return first_sentence

# Input transcript
"My mom fell down the stairs and she's not responding. She hit her head really hard. There's some blood. I don't know what to do."

# Heuristic output (first sentence)
"My mom fell down the stairs and she's not responding"

# OpenAI API Key (required for AI summaries)
OPENAI_API_KEY=sk-...

from openai import AsyncOpenAI

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")

# Create client once if API key exists
_client: Optional[AsyncOpenAI] = None
if OPENAI_API_KEY:
    _client = AsyncOpenAI(api_key=OPENAI_API_KEY)

def refresh_client() -> bool:
    """
    Refresh the OpenAI client if API key is now available.
    
    Useful for runtime configuration changes.
    
    Returns:
        True if client was successfully created, False otherwise
    """
    global _client
    api_key = os.getenv("OPENAI_API_KEY")
    if api_key and not _client:
        _client = AsyncOpenAI(api_key=api_key)
        return True
    return False

try:
    resp = await _client.chat.completions.create(...)
    content = resp.choices[0].message.content
    
    # Handle unlikely edge case of None content
    if content is None:
        print("[summary] OpenAI returned None content, using heuristic")
        return heuristic_summary(transcript)
    
    return content.strip()

except Exception as e:
    print(f"[summary] OpenAI failed: {e}")
    return heuristic_summary(transcript)

Transcript: "Um, so like, my neighbor, I think he's having some kind of emergency. He's like clutching his chest and he looks really pale and sweaty. He's probably in his 60s I'd say."

Summary: "Caller reports 60-year-old neighbor experiencing chest pain with pale appearance and sweating."

Transcript: "Um, so like, my neighbor, I think he's having some kind of emergency."

Summary: "Um, so like, my neighbor, I think he's having some kind of emergency"

max_tokens=80  # Typical summary is 30-60 tokens

import pytest
from app.agents.summary import generate_summary, heuristic_summary

@pytest.mark.asyncio
async def test_generate_summary_medical():
    summary = await generate_summary(
        transcript="My dad is having chest pain and can't breathe",
        category="EMS",
        tags=["CARDIAC_EVENT", "BREATHING_DIFFICULTY"]
    )
    assert len(summary) > 0
    assert len(summary) < 300  # Should be concise

def test_heuristic_summary():
    summary = heuristic_summary(
        "Someone's been shot. They're bleeding badly. We need help now!"
    )
    assert summary == "Someone's been shot"
    assert len(summary) <= 200

def test_heuristic_summary_long():
    long_text = "This is a very long sentence that goes on and on and on for a really long time with lots of unnecessary details about the situation and it just keeps going and going without stopping for quite a while"
    summary = heuristic_summary(long_text)
    assert len(summary) <= 203  # 200 + "..."
    assert summary.endswith("...")

def test_heuristic_summary_empty():
    summary = heuristic_summary("")
    assert summary == "No transcript available."

@pytest.mark.asyncio
async def test_summary_with_service_classify():
    """Test summary generation using real service classification"""
    from app.agents.service_classify import classify_service_and_tags
    
    transcript = "There's a fire in my kitchen and smoke everywhere"
    
    # Classify first
    service = classify_service_and_tags(transcript, distress=0.9)
    
    # Generate summary with classification context
    summary = await generate_summary(
        transcript=transcript,
        category=service['category'],
        tags=service['tags']
    )
    
    assert service['category'] == 'FIRE'
    assert 'fire' in summary.lower() or 'smoke' in summary.lower()
    assert len(summary) > 0

from unittest.mock import AsyncMock, patch

@pytest.mark.asyncio
async def test_summary_openai_failure():
    """Test fallback when OpenAI fails"""
    with patch('app.agents.summary._client') as mock_client:
        mock_client.chat.completions.create = AsyncMock(
            side_effect=Exception("API error")
        )
        
        summary = await generate_summary(
            transcript="Emergency situation here",
            category="EMS",
            tags=[]
        )
        
        # Should fall back to heuristic
        assert summary == "Emergency situation here"

# Good
summary = await generate_summary(
    transcript=text,
    category="EMS",
    tags=["CARDIAC_EVENT"]
)

# Bad (missing context)
summary = await generate_summary(
    transcript=text,
    category="OTHER",
    tags=[]
)

if not transcript or len(transcript) < 10:
    return "Transcript too short for summary"

from functools import lru_cache
import hashlib

@lru_cache(maxsize=1000)
async def cached_summary(transcript_hash: str, category: str, tags_str: str):
    return await generate_summary(transcript, category, tags)

# Usage
transcript_hash = hashlib.md5(transcript.encode()).hexdigest()
tags_str = ",".join(sorted(tags))
summary = await cached_summary(transcript_hash, category, tags_str)

import time

start = time.time()
summary = await generate_summary(...)
latency = time.time() - start

if latency > 3.0:
    print(f"[summary] Slow API response: {latency:.2f}s")