LiveKit Agents：基于WebRTC的实时语音视频AI Agent框架（9.9k Star）

导读

当我们说"AI Agent"时，大多数人想到的是文本聊天框里的对话。但如果Agent能像真人一样加入一场视频通话——听你说话、看你的屏幕、实时回应——交互方式会有显著变化。

LiveKit Agents是一个基于WebRTC的开源Python框架，让AI Agent以"房间参与者"的身份加入实时音视频会话。框架处理了实时语音AI的核心难题：音频流经STT→LLM→TTS管线的流式传输、用户打断的检测与处理、对话轮次的语义判断、多Agent之间的交接。目前GitHub Stars约9.9k，仓库持续活跃，提供64个插件包，发版节奏约为每周一次。

本文将从架构设计、核心能力、插件生态、上手流程四个维度解读这个项目

一、项目概览

———————————————————————————————————————————

LiveKit Agents构建在LiveKit开源实时通信平台之上。LiveKit本身提供WebRTC SFU（Selective Forwarding Unit，选择性转发单元，负责在多人通话中高效转发音视频流）服务器、客户端SDK（覆盖Browser、Swift、Android、Flutter、React Native、Rust、Unity、ESP32等平台）和SIP电话网关。Agents框架在此基础上增加了AI Agent的调度、运行和管理能力。

二、架构设计：Agent如何"入会"

四个核心概念

工作流程

Agent代码启动后，向LiveKit服务器（自托管或LiveKit Cloud）注册为一个AgentServer进程当有用户进入LiveKit Room时，服务器向AgentServer发起调度请求AgentServer启动一个Job子进程（Job是框架分配给单个用户会话的工作单元）Job中的AgentSession加入Room，成为一个WebRTC参与者Agent通过WebRTC接收用户的音频/视频流，处理后将结果以音频/视频/文本的形式发回

WebRTC保障了在不稳定网络下的通信质量（自适应码率、拥塞控制、自动重连）。Agent与后端服务之间通过HTTP和WebSocket通信。

三种运行模式

# 终端模式：本地音频输入输出，无需外部依赖 python myagent.py console # 开发模式：热重载，连接LiveKit服务器 python myagent.py dev # 生产模式：优化部署 python myagent.py start 

console模式特别适合本地调试——直接在终端用麦克风和Agent对话，不需要配置LiveKit服务器。

三、核心能力

语义轮次检测

实时语音对话中最棘手的问题之一是判断用户是否说完了。常见方案用固定的静音时长阈值（如1.5秒无声则视为说完），但这既慢又不准——用户思考时停顿一下就会被误判为说完。

LiveKit Agents内置了一个基于transformer模型的语义轮次检测器（turn-detector插件），通过理解语义上下文来判断用户是否说完，而不仅仅依赖静音。

自适应打断处理

v1.5.0版本引入了ML模型驱动的打断判断，能区分真正的用户打断和非语义声音（咳嗽、叹气、语气词）。根据发布说明，在500ms重叠语音的条件下达到86%精确率和100%召回率。

同时引入了动态端点检测（Dynamic Endpointing），使用指数移动平均值自适应调整静音阈值，替代固定延迟。

工具调用与MCP支持

Agent可以定义工具函数供LLM调用。框架原生支持MCP（Model Context Protocol），一行代码即可接入MCP工具服务。

from livekit.agents import function_tool, RunContext @function_tool async def lookup_weather(context: RunContext, location: str) -> str:     """查询指定地点的天气"""     # 实现逻辑     return f"{location}的天气是晴天，25°C" 

多Agent交接

支持在一个会话中多个Agent之间切换。例如，一个"引导Agent"负责收集用户信息，完成后将会话交接给"业务Agent"继续处理：

class IntroAgent(Agent):     async def on_enter(self):         self.session.generate_reply(             instructions="请收集用户的姓名和需求"         )     @function_tool     async def information_gathered(         self, context: RunContext, name: str, location: str     ):         """信息收集完毕后调用"""         story_agent = StoryAgent()         return story_agent, "Let's start the story!" 

当工具函数返回另一个Agent实例（可附带交接消息）时，框架自动完成会话交接。

内置测试框架

提供基于LLM Judge的Agent测试方案，可以验证Agent的工具调用行为和回复意图：

async def test_agent():     async with AgentSession(llm=llm) as sess:         result = await sess.run(user_input="我要点一份意面")         result.expect.next_event().is_function_call(name="start_order")         result.expect.next_event().is_message(             role="assistant"         ).judge(llm, intent="确认订单内容") 

四、插件生态：64个插件包

———————————————————————————————————————————

LiveKit Agents通过插件机制集成外部服务。仓库livekit-plugins/目录下包含64个插件包（含服务商集成和工具类插件），安装时按需选择：

pip install "livekit-agents[openai,silero,deepgram,cartesia,turn-detector]~=1.4" 

LLM

OpenAI、Anthropic、Google、Groq、Mistral AI、Fireworks AI、AWS Bedrock、Azure、NVIDIA、xAI、SambaNova、Cerebras、Ultravox。

STT（语音识别）

Deepgram、AssemblyAI、Google、Azure、AWS、Gladia、Soniox、Speechmatics、NVIDIA、Clova、Telnyx、Spitch。

TTS（语音合成）

Cartesia、ElevenLabs、Google、Azure、AWS、LMNT、Rime、Speechify、Murf、Neuphonic、Fish Audio、Sarvam、Telnyx、xAI、Smallest AI、Camb.ai。

VAD与轮次检测

Silero（VAD）、turn-detector（语义轮次检测）。

Avatar（数字人）

Hedra、Bey、Bithuman、Simli、Tavus、Keyframe、LiveAvatar、Anam、AvatarIO、AvatarTalk。

其他

Langchain、FAL、Browser（浏览器操作）、NLTK、Hume（情感分析）、Inworld、Resemble、Hamming（监控）等。

五、上手体验

———————————————————————————————————————————

最小示例

from livekit.agents import Agent, AgentSession, AgentServer, JobContext, RunContext, function_tool from livekit.plugins import openai, silero, deepgram, cartesia @function_tool asyncdef lookup_weather(context: RunContext, location: str) -> str:     """查询天气"""     returnf"{location}: 晴天 25°C" # 创建AgentServer server = AgentServer() @server.rtc_session() asyncdef entrypoint(ctx: JobContext):     agent = Agent(         instructions="你是一个友好的语音助手",         tools=[lookup_weather]     )     session = AgentSession(         stt=deepgram.STT(),         llm=openai.LLM(model="gpt-4.1-mini"),         tts=cartesia.TTS(),         vad=silero.VAD.load()     )     await session.start(agent=agent, room=ctx.room)     await session.generate_reply(         instructions="greet the user and ask about their day"     ) 

也支持简写语法，用字符串指定模型：

session.start(     agent=agent,     stt="deepgram/nova-3",     llm="openai/gpt-4.1-mini",     tts="cartesia/sonic-3" ) 

运行

# 设置环境变量 export LIVEKIT_URL=wss://your-livekit-server export LIVEKIT_API_KEY=your-key export LIVEKIT_API_SECRET=your-secret # 本地终端模式（无需LiveKit服务器） python myagent.py console # 开发模式 python myagent.py dev 

示例项目

仓库提供12个示例：

六、总结

———————————————————————————————————————————

项目特点：

Agent以WebRTC参与者身份加入Room，继承WebRTC的网络自适应能力（拥塞控制、自动码率调节、断线重连）内置语义轮次检测（transformer模型）和自适应打断处理（v1.5.0，86%精确率/100%召回率）采用Python代码定义Agent行为，支持IDE调试器、断点、单步执行；console模式可在本地终端直接对话测试64个插件包覆盖主流LLM/STT/TTS/Avatar供应商，原生支持MCP和多Agent交接

适合的场景：

需要构建实时语音/视频AI Agent，特别是需要与已有WebRTC通信系统集成的项目需要精细控制打断和轮次检测行为的语音Agent需要多Agent交接的复杂对话流程