我用 LiveKit Agents 搭了一个全双工 AI 语音助理：从选型到落地的完整思考

写在前面

大语言模型的能力已经不用多说了，但「能聊」和「能通话」是两回事。

在今年之前，要让 AI 真正进入实时语音对话场景，通常需要自己拼凑 STT（语音识别）→ LLM（语义理解）→ TTS（语音合成）三条 pipeline，还要处理 VAD（语音活动检测）、断句、打断、并发等一堆边缘情况。一个不小心，用户说「嗯」的时候 AI 就开始抢话，识别延迟堆到两三秒，语音对话就变成了对讲机。

直到我发现了 LiveKit Agents 这个框架。

这篇文章不打算写教程——网上已经有很多了。我想分享的是：在真实的 IM/呼叫业务中，把 LiveKit Agents 嵌入后端架构时，遇到的那些典型问题和解决思路。

选型：为什么是 LiveKit 而不是 Twilio / Agora

先简单说下背景：我需要在一个已有 IM 和 WebRTC 通话能力的系统中，加入 AI 语音助理的能力。

市面上可选方案大致有三类：

方案	代表	评价
全托管方案	Twilio Voice + OpenAI	省事，但贵，数据不在自己手里，定制空间小
自建 pipeline	自己拼 STT→LLM→TTS	灵活，但工作量巨大，VAD、并发、断句都得自己搞
Agent 框架	LiveKit Agents, Vocode, Pipecat	折中方案，框架处理底层细节，自己写业务逻辑

最终选择 LiveKit Agents 的原因很简单：它的 Agent API 抽象层做得最好。

LiveKit Agents 把「AI 助理加入一个实时音视频房间」这件事，抽象成了一个标准的 Job/Worker 模型：

复制代码

房间创建 → LiveKit Server 发布 Job → Worker 消费 Job → Agent 加入房间 → 开始对话

这个模型天然契合我已有的 IM 呼叫流程——在我的系统中，创建通话房间的逻辑早就写好了，现在只需要在房间创建后，让 AI 作为参与者之一加入即可。

架构拆解

实际的部署架构如下：

复制代码

客户端（浏览器/App）
    ↕ WebRTC 音频流
LiveKit Server（自托管）
    ↕ 音频流
Python Agent Worker（LiveKit Agents）
    ↕ HTTP（拉取上下文 + 推送记录）
Go 主后端（业务服务）

关键设计原则只有一条：Agent 不做业务逻辑，只做「语音 ↔ 文本」的转换。

所有的业务知识、用户信息、聊天记录，都通过 HTTP 接口从主后端拉取，以 System Prompt 的形式注入 LLM。Agent 识别到的文本和 AI 的回复，也通过 HTTP 写回主后端持久化。

Agent 的入口代码非常清晰，核心流程就三步：

python 复制代码

async def entrypoint(ctx: JobContext):
    room_name = ctx.room.name

    # 1. 从业务后端拉取当前房间的上下文
    room_context = await fetch_room_context(room_name)
    system_prompt = build_system_prompt(room_context)

    # 2. 连接房间，等待用户进入
    await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)
    participant = await ctx.wait_for_participant()

    # 3. 构建 Agent，注入 LLM + VAD + TTS
    agent = Agent(
        instructions=system_prompt,
        llm=google.realtime.RealtimeModel(
            api_key=google_api_key, voice="Aoede",
        ),
        tts=google.beta.GeminiTTS(api_key=google_api_key),
        vad=ctx.proc.userdata["vad"],
        allow_interruptions=True,
    )

    # 4. 启动会话
    session = AgentSession()
    await session.start(agent, room=ctx.room)
    await session.say("AI 助理已进入房间")

框架把所有音频流的底层细节都封装了——WebRTC 连接、音频编解码、端到端延迟控制——开发者只需要关注 LLM 的 instructions 和事件回调。

这样做的三个好处：

保持 Agent 的轻量和可替换 —— 未来想换 STT/TTS 服务商，或者甚至把 AI 能力从 LLM 换成其他模型，Agent 代码的改动量很小
业务逻辑统一 —— 所有用户、权限、历史记录的管理，都在主后端，Agent 不需要理解业务
解耦部署 —— Agent Worker 可以独立扩缩容，语音密集场景下可以多开 Worker，不影响主后端

全双工 vs 半双工：一个被低估的细节

早期的大多数 AI 语音方案都是半双工的——用户说完，AI 再回答，谁说完谁「释放麦克风」。这在交互上非常不自然，真人对话中不可能等对方说完再开口。

LiveKit Agents 在 1.x 版本中引入了 allow_interruptions 机制，配合 VAD 模块，实现了真正的全双工对话。实现上，它做了三件事：

VAD（Silero）持续检测 —— 实时判断用户是否在说话，用户何时说完
流式 TTS 支持 —— AI 的回复不是生成完一整句再播放，而是边生成边播放
打断逻辑 —— 用户一旦开口，AI 的播放立即停止，等待用户说完再响应

这三个机制的协作，决定了用户感知到的「对话流畅度」。我在调试阶段最常遇到的问题就是：VAD 灵敏度太高导致频繁打断 AI、或者太低导致用户说完 AI 还在发呆。

VAD 模型需要在 Worker 启动时预加载到内存，避免每次通话都重新下载：

python 复制代码

def prewarm(proc: JobProcess):
    """Worker 进程启动时执行一次，预加载 Silero VAD 模型"""
    proc.userdata["vad"] = silero.VAD.load()

然后在 Worker 入口中注册：

python 复制代码

if __name__ == "__main__":
    cli.run_app(
        WorkerOptions(
            entrypoint_fnc=entrypoint,
            prewarm_fnc=prewarm,
        )
    )

prewarm 机制是 LiveKit Agents 的一个好设计。语音处理模型的加载通常需要几百毫秒到几秒，如果在每次通话开始时才加载，用户的首次语音体验会有一个明显的卡顿。通过进程级预加载，第一个字就能准确检测到。

最终调出来的参数经验是：VAD 的阈值和沉默超时时间，必须根据实际使用场景的语速和背景噪音来调。 文档里的默认参数只能给出一个能跑的结果，离「好用」还有一段距离。

关键决策：选多模态还是纯拆分的三件套

在 Agent 的架构设计上有一个值得讨论的点：LLM 是否负责音频的输入和输出？

方案 A（传统拆分方案）：
用户音频 → STT（转文本） → LLM（生成文本） → TTS（转音频） → 播放
方案 B（多模态方案）：
用户音频 → 多模态 LLM（直接理解音频并生成音频） → 播放

这个决策会影响延迟、成本、以及代码复杂度。

我现在的做法是：使用多模态 RealtimeModel 方案。这样做的好处非常明显：

代码量骤减 —— 不需要注册单独的 STT 和 TTS 组件，LLM 一次性处理输入输出
延迟更低 —— 少了文本序列化和反序列化的环节
天然支持情感和语气 —— 多模态模型能感知用户语调，回复也更有「人情味」

但多模态方案也有代价：模型的选型范围窄，对云服务商的依赖度高。如果未来想换到本地部署的模型，目前多模态方案还不太现实。

对话上下文的同步问题

这是实际落地中遇到的最棘手的工程问题。

当 AI 在一场通话中说话时，这部分对话需要同步到 IM 的消息记录中，让用户可以看到「历史记录」。但如果 AI 和用户之间开启了语音通话，他们在 IM 上看到的消息应该是怎样的？

我最终的设计是：

Agent 实时推送转写内容到后端 —— 用户说的每一句话被识别后，AI 的每句回复生成后，都通过 HTTP 异步写回后端
后端以「消息」的形式持久化 —— 与 IM 中的文本消息共用同一张消息表，带上一个「语音转写」的标识
通话结束时，允许对摘要进行处理 —— 长通话需要生成摘要，以特殊消息格式推送给参与者

实现起来，就是在 Agent 的事件回调中做一次异步 HTTP 推送：

python 复制代码

@session.on("conversation_item_added")
def on_item_added(ev):
    item = ev.item
    if not isinstance(item, llm.ChatMessage):
        return

    content = item.text_content
    if not content:
        return

    if item.role == "user":
        # 用户说的话 → 推送给业务后端持久化
        asyncio.create_task(
            push_utterance(room_name, participant.identity, content)
        )
    elif item.role == "assistant":
        # AI 的回复 → 同样推送给业务后端持久化
        asyncio.create_task(
            push_utterance(room_name, "ai_assistant", content)
        )

而 push_utterance 本身就是一个无状态的 HTTP POST：

python 复制代码

async def push_utterance(room_name: str, identity: str, content: str):
    url = f"{backend_url}/api/agent/utterance"
    payload = {
        "room_name": room_name,
        "speaker_identity": identity,
        "content": content,
        "language": "zh-CN",
    }
    async with session.post(url, json=payload, timeout=aiohttp.ClientTimeout(total=5)) as resp:
        if resp.status != 200:
            logger.warning(f"Push failed: {resp.status}")

单看这两段代码很简单，但实际有几个坑：

异步中的扇出问题 —— Agent 一次可能处理多个房间，回调中的异步任务必须做好异常处理，不能漏掉某一句
顺序保证 —— HTTP 请求是异步的，不能保证到达顺序。但 IM 消息是需要有序的，所以后端需要在前端时间戳的基础上做重排序
重试和幂等 —— 网络抖动可能导致同一句话被推送多次，后端必须能去重

部署和运维的一些经验

Agent Worker 部署我们用 Docker，network_mode: host 是关键决策：

语音流的每一跳都会增加延迟。如果用 bridge 网络模式，音频数据要经过容器内部的 NAT 转发，在带宽敏感的实时场景下不可忽视。host 模式直接共享宿主机网络栈，大约能减少 5-10ms 的往返延迟。

另外，自签名证书的 SSL 跳过也值得一提。

LiveKit Server 在内网部署时几乎都会用自签名证书，而 Python 的 aiohttp/urllib 默认会验证 SSL。如果在每个请求的地方传 ssl=False 太麻烦。在模块级别做一次 monkey-patch，是最省事的做法：

python 复制代码

import ssl
import aiohttp

# 全局跳过 SSL 验证（仅限内网自签名证书场景）
ssl._create_default_https_context = ssl._create_unverified_context

_original_init = aiohttp.TCPConnector.__init__

def _patched_init(self, *args, **kwargs):
    kwargs["ssl"] = False
    _original_init(self, *args, **kwargs)

aiohttp.TCPConnector.__init__ = _patched_init

虽然 monkey-patch 不优雅，但在可信内网中是最实用的方案。

Docker 部署时，network_mode: host 是关键参数：

yaml 复制代码

services:
  axiom-agent:
    build: .
    env_file:
      - .env
    network_mode: "host"          # 直接复用宿主机网络栈
    deploy:
      resources:
        limits:
          memory: 1G
    logging:
      driver: "json-file"
      options:
        max-size: "10m"
        max-file: "3"

语音流的每一跳都会增加延迟。bridge 模式下音频数据要经过容器 NAT 转发，host 模式直接共享宿主机网络栈，大约能减少 5-10ms 的往返延迟。对于实时语音，每一毫秒都算数。

费用分析

用多模态模型的费用结构和传统拆分方案不太一样。传统方案下，一小时活跃通话的大致成本：

STT（语音识别）：约 $0.26
LLM（语义理解）：约 $0.50
TTS（语音合成）：约 $0.30 **合计：约$ 1.06/小时**

而多模态方案下，STT 和 TTS 的成本被「打包」进 LLM 的调用中，单价看似更贵，但省去了多服务串联的开销和网络往返，实际综合成本反而可能会低一些。

当然这些数据只是估算，实际取决于对话密度、模型版本、以及 API 定价变化。但作为一个参考，一小时 AI 通话的成本控制在一到两美元以内，对于很多 B 端场景来说是完全可以接受的。

一些还在探索的方向

目前的方案已经跑通了基本的闭环，但还有几个值得继续深入的方面：

更自然的打断 —— 现在基于 VAD 的打断虽然可用，但在多人对话场景下，AI 不知道「什么时候该插话」，目前只能被动等用户说完。如果能引入更细粒度的对话状态管理，AI 的参与感会更强。
TTS 情感控制 —— 现在的 TTS 支持通过调节参数改变语气，但目前没有和 LLM 的回复内容做联动。如果能让 AI 在安慰用户时语气更柔和，在确认事项时语气更笃定，体验会完全不一样。
长对话的 token 管理 —— 超过一小时的对话，上下文会变得很长，token 消耗和模型响应延迟都会增加。目前简单的滑动窗口策略会丢失早期信息，需要更好的摘要与记忆机制。
语音活性检测与背景降噪 —— 真实环境下的背景噪音（键盘声、环境人声）会影响 VAD 的准确性，甚至导致 Agent 误以为自己被「打断」。加一层降噪预处理可能会显著提升体验。

最后

LiveKit Agents 是一个非常优秀的框架，它把 AI 语音对话从「实验室玩具」推进到了「可投入生产的工具」这个阶段。

但也需要承认，框架只能解决「怎么连接」的问题，不能解决「怎么好用」的问题。真正决定用户体验的，是 VAD 参数的反复调试、是对话上下文如何与现有业务打通、是延迟的每一毫秒优化——这些都需要在具体的业务场景中去磨。

如果你也在做类似的事情，欢迎交流。AI 语音助理还在早期，这个方向上的坑和思路，值得更多人来一起探索。