尧图网页设计 
LiveKit Agents如何快速搭建可上线的实时语音智能体【免费下载链接】agentsA framework for building realtime voice AI agents ️项目地址: https://gitcode.com/GitHub_Trending/agen/agentsLiveKit Agents 是一个在服务器端运行实时语音 AI 智能体的 Python 框架把音频收发、轮次判定、模型调用与任务调度统一进同一条管道。本文按「安装 → 终端里跑通最小示例 → 接入真实客户端 → 生产部署」这条路径展开顺带拆开每一层机制对应的源码位置。为什么值得先看它痛点自己拼装 STT、LLM、TTS 时接口不统一降级与超时逻辑全靠手写。解法70 余个模型插件集中在 livekit-plugins/ 下且实现同一套接口另外inference子包提供统一入口用deepgram/nova-3这样的字符串标识就能切模型管线部件可任意混搭。痛点用户说话中途停顿智能体容易抢话或者该接话时不接。解法框架内置基于 transformer 模型的语义级轮次检测来判断 end-of-turn配合从 voice/turn.py 导出的TurnHandlingOptions、InterruptionOptions、PreemptiveGenerationOptions等开关可以在会话构造时声明式调优。痛点LLM 输出非确定传统断言写不住行为回归。解法内置测试设施——RunResult.expect提供链式事件断言语义层面的正确性交给另一个 LLM 以 judge 方式打分原语定义在 voice/run_result.py。痛点多用户并发下单个会话内存泄漏或崩溃会拖垮整个服务进程。解法核心调度见 worker.pyAgentServer以独立进程Linux 下默认 forkserver 上下文见该文件 L259-261承载每个 job并提供负载阈值、内存上限、优雅排空等进程级保护。五分钟跑起来安装核心库与三个常用插件pip install livekit-agents[openai,deepgram,cartesia]把下面的最小示例存为myagent.py模型标识沿用了 README.md 中的推荐组合换任意已安装插件的等价写法即可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} server 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 built by LiveKit., tools[lookup_weather], ) await session.start(agentagent, roomctx.room) await session.generate_reply(instructionsgreet the user and ask about their day) if __name__ __main__: cli.run_app(server)要点解读function_tool把异步函数包装成FunctionTooldocstring 是 LLM 可见的工具说明类型标注的参数成为 LLM 填的入参context: RunContext由框架注入可读写会话状态server.rtc_session()注册的协程是每会话入口每当一个房间任务被调度AgentServer就调用它并注入JobContext其中ctx.room是智能体加入的 WebRTC 房间AgentSession的构造参数即模型管线vad/stt/llm/tts任意混搭也可传字符串模型标识由框架自动实例化session.start(agent..., room...)挂载智能体随后的generate_reply(instructions...)让智能体主动开口形成先打招呼的开场白。所需环境变量指向 LiveKit Cloud 或自建 LiveKit 服务器export LIVEKIT_URL... export LIVEKIT_API_KEY... export LIVEKIT_API_SECRET...运行python myagent.py console预期现象终端接入本地麦克风与扬声器智能体先主动问候并询问近况你回答天气相关问题时工具调用会返回{weather: sunny, temperature: 70}并被复述出来。console不依赖任何外部调度适合冒烟验证注意版本现状该子命令已标记 deprecated源码提示见 cli/_legacy.py L1534后续将迁移到 LiveKit CLI 的lk agent console。概念地图抽象职责源码位置Agent指令、工具、生命周期钩子如on_entervoice/agent.pyAgentSession会话容器音频 I/O、识别、生成、播放管道voice/agent_session.pyentrypoint每会话入口server.rtc_session()注册worker.pyAgentServerworker 进程、job 调度、连接 LiveKit 服务器worker.py L297 起RunResult测试时的链式事件断言与 judgevoice/run_result.py包顶层init.py 直接导出AgentServer、AgentSession、function_tool、ChatContext、RunResult、TurnHandlingOptions等符号说明调度 会话 工具 测试四块都是一等公民mcp模块则通过__getattr__懒加载避免强依赖。进阶场景场景一多智能体交接场景背景一次会话需要角色分工例如先收集用户姓名与籍贯再由讲故事角色接管。框架的做法是让工具返回值直接携带新智能体。关键代码摘自 README.md 的 handoff 片段完整上下文见 examples/voice_agents/class IntroAgent(Agent): # … 构造省略instructions 要求收集 name 与 location function_tool async def information_gathered( self, context: RunContext, name: str, location: str, ): 信息收集完毕时由 LLM 触发。 context.userdata.name name context.userdata.location location story_agent StoryAgent(name, location) return story_agent, Lets start the story!机制拆解工具返回(Agent 实例, 衔接话术)元组框架检测到返回值是Agent后在同一会话内切换活动智能体并播报衔接语AgentSession[StoryData]用类型参数声明会话级共享数据context.userdata让前一个角色把状态传给后继StoryAgent构造时可传入llmopenai.realtime.RealtimeModel(voiceecho)在交接的同时把管线从STTLLMTTS 级联整体换成端到端 Realtime API。踩坑提示切换后新角色的第一轮回复通常由它的on_enter触发generate_reply而对话历史不会自动带过去需要显式传入chat_ctx保留上下文否则新角色会失忆。场景二用一次测试驱动完整管线场景背景LLM 行为不确定硬编码字符串断言会脆弱。框架把一次sess.run(user_input...)的产物建模为事件流断言按事件逐个推进。关键代码摘自 README.md Testing 一节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在进程内驱动 识别→LLM→工具→合成 的完整管线并返回RunResultL98RunAssertL317与EventAssertL659实现链式断言is_function_call校验工具名L672judgeL953把无法硬编码的语义判断交给裁判模型。若需要脱离 worker 进程、在真实房间里做进程内测试testing.py 的fake_job_context会注入一个 fake 的JobContext使get_job_context()的行为与真实任务一致。踩坑提示skip_next_event_if的存在说明作者预期模型可能先吐一个空消息——写断言时给不确定性分支留口子比追求永远精确的序列更实用judge 断言需要第二个可用 LLM 的凭据。场景三轮次控制与打断行为调优场景背景误打断用户只是换气智能体却抢答和抢话是语音体验的头号投诉。关键代码examples/voice_agents/basic_agent.py 在AgentSession上挂了完整的turn_handlingTurnHandlingOptions(...)配置是仓库内现成的参考实现可直接对照调参。机制拆解可调项全部从 voice/turn.py 导出包括InterruptionOptions是否允许打断、PreemptiveGenerationOptions用户还没说完时预生成回复以降低首字延迟、EndpointingOptions端点判定AgentFalseInterruptionEvent等事件见init.py 的事件导出段可供你监听误打断并恢复播放。踩坑提示这些旋钮互相牵制建议先在console模式人工听感回归再用场景二的方式把关键对话固化成测试避免调好一个指标、崩掉另一个。从本地验证到上线三种运行形态都由cli.run_app(server)暴露的子命令驱动但源码路径已分叉值得分别看。console本地终端走 cli/cli.py 的_run_tcp_consoleL209 起——音频经 TCP 挂接终端worker 以unregisteredTrue运行、不向服务器注册再用server.simulate_job(console-room, ..., fake_jobTrue)L241伪造一个任务驱动 entrypoint这就是它不需要外部服务器的原因。适用边界单人冒烟验证切换成本为零。版本现状已标记 deprecated。dev联调与start同走_run_workerL290 起区别仅在devmodeTrue日志默认 DEBUG、负载阈值关闭、不启动生产侧的排空逻辑。需要设置前文的三个环境变量任何 LiveKit 客户端 SDK 或电话集成都可作为对端。版本现状dev子命令本身已 deprecatedcli/_legacy.py L1811且进程内热重载已移除提示改用lk agent dev获得热重载L1819。start生产同一个_run_worker但退出路径完全不同。收到 SIGINT/SIGTERM 后首次信号只在事件循环上调度退出非 dev 模式会执行server.drain()等待在途任务结束L437-441start提供--drain-timeout配置等待时长若循环被同步代码卡住_EXIT_ESCALATION_TIMEOUT 3.0秒的看门狗L33会升级为强制中断二次 CtrlC 则直接os._exit(1)。生产侧默认值由ServerEnvOption按模式分叉见 worker.py L148-149、L207-248日志 INFO、健康检查端口 8081、负载阈值 0.7 超限即摘流、预热空闲进程池。法务上框架本体为 Apache-2.0见 LICENSE轮次检测模型单独适用 MODEL_LICENSE启用语义轮检时商用前需分别确认。如何确认它真的工作正常跑仓库自带的单元测试在仓库根目录uv run pytest --unit集成测试需要各模型服务商的 API 凭据由 CI 针对维护者 PR 自动执行细节见 .github/workflows/tests.yml。tests/ 目录下的test_agent_session.py、test_false_interruption_resume.py、test_preemptive_pause_deadlock.py等文件本身就是断言写法的示范。针对你自己的智能体建议逐项核对单元测试通过且新增对话路径都补了sess.run用例console下智能体能主动打招呼双向对话不卡死工具触发时能在日志/事件流中看到FunctionCall与执行完成事件打断后智能体停止播报误打断如resume_false_interruption配置能恢复播放start模式下 CtrlC 后进行中的会话被排空而非被截断二次 CtrlC 才强退日志级别符合预期dev 默认 DEBUG、prod 默认 INFO必要时经--log-level/LIVEKIT_LOG_LEVEL覆盖收尾各层构件的职责图景把整个系统看成分层拼图AgentServer是最底层的进程与调度面负责连接 LiveKit 服务器、接受 job 分配、为每个会话拉起隔离进程AgentSession是中间的管道面串联 VAD、识别、生成与播放并持有音频 I/O 与转写同步Agent是最上层的语义面声明指令、工具与生命周期钩子STT/LLM/TTS/Realtime 模型则是可以逐格替换的零件通过统一接口或字符串标识插入。工具系统让返回一个新 Agent成为一等操作测试设施则把这套管道压缩成进程内可断言的事件流。console、dev、start只是同一份代码在三档运行约束下的投影——从终端里的一次冒烟到生产环境带排空与负载摘流的长期服务切换的是部署形态而不是代码结构。【免费下载链接】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 框架把音频收发、轮次判定、模型调用与任务调度统一进同一条管道。本文按「安装 → 终端里跑通最小示例 → 接入真实客户端 → 生产部署」这条路径展开顺带拆开每一层机制对应的源码位置。为什么值得先看它痛点自己拼装 STT、LLM、TTS 时接口不统一降级与超时逻辑全靠手写。解法70 余个模型插件集中在 livekit-plugins/ 下且实现同一套接口另外inference子包提供统一入口用deepgram/nova-3这样的字符串标识就能切模型管线部件可任意混搭。痛点用户说话中途停顿智能体容易抢话或者该接话时不接。解法框架内置基于 transformer 模型的语义级轮次检测来判断 end-of-turn配合从 voice/turn.py 导出的TurnHandlingOptions、InterruptionOptions、PreemptiveGenerationOptions等开关可以在会话构造时声明式调优。痛点LLM 输出非确定传统断言写不住行为回归。解法内置测试设施——RunResult.expect提供链式事件断言语义层面的正确性交给另一个 LLM 以 judge 方式打分原语定义在 voice/run_result.py。痛点多用户并发下单个会话内存泄漏或崩溃会拖垮整个服务进程。解法核心调度见 worker.pyAgentServer以独立进程Linux 下默认 forkserver 上下文见该文件 L259-261承载每个 job并提供负载阈值、内存上限、优雅排空等进程级保护。五分钟跑起来安装核心库与三个常用插件pip install livekit-agents[openai,deepgram,cartesia]把下面的最小示例存为myagent.py模型标识沿用了 README.md 中的推荐组合换任意已安装插件的等价写法即可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} server 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 built by LiveKit., tools[lookup_weather], ) await session.start(agentagent, roomctx.room) await session.generate_reply(instructionsgreet the user and ask about their day) if __name__ __main__: cli.run_app(server)要点解读function_tool把异步函数包装成FunctionTooldocstring 是 LLM 可见的工具说明类型标注的参数成为 LLM 填的入参context: RunContext由框架注入可读写会话状态server.rtc_session()注册的协程是每会话入口每当一个房间任务被调度AgentServer就调用它并注入JobContext其中ctx.room是智能体加入的 WebRTC 房间AgentSession的构造参数即模型管线vad/stt/llm/tts任意混搭也可传字符串模型标识由框架自动实例化session.start(agent..., room...)挂载智能体随后的generate_reply(instructions...)让智能体主动开口形成先打招呼的开场白。所需环境变量指向 LiveKit Cloud 或自建 LiveKit 服务器export LIVEKIT_URL... export LIVEKIT_API_KEY... export LIVEKIT_API_SECRET...运行python myagent.py console预期现象终端接入本地麦克风与扬声器智能体先主动问候并询问近况你回答天气相关问题时工具调用会返回{weather: sunny, temperature: 70}并被复述出来。console不依赖任何外部调度适合冒烟验证注意版本现状该子命令已标记 deprecated源码提示见 cli/_legacy.py L1534后续将迁移到 LiveKit CLI 的lk agent console。概念地图抽象职责源码位置Agent指令、工具、生命周期钩子如on_entervoice/agent.pyAgentSession会话容器音频 I/O、识别、生成、播放管道voice/agent_session.pyentrypoint每会话入口server.rtc_session()注册worker.pyAgentServerworker 进程、job 调度、连接 LiveKit 服务器worker.py L297 起RunResult测试时的链式事件断言与 judgevoice/run_result.py包顶层init.py 直接导出AgentServer、AgentSession、function_tool、ChatContext、RunResult、TurnHandlingOptions等符号说明调度 会话 工具 测试四块都是一等公民mcp模块则通过__getattr__懒加载避免强依赖。进阶场景场景一多智能体交接场景背景一次会话需要角色分工例如先收集用户姓名与籍贯再由讲故事角色接管。框架的做法是让工具返回值直接携带新智能体。关键代码摘自 README.md 的 handoff 片段完整上下文见 examples/voice_agents/class IntroAgent(Agent): # … 构造省略instructions 要求收集 name 与 location function_tool async def information_gathered( self, context: RunContext, name: str, location: str, ): 信息收集完毕时由 LLM 触发。 context.userdata.name name context.userdata.location location story_agent StoryAgent(name, location) return story_agent, Lets start the story!机制拆解工具返回(Agent 实例, 衔接话术)元组框架检测到返回值是Agent后在同一会话内切换活动智能体并播报衔接语AgentSession[StoryData]用类型参数声明会话级共享数据context.userdata让前一个角色把状态传给后继StoryAgent构造时可传入llmopenai.realtime.RealtimeModel(voiceecho)在交接的同时把管线从STTLLMTTS 级联整体换成端到端 Realtime API。踩坑提示切换后新角色的第一轮回复通常由它的on_enter触发generate_reply而对话历史不会自动带过去需要显式传入chat_ctx保留上下文否则新角色会失忆。场景二用一次测试驱动完整管线场景背景LLM 行为不确定硬编码字符串断言会脆弱。框架把一次sess.run(user_input...)的产物建模为事件流断言按事件逐个推进。关键代码摘自 README.md Testing 一节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在进程内驱动 识别→LLM→工具→合成 的完整管线并返回RunResultL98RunAssertL317与EventAssertL659实现链式断言is_function_call校验工具名L672judgeL953把无法硬编码的语义判断交给裁判模型。若需要脱离 worker 进程、在真实房间里做进程内测试testing.py 的fake_job_context会注入一个 fake 的JobContext使get_job_context()的行为与真实任务一致。踩坑提示skip_next_event_if的存在说明作者预期模型可能先吐一个空消息——写断言时给不确定性分支留口子比追求永远精确的序列更实用judge 断言需要第二个可用 LLM 的凭据。场景三轮次控制与打断行为调优场景背景误打断用户只是换气智能体却抢答和抢话是语音体验的头号投诉。关键代码examples/voice_agents/basic_agent.py 在AgentSession上挂了完整的turn_handlingTurnHandlingOptions(...)配置是仓库内现成的参考实现可直接对照调参。机制拆解可调项全部从 voice/turn.py 导出包括InterruptionOptions是否允许打断、PreemptiveGenerationOptions用户还没说完时预生成回复以降低首字延迟、EndpointingOptions端点判定AgentFalseInterruptionEvent等事件见init.py 的事件导出段可供你监听误打断并恢复播放。踩坑提示这些旋钮互相牵制建议先在console模式人工听感回归再用场景二的方式把关键对话固化成测试避免调好一个指标、崩掉另一个。从本地验证到上线三种运行形态都由cli.run_app(server)暴露的子命令驱动但源码路径已分叉值得分别看。console本地终端走 cli/cli.py 的_run_tcp_consoleL209 起——音频经 TCP 挂接终端worker 以unregisteredTrue运行、不向服务器注册再用server.simulate_job(console-room, ..., fake_jobTrue)L241伪造一个任务驱动 entrypoint这就是它不需要外部服务器的原因。适用边界单人冒烟验证切换成本为零。版本现状已标记 deprecated。dev联调与start同走_run_workerL290 起区别仅在devmodeTrue日志默认 DEBUG、负载阈值关闭、不启动生产侧的排空逻辑。需要设置前文的三个环境变量任何 LiveKit 客户端 SDK 或电话集成都可作为对端。版本现状dev子命令本身已 deprecatedcli/_legacy.py L1811且进程内热重载已移除提示改用lk agent dev获得热重载L1819。start生产同一个_run_worker但退出路径完全不同。收到 SIGINT/SIGTERM 后首次信号只在事件循环上调度退出非 dev 模式会执行server.drain()等待在途任务结束L437-441start提供--drain-timeout配置等待时长若循环被同步代码卡住_EXIT_ESCALATION_TIMEOUT 3.0秒的看门狗L33会升级为强制中断二次 CtrlC 则直接os._exit(1)。生产侧默认值由ServerEnvOption按模式分叉见 worker.py L148-149、L207-248日志 INFO、健康检查端口 8081、负载阈值 0.7 超限即摘流、预热空闲进程池。法务上框架本体为 Apache-2.0见 LICENSE轮次检测模型单独适用 MODEL_LICENSE启用语义轮检时商用前需分别确认。如何确认它真的工作正常跑仓库自带的单元测试在仓库根目录uv run pytest --unit集成测试需要各模型服务商的 API 凭据由 CI 针对维护者 PR 自动执行细节见 .github/workflows/tests.yml。tests/ 目录下的test_agent_session.py、test_false_interruption_resume.py、test_preemptive_pause_deadlock.py等文件本身就是断言写法的示范。针对你自己的智能体建议逐项核对单元测试通过且新增对话路径都补了sess.run用例console下智能体能主动打招呼双向对话不卡死工具触发时能在日志/事件流中看到FunctionCall与执行完成事件打断后智能体停止播报误打断如resume_false_interruption配置能恢复播放start模式下 CtrlC 后进行中的会话被排空而非被截断二次 CtrlC 才强退日志级别符合预期dev 默认 DEBUG、prod 默认 INFO必要时经--log-level/LIVEKIT_LOG_LEVEL覆盖收尾各层构件的职责图景把整个系统看成分层拼图AgentServer是最底层的进程与调度面负责连接 LiveKit 服务器、接受 job 分配、为每个会话拉起隔离进程AgentSession是中间的管道面串联 VAD、识别、生成与播放并持有音频 I/O 与转写同步Agent是最上层的语义面声明指令、工具与生命周期钩子STT/LLM/TTS/Realtime 模型则是可以逐格替换的零件通过统一接口或字符串标识插入。工具系统让返回一个新 Agent成为一等操作测试设施则把这套管道压缩成进程内可断言的事件流。console、dev、start只是同一份代码在三档运行约束下的投影——从终端里的一次冒烟到生产环境带排空与负载摘流的长期服务切换的是部署形态而不是代码结构。【免费下载链接】agentsA framework for building realtime voice AI agents ️项目地址: https://gitcode.com/GitHub_Trending/agen/agents创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考