Integração com LLM personalizado

Visão geral

A integração nativa com a Twilio do ElevenAgents abrange o caso em que a ElevenLabs hospeda o LLM. Use este guia quando precisar de controle total sobre o cérebro do LLM no seu próprio servidor — seu próprio modelo, pipeline de RAG, roteamento de chamadas de função ou outro raciocínio no servidor — e o agente ainda estiver em um número de telefone da Twilio.

A parte do LLM personalizado é fornecida pelo SDK Speech Engine, que abre um WebSocket entre a ElevenLabs e seu servidor para que seu LLM possa transmitir respostas conforme a chamada acontece. A parte da Twilio usa o Media Streams para encaminhar o áudio da chamada para o agente.

Arquitetura

O SDK Speech Engine expõe dois endpoints WebSocket no sistema de conversação do agente:

• O WebSocket do cérebro é executado no seu servidor. A ElevenLabs se conecta a ele para enviar transcrições e receber texto gerado pelo LLM.
• O WebSocket de conversação é executado na ElevenLabs. Os clientes se conectam a ele para enviar áudio e receber áudio sintetizado de volta. A ponte da Twilio se conecta por uma URL assinada e encaminha áudio μ-law em ambas as direções.

Como o Twilio Media Streams e o Speech Engine usam ulaw_8000, a ponte encaminha áudio codificado em base64 sem transcodificação.

A ponte e o servidor do cérebro podem ser executados no mesmo processo, se for conveniente — o exemplo abaixo os combina.

Quando usar este padrão

Tanto este guia quanto a integração nativa com a Twilio colocam um agente em um número de telefone da Twilio. A diferença é quem hospeda o LLM:

• Integração nativa: a ElevenLabs hospeda o LLM, e você o configura pelo agente. Mais simples.
• LLM personalizado via SDK Speech Engine (este guia): você hospeda o LLM no seu próprio servidor. Controle total sobre o modelo, RAG, chamadas de função e lógica de negócios. Mais componentes envolvidos.

Se a lógica do seu LLM couber na configuração padrão do agente, prefira a integração nativa. Use este guia quando seu cérebro precisar executar código na sua própria infraestrutura.

Este padrão usa o SDK Speech Engine, que usa uma conexão WebSocket para se comunicar entre seu servidor e a API da ElevenLabs. Você também pode usar o guia de LLM personalizado, que usa um endpoint HTTP compatível com OpenAI em vez do SDK Speech Engine.

A principal diferença entre os dois são WebSockets em vez de solicitações HTTP. Usar WebSockets significa manter uma única conexão, em vez de estabelecer uma nova conexão HTTP a cada turno, o que pode melhorar a latência.

Pré-requisitos

• Uma conta Twilio e um número de telefone com suporte a voz.
• Um recurso do Speech Engine. Siga o início rápido do Speech Engine para criar um e conhecer o padrão de servidor do cérebro.
• Um túnel HTTPS público (por exemplo, ngrok). A Twilio disca para sua ponte pela internet pública.
• Python 3.9+ ou Node.js 18+.

Configure o agente para áudio μ-law

O Twilio Media Streams usa áudio μ-law de 8 kHz. Configure o Speech Engine para aceitar e emitir o mesmo formato, para que a ponte não precise 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 mantém baixa a latência de conversão de texto em voz, algo importante em uma chamada telefônica. O bloco request_headers instrui a ElevenLabs a incluir x-api-key: <shared-secret> em toda conexão WebSocket do cérebro — o servidor do cérebro verifica o cabeçalho para garantir que somente seu Speech Engine possa acessá-lo.

Crie o servidor de ponte

A ponte disponibiliza três rotas:

• POST /incoming-call — webhook da Twilio. Retorna TwiML instruindo a Twilio a abrir um Media Stream para /media-stream.
• GET /media-stream — WebSocket do Twilio Media Streams. Encaminha áudio de e para o WebSocket de conversação do Speech Engine.
• GET /ws — WebSocket do cérebro. A ElevenLabs se conecta aqui quando uma conversa começa. Executa o servidor padrão 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)

Direcione a Twilio para a ponte

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

Considerações para produção

• Validação de webhook: sempre valide o X-Twilio-Signature em /incoming-call. O exemplo acima usa a biblioteca auxiliar da Twilio; não pule esta etapa.
• Segredo compartilhado: exija o segredo compartilhado no WebSocket do cérebro. Sem ele, qualquer pessoa que adivinhar sua URL do ngrok poderá se conectar e se passar pela ElevenLabs.
• Host estável: as URLs do plano gratuito do ngrok mudam a cada reinicialização. Use um domínio ngrok reservado ou um hostname real para não precisar atualizar o ws_url do Speech Engine e o webhook da Twilio após cada reinicialização.
• Latência: cada chamada adiciona dois saltos de rede além do tempo até o primeiro token do LLM. Use um modelo de baixa latência e transmita as respostas por streaming para manter baixa a latência percebida.
• Um processo ou dois: o exemplo coloca a ponte e o cérebro na mesma porta, para que um único túnel do ngrok cubra tudo. Em produção, você pode separá-los em dois serviços, desde que cada um tenha uma URL pública.
• Injeção de prompt: a entrada falada de uma chamada telefônica é uma entrada de usuário não confiável. Valide as transcrições antes que influenciem chamadas de ferramentas ou gravações no banco de dados.

Próximas etapas