OPEN SOURCE DEEP DIVE
LiveKit Agents:把实时语音 AI 智能体做成可编程参与者
LiveKit Agents 是开源的实时语音 AI 智能体框架(Apache-2.0,15.0k star):智能体以服务器端「可编程参与者」身份加入 WebRTC 房间,能听、能看、能说。Agent、AgentSession、entrypoint、AgentServer 四层抽象之上,是 83 个可插拔 STT/LLM/TTS/Realtime 插件与 LiveKit Inference 统一模型字符串;轮转检测 transformer、自适应打断与 AMD 答机检测把对话稳健性做进框架;原生 MCP、多智能体交接、pytest 事件流断言加 LLM 评委,console/dev/start 三种运行模式直通生产调度与电话栈。
把实时语音智能体做成服务器端的「可编程参与者」
LiveKit Agents 是 LiveKit 开源的实时语音 AI 智能体框架(截至 2026-10-30 仓库 14,999 star、4,028 fork,Apache-2.0 协议,版本 1.8.10)。它对自身定位的表述很克制:构建运行在服务器上的「可编程参与者」(programmable participants),用来做能看、能听、能理解的对话式多模态语音智能体。翻译成工程语言就是:智能体不是挂在某个 HTTP 接口后面的聊天机器人,而是像真人一样加入一间 WebRTC 房间,订阅用户的麦克风与摄像头轨,把自己的语音轨发布回房间,整条链路跑在 LiveKit 的实时媒体服务器上。
这个选择决定了框架的形态。既然智能体是房间里的参与者,那么「谁在说话、说到哪了、该不该插话、什么时候回话」这些对话控制问题就必须由框架自己解决,而不是留给应用代码。LiveKit Agents 的答案是四层抽象加一整套对话控制组件:Agent、AgentSession、entrypoint、AgentServer,以及 VAD、语义轮转检测、自适应打断、端点检测、背景音与填充词调度等内建模块。
四层抽象:Agent、AgentSession、entrypoint、AgentServer
README 的「Core concepts」只列了四个名词,但每个都对应一条明确的职责边界。Agent 是带指令的 LLM 应用,装指令与工具;AgentSession 是管理与终端用户交互的容器,VAD、STT、LLM、TTS 的接线和轮转、打断都归它管;entrypoint 是一次交互会话的起点,类比 Web 服务器里的请求处理函数;AgentServer 是主进程,负责作业调度并为每个用户会话拉起智能体。写一个最小语音智能体,就是把这四层各填一格:
@function_tool
async def lookup_weather(context: RunContext, location: str):
"""Used to look up weather information."""
return {"weather": "sunny", "temperature": 70}
server = AgentServer()
@server.rtc_session()
async def entrypoint(ctx: JobContext):
session = AgentSession(
vad=inference.VAD(),
stt=inference.STT("deepgram/nova-3", language="multi"),
llm=inference.LLM("google/gemma-4-31b-it"),
tts=inference.TTS("cartesia/sonic-3", voice="..."),
)
agent = Agent(
instructions="You are a friendly voice assistant built by LiveKit.",
tools=[lookup_weather],
)
await session.start(agent=agent, room=ctx.room)
await session.generate_reply(instructions="greet the user and ask about their day")
工具就是普通的异步函数加一个 @function_tool 装饰器,docstring 即工具说明,类型注解即参数 schema;RunContext 把会话上下文递进工具里。会话启动后,generate_reply 可以用一句临时指令驱动首轮发言,例如先打招呼再问用户今天过得如何。
模型层是插座:STT、LLM、TTS、Realtime 自由组合
框架把语音链路拆成 STT、LLM、TTS 与 Realtime API 四类插槽,README 的第一条特性就是「灵活集成」:按场景混搭最合适的供应商。仓库是 monorepo,核心包 livekit-agents 之外有 83 个插件目录,覆盖 Deepgram、OpenAI、Cartesia、ElevenLabs、Google、Azure、Groq、Silero 等 STT/LLM/TTS/Realtime 提供方,以及 Tavus、Bithuman 等视频化身、电话与轮转检测插件。安装时按 extras 取用:pip install "livekit-agents[openai,deepgram,cartesia]"。
除了直接实例化插件(deepgram.STT(model="nova-3")、openai.LLM(model="gpt-4.1-mini")、cartesia.TTS(model="sonic-3", voice=...)),框架还提供 LiveKit Inference 统一入口:把模型写成 "deepgram/nova-3"、"google/gemma-4-31b-it"、"cartesia/sonic-3:音色ID" 这样的字符串,经 LiveKit Cloud 的推理网关访问,不必自己保管各家密钥。示例里用的 google/gemma-4-31b-it 就是托管在 LiveKit 上的低延迟 Gemma。
对话的硬活内建:轮转检测、自适应打断、答机检测
语音交互的体验差距几乎都出在「什么时候算说完」。LiveKit Agents 用语义轮转检测(semantic turn detection)回答这个问题:一个 transformer 模型判断用户这一轮是否结束,从而减少误打断与迟钝回应;代码里对应 inference.eot.TurnDetector,模型档位是 turn-detector-v1 与更小的 turn-detector-v1-mini,跑在 LiveKit Inference 网关上。打断侧有 AdaptiveInterruptionDetector,在用户插话时停掉正在播放的回复;会话层还有端点检测(endpointing)、背景音(background audio)与填充词调度(filler scheduler),把等待时间用「嗯」「让我查一下」这类自然反馈填上。
电话场景另有专门组件:AMD(Answering Machine Detection,答机检测)用于外呼,在 SIP 参与者创建之前启动,监听接通后的问候语并用 LLM 把它分成五类,避免对着语音信箱念完一整段话。
| AMD 判定 | 含义 |
|---|---|
| human | 真人接听 |
| machine-ivr | IVR / DTMF 菜单提示音 |
| machine-vm | 语音信箱问候,可以留言 |
| machine-unavailable | 信箱满或未开通,无法留言 |
| uncertain | 转写含糊,无法归类 |
多智能体交接:让工具返回下一个 Agent
复杂对话往往不是一个角色演到底。框架的多智能体交接写法很直接:某个 @function_tool 返回「下一个 Agent 实例 + 一句交接语」,会话就切换角色。README 的讲故事示例里,IntroAgent 负责收集姓名与地点,信息齐了就把数据写进 context.userdata,返回 StoryAgent 和「让我们开始故事吧」;StoryAgent 还可以逐智能体覆盖模型,把标准 LLM 换成 openai.realtime.RealtimeModel(voice="echo") 这类 Realtime API。会话容器本身支持泛型 userdata(AgentSession[StoryData]),跨角色共享状态有类型保障。
测试:把会话当事件流来断言,再请 LLM 当评委
LLM 行为不确定,语音智能体尤其难测。框架内建测试集成:在 pytest 里直接 await sess.run(user_input=...) 驱动一次会话,拿到的结果对象支持事件流断言——跳过某条消息、断言下一个事件是名为 start_order 的函数调用、断言其输出,最后对助手消息请 LLM 评委判断意图(.judge(llm, intent="..."))。评委调用被钉在低推理档位(inference_class="low"),不与线上流量抢网关容量。仓库的测试标记体系也反映了这套重心:unit、audio_eot、plugin、stt、tts、realtime、evals、docs 各成一类,插件的集成测试在 CI 里按供应商凭证跑。
async with AgentSession(llm=llm) as sess:
await sess.start(MyAgent())
result = await sess.run(user_input="Hello, I need to place an order.")
result.expect.next_event().is_function_call(name="start_order")
await (result.expect.next_event()
.is_message(role="assistant")
.judge(llm, intent="assistant should be asking the user what they would like"))
三种运行模式与生产调度
| 命令 | 行为 | 外部依赖 |
|---|---|---|
python myagent.py console | 终端模式,本地音频输入输出,快速验证行为 | 无需服务器与凭证 |
python myagent.py dev | 启动智能体服务器并热重载;单进程高效托管多个并发智能体 | LIVEKIT_URL / API_KEY / API_SECRET |
python myagent.py start | 生产模式,带生产级优化 | 同上 |
生产侧的作业分发由内建调度与 dispatch API 承担:把终端用户连到智能体作业,配合 LiveKit 的电话栈(SIP)可以让智能体拨打或接听真实电话;与客户端之间还能用 RPC 与 Data API 交换结构化数据。客户端一侧是 LiveKit 的开源 SDK 生态:浏览器、Swift、Android、Flutter、React Native、Rust、Node.js、Python、Unity、Unity WebGL、ESP32 全覆盖,TypeScript 侧另有 AgentsJS;想先摸效果可以直接打开 Agents Playground。
用智能体写智能体:文档 MCP 与官方技能包
README 专门留了一节给 AI 编码助手:推荐安装 LiveKit Docs MCP 服务器(给编码智能体提供最新文档与跨仓库代码搜索),再装官方 Agent Skill(npx skills add livekit/agent-skills --skill livekit-agents),后者教编码智能体如何做工作流设计、交接、任务与测试。两者的分工被写得很清楚:技能包教「怎么下手」,MCP 服务器给「当前 API 细节」。MCP 在框架里也是一等公民——一行代码接入 MCP 服务器提供的工具。
生态位置与许可
LiveKit Agents 处于「实时传输 + 语音智能体」的交汇点:下面是 LiveKit 服务器这颗被广泛使用的 WebRTC 媒体服务器,上面是各端 SDK 与电话栈,中间这层把对话控制与模型插槽做成框架能力。示例目录给出了电话外呼、前台接待、酒店前台、医疗、drive-thru、暖转移(warm transfer)、多用户转写、视频化身、Gemini Live 视觉等场景骨架。许可方面,框架本体 Apache-2.0,轮转检测模型另按 LiveKit Model License 授权;开发用 uv 管理依赖、ruff 做格式化与 lint、pdoc 生成文档。对需要自建实时语音入口的团队,它是少数能把「传输、对话控制、模型、测试、电话」一次拿齐的开源选项。
flowchart LR
A["用户麦克风/摄像头轨
WebRTC 房间"] --> B["VAD + STT
deepgram/nova-3"]
B --> C{"TurnDetector
turn-detector-v1-mini
这轮说完了吗"}
C -- 未完 --> B
C -- 说完 --> D["Agent / LLM
function_tool + MCP"]
D --> E["TTS
cartesia/sonic-3"]
E --> F["以参与者身份
发布语音轨回房间"]
B -. 并行 .-> G["AdaptiveInterruptionDetector
用户插话即停播"]
H["SIP 外呼"] -. 接通前 .-> I["AMD 答机检测
human / ivr / vm / unavailable"]
I -- human --> A