LangChain et AI transforment les lignes d'urgence vocales

L'importance cruciale des lignes d'assistance d'urgence

Dans le domaine des assistants vocaux, les applications les plus courantes incluent des tâches simples comme commander un repas ou écouter de la musique. Ces usages, bien que pratiques, ne présentent pas de conséquences graves en cas de défaillance. En revanche, les lignes d'assistance d'urgence représentent un défi bien plus complexe et critique. Dans ce contexte, la latence est un facteur essentiel, car le ton de l'assistant vocal peut influencer qui reçoit de l'aide en premier. De plus, il n'est pas possible d'utiliser une méthode alternative pour dépêcher un véhicule d'urgence comme une ambulance. Chaque décision de conception dans ce système peut avoir des répercussions réelles, ce qui en fait un cas d'utilisation précieux pour développer des compétences en conception de systèmes.

Le fonctionnement du pipeline d'urgence

Le système repose sur le modèle Sandwich d'architecture, qui se compose de trois composants indépendants fonctionnant simultanément. Chacun de ces composants démarre son traitement dès que le précédent a terminé sa phase, permettant ainsi une exécution fluide et rapide. Par exemple, la transcription commence dès que l'appelant parle, l'agent de raisonnement traite les réponses pendant que l'appelant continue de parler, et la synthèse vocale produit des réponses pendant que l'agent de raisonnement est encore actif. Si ce processus est bien implémenté, l'ensemble du traitement peut être complété en moins de dix secondes.

Mise en place de l'agent vocal

Pour démarrer avec cet agent vocal, il est nécessaire de disposer de clés API pour AssemblyAI pour la transcription en temps réel et OpenAI pour le raisonnement de l'agent et la synthèse vocale. Ces API peuvent être consolidées en un seul fournisseur pour simplifier le processus. Voici les commandes pour installer les bibliothèques requises :

!pip install langchain assemblyai websockets fastapi uvicorn openai

Instructions pour définir les variables d'environnement :

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

Vous devriez activer Langsmith pour vous assurer que chaque conversation entre votre agent et un client peut être considérée comme un audit et qu'elle peut être utilisée comme un ticket de support potentiel. L'audit permet d'assurer la conformité et le débogage en fournissant une documentation concernant ce que votre agent a dit à quel moment.

Étape 1 : Transcription vocale avec AssemblyAI

La première étape consiste à transcrire la voix de l'appelant en direct grâce à l'API WebSocket d'AssemblyAI. Ce processus suit un modèle producteur-consommateur où les segments audio sont transformés en transcriptions en temps réel.

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()

Deux types d'événements clés sont générés : STT Chunk et STT Output. Le premier permet de surveiller la conversation en temps réel, tandis que le second fournit une transcription finale ponctuée pour déclencher des actions. Lors de l'utilisation d'AssemblyAI pour une ligne d'assistance, il est impératif d'activer le drapeau de détection de sécurité du contenu. Cela fournit des avertissements précoces de signaux de détresse via les métadonnées de la transcription avant que l'agent ne traite le texte, donnant à l'agent plus de temps pour déterminer une réponse appropriée.

Étape 2 : L'agent de triage d'urgence

La deuxième étape implique un agent de triage d'urgence qui analyse la transcription pour évaluer la nécessité d'une assistance. Cet agent utilise quatre outils principaux : la recherche de localisation, le dispatch d'urgence, l'escalade à un opérateur humain, et la désescalade de situations non critiques.

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

# Registre des appels actifs
active_calls = {}

def get_caller_location(caller_id: str) -> str:
    """Rechercher l'adresse enregistrée de l'appelant ou la dernière localisation GPS connue."""
    return locations.get("Location not found. Ask caller to confirm address.")

def dispatch_emergency(service: str, location: str, severity: str) -> str:
    """Dépêcher la police, l'ambulance ou les services d'incendie à une localisation."""
    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:
    """Escalader l'appel à un opérateur humain lorsque la situation dépasse la capacité de l'IA."""
    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:
    """Retourner des instructions de respiration guidée ou de mise à la terre pour les appelants en détresse."""
    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(),
)

Le InMemorySaver joue un rôle crucial ici car il permet à ARIA de se souvenir de l'historique complet de l'appel, y compris ce qui a été dit par l'appelant lors des trois derniers appels, ce qui a déjà été envoyé à l'appelant, et si l'appelant a vérifié sa propre localisation. Sans mémoire, chaque réponse commencerait dans un état vierge, ce qui peut être très problématique dans une situation urgente.

Enfin, l'agent utilise un mode de streaming pour envoyer des tokens à la synthèse vocale dès qu'ils sont produits, permettant à ARIA de commencer à parler avant même que son raisonnement ne soit terminé. Le mode stream_mode="messages" envoie des tokens à TTS au fur et à mesure qu'ils sont produits, ce qui crée une réponse de 400 millisecondes contre une réponse de 2 secondes.

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)

Ce système avancé de gestion des appels d'urgence pourrait transformer la manière dont les urgences sont traitées, en réduisant considérablement les temps de réponse et en améliorant l'efficacité des interventions.