企业官网建设流程全解析

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

【免费下载链接】agentsA framework for building realtime voice AI agents 🤖🎙️📹项目地址: https://gitcode.com/GitHub_Trending/agen/agents

LiveKit 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 Server：

• LIVEKIT_URL
• LIVEKIT_API_KEY
• LIVEKIT_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( vad=inference.VAD(), stt=inference.STT("deepgram/nova-3", language="multi"), llm=inference.LLM("google/gemma-4-31b-it"), 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把普通协程变成 LLM 可调用工具，docstring 即工具说明，类型标注的参数由 LLM 填充，context: RunContext是框架注入的运行期上下文；@server.rtc_session()装饰的entrypoint相当于 Web 服务里的请求处理器，每来一个房间任务就会被调用一次，ctx.room就是智能体要加入的房间；最后用generate_reply主动开口，实现"智能体先打招呼"。cli.run_app(server)则把console/dev/start三个子命令挂到脚本上。

四个概念对照源码：Agent在 voice/agent.py，AgentSession在 voice/agent_session.py，AgentServer在 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( stt=deepgram.STT(model="nova-3"), llm=openai.LLM(model="gpt-4.1-mini"), tts=cartesia.TTS(model="sonic-3", voice="9626c31c-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__( instructions="You are a story teller. Gather the user's name " "and where they are from." ) async def on_enter(self): self.session.generate_reply(instructions="greet 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), "Let's start the story!" class StoryAgent(Agent): def __init__(self, name: str, location: str) -> None: super().__init__( instructions=f"You are a storyteller. The user's name is {name}, " f"from {location}", llm=openai.realtime.RealtimeModel(voice="echo"), chat_ctx=chat_ctx, ) async def on_enter(self): self.session.generate_reply()

这里藏着两个关键设计。第一，information_gathered返回(新智能体, 衔接话术)元组：框架识别到工具输出是Agent后，会在同一会话内切换活动智能体并播出 "Let's start the story!"，用户无感。第二，状态通过userdata延续：entrypoint里用AgentSessionStoryData, ...)声明会话级共享数据，前一个智能体写context.userdata，后一个智能体直接读，对话历史则靠显式携带的chat_ctx保留。另外注意StoryAgent构造时传入了llm=覆盖——交接的同时可以把管线从"STT+LLM+TTS 级联"切到端到端 Realtime API，每个 Agent 都有自己独立的模型管线。

验证智能体行为：链式断言加 LLM 评审

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=...)驱动一次完整的"用户说话→识别→推理→工具→回复"流程并返回RunResult；result.expect逐事件校验（工具名、工具输出、助手消息是否出现），skip_next_event_if用来吸收"模型可能多吐一条空消息"这类不确定分支，最后的.judge(llm, intent=...)把"助手是否追问了用户想点什么"这类无法硬编码的判断委托给另一个 LLM 打分。断言原语定义在 voice/run_result.py（RunResult、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)会注册三个子命令，对应三个阶段：

源码层面，三个命令的差异在 cli/_legacy.py 与 cli/cli.py 里：

• console：_run_console起一个独立线程跑server.run(devmode=True, unregistered=True)——不向 LiveKit 服务器注册，然后server.simulate_job("console-room", agent_identity="console", fake_job=True)伪造一个任务驱动 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）在事件循环被同步代码阻塞时升级强制中断；第二次 Ctrl+C 直接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-dir=docs livekit。若想拉取源码研究或二次开发，仓库地址为https://gitcode.com/GitHub_Trending/agen/agents。

examples/ 目录还有一批可独立运行的成套示例，每个带 Dockerfile 的目录都可以直接容器化部署：voice_agents（基础对话、RAG、Realtime 模型、MCP）、hotel_receptionist（含策略文档与评测场景）、drive_thru（点单）、frontdesk（日程前台）、telephony（IVR 电话）、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