ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

LiveKit Agents 实时语音 AI 智能体框架实战:装好就跑,3 种模式一路部署到生产

LiveKit Agents 实时语音 AI 智能体框架实战:装好就跑,3 种模式一路部署到生产 LiveKit Agents 实时语音 AI 智能体框架实战装好就跑3 种模式一路部署到生产【免费下载链接】agentsA framework for building realtime voice AI agents ️项目地址: https://gitcode.com/GitHub_Trending/agen/agentsLiveKit Agents 是用于构建实时语音 AI 智能体realtime voice AI agents的 Python 框架它让运行在服务器上的智能体能听、能看、能理解、能开口并原生支持工具调用、多智能体交接与自动化测试。本文带你从安装最小依赖开始走通一个可发声的最小智能体再到多角色交接、行为断言测试最后用 3 种运行模式把它部署到生产。先认清 4 个核心构件Agent、AgentSession、entrypoint 与 AgentServer读完本节你能在写任何一行代码前把谁在调度、谁在对话、从哪进入这条链路说清楚——后面所有示例都在复用这四个概念。构件职责源码位置Agent带明确指令instructions的 LLM 应用livekit-agents/livekit/agents/voice/agent.pyAgentSession智能体容器管理音频输入、识别、生成、播放的完整交互管道livekit-agents/livekit/agents/voice/agent_session.pyentrypoint交互式会话的入口函数类似 Web 服务里的请求处理器由server.rtc_session()装饰器注册AgentServer主进程负责任务调度并为用户会话拉起智能体livekit-agents/livekit/agents/worker.py这四个符号在根包 livekit-agents/livekit/agents/init.py 中直接导出与function_tool、ChatContext、RunContext等一起构成年份级 API 表面mcp模块则通过__getattr__懒加载避免对 MCP 的强依赖。一条命令装好依赖跑通最小语音智能体只想跑通最小示例的话只看这一节装依赖、配环境变量、把入口骨架写出来你的智能体就能开口说话。依赖安装一条命令搞定方括号里的 extras 决定一并装哪些模型插件按你的模型栈自由增删pip install livekit-agents[openai,deepgram,cartesia]运行前提准备好 LiveKit Cloud 或自建服务器的三个环境变量LIVEKIT_URL、LIVEKIT_API_KEY、LIVEKIT_API_SECRET想直接跑仓库里的示例可先在examples/下建examples/.env写入凭据模板见examples/.env.example。下面是完整可运行的入口骨架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), # low-latency gemma, hosted on LiveKit 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装饰的异步函数会被自动包装为工具docstring 变成 LLM 可见的工具说明类型标注的参数变成 LLM 填充的入参context: RunContext是框架注入的运行期上下文可访问会话状态与共享数据server.rtc_session()装饰的entrypoint是每会话入口——每有新房间任务被调度AgentServer就调用一次该协程并注入JobContext其中ctx.room即智能体要加入的 WebRTC 房间模型管线可任意混搭vad/stt/llm/tts既能传插件实例也能传字符串模型标识由框架自动实例化。上例使用 LiveKit Inference 统一访问各家模型若想直接用厂商 key可换deepgram.STT、openai.LLM、cartesia.TTS这类插件实例session.start(agent..., room...)把智能体挂入会话并绑定房间紧随其后的generate_reply(instructions...)主动发起首轮回复这就是智能体先打招呼的开场白模式。 想要工程化参数怎么配参考 examples/voice_agents/basic_agent.pyturn_handlingTurnHandlingOptions(...)里resume_false_interruption在误打断后自动恢复播放preemptive_generation在等用户说完的同时预生成回复压低首字延迟aec_warmup_duration给开播初期的回声消除留校准时间tts_text_transforms过滤 emoji 与 markdown、替换特定发音stt_context_options的关键词检测让 LLM 把高频术语自动注入 STT 上下文专治专有名词识别率。多智能体交接怎么做让工具直接换人本节解决一个具体场景一段会话里需要角色分工前者收集信息后者讲故事。关键机制只有一个——工具函数返回新Agent实例框架就完成交接。function_tool async def information_gathered( self, context: RunContext, name: str, location: str, ): Called when the user has provided the information needed to make the story personalized and engaging. Args: name: The name of the user location: The location of the user context.userdata.name name context.userdata.location location story_agent StoryAgent(name, location) return story_agent, Lets start the story!围绕这段代码有三个机制值得拆开看工具即跳转information_gathered返回(story_agent, Lets start the story!)元组。框架发现工具返回值是Agent实例就在当前会话内切换活动智能体并播放元组里的衔接话术userdata 跨智能体共享AgentSession[StoryData]用类型参数声明会话级共享数据context.userdata把前一位收集到的name/location交棒给后继智能体状态不中断每智能体独立模型管线StoryAgent构造时传llmopenai.realtime.RealtimeModel(voiceecho)在交接的同时把模型从STTLLMTTS 级联切到端到端的 Realtime API并显式携带chat_ctx保住对话历史。给智能体行为写断言测试expect 链加 judge 裁判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...)模拟一次用户输入并驱动完整管线识别→LLM→工具调用→合成返回RunResultresult.expect提供链式事件断言is_function_call(namestart_order)校验工具名is_function_call_output()校验工具执行完成is_message(roleassistant)校验助手回复skip_next_event_if用来兼容模型可能先冒一条空消息这类不确定分支.judge(llm, intent...)把助手有没有在询问想点什么这类无法硬编码的判断交给另一个 LLM 做裁判式评分。RunResult、RunAssert、EventAssert都定义在 livekit-agents/livekit/agents/voice/run_result.py 并在根包导出。想脱离 worker 进程做进程内测试testing.py的fake_job_context会注入一个带fake_job的JobContext让get_job_context()及其访问点与真实任务行为一致配合真实房间即可直接session.start(...)。3 种运行模式怎么选console、dev、start从本地验证到生产上线这一节给你一张决策表外加必须知道的两处弃用与许可提示。在终端直接试跑智能体无需服务器python myagent.py console启用本地音频输入输出不依赖外部服务器与依赖适合快速验证行为。用 LiveKit 客户端联调dev 模式python myagent.py dev拉起智能体服务器并接入 LiveKit Cloud 或自建服务器对端可以是任意 LiveKit 客户端 SDK 或电话集成。此模式同样需要LIVEKIT_URL、LIVEKIT_API_KEY、LIVEKIT_API_SECRET三个环境变量。生产环境启动start 模式python myagent.py start以生产级优化运行收到停止信号时会先执行server.drain()等待在途任务结束--drain-timeout可配置等待时长。⚠️弃用提示Python 侧 CLI 整体已标记 deprecated源码注释明确建议改用 LiveKit CLI 的lk agent ...命令其中console与dev子命令均注明将在未来版本移除dev 模式的进程内热重载也已从 Python CLI 移除能力转移到了lk agent dev。选型测试手段时注意这一点。 机制速览不想读源码也能明白为什么这么设计console起独立线程的_ConsoleWorker在独立事件循环里以unregisteredTrue调用server.run(devmodeTrue)且不向服务器注册再用server.simulate_job(console-room, agent_identityconsole, fake_jobTrue)伪造任务驱动 entrypoint——这就是它无需外部服务器的原因控制台音频经TcpAudioInput/TcpAudioOutput挂接支持音频/文本两种模式与--record会话录制dev/start都走 livekit-agents/livekit/agents/cli/cli.py 中的_run_worker先按 CLI 参数server.update_options(ws_url..., api_key..., api_secret...)再调server.run(devmode...)--url/--api-key/--api-secret均声明了对应envvar所以命令行参数与环境变量等价优雅退出有三重保障首次 SIGINT/SIGTERM 只调度退出非 dev 模式同时 drain3 秒看门狗在事件循环被同步代码阻塞时升级为强制中断二次 CtrlC 直接os._exit(1)兜底——保证停止信号不会粗暴截断进行中的语音会话许可证Agents 框架为 Apache-2.0见 LICENSE而 LiveKit 的语义轮次检测模型单独采用 LiveKit Model License见 MODEL_LICENSE两条条款相互独立启用轮次检测后商用前需分别确认。在仓库里找参考实现examples 与 tests 指路这一节帮你跳过从零造轮子直接找到对口径的完整示例和测试示范。examples/ 覆盖多类场景voice_agents/起步智能体、MCP 工具、RAG 等、avatar/基于 Tavus、Bithuman、LemonSlice 等的视频数字人、telephony/电话 IVR、hotel_receptionist/含策略文档与评测场景、frontdesk/日程前台、healthcare/医疗预约、primitives/回声等底层原语带Dockerfile的示例均可容器化部署更多细节见 examples/README.mdtests/ 目录的数百个测试文件既是断言框架的用法示范也覆盖了语音交互的疑难路径如test_false_interruption_resume.py误打断恢复、test_preemptive_pause_deadlock.py预生成死锁、test_llm_fallback.pyLLM 降级想深入某条机制核心库源码集中在 livekit-agents/livekit/agents/voice管会话与打断llm/stt/tts管模型抽象inference管统一模型网关cli管三种运行模式。仓库自身的开发约定uv、单测与格式化如果你要基于这个仓库二次开发或提交贡献这一节把日常命令一次讲全。依赖管理用 uvuv sync --all-extras --dev运行示例建好examples/.envLiveKit Server 与各模型服务商凭据模板见examples/.env.example后uv run examples/voice_agents/basic_agent.py dev单元测试位于 tests/ 目录uv run pytest --unit各插件的集成测试需要相应 API 凭据会在维护者提交的 PR 上由 CI 自动运行详见 .github/workflows/tests.yml代码规范用 ruffuv run ruff format、uv run ruff check --fixAPI 文档可用 pdoc 本地生成uv sync --all-extras --group docs后uv run --active pdoc --skip-errors --html --output-dirdocs livekit。下一步先把console模式的最小示例跑通再对着examples/里最接近你业务的实现逐行看参数给核心行为补上sess.runexpect断言语义分支用judge兜底上线前切到start模式、配齐三个环境变量并留意 Python CLI 向lk agent迁移的弃用节奏。【免费下载链接】agentsA framework for building realtime voice AI agents ️项目地址: https://gitcode.com/GitHub_Trending/agen/agents创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考
返回列表