LiveKit 集成 | ElevenLabs Documentation

本指南介绍如何将 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 的“客户端”。

大脑服务器与 Speech Engine 快速入门中的相同——LiveKit worker 替代浏览器作为音频源，但 LLM 逻辑保持不变。

何时使用此模式

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

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

如果只需要浏览器到 Speech Engine 的语音循环，且没有其他参与者，Speech Engine 快速入门中的 WebRTC 客户端更简单——Speech Engine 可直接通过 WebRTC 与浏览器通信，无需 LiveKit 房间。

前提条件

一个 LiveKit 项目（LiveKit Cloud 或自托管服务器）。worker 需要 LIVEKIT_URL、LIVEKIT_API_KEY 和 LIVEKIT_API_SECRET。
一个 ElevenLabs Speech Engine。请按照 Speech Engine 快速入门创建并运行大脑服务器。
Python 3.9+ 或 Node.js 18+。

Node 桥接 worker 使用 @livekit/rtc-node，该工具目前处于开发者预览阶段。对于生产部署，建议使用 Python worker。

配置 Speech Engine 音频格式

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

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

Speech Engine 的 PCM 始终采用有符号 16 位小端格式。有关其他支持的采样率，请参阅音频格式参考。

构建桥接 worker

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

安装依赖项

pip install "livekit-agents" "livekit-api" "elevenlabs" "aiohttp" "python-dotenv"

创建 Speech Engine 签名 URL

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

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

定义 worker 入口点

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

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

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 bridge.py dev

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

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

将 worker 调度到房间

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

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)

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

从浏览器连接

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

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 可以访问端点：
```
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 快速入门

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

Pipecat 集成

使用 Pipecat 作为 Speech Engine 背后的 LLM 管道。

Python SDK 参考

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

JavaScript SDK 参考

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