LiveKit Agents:实时音视频与LLM集成,构建语音AI助手
这次我们来看一个实时音视频领域的开源项目:LiveKit Agents。它解决的问题非常具体——怎么把大语言模型、语音识别、语音合成快速接进实时通话链路,让 AI 以一个“能听、能说、能实时响应”的智能体形态出现在音视频房间里。如果你关心语音 Agent、实时 AI 助手、WebRTC 接入、自托管部署这些关键词,这篇文章可以直接收藏。
LiveKit Agents 不是传统意义上的“一键 ChatBot 整合包”,它是一套面向语音对话和实时交互场景的 Agent 编排框架。核心思路是让开发者用一个 Python 或 Node.js 的 Worker 进程,去接收 LiveKit 房间里的音频流 / 视频流,把流交给 STT、LLM、TTS 流水线处理,再把结果推回房间。客户端不用关心背后接的是哪个语音模型,只要按标准 WebRTC 协议连进房间就能和 Agent 对话。
这篇文章会从核心能力、适用场景、环境准备、部署启动、功能测试、接口编排、性能观察、常见问题排查和最佳实践这几个维度展开。整体内容更偏“工程落地前的摸底”,适合正在评估 LiveKit Agents 是否能用于自己项目的开发者,也适合想在本地自托管一套语音 Agent Demo 的读者。
1. LiveKit Agents 核心能力速览
先说结论:LiveKit Agents 最大的价值不是某一个单独的语音模型,而是把实时交互链路标准化了。从能力规格上看,它覆盖了“接入实时音视频流、处理 Agent 逻辑、返回语音或视频响应”的完整闭环。
| 能力项 | 说明 |
|---|---|
| 项目类型 | 开源实时音视频 Agent 编排框架 |
| 主要语言 | Python、Node.js |
| 核心功能 | 语音 Agent、实时转写、LLM 对话、TTS 语音回复、多模态输入输出 |
| 平台支持 | LiveKit Cloud、自托管 LiveKit 服务端 |
| 启动方式 | Worker 进程方式启动,Agent 可作为独立服务运行 |
| 是否支持 API | 支持,封装为 Session / Worker 接口,可通过房间事件触发 |
| 是否支持批量任务 | 支持多 Worker 并发,可结合队列编排批量会话 |
| 典型接入模型 | OpenAI、Anthropic、Google Gemini、本地 Ollama、Whisper、ElevenLabs、Azure 等 |
| 适用场景 | 语音客服、面试模拟、实时翻译、语音助手、多模态交互应用 |
| 显存要求 | 取决于接入的模型,使用云端模型 API 时本地几乎无显存压力 |
| 部署形式 | Docker、Python 虚拟环境、Node.js 环境均可 |
从部署角度看,LiveKit Agents 适合两类开发者。第一类是已经在用 LiveKit 做音视频产品的团队,可以直接把 Agent 能力加到现有房间里。第二类是打算从零搭建语音 AI Demo 的开发者,可以先用 LiveKit Cloud 免费额度跑通,再逐步迁移到自托管。
需要注意,LiveKit Agents 的版本更新速度比较快,不同版本的 Worker 接口和工作流配置略有差异。文中的命令和代码适用于当前公开版本的基本模式,实际部署时建议以官方文档和安装包版本为准。
2. 适用场景与使用边界
2.1 适合谁用
LiveKit Agents 最典型的用法是语音对话型 Agent。比如:一个用户打开网页,点击“开始对话”,浏览器通过 WebRTC 连接进 LiveKit 房间,Agent Worker 收到音频流,转写成文本,丢给 LLM,生成回复文本,再通过 TTS 合成语音推回房间。用户听到的就是一个“能实时说话”的 AI 助手。
这个场景尤其适合以下项目形态:
- 语音客服机器人:客户电话进来后,Agent 自动接听、理解意图、查询知识库并回答。
- 面试模拟器:Agent 扮演面试官,根据候选人回答继续追问,甚至能评价表达逻辑。
- 实时翻译助手:一方说中文,Agent 转写成文本、翻译成英文,再通过 TTS 输出英文语音。
- 语音笔记和会议助手:加入会议房间,实时记录、总结要点。
- 多模态看板助手:Agent 不仅能听,还能接收屏幕共享画面,结合视觉模型回答问题。
2.2 不适合什么场景
LiveKit Agents 并不适合离线批量文本对话。虽然它可以处理批量会话,但本质上更侧重“实时性”。如果你只是想做后台的异步 Bot,或者只需要一个简单 HTTP API 调用 LLM,直接用函数调用或消息队列更轻量,没必要引入 WebRTC 链路。
它也不适合对延迟极度敏感、要求纯端侧推理的场景。LiveKit Agents 的服务端 - 客户端架构,天然存在网络传输延迟。STT、LLM、TTS 每个环节都会增加几百毫秒到几秒的延迟。如果要实现“对话无感”,需要做大量模型选型和延迟优化,而不是开箱即用。
2.3 合规与安全边界
这点必须强调。LiveKit Agents 会采集音频和视频数据,涉及用户隐私、企业敏感信息、人脸和声音肖像。开发和测试时要注意:
- 采集录音前必须获得用户明确授权。
- 对话内容涉及个人隐私的,要按最小必要原则存储和加密。
- 使用云端 STT / LLM / TTS 服务时,要确认数据脱敏和存储策略。
- 声音克隆、人脸生成等能力,只能用于获得明确授权的场景,不能用于伪造身份或欺骗。
- 商用项目要特别关注模型服务商的条款,避免把敏感语料直接上传到第三方 API。
LiveKit 本身是一个通用实时通信开源项目,用在哪里取决于业务本身。但作为开发者,我们有责任在架构设计阶段就把隐私、安全和合规考虑进去。
3. LiveKit Agents 本地部署环境准备
3.1 通用环境清单
自托管 LiveKit Agents 需要准备以下环境:
| 依赖项 | 说明 |
|---|---|
| 操作系统 | Linux 优先,Windows / macOS 也可以跑 Worker,但生产环境建议 Linux |
| Python 版本 | 建议 3.9 及以上,部分新版本 Agent 插件需要更高版本 |
| Node.js | LiveKit Agents JS 版本需要 Node 18 及以上 |
| LiveKit 服务端 | 自托管需要部署 LiveKit Server,或用 LiveKit Cloud |
| Redis | 可选,用于多 Worker 之间的任务协调 |
| 模型服务 | 云端 API 或本地模型服务,需要提前准备 API Key |
| 端口 | 7880(LiveKit 默认信号端口)、7881(RTC 端口)等 |
如果你只是想在本地跑通一个 Demo,最省事的方式是用 LiveKit Cloud 分配一个项目,拿到
LIVEKIT_URL
、
LIVEKIT_API_KEY
、
LIVEKIT_API_SECRET
三个凭证,然后把 Agent Worker 跑在自己的电脑上。这样就不需要本地部署 LiveKit Server,适合快速验证。
3.2 Python 项目初始化
以 Python 版本为例,先创建一个虚拟环境:
mkdir livekit-agent-demo
cd livekit-agent-demo
python3 -m venv venv
source venv/bin/activate
然后安装 LiveKit Agents 核心库和必要的插件:
pip install livekit-agents
pip install livekit-plugins-openai
pip install livekit-plugins-silero
livekit-plugins-silero
提供 VAD(语音活动检测)能力,用来判断用户何时开始说话、何时停止说话。这是语音 Agent 里非常重要的一个组件。
livekit-plugins-openai
则提供 STT、LLM、TTS 的封装。
如果你要用本地模型,可以安装
livekit-plugms-ollama
或通过 OpenAI 兼容接口指向本地服务。具体插件名和版本以官方仓库为准。
3.3 环境变量配置
在项目根目录创建一个
.env.local
文件,写入你的 LiveKit 凭证和模型服务凭证:
LIVEKIT_URL=wss://your-project.livekit.cloud
LIVEKIT_API_KEY=your_api_key
LIVEKIT_API_SECRET=your_api_secret
OPENAI_API_KEY=sk-xxxx
OPENAI_MODEL=gpt-4o-mini
注意:
.env.local
文件不要提交到 Git 仓库。生产环境建议通过密钥管理服务注入环境变量。
4. LiveKit Agents 部署启动与服务访问
4.1 最小可运行的 Agent
LiveKit Agents 的代码结构很清晰。核心是定义一个 Agent 入口函数,在函数里初始化会话上下文和 Agent 实例,然后注册各个处理步骤。
下面是一个最小可运行的语音 Agent 示例:
import asyncio
from livekit.agents import AutoSubscribe, JobContext, WorkerOptions, cli
from livekit.agents.llm import function_tool
from livekit.plugins import openai, silero
async def entrypoint(ctx: JobContext):
await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)
session = ctx.session
session.set_audio_input_enabled(True)
vad = silero.VAD.load()
stt = openai.STT.with_groq(model="whisper-large-v3")
llm = openai.LLM(model="gpt-4o-mini")
tts = openai.TTS(model="gpt-4o-mini")
await session.start(stt=stt, llm=llm, tts=tts, vad=vad)
if __name__ == "__main__":
cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))
这个示例做的事情:Agent 连进房间后只订阅音频流,开启麦克风输入,然后启动一套 VAD + STT + LLM + TTS 的实时语音流水线。用户说话后,语音被转成文本,LLM 生成回复,TTS 合成语音送回房间。
4.2 启动 Worker
启动 Worker 的命令:
python agent.py dev
dev
模式会读取当前目录下的
.env.local
文件,并连接到对应的 LiveKit 项目。启动成功后,控制台会显示 Agent 已经注册并等待任务。
生产环境建议用:
python agent.py start
start
模式不会自动加载
.env.local
,需要手动设置环境变量,适合部署到云服务器或容器中。
4.3 客户端连接
Worker 启动后,还需要一个客户端才能测试对话。LiveKit 官方提供了 Web 端、iOS、Android 和命令行工具。最简单的方式是使用 LiveKit Agents Playground,一个网页版调试工具,输入 LiveKit URL 和 Token 后可以直接进入房间和 Agent 对话。
如果你改动了 Agent 代码,需要重启 Worker 进程,新配置才会生效。如果是修改了提示词或工具函数,通常只需要重新注册 Agent,不需要重启 LiveKit Server。
5. LiveKit Agents 功能测试与效果验证
部署完成后,最关心的问题就是:Agent 到底能不能正常对话?识别准不准?回复延迟高不高?下面给出一套可复用的测试流程。
5.1 基础对话测试
测试目的:验证 Agent 能听到用户说话、能生成文本回复、能通过 TTS 播放语音。
操作步骤:
- 启动 Worker 进程。
- 用 LiveKit Agents Playground 或 Web Demo 连接同一个房间。
- 点击“开始对话”,对着麦克风说一句“你好,介绍一下你自己”。
- 观察控制台日志和页面反馈。
预期结果:
- 控制台打印 STT 识别出的文本。
- LLM 生成回复文本。
- 页面端听到 TTS 合成的语音。
判断标准:如果 3 到 5 秒内能听到 Agent 回复,说明链路基本通畅。
常见失败原因:
- 没有正确订阅音频流,Agent 收不到用户声音。
- VAD 模型没有加载成功,无法切分语音片段。
- 环境变量没有加载,模型 API 调用失败。
5.2 多轮对话测试
测试目的:验证 Agent 是否具备多轮对话能力,能不能结合上下文回答后续问题。
操作步骤:
- 先问“今天天气怎么样”。
- 再问“那明天呢”。
- 观察 Agent 是否能理解“明天”指的是“明天天气”。
预期结果:Agent 能把第二句话的上下文关联到第一句话的话题。
实际效果取决于你选的 LLM 模型和提示词设计。LiveKit Agents 本身不限制对话轮数,但要注意长对话会增加 STT / LLM 的令牌消耗和延迟。
5.3 语音打断测试
测试目的:验证用户能否在 Agent 说话时打断它,重新提出新问题。
操作步骤:
- 触发 Agent 说一段较长的话。
- 在 Agent 还没说完时,直接说“停一下,我要问别的”。
- 观察 Agent 是否停止当前回复,开始处理新输入。
预期结果:VAD 检测到新语音输入,Agent 中断当前 TTS 播放,进入下一轮处理。
如果打断不生效,检查 VAD 的语音活动阈值设置,或者查看是否启用了 barge-in 功能。
5.4 自定义 Prompt 测试
测试目的:验证 Agent 是否按照指定角色设定回答问题。
在
entrypoint
中,通过
session.start
的
instructions
参数设置角色提示词:
session = ctx.session
session.set_audio_input_enabled(True)
await session.start(
stt=stt,
llm=llm,
tts=tts,
vad=vad,
instructions="你是一个温柔耐心的英语口语陪练老师,请用简单的英文和学生对话,每次回复不超过三句话。"
)
测试时可以按角色设定问几个问题,检查 Agent 的语气、回复长度和内容是否符合预期。
5.5 工具调用测试
LiveKit Agents 支持在提示词中注册工具函数,让 Agent 能查询数据库、调用外部 API 或执行代码。下面是一个简单示例:
from livekit.agents.llm import function_tool
@function_tool
async def get_weather(city: str) -> str:
"""查询指定城市的天气"""
# 这里可以接入真实的天气 API
return f"{city} 今天晴天,气温 25 摄氏度。"
async def entrypoint(ctx: JobContext):
await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)
session = ctx.session
session.set_audio_input_enabled(True)
await session.start(
stt=stt,
llm=llm,
tts=tts,
vad=vad,
tools=[get_weather]
)
测试时问“北京今天天气怎么样”,如果 Agent 能输出一个带有天气信息的中文回复,说明工具调用链路正常。
6. LiveKit Agents 接口 API 与批量任务编排
6.1 Worker 与 Session 的关系
LiveKit Agents 的“接口”不是传统 REST API,而是一套基于会话的实时事件模型。理解这个模型对做批量任务很有帮助:
- Worker:运行 Agent 代码的进程,可以同时监听多个房间。
- Job:LiveKit 服务端根据房间事件分配给 Worker 的任务。
- Session:Agent 与房间之间的实时通信会话。
- 事件:用户加入房间、Agent 加入房间、用户说话、断开连接等。
你可以把 Worker 理解成一组“接单员”,Job 是订单,Session 是服务过程。LiveKit 服务端会根据房间的情况把 Job 分发给可用的 Worker。
6.2 批量会话设计思路
如果需要同时服务多个用户,比如 10 个人同时进入房间和 Agent 对话,有两种做法:
方案一:单个 Worker 并发处理多个 Session。Agent 代码内部要支持并发会话。适合连接数不多的小规模场景。
方案二:启动多个 Worker 进程,让 LiveKit 服务端自动负载均衡。适合高并发场景,扩展起来更简单。
批量任务的关键在于合理设置 Worker 数。Worker 不是越多越好,因为每个 Session 都会占用 CPU、内存和网络连接。同时,模型 API 的限流是一个容易被忽略的瓶颈。如果一次性有 20 个用户同时说话,而 STT / TTS API 只允许 5 个并发请求,就会出现排队等待。
6.3 运行队列与调度
如果任务不是实时对话,而是异步处理,可以自己加一层队列。比如把待处理音频文件路径写入 Redis 队列,Worker 从队列中取出文件路径,转写、总结、返回结果,再把结果写回存储。
一个通用伪代码示例:
import redis
import json
from livekit.agents import JobContext, WorkerOptions, cli
r = redis.Redis(host="localhost", port=6379, decode_responses=True)
async def process_audio_file(job: JobContext):
# 从 Redis 队列读取任务
task = r.lpop("audio_tasks")
if not task:
return
data = json.loads(task)
file_path = data["file_path"]
# 调用 STT 转写
# 调用 LLM 生成总结
# 把结果写回 Redis 或数据库
if __name__ == "__main__":
cli.run_app(WorkerOptions(entrypoint_fnc=process_audio_file))
异步批量任务要特别注意失败重试。如果 STT 或 LLM 调用失败,任务不应该直接丢失,而应该重新入队或写入失败日志。
7. LiveKit Agents 资源占用与性能观察
7.1 本地资源占用
LiveKit Agents 的资源占用分为两块:Agent 进程本身、模型推理资源。
Agent 进程本身主要消耗 CPU 和内存。VAD 模型是轻量级的,CPU 上就能跑。STT、LLM、TTS 如果调用云端 API,本地几乎不消耗额外的算力,主要是网络请求。如果要在本地跑 Whisper 或本地 LLM,显存和内存占用会明显上升。
具体占用数字与模型版本、并发会话数、音频采样率有关,无法给出统一数值。合理做法是在部署环境里用
nvidia-smi
、
htop
、
docker stats
等工具实测观察。
7.2 显存观察方法
如果你在 Agent 的会话流里接入了本地模型,比如本地 Whisper 或本地 LLM,需要关注显存占用。观察命令:
nvidia-smi --query-gpu=name,memory.used,memory.total,utilization.gpu --format=csv -l 1
这个命令会每秒刷新一次显存占用和使用率。跑一个完整对话流程,记录峰值显存,就能知道当前模型在该分辨率、并发数下是否吃紧。
如果显存不足,可以:
- 换用更小的模型版本。
- 降低并发会话数。
- 使用量化模型。
- 把 STT 和 LLM 部署在不同机器上。
7.3 延迟优化方向
实时语音对话最影响体验的就是端到端延迟。延迟链路大致是:麦克风采集 -> VAD 检测 -> STT 转写 -> LLM 推理 -> TTS 合成 -> 播放。
优化方向:
- VAD 模型尽量选轻量级,Silero VAD 在 CPU 上表现不错。
- STT 优先选流式识别模型,不要等用户说完一整句话再识别。
- LLM 回复尽量短,减少生成时间。
- TTS 选低延迟的流式合成模型。
- 网络链路尽量走就近机房或专线。
7.4 分布式部署的注意事项
生产环境一般不建议把 Agent Worker 和 LiveKit Server 放在同一台机器上,除非只是 Demo。LiveKit Server 要处理媒体流的转发,CPU 和带宽占用和房间数、码率直接相关。Agent Worker 要处理模型请求和回调逻辑,两者混布容易互相干扰。
比较稳妥的结构是:
- LiveKit Server 单独部署。
- Agent Worker 按需横向扩展。
- Redis 或消息队列做任务调度。
- 模型服务通过 API 网关统一管理。
8. LiveKit Agents 常见问题与排查方法
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
Worker 启动报
LIVEKIT_URL
缺失
| 环境变量没配置 |
检查
.env.local
或系统环境变量
| 配置正确的 LiveKit 地址和凭证 |
| Agent 加入房间失败 | Token 过期或权限不足 | 查看服务端日志 | 重新生成 Token,确认具备房间加入权限 |
| 用户说话但 Agent 无响应 | VAD 未加载或音频未订阅 | 检查控制台 VAD 初始化日志 |
确认
AutoSubscribe.AUDIO_ONLY
生效
|
| STT 识别结果为空 | 麦克风权限或静音检测阈值过高 | 检查浏览器麦克风权限 | 调整 VAD 灵敏度,测试时靠近麦克风 |
| LLM 回复超时 | 模型 API 慢或限流 | 查看 API 日志和延迟 | 更换模型或增加超时重试策略 |
| TTS 播放卡顿 | 网络波动或 TTS 并发过高 | 查看房间媒体事件 | 降低并发,升级带宽或换低延迟 TTS |
| 多 Worker 下任务分配不均 | Worker 注册配置不一致 | 查看 Worker 控制台 | 检查 Worker ID 和负载均衡策略 |
| 批量任务卡住 | Redis 队列消费异常 | 检查 Redis 连接和任务日志 | 增加消费超时和失败重试机制 |
| 本地模型显存溢出 | 模型过大或并发过多 |
用
nvidia-smi
观察显存
| 换小模型、降低并发或使用量化版本 |
9. LiveKit Agents 最佳实践与使用建议
9.1 第一次运行先做最小验证
不要一开始就接复杂的多模态链路。先用最简单的 STT + LLM + TTS 跑通一个“你好 -> 回复”的流程,确认 LiveKit Server、Worker、客户端三者连通,再逐步加功能。这样可以更快地定位问题是出在网络链路、模型配置还是提示词设计。
9.2 保留一套最小可运行配置
本地开发环境、测试环境、生产环境要分别维护配置文件。
.env.local
用于本地测试,
.env.test
用于 CI 环境,
.env.prod
用于生产。生产环境不要使用明文密钥,优先用云平台的密钥管理服务。
9.3 模型选择要结合成本和延迟
LiveKit Agents 允许在同一个会话里混用不同模型。比较划算的做法是:用轻量模型做 STT 和 VAD,用中端模型处理日常对话,用高级模型处理复杂推理任务。比如日常闲聊走快速模型,涉及知识库查询时再调用更强模型。
9.4 日志与监控要前置
语音 Agent 调链路长,一个环节出错就很难排查。建议在 Agent 入口处打点日志,记录每次会话的 Session ID、STT 文本、LLM 响应、TTS 状态和总耗时。这样即使出问题,也能快速定位是转写问题、生成问题还是语音合成问题。
9.5 接口服务的安全限制
LiveKit Agents 不像传统 Web API 那样暴露 HTTP 端口,但它依赖 WebSocket 和 RTC 端口。如果部署在公网,要限制端口访问范围,使用防火墙或安全组规则。客户端连接必须校验 Token,不能允许匿名加入房间。
9.6 语音数据合规
录音数据默认会经过 STT 和 TTS 服务。如果接的是云端 API,要确认服务商的隐私政策。如果业务涉及敏感信息,建议选择支持私有化部署的 STT / TTS 服务,并在处理后及时删除临时音频文件。
10. 总结与下一步
LiveKit Agents 最值得尝试的点是它把实时音视频与 LLM 的集成成本压得很低。你不用从零搭建 WebRTC 链路,不用自己处理音频流切片,不用操心房间管理和媒体转发,只需要写 Agent 逻辑本身,就能得到一个能实时对话的语音智能体。
如果你准备上手,建议先跑通一个最小语音 Agent Demo,验证三件事:第一,Worker 能否正常连接 LiveKit 项目;第二,VAD 能否准确检测用户说话;第三,STT + LLM + TTS 的回环延迟在不在你的接受范围内。
最容易踩的坑是环境变量和模型 API 的限制问题。LiveKit 凭证配置错一个字段,Worker 就无法注册;模型 API 并发限制没搞清楚,多用户同时对话就会出现排队延迟。
后续可以扩展的方向很多:把 Agent 从语音扩展到视频流,让 Agent 能“看”到屏幕共享内容;接入外部知识库和数据库,让 Agent 具备业务检索能力;引入任务队列,把 Agent 改造成异步批量处理引擎;甚至可以用 LiveKit 的转写能力做会议纪要自动化。
建议先把这套框架在自己的业务场景里小范围试用,跑通后再决定是否扩大到生产环境。
火山引擎视频云技术社区,是面向 AI 音视频开发者的技术交流平台。这里汇聚源自抖音、豆包等亿级 DAU 产品的 RTC、直播、点播、AI 媒体处理、音视频互动技术,提供接入指南、最佳实践、性能调优、场景案例、Demo 代码、开源项目、白皮书和 API 文档。社区汇聚官方工程师与一线开发者,为 AI 视频通话、数字人、AI 视频处理等应用的开发与落地提供技术支持。
更多推荐
所有评论(0)