Integración de LLM personalizado

Resumen

La integración nativa con Twilio de ElevenAgents cubre el caso en el que ElevenLabs aloja el LLM. Usa esta guía cuando necesites tener control total sobre el cerebro del LLM en tu propio servidor —tu propio modelo, pipeline de RAG, enrutamiento de llamadas a funciones u otro razonamiento del lado del servidor— y el agente siga estando en un número de teléfono de Twilio.

La parte del LLM personalizado se ofrece mediante el SDK de Speech Engine, que abre un WebSocket entre ElevenLabs y tu servidor para que tu LLM pueda transmitir respuestas a medida que transcurre la llamada. La parte de Twilio utiliza Media Streams para transmitir el audio de la llamada al agente.

Arquitectura

El SDK de Speech Engine expone dos rutas de WebSocket en el sistema de conversación del agente:

• El WebSocket del cerebro se ejecuta en tu servidor. ElevenLabs se conecta a él para enviar transcripciones y recibir texto generado por el LLM.
• El WebSocket de conversación se ejecuta en ElevenLabs. Los clientes se conectan a él para enviar audio y recibir audio sintetizado. El puente de Twilio se conecta mediante una URL firmada y retransmite audio μ-law en ambas direcciones.

Como Twilio Media Streams y Speech Engine utilizan ulaw_8000, el puente retransmite audio codificado en base64 sin transcodificación.

El puente y el servidor del cerebro pueden ejecutarse en el mismo proceso si te resulta práctico; el ejemplo siguiente los combina.

Cuándo usar este patrón

Tanto esta guía como la integración nativa con Twilio colocan un agente en un número de teléfono de Twilio. La diferencia está en quién se encarga del LLM:

• Integración nativa: ElevenLabs aloja el LLM y tú lo configuras a través del agente. Más sencillo.
• LLM personalizado mediante el SDK de Speech Engine (esta guía): alojas el LLM en tu propio servidor. Control total sobre el modelo, RAG, llamadas a funciones y lógica de negocio. Más componentes.

Si la lógica de tu LLM se ajusta a la configuración estándar del agente, prioriza la integración nativa. Usa esta guía cuando tu cerebro necesite ejecutar código en tu propia infraestructura.

Este patrón utiliza el SDK de Speech Engine, que emplea una conexión WebSocket para comunicarse entre tu servidor y la API de ElevenLabs. También puedes usar la guía de LLM personalizado, que utiliza una ruta HTTP compatible con OpenAI en lugar del SDK de Speech Engine.

La principal diferencia entre ambos son los WebSockets frente a las solicitudes HTTP. Usar WebSockets implica mantener una única conexión en lugar de establecer una nueva conexión HTTP en cada turno, lo que puede reducir la latencia.

Requisitos previos

• Una cuenta de Twilio y un número de teléfono con capacidad de voz.
• Un recurso de Speech Engine. Sigue la guía de inicio rápido de Speech Engine para crear uno y conocer el patrón de servidor del cerebro.
• Un túnel HTTPS público (por ejemplo, ngrok). Twilio llama a tu puente a través de Internet público.
• Python 3.9+ o Node.js 18+.

Configura el agente para audio μ-law

Twilio Media Streams utiliza audio μ-law de 8 kHz. Configura Speech Engine para aceptar y emitir el mismo formato, de modo que el puente no tenga que transcodificar.

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 baja la latencia de texto a voz, algo importante en una llamada telefónica. El bloque request_headers indica a ElevenLabs que incluya x-api-key: <shared-secret> en cada conexión WebSocket del cerebro; el servidor del cerebro comprueba la cabecera para asegurar que solo tu Speech Engine pueda acceder a él.

Crea el servidor puente

El puente ofrece tres rutas:

• POST /incoming-call — Webhook de Twilio. Devuelve TwiML que indica a Twilio que abra un Media Stream hacia /media-stream.
• GET /media-stream — WebSocket de Twilio Media Streams. Retransmite audio desde y hacia el WebSocket de conversación de Speech Engine.
• GET /ws — WebSocket del cerebro. ElevenLabs se conecta aquí cuando empieza una conversación. Ejecuta el servidor estándar 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)

Dirige Twilio al puente

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

Consideraciones para producción

• Validación de webhooks: valida siempre X-Twilio-Signature en /incoming-call. El ejemplo anterior usa la biblioteca auxiliar de Twilio; no omitas este paso.
• Secreto compartido: aplica el secreto compartido en el WebSocket del cerebro. Sin él, cualquiera que adivine tu URL de ngrok puede conectarse y hacerse pasar por ElevenLabs.
• Host estable: las URL del nivel gratuito de ngrok cambian con cada reinicio. Usa un dominio de ngrok reservado o un nombre de host real para no tener que actualizar ws_url de Speech Engine y el webhook de Twilio después de cada reinicio.
• Latencia: cada llamada añade dos saltos de red además del tiempo hasta el primer token del LLM. Usa un modelo de baja latencia y transmite las respuestas en streaming para mantener baja la latencia percibida.
• Un proceso o dos: el ejemplo ubica el puente y el cerebro en el mismo puerto, de modo que un único túnel de ngrok cubre todo. En producción, puedes dividirlos en dos servicios siempre que cada uno tenga una URL pública.
• Inyección de prompts: la entrada de voz de una llamada telefónica es una entrada de usuario no fiable. Valida las transcripciones antes de que influyan en llamadas a herramientas o escrituras en la base de datos.

Próximos pasos