Intégration LLM personnalisée

Vue d’ensemble

L’intégration Twilio native d’ElevenAgents couvre le cas où ElevenLabs héberge le LLM. Consultez ce guide lorsque vous avez besoin de contrôler entièrement le moteur LLM sur votre propre serveur, avec votre modèle, pipeline RAG, routage des appels de fonction ou autre logique côté serveur, tout en conservant l’agent sur un numéro de téléphone Twilio.

La partie LLM personnalisé est assurée par le SDK Speech Engine, qui ouvre un WebSocket entre ElevenLabs et votre serveur afin que votre LLM puisse diffuser les réponses à mesure que l’appel se déroule. La partie Twilio utilise Media Streams pour relayer l’audio des appels vers l’agent.

Architecture

Le SDK Speech Engine expose deux points de terminaison WebSocket dans le système de conversation de l’agent :

• Le WebSocket du moteur s’exécute sur votre serveur. ElevenLabs s’y connecte pour transmettre les transcriptions et recevoir le texte généré par le LLM.
• Le WebSocket de conversation s’exécute chez ElevenLabs. Les clients s’y connectent pour envoyer de l’audio et recevoir l’audio synthétisé en retour. Le pont Twilio se connecte via une URL signée et relaie l’audio μ-law dans les deux sens.

Comme Twilio Media Streams et Speech Engine utilisent tous deux ulaw_8000, le pont relaie l’audio encodé en base64 sans transcodage.

Le pont et le serveur du moteur peuvent s’exécuter dans le même processus si cela vous convient. L’exemple ci-dessous les combine.

Quand utiliser ce modèle

Ce guide comme l’intégration Twilio native placent un agent sur un numéro de téléphone Twilio. La différence réside dans l’hébergement du LLM :

• Intégration native : ElevenLabs héberge le LLM, que vous configurez via l’agent. Plus simple.
• LLM personnalisé via le SDK Speech Engine (ce guide) : vous hébergez le LLM sur votre propre serveur. Contrôle total du modèle, du RAG, des appels de fonction et de la logique métier. Plus d’éléments à gérer.

Si votre logique LLM correspond à la configuration standard d’un agent, privilégiez l’intégration native. Consultez ce guide lorsque votre moteur doit exécuter du code sur votre propre infrastructure.

Ce modèle utilise le SDK Speech Engine, qui s’appuie sur une connexion WebSocket pour communiquer entre votre serveur et l’API ElevenLabs. Vous pouvez également utiliser le guide LLM personnalisé, qui emploie un point de terminaison HTTP compatible avec OpenAI à la place du SDK Speech Engine.

La principale différence entre les deux réside dans les WebSockets et les requêtes HTTP. Les WebSockets maintiennent une connexion unique plutôt que d’établir une nouvelle connexion HTTP à chaque tour, ce qui peut réduire la latence.

Prérequis

• Un compte Twilio et un numéro de téléphone compatible avec les appels vocaux.
• Une ressource Speech Engine. Suivez le guide de démarrage rapide Speech Engine pour en créer une et découvrir le modèle de serveur du moteur.
• Un tunnel HTTPS public (par exemple, ngrok). Twilio contacte votre pont via l’internet public.
• Python 3.9+ ou Node.js 18+.

Configurer l’agent pour l’audio μ-law

Twilio Media Streams utilise un audio μ-law à 8 kHz. Configurez Speech Engine pour accepter et émettre ce même format afin que le pont n’ait pas à effectuer de transcodage.

import asyncio
import os
from elevenlabs import AsyncElevenLabs
elevenlabs = AsyncElevenLabs(api_key=os.environ["ELEVENLABS_API_KEY"])
async def update_engine():
    await elevenlabs.speech_engine.update(
        speech_engine_id="seng_8k3m9xr4hjnfg983brhmhkd98n6",
        asr={"user_input_audio_format": "ulaw_8000"},
        tts={
            "model_id": "eleven_flash_v2",
            "agent_output_audio_format": "ulaw_8000",
        },
        speech_engine={
            "request_headers": {"x-api-key": os.environ["SHARED_SECRET"]},
        },
    )
asyncio.run(update_engine())

eleven_flash_v2 maintient une faible latence de synthèse vocale, un point essentiel lors d’un appel téléphonique. Le bloc request_headers indique à ElevenLabs d’inclure x-api-key: <shared-secret> sur chaque connexion WebSocket du moteur. Le serveur du moteur vérifie cet en-tête afin de s’assurer que seul votre Speech Engine peut s’y connecter.

Créer le serveur de pont

Le pont fournit trois routes :

• POST /incoming-call : webhook Twilio. Renvoie du TwiML demandant à Twilio d’ouvrir un Media Stream vers /media-stream.
• GET /media-stream : WebSocket Twilio Media Streams. Relaye l’audio vers et depuis le WebSocket de conversation Speech Engine.
• GET /ws : WebSocket du moteur. ElevenLabs s’y connecte lorsqu’une conversation démarre. Exécute le serveur standard engine.serve() / engine.attach().

pip install "elevenlabs" "aiohttp" "twilio" "python-dotenv"

from elevenlabs import AsyncElevenLabs
elevenlabs = AsyncElevenLabs(api_key=os.environ["ELEVENLABS_API_KEY"])
async def signed_url() -> str:
    response = await elevenlabs.conversational_ai.conversations.get_signed_url(
        agent_id=os.environ["SPEECH_ENGINE_ID"],
    )
    return response.signed_url

from aiohttp import web
from twilio.request_validator import RequestValidator
validator = RequestValidator(os.environ["TWILIO_AUTH_TOKEN"])
async def incoming_call(request: web.Request) -> web.Response:
    form = await request.post()
    signature = request.headers.get("X-Twilio-Signature", "")
    url = str(request.url)
    if not validator.validate(url, dict(form), signature):
        return web.Response(status=403, text="forbidden")
    host = request.headers.get("X-Forwarded-Host") or request.host
    twiml = (
        '<?xml version="1.0" encoding="UTF-8"?>'
        "<Response><Connect>"
        f'<Stream url="wss://{host}/media-stream"/>'
        "</Connect></Response>"
    )
    return web.Response(text=twiml, content_type="text/xml")

import asyncio
import json
import aiohttp
from aiohttp import web
async def media_stream(request: web.Request) -> web.WebSocketResponse:
    twilio_ws = web.WebSocketResponse()
    await twilio_ws.prepare(request)
    stream_sid: str | None = None
    el_session: aiohttp.ClientSession | None = None
    el_ws: aiohttp.ClientWebSocketResponse | None = None
    pump_task: asyncio.Task | None = None
    async def pump_el_to_twilio(el: aiohttp.ClientWebSocketResponse):
        async for msg in el:
            if msg.type != aiohttp.WSMsgType.TEXT:
                continue
            event = json.loads(msg.data)
            etype = event.get("type")
            if etype == "audio":
                await twilio_ws.send_str(json.dumps({
                    "event": "media",
                    "streamSid": stream_sid,
                    "media": {"payload": event["audio_event"]["audio_base_64"]},
                }))
            elif etype == "interruption":
                await twilio_ws.send_str(json.dumps({
                    "event": "clear",
                    "streamSid": stream_sid,
                }))
            elif etype == "ping":
                event_id = event.get("ping_event", {}).get("event_id")
                await el.send_str(json.dumps({
                    "type": "pong", "event_id": event_id,
                }))
    try:
        async for msg in twilio_ws:
            if msg.type != aiohttp.WSMsgType.TEXT:
                continue
            event = json.loads(msg.data)
            if event["event"] == "start":
                stream_sid = event["start"]["streamSid"]
                el_session = aiohttp.ClientSession()
                el_ws = await el_session.ws_connect(await signed_url())
                await el_ws.send_str(json.dumps({
                    "type": "conversation_initiation_client_data",
                }))
                pump_task = asyncio.create_task(pump_el_to_twilio(el_ws))
            elif event["event"] == "media" and el_ws is not None:
                await el_ws.send_str(json.dumps({
                    "user_audio_chunk": event["media"]["payload"],
                }))
            elif event["event"] == "stop":
                break
    finally:
        if pump_task:
            pump_task.cancel()
        if el_ws and not el_ws.closed:
            await el_ws.close()
        if el_session and not el_session.closed:
            await el_session.close()
    return twilio_ws

import os
from elevenlabs import AsyncElevenLabs
elevenlabs = AsyncElevenLabs(api_key=os.environ["ELEVENLABS_API_KEY"])
SHARED_SECRET = os.environ["SHARED_SECRET"]
async def brain_ws(request: web.Request) -> web.WebSocketResponse:
    if request.headers.get("x-api-key") != SHARED_SECRET:
        return web.Response(status=401, text="unauthorized")
    ws = web.WebSocketResponse()
    await ws.prepare(request)
    engine = await elevenlabs.speech_engine.get(os.environ["SPEECH_ENGINE_ID"])
    session = engine.create_session(ws)
    async def on_transcript(transcript):
        # Replace this with your own LLM call; see the quickstart.
        await session.send_response("Hello, you've reached the demo.")
    session.on("user_transcript", on_transcript)
    await session.run()
    return ws
def make_app() -> web.Application:
    app = web.Application()
    app.router.add_post("/incoming-call", incoming_call)
    app.router.add_get("/media-stream", media_stream)
    app.router.add_get("/ws", brain_ws)
    return app
if __name__ == "__main__":
    web.run_app(make_app(), port=3001)

Diriger Twilio vers le pont

ngrok http 3001
python bridge.py

await elevenlabs.speech_engine.update(
    speech_engine_id="seng_8k3m9xr4hjnfg983brhmhkd98n6",
    speech_engine={"ws_url": "wss://abc123.ngrok.io/ws"},
)

Considérations pour la production

• Validation des webhooks : validez systématiquement X-Twilio-Signature sur /incoming-call. L’exemple ci-dessus utilise la bibliothèque d’assistance Twilio, ne sautez pas cette étape.
• Secret partagé : appliquez le secret partagé sur le WebSocket du moteur. Sans cela, toute personne qui devine votre URL ngrok peut se connecter et se faire passer pour ElevenLabs.
• Hôte stable : les URL du forfait gratuit ngrok changent à chaque redémarrage. Utilisez un domaine ngrok réservé ou un véritable nom d’hôte afin de ne pas avoir à mettre à jour le ws_url de Speech Engine et le webhook Twilio après chaque redémarrage.
• Latence : chaque appel ajoute deux sauts réseau au délai avant le premier jeton du LLM. Utilisez un modèle à faible latence et diffusez les réponses pour maintenir une faible latence perçue.
• Un ou deux processus : l’exemple place le pont et le moteur sur le même port, de sorte qu’un seul tunnel ngrok couvre l’ensemble. En production, vous pouvez les répartir sur deux services, à condition que chacun dispose d’une URL publique.
• Injection de prompt : l’entrée vocale d’un appel téléphonique est une entrée utilisateur non fiable. Validez les transcriptions avant qu’elles n’influencent les appels d’outils ou les écritures en base de données.

Prochaines étapes