LangChain and AI Transform Emergency Voice Lines

The Crucial Importance of Emergency Assistance Lines

In the realm of voice assistants, the most common applications involve simple tasks like ordering a meal or listening to music. While these uses are convenient, they do not carry severe consequences in the event of a failure. In contrast, emergency assistance lines present a far more complex and critical challenge. In this context, latency is a key factor, as the tone of the voice assistant can influence who receives help first. Moreover, there is no alternative method to dispatch an emergency vehicle like an ambulance. Every design decision in this system can have real-world repercussions, making it a valuable use case for developing system design skills.

How the Emergency Pipeline Works

The system relies on the Sandwich Architecture Model, which consists of three independent components operating simultaneously. Each of these components begins processing as soon as the previous one has completed its phase, allowing for smooth and rapid execution. For example, transcription starts as soon as the caller speaks, the reasoning agent processes responses while the caller continues to talk, and the voice synthesis produces responses while the reasoning agent is still active. If this process is well implemented, the entire processing can be completed in under ten seconds.

Setting Up the Voice Agent

To get started with this voice agent, you need API keys for AssemblyAI for real-time transcription and OpenAI for the agent's reasoning and voice synthesis. These APIs can be consolidated into a single provider to simplify the process. Here are the commands to install the required libraries:

!pip install langchain assemblyai websockets fastapi uvicorn openai

Instructions for setting environment variables:

export ASSEMBLYAI_API_KEY="your_key"
export OPENAI_API_KEY="your_key"
export LANGSMITH_TRACING="true"
export LANGSMITH_API_KEY="your_key"

You should enable Langsmith to ensure that every conversation between your agent and a client can be treated as an audit and can be used as a potential support ticket. The audit ensures compliance and debugging by providing documentation regarding what your agent said at what time.

Step 1: Voice Transcription with AssemblyAI

The first step is to transcribe the caller's voice live using AssemblyAI's WebSocket API. This process follows a producer-consumer model where audio segments are transformed into real-time transcriptions.

from typing import AsyncIterator
import contextlib

async def stt_stream(audio_stream: AsyncIterator[bytes]) -> AsyncIterator[VoiceAgentEvent]:
    stt = AssemblyAISTT(sample_rate=16000)

    async def send_audio():
        async for chunk in audio_stream:
            await stt.send_audio(chunk)
        await stt.close()

    send_task = asyncio.create_task(send_audio())
    async for event in stt.receive_events():
        send_task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await stt.close()

Two key types of events are generated: STT Chunk and STT Output. The former allows monitoring of the conversation in real time, while the latter provides a final punctuated transcription to trigger actions. When using AssemblyAI for an assistance line, it is imperative to enable the content safety detection flag. This provides early warnings of distress signals via the transcription metadata before the agent processes the text, giving the agent more time to determine an appropriate response.

Step 2: The Emergency Triage Agent

The second step involves an emergency triage agent that analyzes the transcription to assess the need for assistance. This agent uses four main tools: location lookup, emergency dispatch, escalation to a human operator, and de-escalation of non-critical situations.

from uuid import uuid4
from langchain.agents import create_agent
from langchain.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver

# Active call registry
active_calls = {}

def get_caller_location(caller_id: str) -> str:
    """Look up the caller's registered address or last known GPS location."""
    return locations.get("Location not found. Ask caller to confirm address.")

def dispatch_emergency(service: str, location: str, severity: str) -> str:
    """Dispatch police, ambulance, or fire services to a location."""
    valid_services = ["ambulance", "police", "fire"]
    if service.lower() not in valid_services:
        return f"Unknown service: {service}. Use ambulance, police, or fire."
    return f"{service.capitalize()} dispatched to {location}. Severity: {severity}. ETA: 8-12 minutes."

def escalate_to_human(caller_id: str, reason: str) -> str:
    """Escalate the call to a human operator when the situation exceeds AI capabilities."""
    active_calls[caller_id] = {
        "status": "escalated",
        "reason": reason,
    }
    return f"Escalating call {caller_id} to human operator. Reason: {reason}. Hold time: under 2 minutes."

def calming_protocol(situation: str) -> str:
    """Return guided breathing or grounding instructions for distressed callers."""
    return "I hear you. You are safe right now. Take a slow breath in for 4 counts, hold for 4, out for 4. I am here with you."

agent = create_agent(
    model="openai:[gpt](/glossaire/gpt)-4o-mini",
    get_caller_location,
    dispatch_emergency,
    escalate_to_human,
    calming_protocol,
    system_prompt="""You are ARIA, an AI emergency response assistant for a 24/7 helpline.
    Your job is to stay calm, assess the situation quickly, and take the right action.
    Rules you must always follow:
    - Always acknowledge the caller's distress before asking questions.
    - Ask only one question at a time. Never overwhelm a panicking caller.
    - If someone mentions chest pain, difficulty breathing, or unconsciousness — dispatch ambulance immediately.
    - If someone mentions violence, threats, or break-in — dispatch police immediately.
    - If the situation is unclear or emotional crisis — use calming protocol first.
    - Escalate to a human operator if the caller is unresponsive or the situation is ambiguous.
    - Keep every response under 3 sentences. Short and clear saves lives.
    - Do NOT use emojis, asterisks, bullet points, or markdown. You are speaking aloud.""",
    checkpointer=InMemorySaver(),
)

The InMemorySaver plays a crucial role here as it allows ARIA to remember the complete call history, including what has been said by the caller during the last three calls, what has already been sent to the caller, and whether the caller has confirmed their own location. Without memory, each response would start in a blank state, which can be very problematic in an urgent situation.

Finally, the agent uses a streaming mode to send tokens to the voice synthesis as soon as they are produced, allowing ARIA to start speaking even before its reasoning is complete. The stream_mode="messages" sends tokens to TTS as they are produced, resulting in a response time of 400 milliseconds compared to a 2-second response.

async def agent_stream(event_stream: AsyncIterator[VoiceAgentEvent]) -> AsyncIterator[VoiceAgentEvent]:
    thread_id = str(uuid4())  # Unique per call session
    async for event in event_stream:
        if event.type == "stt_output":
            stream = agent.astream(
                {"messages": [HumanMessage(content=event.transcript)]},
                {"configurable": {"thread_id": thread_id}},
            )
            async for message, _ in stream:
                if message.text:
                    yield AgentChunkEvent.create(message.text)

This advanced emergency call management system could transform the way emergencies are handled, significantly reducing response times and improving the efficiency of interventions.