LiveKit Agents 实时语音 AI 智能体框架实战:装好就跑,3 种模式一路部署到生产
【免费下载链接】agentsA framework for building realtime voice AI agents 🤖🎙️📹项目地址: https://gitcode.com/GitHub_Trending/agen/agents
LiveKit Agents 是用于构建实时语音 AI 智能体(realtime voice AI agents)的 Python 框架,它让运行在服务器上的智能体能听、能看、能理解、能开口,并原生支持工具调用、多智能体交接与自动化测试。本文带你从安装最小依赖开始,走通一个可发声的最小智能体,再到多角色交接、行为断言测试,最后用 3 种运行模式把它部署到生产。
先认清 4 个核心构件:Agent、AgentSession、entrypoint 与 AgentServer
读完本节,你能在写任何一行代码前,把"谁在调度、谁在对话、从哪进入"这条链路说清楚——后面所有示例都在复用这四个概念。
| 构件 | 职责 | 源码位置 |
|---|---|---|
| Agent | 带明确指令(instructions)的 LLM 应用 | livekit-agents/livekit/agents/voice/agent.py |
| AgentSession | 智能体容器,管理音频输入、识别、生成、播放的完整交互管道 | livekit-agents/livekit/agents/voice/agent_session.py |
| entrypoint | 交互式会话的入口函数,类似 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( vad=inference.VAD(), stt=inference.STT("deepgram/nova-3", language="multi"), llm=inference.LLM("google/gemma-4-31b-it"), # low-latency gemma, hosted on LiveKit tts=inference.TTS("cartesia/sonic-3", voice="9626c31c-bec5-4cca-baa8-f8ba9e84c8bc"), ) agent = Agent( instructions="You are a friendly voice assistant built by LiveKit.", tools=[lookup_weather], ) await session.start(agent=agent, room=ctx.room) await session.generate_reply(instructions="greet 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.py:turn_handling=TurnHandlingOptions(...)里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, "Let's start the story!"围绕这段代码,有三个机制值得拆开看:
- 工具即跳转:
information_gathered返回(story_agent, "Let's start the story!")元组。框架发现工具返回值是Agent实例,就在当前会话内切换活动智能体,并播放元组里的衔接话术; - userdata 跨智能体共享:
AgentSession[StoryData]用类型参数声明会话级共享数据,context.userdata把前一位收集到的name/location交棒给后继智能体,状态不中断; - 每智能体独立模型管线:
StoryAgent构造时传llm=openai.realtime.RealtimeModel(voice="echo"),在交接的同时把模型从"STT+LLM+TTS 级联"切到端到端的 Realtime API,并显式携带chat_ctx保住对话历史。
给智能体行为写断言测试:expect 链加 judge 裁判
LLM 输出天生不确定,这一节给你一套能自动化验收智能体行为的写法——硬断言管事件序列,judge 管语义正确性。
@pytest.mark.asyncio async def test_no_availability() -> None: llm = google.LLM() async with AgentSession(llm=llm) as sess: await sess.start(MyAgent()) result = await sess.run( user_input="Hello, I need to place an order." ) result.expect.skip_next_event_if(type="message", role="assistant") result.expect.next_event().is_function_call(name="start_order") result.expect.next_event().is_function_call_output() await ( result.expect.next_event() .is_message(role="assistant") .judge(llm, intent="assistant should be asking the user what they would like") )sess.run(user_input=...)模拟一次用户输入并驱动完整管线(识别→LLM→工具调用→合成),返回RunResult;result.expect提供链式事件断言:is_function_call(name="start_order")校验工具名,is_function_call_output()校验工具执行完成,is_message(role="assistant")校验助手回复;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,在独立事件循环里以unregistered=True调用server.run(devmode=True)且不向服务器注册,再用server.simulate_job("console-room", agent_identity="console", fake_job=True)伪造任务驱动 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 模式同时 drain),3 秒看门狗在事件循环被同步代码阻塞时升级为强制中断,二次 Ctrl+C 直接
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.md; - tests/ 目录的数百个测试文件既是断言框架的用法示范,也覆盖了语音交互的疑难路径,如
test_false_interruption_resume.py(误打断恢复)、test_preemptive_pause_deadlock.py(预生成死锁)、test_llm_fallback.py(LLM 降级); - 想深入某条机制,核心库源码集中在 livekit-agents/livekit/agents/:
voice管会话与打断,llm/stt/tts管模型抽象,inference管统一模型网关,cli管三种运行模式。
仓库自身的开发约定:uv、单测与格式化
如果你要基于这个仓库二次开发或提交贡献,这一节把日常命令一次讲全。
依赖管理用 uv:
uv sync --all-extras --dev运行示例:建好
examples/.env(LiveKit 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;
代码规范用 ruff(
uv run ruff format、uv run ruff check --fix);API 文档可用 pdoc 本地生成(uv sync --all-extras --group docs后uv run --active pdoc --skip-errors --html --output-dir=docs livekit)。
下一步
- 先把
console模式的最小示例跑通,再对着examples/里最接近你业务的实现逐行看参数; - 给核心行为补上
sess.run+expect断言,语义分支用judge兜底; - 上线前切到
start模式、配齐三个环境变量,并留意 Python CLI 向lk agent迁移的弃用节奏。
【免费下载链接】agentsA framework for building realtime voice AI agents 🤖🎙️📹项目地址: https://gitcode.com/GitHub_Trending/agen/agents
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考