Aangepaste LLM-integratie

Overzicht

De native Twilio-integratie van ElevenAgents is bedoeld voor situaties waarin ElevenLabs de LLM host. Gebruik deze handleiding wanneer je volledige controle nodig hebt over het LLM-brein op je eigen server — je eigen model, RAG-pipeline, routing van functieaanroepen of andere server-side redenering — en de agent nog steeds aan een Twilio-telefoonnummer is gekoppeld.

Het aangepaste LLM-gedeelte wordt geleverd door de Speech Engine SDK, die een WebSocket opent tussen ElevenLabs en je server zodat je LLM reacties kan streamen terwijl het gesprek verloopt. Het Twilio-gedeelte gebruikt Media Streams om gespreksaudio naar de agent door te sturen.

Architectuur

De Speech Engine SDK stelt twee WebSocket-endpoints beschikbaar in het gesprekssysteem van de agent:

• De brain WebSocket draait op je server. ElevenLabs maakt er verbinding mee om transcripten af te leveren en door de LLM gegenereerde tekst te ontvangen.
• De conversation WebSocket draait op ElevenLabs. Clients maken er verbinding mee om audio te versturen en gesynthetiseerde audio terug te ontvangen. De Twilio-bridge maakt verbinding via een ondertekende URL en stuurt μ-law-audio in beide richtingen door.

Omdat Twilio Media Streams en Speech Engine beide ulaw_8000 gebruiken, stuurt de bridge base64-gecodeerde audio door zonder transcoding.

De bridge en de brain-server kunnen in hetzelfde proces draaien als dat handig is — in het onderstaande voorbeeld worden ze gecombineerd.

Wanneer je dit patroon gebruikt

Zowel deze handleiding als de native Twilio-integratie koppelen een agent aan een Twilio-telefoonnummer. Het verschil is wie eigenaar is van de LLM:

• Native integratie: ElevenLabs host de LLM en je configureert deze via de agent. Eenvoudiger.
• Aangepaste LLM via Speech Engine SDK (deze handleiding): je host de LLM op je eigen server. Volledige controle over het model, RAG, functieaanroepen en bedrijfslogica. Meer onderdelen.

Als je LLM-logica binnen de standaardagentconfiguratie past, geef dan de voorkeur aan native integratie. Gebruik deze handleiding wanneer je brain code op je eigen infrastructuur moet uitvoeren.

Dit patroon gebruikt de Speech Engine SDK, die een WebSocket-verbinding gebruikt om tussen je server en de ElevenLabs API te communiceren. Je kunt ook de handleiding Custom LLM gebruiken, die een OpenAI-compatibel HTTP-endpoint gebruikt in plaats van de Speech Engine SDK.

Het belangrijkste verschil tussen de twee is WebSockets versus HTTP-requests. Met WebSockets onderhoud je één verbinding in plaats van voor elke beurt een nieuwe HTTP-verbinding op te zetten, wat de latentie mogelijk verbetert.

Vereisten

• Een Twilio-account en een telefoonnummer waarmee je kunt bellen.
• Een Speech Engine-resource. Volg de Speech Engine-snelstart om er een te maken en het brain-serverpatroon te leren kennen.
• Een openbare HTTPS-tunnel (bijv. ngrok). Twilio belt je bridge via het openbare internet.
• Python 3.9+ of Node.js 18+.

De agent configureren voor μ-law-audio

Twilio Media Streams gebruikt μ-law-audio van 8 kHz. Configureer de Speech Engine om dezelfde indeling te accepteren en uit te voeren, zodat de bridge niet hoeft te transcoderen.

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 houdt de tekst-naar-spraaklatentie laag, wat belangrijk is tijdens een telefoongesprek. Het blok request_headers vertelt ElevenLabs om bij elke brain WebSocket-verbinding x-api-key: <shared-secret> mee te sturen — de brain-server controleert de header om ervoor te zorgen dat alleen je Speech Engine deze kan bereiken.

De bridge-server bouwen

De bridge bedient drie routes:

• POST /incoming-call — Twilio-webhook. Geeft TwiML terug die Twilio vertelt een Media Stream naar /media-stream te openen.
• GET /media-stream — Twilio Media Streams-WebSocket. Stuurt audio door naar en van de Speech Engine conversation WebSocket.
• GET /ws — Brain WebSocket. ElevenLabs maakt hier verbinding wanneer een gesprek begint. Draait de standaardserver 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)

Twilio naar de bridge verwijzen

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"},
)

Overwegingen voor productie

• Webhookvalidatie: valideer altijd de X-Twilio-Signature op /incoming-call. Het bovenstaande voorbeeld gebruikt de hulplib van Twilio; sla deze stap niet over.
• Gedeeld geheim: dwing het gedeelde geheim af op de brain WebSocket. Zonder dit kan iedereen die je ngrok-URL raadt verbinding maken en zich voordoen als ElevenLabs.
• Stabiele host: URL’s van de gratis ngrok-laag veranderen bij elke herstart. Gebruik een gereserveerd ngrok-domein of een echte hostnaam, zodat je na elke herstart niet de ws_url van Speech Engine en de Twilio-webhook hoeft bij te werken.
• Latentie: elk gesprek voegt boven op de tijd tot het eerste token van de LLM twee netwerksprongen toe. Gebruik een model met lage latentie en stream reacties om de waargenomen latentie laag te houden.
• Eén of twee processen: in het voorbeeld bevinden de bridge en de brain zich op dezelfde poort, zodat één ngrok-tunnel alles dekt. In productie kun je ze over twee services verdelen, zolang elk een openbare URL heeft.
• Promptinjectie: gesproken invoer uit een telefoongesprek is onbetrouwbare gebruikersinvoer. Valideer transcripten voordat ze toolaanroepen of schrijfacties naar de database beïnvloeden.

Volgende stappen