Integrazione LLM personalizzato

Panoramica

L’integrazione nativa con Twilio di ElevenAgents copre il caso in cui ElevenLabs fornisce il modello LLM. Usa questa guida quando hai bisogno del pieno controllo del motore LLM sul tuo server — il tuo modello, pipeline RAG, instradamento delle function call o altri ragionamenti lato server — e l’agente si trova comunque su un numero di telefono Twilio.

La parte relativa al modello LLM personalizzato è fornita da Speech Engine SDK, che apre un WebSocket tra ElevenLabs e il tuo server, in modo che il tuo LLM possa trasmettere le risposte man mano che la chiamata procede. La parte relativa a Twilio usa Media Streams per inoltrare l’audio della chiamata all’agente.

Architettura

Speech Engine SDK espone due endpoint WebSocket nel sistema di conversazione dell’agente:

• Il WebSocket del motore viene eseguito sul tuo server. ElevenLabs si connette per inviare le trascrizioni e ricevere il testo generato dall’LLM.
• Il WebSocket della conversazione viene eseguito su ElevenLabs. I client vi si connettono per inviare audio e ricevere in risposta audio sintetizzato. Il bridge Twilio si connette tramite un URL firmato e inoltra l’audio μ-law in entrambe le direzioni.

Poiché Twilio Media Streams e Speech Engine usano entrambi ulaw_8000, il bridge inoltra audio codificato in base64 senza transcodifica.

Il bridge e il server del motore possono essere eseguiti nello stesso processo, se è più comodo: l’esempio seguente li combina.

Quando usare questo schema

Sia questa guida sia l’integrazione nativa con Twilio collocano un agente su un numero di telefono Twilio. La differenza sta in chi gestisce il modello LLM:

• Integrazione nativa: ElevenLabs fornisce il modello LLM, che configuri tramite l’agente. Più semplice.
• LLM personalizzato tramite Speech Engine SDK (questa guida): fornisci il modello LLM sul tuo server. Controllo completo sul modello, RAG, function call e logica di business. Più componenti da gestire.

Se la logica del tuo LLM rientra nella configurazione standard dell’agente, preferisci l’integrazione nativa. Usa questa guida quando il tuo motore deve eseguire codice sulla tua infrastruttura.

Questo schema usa Speech Engine SDK, che utilizza una connessione WebSocket per comunicare tra il tuo server e l’API ElevenLabs. Puoi anche usare la guida LLM personalizzato, che utilizza un endpoint HTTP compatibile con OpenAI anziché Speech Engine SDK.

La principale differenza tra i due è WebSocket rispetto alle richieste HTTP. L’uso di WebSocket consente di mantenere un’unica connessione invece di stabilire una nuova connessione HTTP per ogni turno, con possibili miglioramenti della latenza.

Prerequisiti

• Un account Twilio e un numero di telefono abilitato alle chiamate vocali.
• Una risorsa Speech Engine. Segui la guida rapida di Speech Engine per crearne una e conoscere lo schema del server del motore.
• Un tunnel HTTPS pubblico (ad esempio, ngrok). Twilio chiama il tuo bridge tramite Internet pubblico.
• Python 3.9+ o Node.js 18+.

Configura l’agente per l’audio μ-law

Twilio Media Streams usa audio μ-law a 8 kHz. Configura Speech Engine affinché accetti ed emetta lo stesso formato, così il bridge non dovrà effettuare transcodifica.

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 mantiene bassa la latenza della sintesi vocale, un aspetto importante in una chiamata telefonica. Il blocco request_headers indica a ElevenLabs di includere x-api-key: <shared-secret> in ogni connessione WebSocket del motore: il server del motore verifica l’header per assicurarsi che possa raggiungerlo solo il tuo Speech Engine.

Crea il server bridge

Il bridge espone tre route:

• POST /incoming-call — webhook Twilio. Restituisce TwiML che indica a Twilio di aprire un Media Stream verso /media-stream.
• GET /media-stream — WebSocket di Twilio Media Streams. Inoltra l’audio da e verso il WebSocket della conversazione Speech Engine.
• GET /ws — WebSocket del motore. ElevenLabs si connette qui quando inizia una conversazione. Esegue il server 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)

Indica a Twilio il bridge

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

Considerazioni per la produzione

• Convalida del webhook: convalida sempre X-Twilio-Signature su /incoming-call. L’esempio precedente usa la libreria di supporto di Twilio; non saltare questo passaggio.
• Segreto condiviso: applica il segreto condiviso sul WebSocket del motore. Senza di esso, chiunque indovini il tuo URL ngrok può connettersi e impersonare ElevenLabs.
• Host stabile: gli URL del piano gratuito di ngrok cambiano a ogni riavvio. Usa un dominio ngrok riservato o un hostname reale per non dover aggiornare ws_url di Speech Engine e il webhook Twilio dopo ogni riavvio.
• Latenza: ogni chiamata aggiunge due passaggi di rete al tempo al primo token dell’LLM. Usa un modello a bassa latenza e trasmetti le risposte in streaming per ridurre la latenza percepita.
• Un processo o due: l’esempio colloca il bridge e il motore sulla stessa porta, così un unico tunnel ngrok copre tutto. In produzione, puoi suddividerli in due servizi, purché ciascuno disponga di un URL pubblico.
• Prompt injection: l’input vocale di una chiamata telefonica è input utente non attendibile. Convalida le trascrizioni prima che influenzino chiamate agli strumenti o scritture nel database.

Passaggi successivi