自定义 LLM 集成 | ElevenLabs Documentation

概览

ElevenAgents 的原生 Twilio 集成适用于由 ElevenLabs 托管 LLM 的场景。如果需要在自己的服务器上完全控制 LLM 大脑——使用自有模型、RAG 管道、函数调用路由或其他服务端推理——同时智能体仍使用 Twilio 电话号码，请参考本指南。

自定义 LLM 部分由 Speech Engine SDK提供。它会在 ElevenLabs 和服务器之间建立 WebSocket，让 LLM 能够随着通话进行流式返回响应。Twilio 部分使用 Media Streams，将通话音频转发给智能体。

架构

Speech Engine SDK 会在智能体的对话系统中公开两个 WebSocket 端点：

大脑 WebSocket 在服务器上运行。ElevenLabs 连接到它，以传递转写文本并接收 LLM 生成的文本。
对话 WebSocket 在 ElevenLabs 上运行。客户端连接到它以发送音频并接收合成音频。Twilio 桥接器通过签名 URL 连接，并双向转发 μ-law 音频。

由于 Twilio Media Streams 和 Speech Engine 都使用 ulaw_8000，桥接器可直接转发 base64 编码的音频，无需转码。

如果方便，桥接器和大脑服务器可以在同一进程中运行——以下示例将它们合并在一起。

何时使用此模式

本指南和原生 Twilio 集成都能将智能体部署到 Twilio 电话号码。区别在于谁负责 LLM：

原生集成：ElevenLabs 托管 LLM，你通过智能体进行配置。更简单。
通过 Speech Engine SDK 使用自定义 LLM（本指南）：你在自己的服务器上托管 LLM。可完全控制模型、RAG、函数调用和业务逻辑。涉及更多组件。

如果 LLM 逻辑可在标准智能体配置中实现，建议使用原生集成。如果大脑需要在自己的基础设施上运行代码，请使用本指南。

此模式使用 Speech Engine SDK，它通过 WebSocket 连接在服务器与 ElevenLabs API 之间通信。你还可以使用 Custom LLM 指南，该指南使用兼容 OpenAI 的 HTTP 端点，而不是 Speech Engine SDK。

两者的主要区别是 WebSocket 与 HTTP 请求。WebSocket 只需维持一个连接，而不必为每轮对话建立新的 HTTP 连接，因此可能降低延迟。

前提条件

一个 Twilio 账户和一个支持语音的电话号码。
一个 Speech Engine 资源。请按照 Speech Engine 快速入门创建资源，并了解大脑服务器模式。
一个公开 HTTPS 隧道（例如 ngrok）。Twilio 会通过公网连接到桥接器。
Python 3.9+ 或 Node.js 18+。

为 μ-law 音频配置智能体

Twilio Media Streams 使用 8 kHz μ-law 音频。将 Speech Engine 配置为接受和输出相同格式，以便桥接器无需转码。

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 可保持较低的文本转语音延迟，这对电话通话很重要。request_headers 块会让 ElevenLabs 在每个大脑 WebSocket 连接中包含 x-api-key: <shared-secret>——大脑服务器会检查此标头，确保只有 Speech Engine 能连接。

构建桥接服务器

桥接器提供 3 条路由：

POST /incoming-call — Twilio webhook。返回 TwiML，指示 Twilio 向 /media-stream 打开 Media Stream。
GET /media-stream — Twilio Media Streams WebSocket。在 Speech Engine 对话 WebSocket 与通话之间转发音频。
GET /ws — 大脑 WebSocket。对话开始时 ElevenLabs 会连接到此处。运行标准的 engine.serve() / engine.attach() 服务器。

安装依赖

pip install "elevenlabs" "aiohttp" "twilio" "python-dotenv"

为 Speech Engine 生成签名 URL

每次新通话到达时，桥接器都会请求签名 URL。该 URL 包含 Speech Engine ID 和一次性签名，因此桥接器无需使用原始 API 密钥。

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

提供 TwiML 响应

通话到达时，Twilio 会向 /incoming-call 发送 POST 请求。响应是 TwiML，它会向桥接器自身的 /media-stream WebSocket 打开 Media Stream。

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

RequestValidator（Python）和 twilio.webhook({ validate: true })（Node）会根据 TWILIO_AUTH_TOKEN 检查 X-Twilio-Signature 标头。若不进行验证，公网中的任何人都能向 /incoming-call 发送 POST 请求，并让账户承担通话费用。

桥接 Media Stream

Media Stream 是一个 WebSocket，会发送一系列 JSON 事件：connected、start、media（音频负载）和 stop。桥接器会在 start 时打开 Speech Engine 对话 WebSocket，并在流关闭前双向转发音频。

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

Speech Engine 的 interruption 事件会在 Twilio 流上触发 clear 事件，丢弃所有缓冲音频，从而确保插话功能正常工作。ping 事件会以 pong 响应，以保持对话 WebSocket 连接存活。

同时运行大脑服务器

大脑服务器是快速入门中展示的标准 Speech Engine 服务器。唯一的新增内容是在 WebSocket 升级时检查共享密钥——仅当 x-api-key 与为 Speech Engine 设置的值匹配时才接受连接。

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)

有关完整的 on_transcript 实现，包括 LLM 调用和流式响应，请参阅 Speech Engine 快速入门。

将 Twilio 指向桥接器

启动桥接器和公开隧道

ngrok http 3001
python bridge.py

记下 ngrok 输出的 https:// URL——Twilio 将向其发送 POST 请求。

更新 Speech Engine ws_url

将 speech_engine.ws_url 设为大脑端点的公开 WebSocket URL，以便 ElevenLabs 知道连接位置。

await elevenlabs.speech_engine.update(
    speech_engine_id="seng_8k3m9xr4hjnfg983brhmhkd98n6",
    speech_engine={"ws_url": "wss://abc123.ngrok.io/ws"},
)

配置 Twilio 号码

在 Twilio 控制台中，打开电话号码的 Voice Configuration：

A call comes in：Webhook
URL：https://abc123.ngrok.io/incoming-call
HTTP method：POST

如果该号码已关联 Elastic SIP Trunk，请先解除关联——Twilio 号码只能路由到中继或 webhook 之一，不能同时使用两者。

拨打号码

使用任意电话拨打该号码。智能体会接听；对着电话说话，你应能听到智能体回应。启用调试日志后，桥接器会记录每一轮的通话 SID、对话 ID 和音频格式。

生产环境注意事项

Webhook 验证：始终验证 /incoming-call 上的 X-Twilio-Signature。上例使用 Twilio 的辅助库；请勿跳过此步骤。
共享密钥：在大脑 WebSocket 上强制执行共享密钥验证。否则，任何猜到 ngrok URL 的人都能连接并冒充 ElevenLabs。
稳定主机：ngrok 免费套餐的 URL 每次重启都会变化。请使用保留的 ngrok 域名或真实主机名，以免每次重启后都要更新 Speech Engine ws_url 和 Twilio webhook。
延迟：每通电话都会在 LLM 首个 token 生成时间的基础上增加两次网络跳转。使用低延迟模型并流式返回响应，以保持较低的感知延迟。
一个进程还是两个：该示例将桥接器和大脑部署在同一端口，因此一个 ngrok 隧道即可覆盖所有内容。在生产环境中，只要各自都有公开 URL，也可将它们拆分为两个服务。
提示词注入：电话通话中的语音输入是不受信任的用户输入。在它们影响工具调用或数据库写入前，请验证转写文本。

后续步骤

Twilio 原生集成

使用托管 LLM，而非自定义 LLM。

Speech Engine 快速入门

使用流式 LLM 端到端构建大脑服务器。

自定义 LLM（兼容 OpenAI）

使用兼容 OpenAI 的 HTTP 端点实现自定义 LLM 的另一种方式。

Python SDK 参考文档

Speech Engine Python SDK 的类、方法和事件。

JavaScript SDK 参考文档

Speech Engine JavaScript SDK 的类、方法和事件。