> This is a page from the ElevenLabs documentation. For a complete page index, fetch https://elevenlabs.io/docs/llms.txt. For the full documentation in a single file, fetch https://elevenlabs.io/docs/llms-full.txt.

# 自定义 LLM 集成

## 概览

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

自定义 LLM 部分由 [Speech Engine SDK](/docs/zh/eleven-api/guides/cookbooks/speech-engine)提供。它会在 ElevenLabs 和服务器之间建立 WebSocket，让 LLM 能够随着通话进行流式返回响应。Twilio 部分使用 [Media Streams](https://www.twilio.com/docs/voice/media-streams)，将通话音频转发给智能体。

## 架构

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

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

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

```mermaid
sequenceDiagram
    participant Caller
    participant Twilio
    participant Bridge as Bridge Server
    participant EL as ElevenLabs (conversation WS)
    participant Brain as Brain Server

    Caller->>Twilio: Dial number
    Twilio->>Bridge: POST /incoming-call
    Bridge-->>Twilio: TwiML <Connect><Stream>
    Twilio->>Bridge: WebSocket /media-stream
    Bridge->>EL: Open conversation WebSocket (signed URL)

    loop Conversation
        Caller->>Twilio: Speak
        Twilio->>Bridge: media event (μ-law base64)
        Bridge->>EL: user_audio_chunk
        EL->>Brain: user_transcript
        Brain-->>EL: agent_response (streamed)
        EL->>Bridge: audio event (μ-law base64)
        Bridge->>Twilio: media event
        Twilio->>Caller: Play audio
    end
```

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

## 何时使用此模式

本指南和[原生 Twilio 集成](/docs/zh/eleven-agents/phone-numbers/twilio-integration/native-integration)都能将智能体部署到 Twilio 电话号码。区别在于谁负责 LLM：

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

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

此模式使用 Speech Engine SDK，它通过 WebSocket 连接在服务器与 ElevenLabs API 之间通信。你还可以使用 [Custom LLM](/docs/zh/eleven-agents/customization/llm/custom-llm) 指南，该指南使用兼容 OpenAI 的 HTTP 端点，而不是 Speech Engine SDK。

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

## 前提条件

* 一个 [Twilio](https://www.twilio.com/) 账户和一个支持语音的电话号码。
* 一个 Speech Engine 资源。请按照 [Speech Engine 快速入门](/docs/zh/eleven-api/guides/cookbooks/speech-engine)创建资源，并了解大脑服务器模式。
* 一个公开 HTTPS 隧道（例如 [ngrok](https://ngrok.com)）。Twilio 会通过公网连接到桥接器。
* Python 3.9+ 或 Node.js 18+。

## 为 μ-law 音频配置智能体

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

**`configure_engine.py`**

```python title="configure_engine.py"
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())
```

**`configure-engine.mts`**

```typescript title="configure-engine.mts"
import { ElevenLabsClient } from "@elevenlabs/elevenlabs-js";
import "dotenv/config";

const elevenlabs = new ElevenLabsClient({
  apiKey: process.env.ELEVENLABS_API_KEY,
});

await elevenlabs.speechEngine.update("seng_8k3m9xr4hjnfg983brhmhkd98n6", {
  asr: { userInputAudioFormat: "ulaw_8000" },
  tts: {
    modelId: "eleven_flash_v2",
    agentOutputAudioFormat: "ulaw_8000",
  },
  speechEngine: {
    requestHeaders: { "x-api-key": process.env.SHARED_SECRET! },
  },
});
```

`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()` 服务器。

#### 安装依赖

**`Python`**

```bash title="Python"
pip install "elevenlabs" "aiohttp" "twilio" "python-dotenv"
```

**`Node`**

```bash title="Node"
npm install @elevenlabs/elevenlabs-js express ws twilio dotenv openai
```

#### 为 Speech Engine 生成签名 URL

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

**`bridge.py`**

```python title="bridge.py"
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
```

**`bridge.mts`**

```typescript title="bridge.mts"
import { ElevenLabsClient } from "@elevenlabs/elevenlabs-js";

const elevenlabs = new ElevenLabsClient({
  apiKey: process.env.ELEVENLABS_API_KEY,
});

async function signedUrl(): Promise<string> {
  const response = await elevenlabs.conversationalAi.conversations.getSignedUrl({
    agentId: process.env.SPEECH_ENGINE_ID!,
  });
  return response.signedUrl;
}
```

#### 提供 TwiML 响应

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

**`bridge.py`**

```python title="bridge.py"
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")
```

**`bridge.mts`**

```typescript title="bridge.mts"
import express from "express";
import twilio from "twilio";

const app = express();
app.use(express.urlencoded({ extended: false }));

app.post(
  "/incoming-call",
  twilio.webhook({ validate: true }),
  (req, res) => {
    const host = req.headers["x-forwarded-host"] ?? req.get("host");
    const twiml = `<?xml version="1.0" encoding="UTF-8"?>
      <Response>
        <Connect>
          <Stream url="wss://${host}/media-stream"/>
        </Connect>
      </Response>`;
    res.type("text/xml").send(twiml);
  },
);
```

`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，并在流关闭前双向转发音频。

**`bridge.py`**

```python title="bridge.py" maxLines=0
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
```

**`bridge.mts`**

```typescript title="bridge.mts" maxLines=0
import { WebSocket, WebSocketServer } from "ws";
import { createServer } from "node:http";

const httpServer = createServer(app);
const wss = new WebSocketServer({ noServer: true });

httpServer.on("upgrade", (req, socket, head) => {
  if (req.url === "/media-stream") {
    wss.handleUpgrade(req, socket, head, (ws) => handleMediaStream(ws));
  } else {
    socket.destroy();
  }
});

async function handleMediaStream(twilioWs: WebSocket) {
  let streamSid: string | null = null;
  let elReady: Promise<WebSocket | null> | null = null;

  twilioWs.on("message", async (raw) => {
    const event = JSON.parse(raw.toString());

    if (event.event === "start") {
      streamSid = event.start.streamSid;
      // Convert rejection into a clean null + close so a failed signed-URL
      // fetch doesn't become an unhandled rejection on the next media event.
      elReady = openElevenLabsWebSocket(twilioWs, () => streamSid).catch((err) => {
        console.error("Failed to open Speech Engine conversation:", err);
        twilioWs.close();
        return null;
      });
    } else if (event.event === "media" && elReady) {
      const elWs = await elReady;
      if (!elWs) return;
      elWs.send(JSON.stringify({
        user_audio_chunk: event.media.payload,
      }));
    } else if (event.event === "stop") {
      twilioWs.close();
    }
  });

  twilioWs.on("close", async () => {
    (await elReady)?.close();
  });
}

async function openElevenLabsWebSocket(
  twilioWs: WebSocket,
  getStreamSid: () => string | null,
): Promise<WebSocket> {
  const elWs = new WebSocket(await signedUrl());
  await new Promise<void>((resolve, reject) => {
    elWs.once("open", () => resolve());
    elWs.once("error", reject);
  });
  elWs.send(JSON.stringify({
    type: "conversation_initiation_client_data",
  }));

  elWs.on("message", (raw) => {
    const event = JSON.parse(raw.toString());
    const streamSid = getStreamSid();
    if (event.type === "audio") {
      twilioWs.send(JSON.stringify({
        event: "media",
        streamSid,
        media: { payload: event.audio_event.audio_base_64 },
      }));
    } else if (event.type === "interruption") {
      twilioWs.send(JSON.stringify({ event: "clear", streamSid }));
    } else if (event.type === "ping") {
      elWs.send(JSON.stringify({
        type: "pong", event_id: event.ping_event?.event_id,
      }));
    }
  });

  return elWs;
}
```

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

#### 同时运行大脑服务器

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

**`bridge.py`**

```python title="bridge.py" maxLines=0
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)
```

**`bridge.mts`**

```typescript title="bridge.mts" maxLines=0
httpServer.on("upgrade", async (req, socket, head) => {
  if (req.url === "/ws") {
    if (req.headers["x-api-key"] !== process.env.SHARED_SECRET) {
      socket.write("HTTP/1.1 401 Unauthorized\r\n\r\n");
      socket.destroy();
      return;
    }
    // Hand off to engine.attach() — see the Speech Engine quickstart.
  } else if (req.url === "/media-stream") {
    wss.handleUpgrade(req, socket, head, (ws) => handleMediaStream(ws));
  } else {
    socket.destroy();
  }
});

httpServer.listen(3001);
```

有关完整的 `on_transcript` 实现，包括 LLM 调用和流式响应，请参阅 [Speech Engine 快速入门](/docs/zh/eleven-api/guides/cookbooks/speech-engine#server-setup)。

## 将 Twilio 指向桥接器

#### 启动桥接器和公开隧道

```bash
ngrok http 3001
python bridge.py
```

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

#### 更新 Speech Engine ws_url

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

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

```typescript
await elevenlabs.speechEngine.update("seng_8k3m9xr4hjnfg983brhmhkd98n6", {
  speechEngine: { wsUrl: "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 原生集成](/docs/zh/eleven-agents/phone-numbers/twilio-integration/native-integration)

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

#### [Speech Engine 快速入门](/docs/zh/eleven-api/guides/cookbooks/speech-engine)

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

#### [自定义 LLM（兼容 OpenAI）](/docs/zh/eleven-agents/customization/llm/custom-llm)

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

#### [Python SDK 参考文档](/docs/zh/eleven-api/resources/libraries/speech-engine/python-sdk-reference)

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

#### [JavaScript SDK 参考文档](/docs/zh/eleven-api/resources/libraries/speech-engine/javascript-sdk-reference)

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