> 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.

# LiveKit 集成

本指南介绍如何将 ElevenLabs Speech Engine 用作 LiveKit 房间的语音层。一个 LiveKit Agents worker 会作为参与者加入房间，订阅用户的音频轨道，打开与 Speech Engine 的 WebSocket，并将 Speech Engine 合成的音频作为自己的轨道发布回房间。

## 架构

Speech Engine 接受两类 WebSocket 连接：

* ElevenLabs API 连接的 **大脑 WebSocket**。服务器通过 Speech Engine SDK（`engine.serve()` / `engine.attach()`）运行此连接，并接收需要响应的转写文本。
* 客户端连接的 **对话 WebSocket**。浏览器通过 WebRTC 令牌连接；非浏览器客户端（如 LiveKit Agents worker）则通过签名 URL 连接，并双向流式传输原始 PCM 音频。

LiveKit worker 使用第二种连接。它代表 LiveKit 房间中的参与者充当 Speech Engine 的“客户端”。

```mermaid
sequenceDiagram
    participant Browser
    participant LK as LiveKit Room
    participant Worker as Agents Worker
    participant EL as ElevenLabs (conversation WS)
    participant Brain as Brain Server

    Browser->>LK: Join room (LiveKit token)
    Worker->>LK: Join room (dispatched)
    Worker->>EL: Open conversation WebSocket (signed URL)

    loop Conversation
        Browser->>LK: Microphone audio (Opus)
        LK->>Worker: Decoded PCM frames
        Worker->>EL: user_audio_chunk (base64 PCM)
        EL->>Brain: user_transcript
        Brain-->>EL: agent_response (streamed)
        EL->>Worker: audio (base64 PCM)
        Worker->>LK: Publish PCM frames
        LK->>Browser: Audio (Opus)
    end
```

大脑服务器与 [Speech Engine 快速入门](/docs/zh/eleven-api/guides/cookbooks/speech-engine)中的相同——LiveKit worker 替代浏览器作为音频源，但 LLM 逻辑保持不变。

## 何时使用此模式

当房间本身是体验的一部分时，请使用 LiveKit 桥接：

* 多参与者会话，用户可与智能体一起交流
* 已有的 LiveKit 部署，切换传输方式会导致客户端无法使用
* 与屏幕共享、视频或文字聊天共处同一房间的语音智能体
* 需要 AI 智能体参与通话的 SIP 至 LiveKit 调度通话

如果只需要浏览器到 Speech Engine 的语音循环，且没有其他参与者，[Speech Engine 快速入门](/docs/zh/eleven-api/guides/cookbooks/speech-engine#client-setup)中的 WebRTC 客户端更简单——Speech Engine 可直接通过 WebRTC 与浏览器通信，无需 LiveKit 房间。

## 前提条件

* 一个 LiveKit 项目（[LiveKit Cloud](https://cloud.livekit.io/) 或自托管服务器）。worker 需要 `LIVEKIT_URL`、`LIVEKIT_API_KEY` 和 `LIVEKIT_API_SECRET`。
* 一个 ElevenLabs Speech Engine。请按照 [Speech Engine 快速入门](/docs/zh/eleven-api/guides/cookbooks/speech-engine)创建并运行大脑服务器。
* Python 3.9+ 或 Node.js 18+。

> **Note**
>
> Node 桥接 worker 使用
> [`@livekit/rtc-node`](https://www.npmjs.com/package/@livekit/rtc-node)，该工具目前处于
> 开发者预览阶段。对于生产部署，建议使用 Python worker。

## 配置 Speech Engine 音频格式

LiveKit 的 `AudioStream` 会将传入的 Opus 轨道重采样为你请求的任意 PCM 采样率，因此可以直接匹配 Speech Engine 的输入。更新 Speech Engine，使其接受用于 ASR 输入的 16 kHz PCM，并输出用于 TTS 的 24 kHz PCM。

**`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": "pcm_16000"},
        tts={"agent_output_audio_format": "pcm_24000"},
    )


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: "pcm_16000" },
  tts: { agentOutputAudioFormat: "pcm_24000" },
});
```

Speech Engine 的 PCM 始终采用有符号 16 位小端格式。有关其他支持的采样率，请参阅[音频格式参考](#audio-format-reference)。

## 构建桥接 worker

该 worker 是一个长期运行的进程，它会连接到 LiveKit 服务器、等待任务、加入分配的房间，并在房间与 Speech Engine 之间桥接音频。

#### 安装依赖项

**`Python`**

```bash title="Python"
pip install "livekit-agents" "livekit-api" "elevenlabs" "aiohttp" "python-dotenv"
```

**`Node`**

```bash title="Node"
npm install @livekit/agents @livekit/rtc-node @elevenlabs/elevenlabs-js ws dotenv
```

#### 创建 Speech Engine 签名 URL

worker 会为 Speech Engine 对话 WebSocket 请求一个短期有效的签名 URL。该签名 URL 嵌入引擎 ID 和一次性签名，因此 worker 可以在不暴露 API 密钥的情况下打开 WebSocket。

**`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;
}
```

#### 定义 worker 入口点

每次 worker 被调度到房间时，其入口点都会运行。入口点连接到房间，打开 Speech Engine 对话 WebSocket，并启动两个音频桥接：一个将呼叫方音频发送至 Speech Engine，另一个接收返回的合成音频。

**`bridge.py`**

```python title="bridge.py" maxLines=0
import asyncio
import base64
import json
import os

import aiohttp
from dotenv import load_dotenv
from elevenlabs import AsyncElevenLabs
from livekit import agents, rtc
from livekit.agents import JobContext, WorkerOptions, cli

load_dotenv()

elevenlabs = AsyncElevenLabs(api_key=os.environ["ELEVENLABS_API_KEY"])
SPEECH_ENGINE_ID = os.environ["SPEECH_ENGINE_ID"]

USER_INPUT_RATE = 16000
AGENT_OUTPUT_RATE = 24000


async def signed_url() -> str:
    response = await elevenlabs.conversational_ai.conversations.get_signed_url(
        agent_id=SPEECH_ENGINE_ID,
    )
    return response.signed_url


async def entrypoint(ctx: JobContext):
    el_ws_ready: asyncio.Future[aiohttp.ClientWebSocketResponse] = (
        asyncio.get_running_loop().create_future()
    )

    async def pump_user_audio(track: rtc.Track):
        el_ws = await el_ws_ready
        stream = rtc.AudioStream(
            track, sample_rate=USER_INPUT_RATE, num_channels=1,
        )
        async for event in stream:
            payload = base64.b64encode(bytes(event.frame.data)).decode()
            await el_ws.send_str(json.dumps({"user_audio_chunk": payload}))

    # Register the subscriber BEFORE ctx.connect() so we don't miss tracks
    # that get auto-subscribed during the connection handshake.
    @ctx.room.on("track_subscribed")
    def on_track_subscribed(track, publication, participant):
        if track.kind != rtc.TrackKind.KIND_AUDIO:
            return
        if participant.identity == ctx.room.local_participant.identity:
            return
        asyncio.create_task(pump_user_audio(track))

    await ctx.connect()

    # Publish a track for the agent's synthesized audio.
    source = rtc.AudioSource(sample_rate=AGENT_OUTPUT_RATE, num_channels=1)
    track = rtc.LocalAudioTrack.create_audio_track("elevenlabs-agent", source)
    await ctx.room.local_participant.publish_track(
        track,
        rtc.TrackPublishOptions(source=rtc.TrackSource.SOURCE_MICROPHONE),
    )

    # Open the Speech Engine conversation WebSocket.
    http = aiohttp.ClientSession()
    el_ws = await http.ws_connect(await signed_url())
    await el_ws.send_str(json.dumps({"type": "conversation_initiation_client_data"}))
    el_ws_ready.set_result(el_ws)

    async def el_to_room():
        async for msg in el_ws:
            if msg.type != aiohttp.WSMsgType.TEXT:
                continue
            event = json.loads(msg.data)
            etype = event.get("type")
            if etype == "audio":
                pcm = base64.b64decode(event["audio_event"]["audio_base_64"])
                samples_per_channel = len(pcm) // 2
                frame = rtc.AudioFrame(
                    pcm, AGENT_OUTPUT_RATE, 1, samples_per_channel,
                )
                await source.capture_frame(frame)
            elif etype == "interruption":
                source.clear_queue()
            elif etype == "ping":
                event_id = event.get("ping_event", {}).get("event_id")
                await el_ws.send_str(json.dumps({
                    "type": "pong", "event_id": event_id,
                }))

    pump_task = asyncio.create_task(el_to_room())

    async def cleanup():
        pump_task.cancel()
        await el_ws.close()
        await http.close()

    ctx.add_shutdown_callback(cleanup)


if __name__ == "__main__":
    cli.run_app(WorkerOptions(
        entrypoint_fnc=entrypoint,
        agent_name="elevenlabs-bridge",
    ))
```

**`bridge.mts`**

```typescript title="bridge.mts" maxLines=0
import {
  type JobContext,
  WorkerOptions,
  cli,
  defineAgent,
} from "@livekit/agents";
import {
  AudioFrame,
  AudioSource,
  AudioStream,
  LocalAudioTrack,
  RoomEvent,
  TrackKind,
  TrackPublishOptions,
  TrackSource,
} from "@livekit/rtc-node";
import { ElevenLabsClient } from "@elevenlabs/elevenlabs-js";
import WebSocket from "ws";
import { fileURLToPath } from "node:url";
import "dotenv/config";

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

const USER_INPUT_RATE = 16000;
const AGENT_OUTPUT_RATE = 24000;

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

export default defineAgent({
  entry: async (ctx: JobContext) => {
    let resolveElReady: (ws: WebSocket) => void;
    const elReady = new Promise<WebSocket>((resolve) => {
      resolveElReady = resolve;
    });

    // Register the subscriber BEFORE ctx.connect() so we don't miss
    // tracks that get auto-subscribed during the connection handshake.
    ctx.room.on(RoomEvent.TrackSubscribed, (track, _pub, participant) => {
      if (track.kind !== TrackKind.KIND_AUDIO) return;
      if (participant.identity === ctx.room.localParticipant?.identity) return;

      (async () => {
        const ws = await elReady;
        const stream = new AudioStream(track, {
          sampleRate: USER_INPUT_RATE,
          numChannels: 1,
        });
        for await (const frame of stream) {
          const payload = Buffer.from(
            frame.data.buffer,
            frame.data.byteOffset,
            frame.data.byteLength,
          ).toString("base64");
          ws.send(JSON.stringify({ user_audio_chunk: payload }));
        }
      })();
    });

    await ctx.connect();

    const source = new AudioSource(AGENT_OUTPUT_RATE, 1);
    const track = LocalAudioTrack.createAudioTrack("elevenlabs-agent", source);
    const publishOptions = new TrackPublishOptions();
    publishOptions.source = TrackSource.SOURCE_MICROPHONE;
    await ctx.room.localParticipant!.publishTrack(track, publishOptions);

    const ws = new WebSocket(await signedUrl());
    await new Promise<void>((resolve, reject) => {
      ws.once("open", () => resolve());
      ws.once("error", reject);
    });
    ws.send(JSON.stringify({ type: "conversation_initiation_client_data" }));
    resolveElReady!(ws);

    // Serialize captureFrame calls — concurrent captures throw
    // InvalidState in the rtc-node native layer.
    let captureChain: Promise<unknown> = Promise.resolve();

    ws.on("message", (raw) => {
      const event = JSON.parse(raw.toString());
      if (event.type === "audio") {
        const pcm = Buffer.from(event.audio_event.audio_base_64, "base64");
        const samples = new Int16Array(
          pcm.buffer, pcm.byteOffset, pcm.byteLength / 2,
        );
        const frame = new AudioFrame(
          samples, AGENT_OUTPUT_RATE, 1, samples.length,
        );
        captureChain = captureChain
          .then(() => source.captureFrame(frame))
          .catch((err) => console.warn("captureFrame:", err.message));
      } else if (event.type === "interruption") {
        source.clearQueue();
      } else if (event.type === "ping") {
        ws.send(JSON.stringify({
          type: "pong", event_id: event.ping_event?.event_id,
        }));
      }
    });

    ctx.addShutdownCallback(async () => {
      ws.close();
    });
  },
});

cli.runApp(new WorkerOptions({
  agent: fileURLToPath(import.meta.url),
  agentName: "elevenlabs-bridge",
}));
```

worker 会在 `track_subscribed` 处理程序中通过与本地参与者的身份进行比较，过滤掉自己发布的音频。没有此检查，worker 会尝试将自己合成的音频发送回 Speech Engine。

有两个顺序细节会影响正确性：

* **监听器时机**：在 `ctx.connect()` 之前注册 `TrackSubscribed`。LiveKit 会在连接握手期间自动订阅已有轨道，之后注册的监听器可能会错过该事件。音频泵会等待 Speech Engine WebSocket 的 `Future` / `Promise`，从而能立即订阅，并在连接打开后立即转发音频。
* **仅 TypeScript：采集序列化**：如果并发调用，`@livekit/rtc-node` 的 `AudioSource.captureFrame` 会抛出 `InvalidState`。TypeScript 处理程序通过 Promise 链将采集操作串行化。Python 的单个 `async for el_to_room` 循环天然是顺序执行，无需这样做。

#### 启动 worker

**`Python`**

```bash title="Python"
python bridge.py dev
```

**`Node`**

```bash title="Node"
npx tsx bridge.mts dev
```

`dev` 会启用热重载和彩色日志。生产环境请使用 `start`，以获得 JSON 日志和优雅关闭。

worker 会连接到 LiveKit 服务器并等待任务分配。在被调度前，它不会加入任何房间。

## 将 worker 调度到房间

由于 worker 具有 `agent_name`，它使用显式调度——只有后端通知时才会加入房间。最简单的模式是在浏览器用于连接的 LiveKit 访问令牌中包含 `RoomAgentDispatch`。

**`token_server.py`**

```python title="token_server.py"
import os

from dotenv import load_dotenv
from flask import Flask, jsonify, request
from livekit.api import AccessToken, RoomAgentDispatch, VideoGrants

load_dotenv()

app = Flask(**name**)

@app.route("/api/livekit-token")
def get_token():
room_name = request.args.get("room", "demo-room")
identity = request.args.get("identity", "web-user")

    token = (
        AccessToken(
            os.environ["LIVEKIT_API_KEY"],
            os.environ["LIVEKIT_API_SECRET"],
        )
        .with_identity(identity)
        .with_grants(VideoGrants(room_join=True, room=room_name))
        .with_room_config(
            room_configuration={
                "agents": [RoomAgentDispatch(agent_name="elevenlabs-bridge")],
            },
        )
    )

    return jsonify(token=token.to_jwt(), url=os.environ["LIVEKIT_URL"])

if **name** == "**main**":
app.run(port=3002)

```

**`token-server.mts`**

```typescript title="token-server.mts"
import express from "express";
import { AccessToken } from "livekit-server-sdk";
import "dotenv/config";

const app = express();

app.get("/api/livekit-token", async (req, res) => {
  const room = (req.query.room as string) ?? "demo-room";
  const identity = (req.query.identity as string) ?? "web-user";

  const token = new AccessToken(
    process.env.LIVEKIT_API_KEY!,
    process.env.LIVEKIT_API_SECRET!,
    { identity },
  );
  token.addGrant({ roomJoin: true, room });
  token.roomConfig = {
    agents: [{ agentName: "elevenlabs-bridge" }],
  };

  res.json({
    token: await token.toJwt(),
    url: process.env.LIVEKIT_URL,
  });
});

app.listen(3002, () => {
  console.log("Token server listening on port 3002");
});
```

当浏览器使用此令牌创建或加入房间时，LiveKit 会自动将桥接 worker 调度到同一房间。

## 从浏览器连接

浏览器只需要标准的 LiveKit 客户端——无需直接与 Speech Engine 交互。

**`App.tsx`**

```typescript title="App.tsx"
import { Room, RoomEvent, Track } from "livekit-client";
import { useCallback, useState } from "react";

export default function App() {
  const [room] = useState(() => new Room());

  const join = useCallback(async () => {
    const response = await fetch("/api/livekit-token");
    const { token, url } = await response.json();

    room.on(RoomEvent.TrackSubscribed, (track) => {
      if (track.kind === Track.Kind.Audio) {
        document.body.appendChild(track.attach());
      }
    });

    await room.connect(url, token);
    await room.localParticipant.setMicrophoneEnabled(true);
  }, [room]);

  return <button onClick={join}>Start conversation</button>;
}
```

点击按钮后，浏览器会获取 LiveKit 令牌，启用麦克风加入房间，并开始接收智能体的音频轨道。worker 会被调度，打开其 Speech Engine 会话，并双向桥接音频。

## 音频格式参考

Speech Engine 支持以下音频格式。通过引擎上的 `asr.user_input_audio_format` 和 `tts.agent_output_audio_format` 进行配置。

| 格式          | 采样率       | 编码              | 说明                                      |
| ----------- | --------- | --------------- | --------------------------------------- |
| `pcm_8000`  | 8 kHz     | 有符号 16 位 LE PCM | 仅用于 ASR 输入。                             |
| `pcm_16000` | 16 kHz    | 有符号 16 位 LE PCM | 推荐用于 LiveKit 用户输入。                      |
| `pcm_22050` | 22.05 kHz | 有符号 16 位 LE PCM |                                         |
| `pcm_24000` | 24 kHz    | 有符号 16 位 LE PCM | 推荐用于 LiveKit 智能体输出。                     |
| `pcm_44100` | 44.1 kHz  | 有符号 16 位 LE PCM | TTS 输出需要 Independent Publisher 层级或更高版本。 |
| `pcm_48000` | 48 kHz    | 有符号 16 位 LE PCM | 仅用于 ASR 输入。                             |
| `ulaw_8000` | 8 kHz     | μ-law           | 由 Twilio Media Streams 使用。              |

LiveKit 中的 `AudioStream` 和 `AudioSource` 会为你处理重采样——你可以从 `AudioStream` 请求任意采样率，SDK 会从底层的 48 kHz Opus 轨道进行转换。

## 生产环境注意事项

* **显式调度**：始终在 `WorkerOptions` 上设置 `agent_name` / `agentName`。自动调度会为 LiveKit 项目中创建的每个房间启动 worker，这通常不是你想要的。
* **大脑服务器身份验证**：在 Speech Engine 上设置共享密钥，并在大脑服务器中验证它，确保只有 Speech Engine 可以访问端点：
  ```python
  await elevenlabs.speech_engine.update(
      speech_engine_id="seng_8k3m9xr4hjnfg983brhmhkd98n6",
      speech_engine={"request_headers": {"x-api-key": os.environ["SHARED_SECRET"]}},
  )
  ```
  然后，大脑服务器会在接受 WebSocket 升级前检查 `request.headers["x-api-key"]`。
* **令牌服务器**：在服务端创建 LiveKit 和 Speech Engine 令牌。绝不要将 `LIVEKIT_API_SECRET` 或 `ELEVENLABS_API_KEY` 暴露给浏览器。
* **事件循环规范**：不要在 worker 的事件循环中执行 CPU 密集型工作。`AudioSource.capture_frame` 和 `AudioStream` 迭代对时间敏感；较长的同步调用会延迟或丢失打断事件。对于阻塞性工作，请使用 `asyncio.to_thread()`（Python）或 `worker_threads`（Node）。
* **关闭**：注册 `ctx.add_shutdown_callback` / `ctx.addShutdownCallback`，以干净地关闭 ElevenLabs WebSocket。默认情况下，最后一个非智能体参与者离开时，房间（以及任务）会终止。

## 后续步骤

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

构建用于响应转写文本的大脑服务器。

#### [Pipecat 集成](/docs/zh/eleven-api/guides/how-to/speech-engine/pipecat-integration)

使用 Pipecat 作为 Speech Engine 背后的 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 的类、方法和事件。