拓冰网站定制 
LiveKit Agents 实时语音智能体框架深度实战指南【免费下载链接】agentsA framework for building realtime voice AI agents ️项目地址: https://gitcode.com/GitHub_Trending/agen/agentsLiveKit Agents 是构建运行在服务器上的实时语音 AI 智能体的 Python 框架。本文从源码层剖析 AgentServer、AgentSession 与可插拔 STT/LLM/TTS 管线的协作机制覆盖最小闭环搭建、多智能体交接、judge 测试与 console/dev/start 三种部署形态。语音智能体绕不开的三个硬问题误打断、首字延迟与非确定性文本对话产品很少需要回答用户到底说完了没有而语音产品必须在毫秒级做这个判断。从 livekit-agents/livekit/agents/voice/turn.py 暴露的TurnHandlingOptions、InterruptionOptions、PreemptiveGenerationOptions等类型可以看出框架把三个高频痛点做成了显式配置面误打断用户一句嗯啊或背景噪声就可能被 VAD语音活动检测判为新一轮发言粗暴切断正在播放的回复。框架提供两条防线——end-of-turn语义级轮次检测由 transformer 模型判断一句话是否结束降低 VAD 误判resume_false_interruption在确认为假打断后自动续播。首字延迟级联管线 STT→LLM→TTS 的延迟是三段相加。preemptive_generation允许在用户尚未完全说完时就让 LLM 预生成回复用一次投机换首字延迟。行为非确定性LLM 的工具调用顺序、空消息、措辞都无法硬编码断言需要事件链断言 模型裁判的测试机制兜底。此外还有一个进程级痛点一个 worker 要同时服务多个用户房间收到停止信号时不能粗暴截断在途语音会话。这三个问题分别对应本文第 5、7、8 节的机制拆解。架构全景一次通话在四层构件间的流动路径从 livekit-agents/livekit/agents/worker.py 的AgentServer约 L297与 livekit-agents/livekit/agents/init.py 的导出清单看一次实时通话的时序如下用户通过任意 LiveKit 客户端 SDK 或电话telephony集成加入 WebRTC 房间LiveKit 服务器把房间作为 job 派发给已注册的AgentServer进程AgentServer调用server.rtc_session()装饰的 entrypoint注入JobContextctx.room即目标房间entrypoint 内构造AgentSession挂载 vad/stt/llm/tts 管线与Agent指令 工具session.start(agent..., room...)后管线接管音频输入→识别→生成→播放工具调用、智能体交接、指标采集metrics都发生在会话生命周期内。用户/电话 ── WebRTC/SIP ── LiveKit Server房间与 job 分发 │ 派发 job ▼ AgentServerworker.py │ 调用 entrypoint(ctx: JobContext) ▼ Agentinstructions/tools◄── AgentSessionvad/stt/llm/tts 管线──► 房间音频仓库布局印证核心库 插件生态核心在 livekit-agents/livekit/agents/voice、llm、stt、tts、cli、inference等子包70 余个模型插件集中在 livekit-plugins/如livekit-plugins-openai、livekit-plugins-deepgram、livekit-plugins-cartesia、livekit-plugins-turn-detector。六个核心构件的职责与源码位置对照构件职责源码位置Agent带 instructions 的 LLM 应用含on_enter生命周期钩子与工具集livekit-agents/livekit/agents/voice/agent.pyAgentSession会话容器音频 I/O、识别、生成、播放的完整管道livekit-agents/livekit/agents/voice/agent_session.pyAgentServer主进程job 调度与分发为每个用户会话拉起 entrypointlivekit-agents/livekit/agents/worker.pyJobContext/JobRequest任务上下文ctx.room为智能体加入的房间livekit-agents/livekit/agents/job.pyfunction_tool/RunContext工具系统装饰器包装函数docstring 成为 LLM 可见说明livekit-agents/livekit/agents/llm/tool_context.pyRunResult/RunAssert测试断言链result.expect.next_event()逐事件校验livekit-agents/livekit/agents/voice/run_result.py从__init__.py的__all__清单还能看到一个机制细节mcp不在顶层直接导入而是通过模块级__getattr__懒加载避免使用 MCP 功能之外的场景强依赖 MCP 包——插件能力与核心库的耦合被刻意压到最低。三步跑通最小闭环装依赖、写入口、起会话第 1 步获取代码与依赖。核心库 常用模型插件一条命令装齐extras 决定安装哪些模型栈git clone https://gitcode.com/GitHub_Trending/agen/agents pip install livekit-agents[openai,deepgram,cartesia]第 2 步写工具与入口。工具函数用function_tool装饰context: RunContext由框架注入类型标注的参数是 LLM 填充的入参from livekit.agents import ( Agent, AgentServer, AgentSession, JobContext, RunContext, cli, function_tool, inference, ) function_tool async def lookup_weather(context: RunContext, location: str): Used to look up weather information. return {weather: sunny, temperature: 70}第 3 步注册 entrypoint 并启动。完整文件见 examples/voice_agents/basic_agent.pyserver AgentServer() server.rtc_session() async def entrypoint(ctx: JobContext): session AgentSession( vadinference.VAD(), sttinference.STT(deepgram/nova-3, languagemulti), llminference.LLM(google/gemma-4-31b-it), ttsinference.TTS(cartesia/sonic-3, voice9626c31c-bec5-4cca-baa8-f8ba9e84c8bc), ) agent Agent(instructionsYou are a friendly voice assistant., tools[lookup_weather]) await session.start(agentagent, roomctx.room) # 省略await session.generate_reply(instructionsgreet the user ...) 开场白 if __name__ __main__: cli.run_app(server)要点解读inference.STT/LLM/TTS是 LiveKit Inference 的统一模型 API参数是服务商/模型名字符串也可以换成deepgram.STT(modelnova-3)、openai.LLM(modelgpt-4.1-mini)等插件实例管线位置不变每当有新房间 job 被调度AgentServer都会为它调用一次entrypoint——它等价于 Web 框架里的请求处理器所以会话内状态必须自建不能依赖模块级全局变量session.generate_reply(instructions...)主动发起第一轮生成实现智能体先开口的开场白模式运行需要三个环境变量LIVEKIT_URL、LIVEKIT_API_KEY、LIVEKIT_API_SECRET指向 LiveKit Cloud 或自建服务器。叠加 TurnHandling 与文本变换把误打断和首字延迟调下来最小闭环跑通后工程化配置决定听感。basic_agent.py的 entrypoint 展示了TurnHandlingOptions的完整用法节选session AgentSession( sttinference.STT(deepgram/nova-3, languagemulti), llminference.LLM(openai/gpt-4.1-mini), ttsinference.TTS(cartesia/sonic-3, voice9626c31c-bec5-4cca-baa8-f8ba9e84c8bc), turn_handlingTurnHandlingOptions( interruption{ resume_false_interruption: True, false_interruption_timeout: 1.0, }, preemptive_generation{enabled: True, max_retries: 3}, ), aec_warmup_duration3.0, tts_text_transforms[filter_emoji, filter_markdown, text_transforms.replace({LiveKit: ˈ|l|aɪ|v|k|ɪ|t})], stt_context_options{keyterms: [LiveKit], keyterm_detection: {enabled: True, turn_interval: 1}}, )要点解读resume_false_interruption: Truefalse_interruption_timeout: 1.01 秒内判定为假打断如背景噪声就自动恢复播放对应 tests/test_false_interruption_resume.py 专门覆盖的路径preemptive_generation在等待 end-of-turn 的同时让 LLM 预生成max_retries: 3控制用户持续说话时的重试上限aec_warmup_duration: 3.0开播前 3 秒屏蔽打断留给客户端回声消除AEC校准防止自己的声音触发打断tts_text_transforms在合成前过滤 emoji/markdown并用ˈ|l|aɪ|v|k|ɪ|t这种发音标记替换特定品牌词的读法——直接作用于 TTS 输入文本stt_context_options的keyterm_detection每turn_interval1个轮次让 LLM 提取关键词注入 STT 上下文提高专有名词识别率。另外注意AgentSession的 stt/llm/tts 参数接受字符串模型标识如deepgram/nova-3框架会自动实例化对应插件——交接示例中就是这种写法。多智能体交接让工具返回值充当路由指令当一次会话需要分工信息收集→讲故事LiveKit Agents 的交接不靠显式状态机而是工具返回一个 Agent 实例即跳转。从 README 的节选示例看function_tool async def information_gathered(self, context: RunContext, name: str, location: str): Called when the user has provided the needed information. context.userdata.name name context.userdata.location location return StoryAgent(name, location), Lets start the story! class StoryAgent(Agent): def __init__(self, name: str, location: str) - None: super().__init__( instructionsf...The users name is {name}, from {location}, llmopenai.realtime.RealtimeModel(voiceecho), # 省略chat_ctxchat_ctx )要点解读information_gathered返回(StoryAgent(...), Lets start the story!)元组框架检测到返回值是Agent实例后在当前会话内切换活动智能体并播放元组第二项作为衔接话术——工具调用与路由复用同一条机制状态延续靠类型化 userdataentrypoint 里AgentSessionStoryData, ...)声明会话级共享数据context.userdata让前一个智能体把收集到的name/location传给后继StoryAgent构造时传llmopenai.realtime.RealtimeModel(voiceecho)在交接的同时把管线从STTLLMTTS 级联切到端到端 Realtime API并用chat_ctx携带既有对话历史on_enter钩子在智能体进入会话时触发self.session.generate_reply(...)是新角色开口的标准挂点。用模型当裁判run_result 断言链与 judge 评估LLM 非确定性意味着工具必须叫start_order可以硬断言但助手是否在追问想点什么只能语义判断。框架的做法是把两类断言放进同一条链README 测试示例pytest.mark.asyncio async def test_no_availability() - None: llm google.LLM() async with AgentSession(llmllm) as sess: await sess.start(MyAgent()) result await sess.run(user_inputHello, I need to place an order.) result.expect.skip_next_event_if(typemessage, roleassistant) result.expect.next_event().is_function_call(namestart_order) result.expect.next_event().is_function_call_output() await (result.expect.next_event() .is_message(roleassistant) .judge(llm, intentassistant should be asking the user what they would like))要点解读sess.run(user_input...)在进程内驱动识别→LLM→工具→合成完整管线返回RunResult——不需要真实房间或 worker 进程result.expect.next_event()逐事件消费is_function_call(name...)校验工具名、is_function_call_output()校验执行完成、is_message(roleassistant)校验回复存在.judge(llm, intent...)把语义正确性委托给另一个 LLM 裁判打分intent是自然语言判据skip_next_event_if处理模型可能先吐一条空 assistant 消息这类不确定性分支避免断言被随机事件顶掉。源码映射断言原语RunResult、RunAssert、EventAssert、mock_tools定义在 livekit-agents/livekit/agents/voice/run_result.py 并由根包导出进程内脱离 worker 做测试可用 livekit-agents/livekit/agents/testing.py 的fake_job_context上下文管理器注入假任务上下文。仓库自身 tests/ 目录的数百个测试即是示范覆盖了打断恢复test_false_interruption_resume.py、预生成死锁test_preemptive_pause_deadlock.py、LLM 回退test_llm_fallback.py等疑难路径。console、dev、start 三种运行形态的底层机制差异三个子命令都由cli.run_app(server)挂到 typer 应用上定义在 livekit-agents/livekit/agents/cli/_legacy.pyconsole约 L1652、start约 L1710、dev约 L1769形态命令依赖 LiveKit Server底层机制现状终端调试python myagent.py console否独立线程内server.run(devmodeTrue, unregisteredTrue)伪造 job 驱动 entrypoint已标注 deprecated源码提示改用lk agent console开发联调python myagent.py dev是_run_worker(devmodeTrue)向服务器注册已 deprecated进程内热重载已移除lk agent dev提供生产运行python myagent.py start是_run_worker(devmodeFalse)含 drain 与看门狗生产形态支持--drain-timeout从 livekit-agents/livekit/agents/cli/cli.py 的_run_worker约 L290与_legacy.py的_ConsoleWorker约 L1463看机制差异有三处console 为何不需要服务器_ConsoleWorker在独立线程/事件循环中调用server.run(devmodeTrue, unregisteredTrue)——unregisteredTrue表示不向 LiveKit 服务器注册监听worker_started事件后调用server.simulate_job(console-room, agent_identityconsole, fake_jobTrue)伪造一个任务entrypoint 照常执行。音频经AgentsConsole的acquire_io挂接本地设备支持--text文本模式与--record会话录制凭据解析start/dev的--url/--api-key/--api-secret选项都声明了envvarLIVEKIT_URL/LIVEKIT_API_KEY/LIVEKIT_API_SECRET命令行与环境变量等价--log-level对应LIVEKIT_LOG_LEVEL。_run_worker先server.update_options(ws_url..., api_key..., api_secret...)再server.run(devmode...)优雅退出时序start最关键首次 SIGINT/SIGTERM 只在事件循环上调度退出非 dev 模式执行server.drain()等待在途 job 结束--drain-timeout可配置等待秒数超时打印 drain timed out, forcing shutdown_EXIT_ESCALATION_TIMEOUT 3.0秒的看门狗在退出回调迟迟不执行事件循环被同步代码阻塞时升级为强制中断二次 CtrlC 直接强制退出。另外AgentServer生产默认值从ServerEnvOption可见空闲进程数为 CPU 核数、默认端口8081开发默认 0 即不监听。uv 依赖管理与 ruff 规范下的工程化流程仓库自身即按此流程开发二次开发团队可直接沿用见 makefile 与 pyproject.tomluv sync --all-extras --dev # 安装开发依赖 # 在 examples/.env 写入凭据模板见 examples/.env.example uv run examples/voice_agents/basic_agent.py dev uv run pytest --unit # 运行单元测试 uv run ruff format uv run ruff check --fix要点解读示例凭据统一放examples/.env字段覆盖 LiveKit Server 与所选模型服务商模板文件为 examples/.env.example各插件的集成测试需要对应 API 凭据由 CI 在维护者 PR 上自动运行见 .github/workflows/tests.yml本地只跑--unit即可API 文档用 pdoc 本地生成uv sync --all-extras --group docs后执行uv run --active pdoc --skip-errors --html --output-dirdocs livekit。示例库导航均在 examples/ 下带Dockerfile的可容器化部署示例说明路径Starter Agent语音对话调优过的起步智能体examples/voice_agents/basic_agent.pyMCP support一行接入 MCP 服务器工具examples/voice_agents/mcp/Multi-user transcriber输出房间内所有用户的转写examples/other/transcription/multi-user-transcriber.pyVideo avatarsTavus、Bithuman、LemonSlice 等数字人examples/avatar/Hotel receptionist酒店前台策略文档 评测场景examples/hotel_receptionist/Telephony电话 IVR 示例examples/telephony/适用边界与双许可证提示能力边界有几处需要明确Python CLI 的 dev/console 正在退役源码中两条命令都打印 deprecated 警告热重载能力整体转移到 LiveKit CLI 工具链的lk agent dev/lk agent console直接在 Python CLI 跑dev时不要依赖热重载语义级轮次检测是独立许可框架本体为 Apache-2.0见 LICENSE但 turn detection 模型单独适用 LiveKit Model License见 MODEL_LICENSE启用livekit-plugins-turn-detector商用前需分别确认两份条款测试不等于真实听感sess.run(user_input...)走文本输入驱动管线适合验证工具路由与事件序列回声、AEC 校准、真打断恢复这类音频路径必须用 console 真麦克风或真实房间验证MCP 为懒加载能力不接入 MCP 时核心库不产生相应依赖但接入后工具发现、调用链路完全交给 MCP 服务器侧实现。小结从终端验证到生产调度的同一条代码路径LiveKit Agents 把实时语音智能体拆成四层可独立替换的构件AgentServer管进程与 job 调度AgentSession管交互管道与音频 I/OAgent管指令、工具与生命周期STT/LLM/TTS/Realtime 模型作为可插拔零件按服务商/模型名字符串或插件实例混入。TurnHandlingOptions把误打断与首字延迟变成可调参数RunResult.expectjudge把非确定性行为变成可断言事件链而 console 的伪造 job、dev/start 的 drain 看门狗让同一份 entrypoint 代码在终端冒烟、客户端联调、生产调度三种形态间平移。上手顺序建议先用 console 跑通最小闭环再叠加 turn_handling 调听感最后用 judge 测试固化工具路由行为。【免费下载链接】agentsA framework for building realtime voice AI agents ️项目地址: https://gitcode.com/GitHub_Trending/agen/agents创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 LiveKit Agents 实时语音智能体框架深度实战指南
LiveKit Agents 实时语音智能体框架深度实战指南【免费下载链接】agentsA framework for building realtime voice AI agents ️项目地址: https://gitcode.com/GitHub_Trending/agen/agentsLiveKit Agents 是构建运行在服务器上的实时语音 AI 智能体的 Python 框架。本文从源码层剖析 AgentServer、AgentSession 与可插拔 STT/LLM/TTS 管线的协作机制覆盖最小闭环搭建、多智能体交接、judge 测试与 console/dev/start 三种部署形态。语音智能体绕不开的三个硬问题误打断、首字延迟与非确定性文本对话产品很少需要回答用户到底说完了没有而语音产品必须在毫秒级做这个判断。从 livekit-agents/livekit/agents/voice/turn.py 暴露的TurnHandlingOptions、InterruptionOptions、PreemptiveGenerationOptions等类型可以看出框架把三个高频痛点做成了显式配置面误打断用户一句嗯啊或背景噪声就可能被 VAD语音活动检测判为新一轮发言粗暴切断正在播放的回复。框架提供两条防线——end-of-turn语义级轮次检测由 transformer 模型判断一句话是否结束降低 VAD 误判resume_false_interruption在确认为假打断后自动续播。首字延迟级联管线 STT→LLM→TTS 的延迟是三段相加。preemptive_generation允许在用户尚未完全说完时就让 LLM 预生成回复用一次投机换首字延迟。行为非确定性LLM 的工具调用顺序、空消息、措辞都无法硬编码断言需要事件链断言 模型裁判的测试机制兜底。此外还有一个进程级痛点一个 worker 要同时服务多个用户房间收到停止信号时不能粗暴截断在途语音会话。这三个问题分别对应本文第 5、7、8 节的机制拆解。架构全景一次通话在四层构件间的流动路径从 livekit-agents/livekit/agents/worker.py 的AgentServer约 L297与 livekit-agents/livekit/agents/init.py 的导出清单看一次实时通话的时序如下用户通过任意 LiveKit 客户端 SDK 或电话telephony集成加入 WebRTC 房间LiveKit 服务器把房间作为 job 派发给已注册的AgentServer进程AgentServer调用server.rtc_session()装饰的 entrypoint注入JobContextctx.room即目标房间entrypoint 内构造AgentSession挂载 vad/stt/llm/tts 管线与Agent指令 工具session.start(agent..., room...)后管线接管音频输入→识别→生成→播放工具调用、智能体交接、指标采集metrics都发生在会话生命周期内。用户/电话 ── WebRTC/SIP ── LiveKit Server房间与 job 分发 │ 派发 job ▼ AgentServerworker.py │ 调用 entrypoint(ctx: JobContext) ▼ Agentinstructions/tools◄── AgentSessionvad/stt/llm/tts 管线──► 房间音频仓库布局印证核心库 插件生态核心在 livekit-agents/livekit/agents/voice、llm、stt、tts、cli、inference等子包70 余个模型插件集中在 livekit-plugins/如livekit-plugins-openai、livekit-plugins-deepgram、livekit-plugins-cartesia、livekit-plugins-turn-detector。六个核心构件的职责与源码位置对照构件职责源码位置Agent带 instructions 的 LLM 应用含on_enter生命周期钩子与工具集livekit-agents/livekit/agents/voice/agent.pyAgentSession会话容器音频 I/O、识别、生成、播放的完整管道livekit-agents/livekit/agents/voice/agent_session.pyAgentServer主进程job 调度与分发为每个用户会话拉起 entrypointlivekit-agents/livekit/agents/worker.pyJobContext/JobRequest任务上下文ctx.room为智能体加入的房间livekit-agents/livekit/agents/job.pyfunction_tool/RunContext工具系统装饰器包装函数docstring 成为 LLM 可见说明livekit-agents/livekit/agents/llm/tool_context.pyRunResult/RunAssert测试断言链result.expect.next_event()逐事件校验livekit-agents/livekit/agents/voice/run_result.py从__init__.py的__all__清单还能看到一个机制细节mcp不在顶层直接导入而是通过模块级__getattr__懒加载避免使用 MCP 功能之外的场景强依赖 MCP 包——插件能力与核心库的耦合被刻意压到最低。三步跑通最小闭环装依赖、写入口、起会话第 1 步获取代码与依赖。核心库 常用模型插件一条命令装齐extras 决定安装哪些模型栈git clone https://gitcode.com/GitHub_Trending/agen/agents pip install livekit-agents[openai,deepgram,cartesia]第 2 步写工具与入口。工具函数用function_tool装饰context: RunContext由框架注入类型标注的参数是 LLM 填充的入参from livekit.agents import ( Agent, AgentServer, AgentSession, JobContext, RunContext, cli, function_tool, inference, ) function_tool async def lookup_weather(context: RunContext, location: str): Used to look up weather information. return {weather: sunny, temperature: 70}第 3 步注册 entrypoint 并启动。完整文件见 examples/voice_agents/basic_agent.pyserver AgentServer() server.rtc_session() async def entrypoint(ctx: JobContext): session AgentSession( vadinference.VAD(), sttinference.STT(deepgram/nova-3, languagemulti), llminference.LLM(google/gemma-4-31b-it), ttsinference.TTS(cartesia/sonic-3, voice9626c31c-bec5-4cca-baa8-f8ba9e84c8bc), ) agent Agent(instructionsYou are a friendly voice assistant., tools[lookup_weather]) await session.start(agentagent, roomctx.room) # 省略await session.generate_reply(instructionsgreet the user ...) 开场白 if __name__ __main__: cli.run_app(server)要点解读inference.STT/LLM/TTS是 LiveKit Inference 的统一模型 API参数是服务商/模型名字符串也可以换成deepgram.STT(modelnova-3)、openai.LLM(modelgpt-4.1-mini)等插件实例管线位置不变每当有新房间 job 被调度AgentServer都会为它调用一次entrypoint——它等价于 Web 框架里的请求处理器所以会话内状态必须自建不能依赖模块级全局变量session.generate_reply(instructions...)主动发起第一轮生成实现智能体先开口的开场白模式运行需要三个环境变量LIVEKIT_URL、LIVEKIT_API_KEY、LIVEKIT_API_SECRET指向 LiveKit Cloud 或自建服务器。叠加 TurnHandling 与文本变换把误打断和首字延迟调下来最小闭环跑通后工程化配置决定听感。basic_agent.py的 entrypoint 展示了TurnHandlingOptions的完整用法节选session AgentSession( sttinference.STT(deepgram/nova-3, languagemulti), llminference.LLM(openai/gpt-4.1-mini), ttsinference.TTS(cartesia/sonic-3, voice9626c31c-bec5-4cca-baa8-f8ba9e84c8bc), turn_handlingTurnHandlingOptions( interruption{ resume_false_interruption: True, false_interruption_timeout: 1.0, }, preemptive_generation{enabled: True, max_retries: 3}, ), aec_warmup_duration3.0, tts_text_transforms[filter_emoji, filter_markdown, text_transforms.replace({LiveKit: ˈ|l|aɪ|v|k|ɪ|t})], stt_context_options{keyterms: [LiveKit], keyterm_detection: {enabled: True, turn_interval: 1}}, )要点解读resume_false_interruption: Truefalse_interruption_timeout: 1.01 秒内判定为假打断如背景噪声就自动恢复播放对应 tests/test_false_interruption_resume.py 专门覆盖的路径preemptive_generation在等待 end-of-turn 的同时让 LLM 预生成max_retries: 3控制用户持续说话时的重试上限aec_warmup_duration: 3.0开播前 3 秒屏蔽打断留给客户端回声消除AEC校准防止自己的声音触发打断tts_text_transforms在合成前过滤 emoji/markdown并用ˈ|l|aɪ|v|k|ɪ|t这种发音标记替换特定品牌词的读法——直接作用于 TTS 输入文本stt_context_options的keyterm_detection每turn_interval1个轮次让 LLM 提取关键词注入 STT 上下文提高专有名词识别率。另外注意AgentSession的 stt/llm/tts 参数接受字符串模型标识如deepgram/nova-3框架会自动实例化对应插件——交接示例中就是这种写法。多智能体交接让工具返回值充当路由指令当一次会话需要分工信息收集→讲故事LiveKit Agents 的交接不靠显式状态机而是工具返回一个 Agent 实例即跳转。从 README 的节选示例看function_tool async def information_gathered(self, context: RunContext, name: str, location: str): Called when the user has provided the needed information. context.userdata.name name context.userdata.location location return StoryAgent(name, location), Lets start the story! class StoryAgent(Agent): def __init__(self, name: str, location: str) - None: super().__init__( instructionsf...The users name is {name}, from {location}, llmopenai.realtime.RealtimeModel(voiceecho), # 省略chat_ctxchat_ctx )要点解读information_gathered返回(StoryAgent(...), Lets start the story!)元组框架检测到返回值是Agent实例后在当前会话内切换活动智能体并播放元组第二项作为衔接话术——工具调用与路由复用同一条机制状态延续靠类型化 userdataentrypoint 里AgentSessionStoryData, ...)声明会话级共享数据context.userdata让前一个智能体把收集到的name/location传给后继StoryAgent构造时传llmopenai.realtime.RealtimeModel(voiceecho)在交接的同时把管线从STTLLMTTS 级联切到端到端 Realtime API并用chat_ctx携带既有对话历史on_enter钩子在智能体进入会话时触发self.session.generate_reply(...)是新角色开口的标准挂点。用模型当裁判run_result 断言链与 judge 评估LLM 非确定性意味着工具必须叫start_order可以硬断言但助手是否在追问想点什么只能语义判断。框架的做法是把两类断言放进同一条链README 测试示例pytest.mark.asyncio async def test_no_availability() - None: llm google.LLM() async with AgentSession(llmllm) as sess: await sess.start(MyAgent()) result await sess.run(user_inputHello, I need to place an order.) result.expect.skip_next_event_if(typemessage, roleassistant) result.expect.next_event().is_function_call(namestart_order) result.expect.next_event().is_function_call_output() await (result.expect.next_event() .is_message(roleassistant) .judge(llm, intentassistant should be asking the user what they would like))要点解读sess.run(user_input...)在进程内驱动识别→LLM→工具→合成完整管线返回RunResult——不需要真实房间或 worker 进程result.expect.next_event()逐事件消费is_function_call(name...)校验工具名、is_function_call_output()校验执行完成、is_message(roleassistant)校验回复存在.judge(llm, intent...)把语义正确性委托给另一个 LLM 裁判打分intent是自然语言判据skip_next_event_if处理模型可能先吐一条空 assistant 消息这类不确定性分支避免断言被随机事件顶掉。源码映射断言原语RunResult、RunAssert、EventAssert、mock_tools定义在 livekit-agents/livekit/agents/voice/run_result.py 并由根包导出进程内脱离 worker 做测试可用 livekit-agents/livekit/agents/testing.py 的fake_job_context上下文管理器注入假任务上下文。仓库自身 tests/ 目录的数百个测试即是示范覆盖了打断恢复test_false_interruption_resume.py、预生成死锁test_preemptive_pause_deadlock.py、LLM 回退test_llm_fallback.py等疑难路径。console、dev、start 三种运行形态的底层机制差异三个子命令都由cli.run_app(server)挂到 typer 应用上定义在 livekit-agents/livekit/agents/cli/_legacy.pyconsole约 L1652、start约 L1710、dev约 L1769形态命令依赖 LiveKit Server底层机制现状终端调试python myagent.py console否独立线程内server.run(devmodeTrue, unregisteredTrue)伪造 job 驱动 entrypoint已标注 deprecated源码提示改用lk agent console开发联调python myagent.py dev是_run_worker(devmodeTrue)向服务器注册已 deprecated进程内热重载已移除lk agent dev提供生产运行python myagent.py start是_run_worker(devmodeFalse)含 drain 与看门狗生产形态支持--drain-timeout从 livekit-agents/livekit/agents/cli/cli.py 的_run_worker约 L290与_legacy.py的_ConsoleWorker约 L1463看机制差异有三处console 为何不需要服务器_ConsoleWorker在独立线程/事件循环中调用server.run(devmodeTrue, unregisteredTrue)——unregisteredTrue表示不向 LiveKit 服务器注册监听worker_started事件后调用server.simulate_job(console-room, agent_identityconsole, fake_jobTrue)伪造一个任务entrypoint 照常执行。音频经AgentsConsole的acquire_io挂接本地设备支持--text文本模式与--record会话录制凭据解析start/dev的--url/--api-key/--api-secret选项都声明了envvarLIVEKIT_URL/LIVEKIT_API_KEY/LIVEKIT_API_SECRET命令行与环境变量等价--log-level对应LIVEKIT_LOG_LEVEL。_run_worker先server.update_options(ws_url..., api_key..., api_secret...)再server.run(devmode...)优雅退出时序start最关键首次 SIGINT/SIGTERM 只在事件循环上调度退出非 dev 模式执行server.drain()等待在途 job 结束--drain-timeout可配置等待秒数超时打印 drain timed out, forcing shutdown_EXIT_ESCALATION_TIMEOUT 3.0秒的看门狗在退出回调迟迟不执行事件循环被同步代码阻塞时升级为强制中断二次 CtrlC 直接强制退出。另外AgentServer生产默认值从ServerEnvOption可见空闲进程数为 CPU 核数、默认端口8081开发默认 0 即不监听。uv 依赖管理与 ruff 规范下的工程化流程仓库自身即按此流程开发二次开发团队可直接沿用见 makefile 与 pyproject.tomluv sync --all-extras --dev # 安装开发依赖 # 在 examples/.env 写入凭据模板见 examples/.env.example uv run examples/voice_agents/basic_agent.py dev uv run pytest --unit # 运行单元测试 uv run ruff format uv run ruff check --fix要点解读示例凭据统一放examples/.env字段覆盖 LiveKit Server 与所选模型服务商模板文件为 examples/.env.example各插件的集成测试需要对应 API 凭据由 CI 在维护者 PR 上自动运行见 .github/workflows/tests.yml本地只跑--unit即可API 文档用 pdoc 本地生成uv sync --all-extras --group docs后执行uv run --active pdoc --skip-errors --html --output-dirdocs livekit。示例库导航均在 examples/ 下带Dockerfile的可容器化部署示例说明路径Starter Agent语音对话调优过的起步智能体examples/voice_agents/basic_agent.pyMCP support一行接入 MCP 服务器工具examples/voice_agents/mcp/Multi-user transcriber输出房间内所有用户的转写examples/other/transcription/multi-user-transcriber.pyVideo avatarsTavus、Bithuman、LemonSlice 等数字人examples/avatar/Hotel receptionist酒店前台策略文档 评测场景examples/hotel_receptionist/Telephony电话 IVR 示例examples/telephony/适用边界与双许可证提示能力边界有几处需要明确Python CLI 的 dev/console 正在退役源码中两条命令都打印 deprecated 警告热重载能力整体转移到 LiveKit CLI 工具链的lk agent dev/lk agent console直接在 Python CLI 跑dev时不要依赖热重载语义级轮次检测是独立许可框架本体为 Apache-2.0见 LICENSE但 turn detection 模型单独适用 LiveKit Model License见 MODEL_LICENSE启用livekit-plugins-turn-detector商用前需分别确认两份条款测试不等于真实听感sess.run(user_input...)走文本输入驱动管线适合验证工具路由与事件序列回声、AEC 校准、真打断恢复这类音频路径必须用 console 真麦克风或真实房间验证MCP 为懒加载能力不接入 MCP 时核心库不产生相应依赖但接入后工具发现、调用链路完全交给 MCP 服务器侧实现。小结从终端验证到生产调度的同一条代码路径LiveKit Agents 把实时语音智能体拆成四层可独立替换的构件AgentServer管进程与 job 调度AgentSession管交互管道与音频 I/OAgent管指令、工具与生命周期STT/LLM/TTS/Realtime 模型作为可插拔零件按服务商/模型名字符串或插件实例混入。TurnHandlingOptions把误打断与首字延迟变成可调参数RunResult.expectjudge把非确定性行为变成可断言事件链而 console 的伪造 job、dev/start 的 drain 看门狗让同一份 entrypoint 代码在终端冒烟、客户端联调、生产调度三种形态间平移。上手顺序建议先用 console 跑通最小闭环再叠加 turn_handling 调听感最后用 judge 测试固化工具路由行为。【免费下载链接】agentsA framework for building realtime voice AI agents ️项目地址: https://gitcode.com/GitHub_Trending/agen/agents创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考