ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

LiveKit Agents 实时语音智能体框架实战:从最小示例到生产部署的完整教程

LiveKit Agents 实时语音智能体框架实战:从最小示例到生产部署的完整教程 LiveKit Agents 实时语音智能体框架实战从最小示例到生产部署的完整教程【免费下载链接】agentsA framework for building realtime voice AI agents ️项目地址: https://gitcode.com/GitHub_Trending/agen/agentsLiveKit Agents 是一个开源的实时语音智能体框架Agent Framework目标是把听得见、说得出、看得见的对话式 AI 参与者部署在你自己的服务器上。它把任务调度AgentServer、会话管道AgentSession、指令与工具Agent以及可插拔的 STT/LLM/TTS 模型拆成四层让你用同一套代码完成从终端调试、客户端联调到生产运行的全过程。LiveKit Agents 是什么运行在服务器上的实时参与者官方对它的定位是构建realtime, programmable participants实时、可编程参与者。相比只能聊天窗口的文本机器人它解决的是语音交互链路问题用户的音频进来经过 VAD语音活动检测判断是否说话、STT语音识别转文字、LLM大语言模型推理、TTS语音合成再播出去全程在房间内实时完成。框架的能力面覆盖了语音应用的常见刚需模型自由混搭STT/LLM/TTS 与 Realtime API 之间任意组合背后是 livekit-plugins/ 下 70 余个模型服务商插件OpenAI、Deepgram、Cartesia、ElevenLabs 等内置任务调度AgentServer 负责把每个用户会话job分发给智能体电话与 WebRTC 双通道既能对接 LiveKit 客户端 SDK也能走 telephony 栈打接电话语义级轮次检测用 transformer 模型判断用户是否说完这句话减少误打断MCP 原生支持一行代码挂上 MCP 服务器提供的工具内置测试框架断言 LLM 评审judge应对 LLM 输出不确定的问题。核心库位于 livekit-agents/livekit/agents/含voice、llm、stt、tts、cli、inference等子包插件全部在 livekit-plugins/核心 插件的目录结构一目了然。跑通第一次对话安装、环境变量与最小示例安装只需一条命令方括号里的 extras 决定附带哪些模型插件pip install livekit-agents[openai,deepgram,cartesia]跑智能体前先准备三个环境变量指向 LiveKit Cloud 或自建 LiveKit ServerLIVEKIT_URLLIVEKIT_API_KEYLIVEKIT_API_SECRET下面是能跑起来的最小闭环一个会查天气的语音助手。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把普通协程变成 LLM 可调用工具docstring 即工具说明类型标注的参数由 LLM 填充context: RunContext是框架注入的运行期上下文server.rtc_session()装饰的entrypoint相当于 Web 服务里的请求处理器每来一个房间任务就会被调用一次ctx.room就是智能体要加入的房间最后用generate_reply主动开口实现智能体先打招呼。cli.run_app(server)则把console/dev/start三个子命令挂到脚本上。四个概念对照源码Agent在 voice/agent.pyAgentSession在 voice/agent_session.pyAgentServer在 worker.py它们都由根包 livekit-agents/livekit/agents/init.py 顶层导出mcp模块则是懒加载以避免强依赖。模型管线怎么换Inference 统一入口与直接插插件AgentSession的构造参数就是模型管线而管线里每个位置都是可插拔的。上例用的是inference.*这一路通过 LiveKit Cloud 的统一 API 访问不同模型好处是不用为每家服务商分别管 key 和 SDK坏处是 Realtime 模型如openai.realtime.RealtimeModel不在 Inference 支持范围内必须直接用对应插件。换成直接调用服务商插件只改构造函数from livekit.plugins import deepgram, openai, cartesia session AgentSession( sttdeepgram.STT(modelnova-3), llmopenai.LLM(modelgpt-4.1-mini), ttscartesia.TTS(modelsonic-3, voice9626c31c-bec5-4cca-baa8-f8ba9e84c8bc), )更值得注意的是工程化旋钮。examples/voice_agents/basic_agent.py 在最小骨架之上演示了一组针对真实语音体验的配置turn_handling里的resume_false_interruption误打断后自动续播和preemptive_generation预判用户说完前让 LLM 预生成压低首字延迟、aec_warmup_duration开播前几秒屏蔽打断留给客户端回声消除校准、tts_text_transforms过滤 emoji/markdown、纠正特定发音以及stt_context_options关键词注入把高频术语喂给 STT 上下文提高专有名词识别率。这些正是减少打断语义轮次检测特性在 API 上的具体落点。多智能体交接工具返回值换角色userdata 传状态一次会话常需要分工前一段收集信息后一段执行任务。LiveKit Agents 的交接机制很直接——工具返回一个 Agent 实例框架就完成切换。class IntroAgent(Agent): def __init__(self) - None: super().__init__( instructionsYou are a story teller. Gather the users name and where they are from. ) async def on_enter(self): self.session.generate_reply(instructionsgreet the user and gather information) 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__( instructionsfYou are a storyteller. The users name is {name}, ffrom {location}, llmopenai.realtime.RealtimeModel(voiceecho), chat_ctxchat_ctx, ) async def on_enter(self): self.session.generate_reply()这里藏着两个关键设计。第一information_gathered返回(新智能体, 衔接话术)元组框架识别到工具输出是Agent后会在同一会话内切换活动智能体并播出 Lets start the story!用户无感。第二状态通过userdata延续entrypoint里用AgentSessionStoryData, ...)声明会话级共享数据前一个智能体写context.userdata后一个智能体直接读对话历史则靠显式携带的chat_ctx保留。另外注意StoryAgent构造时传入了llm覆盖——交接的同时可以把管线从STTLLMTTS 级联切到端到端 Realtime API每个 Agent 都有自己独立的模型管线。验证智能体行为链式断言加 LLM 评审LLM 输出不确定硬断言容易脆纯人工验收又不可扩展。框架的测试集成把两件事都做了确定性事件用链式断言语义正确性交给 judge评审模型。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...)驱动一次完整的用户说话→识别→推理→工具→回复流程并返回RunResultresult.expect逐事件校验工具名、工具输出、助手消息是否出现skip_next_event_if用来吸收模型可能多吐一条空消息这类不确定分支最后的.judge(llm, intent...)把助手是否追问了用户想点什么这类无法硬编码的判断委托给另一个 LLM 打分。断言原语定义在 voice/run_result.pyRunResult、RunAssert、EventAssert若想完全绕开 AgentServer/worker 做进程内测试testing.py 提供fake_job_context注入一个伪JobContext可配合真实房间直接session.start(...)。仓库自带的 tests/ 目录有数百个测试test_agent_session.py、test_false_interruption_resume.py、test_preemptive_pause_deadlock.py等覆盖打断恢复、预生成死锁等语音交互的疑难路径本身就是很好的用法参考。console / dev / start 三种运行模式怎么选脚本末尾的cli.run_app(server)会注册三个子命令对应三个阶段模式命令适用场景前置条件终端调试python myagent.py console本地音频输入/输出快速验证无需外部服务器客户端联调python myagent.py dev让 LiveKit 客户端 SDK 或电话集成作为对端接入LIVEKIT_URL等三个环境变量生产运行python myagent.py start生产级优化部署同 dev源码层面三个命令的差异在 cli/_legacy.py 与 cli/cli.py 里console_run_console起一个独立线程跑server.run(devmodeTrue, unregisteredTrue)——不向 LiveKit 服务器注册然后server.simulate_job(console-room, agent_identityconsole, fake_jobTrue)伪造一个任务驱动 entrypoint这就是它零服务器依赖的原因音频走AgentsConsole挂接本地设备支持音频/文本两种模式与--record录制。dev / start都走_run_worker先按参数执行server.update_options(ws_url..., api_key..., api_secret...)再server.run(devmode...)。--url/--api-key/--api-secret都声明了对应的envvar即LIVEKIT_URL等--log-level同理读LIVEKIT_LOG_LEVEL所以只设环境变量也能跑。退出路径做了完整保护首次 SIGINT/SIGTERM 只调度退出非 dev 模式会先执行server.drain()start可用--drain-timeout配置等待时长等在途会话自然结束3 秒看门狗_EXIT_ESCALATION_TIMEOUT 3.0在事件循环被同步代码阻塞时升级强制中断第二次 CtrlC 直接os._exit(1)。版本现状要注意源码中console与dev子命令均已标注 deprecated内置 Python CLI 自 1.5.10 起整体建议迁移到 LiveKit CLI 的lk agent console/lk agent dev且dev的进程内自动热重载已从 Python CLI 移除热重载能力由lk agent dev提供。README 中dev 支持热重载的说法对应的是 LiveKit CLI 工具链在 Python 脚本里直接跑dev是拿不到该能力的。选型建议写提示词阶段用console或lk agent console联调前端/电话用dev上线用start。本地开发流uv 管依赖pytest 跑测试ruff 管风格仓库自身用 uv 做包管理二次开发时同样适用uv sync --all-extras --dev # 安装开发依赖 uv run pytest --unit # 跑单元测试 uv run ruff format uv run ruff check --fix # 格式化与 lint跑示例需要先在 examples/ 下建.env模板见 examples/.env.example填 LiveKit Server 与各模型服务商的凭据然后uv run examples/voice_agents/basic_agent.py dev各插件的集成测试依赖相应 API 凭据维护者的 PR 会由 CI 自动执行。需要 API 文档时可用 pdoc 本地生成uv sync --all-extras --group docs后uv run --active pdoc --skip-errors --html --output-dirdocs livekit。若想拉取源码研究或二次开发仓库地址为https://gitcode.com/GitHub_Trending/agen/agents。examples/ 目录还有一批可独立运行的成套示例每个带 Dockerfile 的目录都可以直接容器化部署voice_agents基础对话、RAG、Realtime 模型、MCP、hotel_receptionist含策略文档与评测场景、drive_thru点单、frontdesk日程前台、telephonyIVR 电话、warm-transfer人工坐席暖交接、avatar数字人视频等详见 examples/README.md。许可证与合规提醒框架本体采用 Apache-2.0 许可见 LICENSE但 LiveKit 的轮次检测turn detection模型单独适用 LiveKit Model License见 MODEL_LICENSE。如果你的产品启用了语义级轮次检测模型侧条款与框架本体是分离的商用前两份协议都要各自确认。【免费下载链接】agentsA framework for building realtime voice AI agents ️项目地址: https://gitcode.com/GitHub_Trending/agen/agents创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考
RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进