跳到主要内容

LiveKit Agents — 架构与原理

30 秒导读: LiveKit Agents 是一个用来搭"能打电话、能语音对话的 AI 参与者"的 Python 框架。它跑在服务器上,接进一个 WebRTC 房间,实时听用户说话、用 LLM 想、再用合成语音回答——而且用户随时能打断它。它最难、也最值得学的地方不是"调 LLM",而是时序:什么时候算用户"说完了"、机器该不该现在开口、被打断了怎么优雅停下。整套框架就是围绕这几个时序问题设计的。

1. 这是什么(零基础也能懂)​

一句话定义: LiveKit Agents 是一个实时语音 agent 框架——你写一个 Agent(给它一句 instructions 和几个工具),它就能接进一个音视频房间,和真人来回语音对话。

解决什么问题 / 给谁用:

假设你要做一个"AI 前台"或"外呼机器人":用户打进来,你的程序要一边听他说话、一边把语音转成文字喂给大模型、再把模型的回复合成成语音播回去。这中间全是麻烦的实时问题:

  • 用户说到一半停顿了 0.5 秒,是"说完了"还是"在想词"?你不能一停顿就抢话。
  • 机器正在念一大段回复,用户突然插话——得立刻停下来听他说,而不是把这段念完。
  • 网络有延迟、模型有延迟,怎么让整体对话"感觉流畅"?

LiveKit Agents 把这些都封装好了,面向要构建生产级语音 / 电话 agent的开发者。

它能做什么:

  • 把任意 STT / LLM / TTS 供应商像积木一样拼起来(50+ 插件在 livekit-plugins/)。
  • 语义化的轮次检测(用一个 transformer 模型判断"用户这句是不是说完了"),减少误打断。
  • 原生打断(barge-in):用户一开口,正在播的语音立即停。
  • 工具调用、多 agent 交接(handoff)、MCP 工具。
  • 也支持"实时多模态模型"(如 OpenAI Realtime),此时语音直接进模型、绕过 STT/TTS 流水线。

用起来什么样:

核心就三步:定义 Agent、组装 AgentSession、在 entrypoint 里 start。摘自 examples/voice_agents/basic_agent.py:71:

@server.rtc_session()
async def entrypoint(ctx: JobContext) -> None:
    session = AgentSession(
        stt=inference.STT("deepgram/nova-3", language="multi"),  # 耳朵:语音转文字
        llm=inference.LLM("openai/gpt-4.1-mini"),                # 脑子
        tts=inference.TTS("cartesia/sonic-3", voice="..."),      # 嘴:文字转语音
    )
    await session.start(agent=MyAgent(), room=ctx.room)

MyAgent 只是继承 Agent,写上 instructions,再用 @function_tool 标注几个方法当工具(basic_agent.py:30)。

一句话直觉/类比: 把它想成一个装了"三块可换器官"的对话机器人——STT 是耳朵、LLM 是脑、TTS 是嘴;框架本身是神经系统,负责协调"什么时候听、什么时候说、被打断了怎么反应"。器官谁家的都行(换插件),神经系统是它的核心价值。

2. 顶层全景(它大概怎么转)​

先分清两层:外层是"作业调度"(一个 Worker 进程接房间、派活到子进程),内层是"一次对话"(AgentSession 里 STT→LLM→TTS 的实时循环)。这份文档的重点是内层。

一次"用户说一句、机器答一句"的完整数据流(语音流水线模式)长这样:

主要部件一句话职责:

部件干什么在哪个文件
AgentServer(旧名 Worker)顶层进程:注册到 LiveKit、接作业、派给子进程池worker.py:296 / run 在 worker.py:535
JobContext传给 entrypoint 的上下文,ctx.connect() 接房间job.py:154
Agent你的"应用":instructions + 工具 + 可覆盖的节点方法voice/agent.py:38
AgentSession一次对话的容器:管 I/O、状态、当前活跃的 activityvoice/agent_session.py:209
AgentActivity真正的循环大脑:调度语音、跑生成、处理打断/工具voice/agent_activity.py:163
AudioRecognition听觉子系统:VAD + STT + 端点/轮次检测,产出"用户轮次"voice/audio_recognition.py:195
SpeechHandle一段机器语音的句柄:可等待、可打断、带优先级voice/speech_handle.py:25
生成原语perform_llm_inference / perform_tts_inference 等voice/generation.py:118

主线走一遍(高层,不进代码):

  1. 音频进来。 房间里的用户音频帧,由 AgentSession._forward_audio_task(agent_session.py:1556)不断喂给 AgentActivity.push_audio,再进 AudioRecognition。
  2. 切轮次。 AudioRecognition 一边用 VAD 判断"在不在说话",一边用 STT 出文字;当检测到停顿,启动端点判定决定"是不是真说完了"(见 §3.1)。
  3. 触发回复。 判定说完 → 回调 AgentActivity.on_end_of_turn(agent_activity.py:2086)→ 排一个 _pipeline_reply_task 生成回复。
  4. 流式生成。 LLM 流式吐字,按句切段,每段边出边喂 TTS 合成音频,音频流式播回房间(agent_activity.py:2663)。
  5. 随时可断。 播的过程中用户一开口,STT/打断检测触发 on_interruption(agent_activity.py:1951),当前 SpeechHandle 被 interrupt(),音频立即停,回到第 1 步。

一个关键认知:框架有"两条生成路径"。 上面讲的是语音流水线(STT+LLM+TTS 三件套)。如果你给的 llm 是一个实时多模态模型(llm.RealtimeModel),音频会直接 push_audio 进模型会话(agent_activity.py:1175),走 _realtime_reply_task(agent_activity.py:3191),STT/TTS 被绕开。理解"同一套调度、两条生成路径"是看懂这个框架的关键。

3. 核心机制​

挑三个最能体现设计精华的子系统:(1) 轮次检测与端点判定——它怎么知道"该我说话了";(2) 打断与单活跃 SpeechHandle——它怎么做到"随时能停、一次只说一句";(3) 节点流水线与多步工具循环——它怎么把可换器官串成一条流式、可 agentic 的生成链。

3.1 轮次检测:怎么判断"用户说完了"​

要解决的小问题: 麦克风给的是连续音频。人说话中间有大量停顿(想词、换气)。如果一停顿就抢话,体验很差;如果等太久,又显得反应慢。必须判断"这次停顿是不是一个轮次(turn)的结束"。

思路: 用三种信号叠加判断,而不是只看静音时长。

  • VAD(Voice Activity Detection,语音活动检测): 出"现在在不在说话"的粗信号,给出静音起点。
  • STT: 出转写文字;文字本身也是"说没说完"的线索。
  • 端点模型(turn detector): 一个 transformer,吃最近几轮对话文本,输出一个 end_of_turn_probability(说完的概率)。

巧妙的地方——用概率去"调节等待时间",而不是直接决定。 框架不是拿模型概率当"说完/没说完"的开关,而是用它去动态改变端点延迟。核心逻辑在 audio_recognition.py:1399:

# 示意,非源码;真实见 audio_recognition.py:1399-1404
if end_of_turn_probability < unlikely_threshold:
    endpointing_delay = self._endpointing.max_delay   # 模型觉得"多半没说完" → 多等
else:
    endpointing_delay = self._endpointing.min_delay   # 否则用短延迟,反应快

也就是说:模型越倾向"没说完",就等更久再触发回复(默认 min_delay=0.5s、max_delay=3.0s,见 voice/turn.py:137)。这把"要不要抢话"从硬开关变成了软性的、随信心调节的等待,是减少误打断的关键。

延迟等待的实现也讲究:它不是死 sleep,而是 await asyncio.wait_for(self._closing.wait(), timeout=extra_sleep)(audio_recognition.py:1515)——这样一旦会话要关或状态变化能立刻醒来。等待结束后回调 on_end_of_turn(audio_recognition.py:1535),把新转写、置信度、时序指标一起交给上层。

关键细节: 端点判定跑在一个可被取消/重启的 "bounce" 任务里(_bounce_eou_task,audio_recognition.py:1338)——用户如果又开口,旧任务被取消、重排,避免在用户明明还在说的时候误触发。流式端点模型还支持预测缓存(_turn_detector_prediction_fut),说到一半就先算,真到端点时直接取结果(audio_recognition.py:1360)。

3.2 打断与单活跃 SpeechHandle:一次只说一句、随时能停​

要解决的小问题: 机器可能同时有好几段想说的话(问候、对上一句的回复、工具执行后的追加回复)。但真实对话里一次只能说一句,而且正在说的这句得能被用户打断。需要一个统一的"语音调度器"。

思路:每段机器语音都是一个 SpeechHandle,进一个优先级队列,一次只放行一个。

SpeechHandle(voice/speech_handle.py:25)是一段语音的可等待句柄,内部就是几个 asyncio.Future:_interrupt_fut(被打断)、_done_fut(播完)、_scheduled_fut(轮到它)、_authorize_event(获准开播)。它带三档优先级常量:LOW=0 / NORMAL=5 / HIGH=10(speech_handle.py:26)。

调度用一个最大堆(agent_activity.py:1482):

# 示意,非源码;真实见 agent_activity.py:1485
# 优先级取负 → heapq(最小堆)当最大堆用;时间戳做次序 tie-break
heapq.heappush(self._speech_q, (-priority, time.perf_counter_ns(), speech))

调度主循环 _scheduling_task(agent_activity.py:1495)每次只 heappop 一个,设为 self._current_speech,等它播完再取下一个——这就是"同一时刻只有一个活跃语音"的机制。

打断怎么做: AgentActivity.interrupt(agent_activity.py:1400)对当前语音 + 队列里所有语音逐个 speech.interrupt(),并等它们都 done() 后才让返回的 Future 完成——保证"打断"这件事直到聊天上下文更新完毕才算数。触发源可以是 VAD、STT 首字、或专门的重叠语音检测(on_interruption,agent_activity.py:1951)。

关键细节——防"假打断"。 背景噪音可能被误当成用户插话。框架有 resume_false_interruption 机制:被打断后如果用户其实没真说话,会在 false_interruption_timeout 后恢复刚才被暂停的语音(on_interim_transcript 里的 _start_false_interruption_timer,agent_activity.py:1990)。还有 min_words:新转写词数太少就不算打断(agent_activity.py:2124),避免一个"嗯"就把机器打断。

3.3 节点流水线与多步工具循环​

要解决的小问题: STT/LLM/TTS 得能任意替换,还要能让用户插进自己的逻辑(比如在喂给 LLM 前改写上下文);同时 LLM 可能要调工具、再根据工具结果继续说,这是个 agentic 循环。

思路一:把每个模型调用包成一个可覆盖的"节点(Node)"。 Agent 上有 stt_node / llm_node / tts_node / transcription_node 等 staticmethod(voice/agent.py:421 起),默认实现只是"拿 activity 里配的模型、流式跑一遍"。比如默认 llm_node(agent.py:469)就是 async with activity_llm.chat(...) as stream: yield chunk。你想插逻辑,子类覆盖这个方法即可——框架调的是节点,不是模型本身。

默认节点还顺手做了能力适配:STT 不支持流式就自动套 StreamAdapter 用 VAD 补齐(agent.py:437);TTS 不支持流式就套句子分词的 StreamAdapter(agent.py:507)。

思路二:LLM 流按"句"切段,边出边合成,让说话尽早开始。 _pipeline_reply_task_impl(agent_activity.py:2663)把 LLM 的文本流按 FlushSentinel 切成 _SpeechSegment,每段独立起一次 TTS,而且"下一段的合成在上一段还在播时就开始"(_start_segment,agent_activity.py:2757)——这是降低首字延迟、让语音听起来连续的关键。

思路三:工具调用是个有上限的递归循环。 LLM 若吐了工具调用,perform_tool_executions(generation.py:586)去执行,拿到结果后再排一次 _pipeline_reply_task 生成对工具结果的回复(agent_activity.py:3159)。步数受 max_tool_steps(默认 3,agent_session.py:237)限制:

# 示意,非源码;真实见 agent_activity.py:3088
max_steps_reached = speech_handle.num_steps >= max_tool_steps + 1
# 到上限就把 tool_choice 强制成 "none",逼模型给最终文本、别再调工具
tool_choice = "none" if max_steps_reached else "auto"

思路四:工具还能触发 agent 交接(handoff)。 一个工具若返回了新的 Agent,框架识别为 _handoff_required,调 update_agent 换掉当前 agent(agent_activity.py:3130),后续对话由新 agent 接管。这就是多 agent 系统(如"前台 → 转专家")的实现基础。

4. 巧妙之处与局限​

值得偷师的设计:

  • 把"时序"当一等公民。 大多数教程级语音 demo 是"录完 → 识别 → 回答"的回合制;这个框架从底层就是流式 + 可打断的,SpeechHandle 用四个 Future 精确表达"已排期/获准/被打断/已完成"四个时刻(speech_handle.py:40),整套逻辑围绕这些时刻转。
  • 端点用概率调节等待,而非硬判决。 §3.1 那招"模型没信心就多等一会儿",比"概率过阈值就触发"鲁棒得多,直接影响对话自然度。
  • 节点 = 可覆盖 staticmethod。 用最轻的方式同时满足"换供应商"和"插自定义逻辑"两个需求,没有引入重量级插件接口(agent.py:421)。
  • 预生成(preemptive generation)。 在端点判定还在等延迟时就先跑 LLM,真到端点时若上下文没变就直接用这份结果(on_preemptive_generation,agent_activity.py:2028;复用校验在 agent_activity.py:2254),用"赌一把"换更低延迟。
  • 同步的 on_end_of_turn 钩子。 它特意写成同步、内部再手动起 task(agent_activity.py:2087 的注释),就是为了不被 AudioRecognition 的取消波及——细节里见工程克制。

边界与坑:

  • 不是零延迟。 端到端延迟是 VAD 静音 + 端点延迟(0.5~3s)+ LLM 首字 + TTS 首帧的叠加;max_delay=3.0s 意味着模型不确定时最坏要等 3 秒才开口。
  • 强绑 LiveKit 房间 / WebRTC。 核心 I/O 面向 LiveKit 的 rtc.Room;脱离这套媒体栈用它,要自己接 I/O(虽然 AgentSession 的 input/output 是抽象的,但生态默认是 LiveKit)。
  • 两条生成路径行为有差异。 实时模型路径下,STT 转写、min_words 之类的语音流水线特性部分被绕过(on_final_transcript 里对 RealtimeModel 直接 return,agent_activity.py:1993),调试时容易踩"为什么这个配置在 realtime 下不生效"。
  • 工具步数硬上限。 max_tool_steps 到顶会被强制 tool_choice="none"(§3.3),复杂多步工具任务可能被截断成"提前收尾"。

代码地图(导航索引)​

关注点文件说明
顶层作业进程livekit-agents/livekit/agents/worker.py:296AgentServer(旧名 Worker):注册、接作业、派进程池;run 在 worker.py:535
作业上下文livekit-agents/livekit/agents/job.py:154JobContext:connect() 接房间、add_shutdown_callback
进程池livekit-agents/livekit/agents/ipc/proc_pool.py:28ProcPool.launch_job:预热空闲子进程、隔离每个作业
你的应用livekit-agents/livekit/agents/voice/agent.py:38Agent:instructions/工具 + 可覆盖节点(stt_node/llm_node/tts_node,agent.py:421)
对话容器livekit-agents/livekit/agents/voice/agent_session.py:209AgentSession:start / generate_reply / interrupt / I/O 转发
循环大脑livekit-agents/livekit/agents/voice/agent_activity.py:163AgentActivity:调度、生成、打断、工具循环、handoff
回复生成livekit-agents/livekit/agents/voice/agent_activity.py:2663_pipeline_reply_task_impl:LLM→分段→TTS→播放的主体
工具执行livekit-agents/livekit/agents/voice/agent_activity.py:3087多步工具循环 + max_tool_steps + agent handoff
听觉子系统livekit-agents/livekit/agents/voice/audio_recognition.py:195AudioRecognition:VAD+STT+端点;EOU 在 _run_eou_detection(:1311)
端点策略livekit-agents/livekit/agents/voice/endpointing.py:10BaseEndpointing / DynamicEndpointing:min/max delay
端点默认值livekit-agents/livekit/agents/voice/turn.py:137EndpointingOptions 默认 min_delay=0.5、max_delay=3.0
语音句柄livekit-agents/livekit/agents/voice/speech_handle.py:25SpeechHandle:四个 Future + 三档优先级
生成原语livekit-agents/livekit/agents/voice/generation.py:118perform_llm_inference / perform_tts_inference / perform_tool_executions
最小示例examples/voice_agents/basic_agent.py:71一个可跑的完整语音 agent(STT/LLM/TTS + 工具)