Integration eines benutzerdefinierten LLM

Überblick

Die native Twilio-Integration von ElevenAgents deckt den Fall ab, in dem ElevenLabs das LLM hostet. Nutzen Sie diesen Leitfaden, wenn Sie die volle Kontrolle über das LLM-Gehirn auf Ihrem eigenen Server benötigen — mit Ihrem eigenen Modell, einer RAG-Pipeline, dem Routing von Funktionsaufrufen oder anderer serverseitiger Logik — und der Agent weiterhin über eine Twilio-Telefonnummer erreichbar ist.

Der Teil mit dem benutzerdefinierten LLM wird über das Speech Engine SDK bereitgestellt. Es öffnet einen WebSocket zwischen ElevenLabs und Ihrem Server, sodass Ihr LLM Antworten während des Gesprächs streamen kann. Der Twilio-Teil nutzt Media Streams, um Gesprächsaudio an den Agenten weiterzuleiten.

Architektur

Das Speech Engine SDK stellt im Gesprächssystem des Agenten zwei WebSocket-Endpunkte bereit:

• Der Brain-WebSocket läuft auf Ihrem Server. ElevenLabs verbindet sich damit, um Transkripte zu übermitteln und vom LLM generierten Text zu empfangen.
• Der Gesprächs-WebSocket läuft bei ElevenLabs. Clients verbinden sich damit, um Audio zu senden und synthetisiertes Audio zurückzuerhalten. Die Twilio-Bridge verbindet sich über eine signierte URL und leitet μ-law-Audio in beide Richtungen weiter.

Da Twilio Media Streams und die Speech Engine beide ulaw_8000 verwenden, leitet die Bridge base64-kodiertes Audio ohne Transkodierung weiter.

Bridge und Brain-Server können im selben Prozess laufen, falls das praktischer ist — das folgende Beispiel kombiniert beide.

Wann Sie dieses Muster verwenden sollten

Sowohl dieser Leitfaden als auch die native Twilio-Integration stellen einen Agenten über eine Twilio-Telefonnummer bereit. Der Unterschied besteht darin, wer das LLM betreibt:

• Native Integration: ElevenLabs hostet das LLM, Sie konfigurieren es über den Agenten. Einfacher.
• Benutzerdefiniertes LLM über das Speech Engine SDK (dieser Leitfaden): Sie hosten das LLM auf Ihrem eigenen Server. Volle Kontrolle über Modell, RAG, Funktionsaufrufe und Geschäftslogik. Mehr Komponenten.

Wenn Ihre LLM-Logik in die Standard-Agentenkonfiguration passt, nutzen Sie die native Integration. Verwenden Sie diesen Leitfaden, wenn Ihr Gehirn Code auf Ihrer eigenen Infrastruktur ausführen muss.

Dieses Muster verwendet das Speech Engine SDK, das über eine WebSocket-Verbindung zwischen Ihrem Server und der ElevenLabs API kommuniziert. Sie können auch den Leitfaden Benutzerdefiniertes LLM verwenden, der statt des Speech Engine SDK einen OpenAI-kompatiblen HTTP-Endpunkt nutzt.

Der Hauptunterschied zwischen beiden sind WebSockets gegenüber HTTP-Anfragen. WebSockets halten eine einzelne Verbindung aufrecht, statt für jeden Gesprächszug eine neue HTTP-Verbindung aufzubauen. Das kann die Latenz verringern.

Voraussetzungen

• Ein Twilio-Konto und eine sprachfähige Telefonnummer.
• Eine Speech-Engine-Ressource. Folgen Sie dem Speech-Engine-Schnellstart, um eine zu erstellen und das Brain-Server-Muster kennenzulernen.
• Einen öffentlichen HTTPS-Tunnel, zum Beispiel ngrok. Twilio ruft Ihre Bridge über das öffentliche Internet auf.
• Python 3.9+ oder Node.js 18+.

Agenten für μ-law-Audio konfigurieren

Twilio Media Streams verwendet μ-law-Audio mit 8 kHz. Konfigurieren Sie die Speech Engine so, dass sie dasselbe Format akzeptiert und ausgibt, damit die Bridge nicht transkodieren muss.

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 hält die Text-to-Speech-Latenz niedrig, was bei einem Telefongespräch wichtig ist. Der Block request_headers weist ElevenLabs an, bei jeder Brain-WebSocket-Verbindung x-api-key: <shared-secret> einzuschließen — der Brain-Server prüft den Header, damit nur Ihre Speech Engine ihn erreichen kann.

Bridge-Server erstellen

Die Bridge stellt drei Routen bereit:

• POST /incoming-call — Twilio-Webhook. Gibt TwiML zurück, das Twilio anweist, einen Media Stream zu /media-stream zu öffnen.
• GET /media-stream — Twilio-Media-Streams-WebSocket. Leitet Audio zum und vom Speech-Engine-Gesprächs-WebSocket weiter.
• GET /ws — Brain-WebSocket. ElevenLabs verbindet sich hier, wenn ein Gespräch startet. Führt den Standardserver engine.serve() / engine.attach() aus.

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 auf die Bridge verweisen

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

Hinweise für die Produktion

• Webhook-Validierung: Validieren Sie stets die X-Twilio-Signature auf /incoming-call. Das obige Beispiel verwendet die Hilfsbibliothek von Twilio. Überspringen Sie diesen Schritt nicht.
• Gemeinsames Secret: Erzwingen Sie das gemeinsame Secret auf dem Brain-WebSocket. Andernfalls kann sich jeder, der Ihre ngrok-URL errät, verbinden und ElevenLabs imitieren.
• Stabiler Host: URLs der kostenlosen ngrok-Version ändern sich bei jedem Neustart. Verwenden Sie eine reservierte ngrok-Domain oder einen echten Hostnamen, damit Sie die ws_url der Speech Engine und den Twilio-Webhook nicht nach jedem Neustart aktualisieren müssen.
• Latenz: Jeder Aufruf fügt zusätzlich zur Zeit bis zum ersten Token des LLM zwei Netzwerk-Hops hinzu. Verwenden Sie ein Modell mit geringer Latenz und streamen Sie Antworten, um die wahrgenommene Latenz niedrig zu halten.
• Ein oder zwei Prozesse: Das Beispiel platziert Bridge und Brain auf demselben Port, sodass ein einzelner ngrok-Tunnel alles abdeckt. In der Produktion können Sie sie auf zwei Services aufteilen, sofern jeder eine öffentliche URL hat.
• Prompt-Injection: Gesprochene Eingaben aus einem Telefonat sind nicht vertrauenswürdige Nutzereingaben. Validieren Sie Transkripte, bevor sie Tool-Aufrufe oder Datenbankeinträge beeinflussen.

Nächste Schritte