如果你想在国内跑一套实时语音对话 AgentLiveKit 加阿里云语音交互服务是我实测下来最稳的组合之一。LiveKit 负责 WebRTC 信令、媒体传输和 Agent 会话管理阿里云负责语音识别ASR和语音合成TTS中间接什么大模型可以随便换。这篇文章把整套方案的架构设计、代码接入、部署上线和踩坑修复写明白适合已经在跑 LiveKit Agents 但头疼海外语音服务延迟的人也适合刚准备做会话式 AI 项目、想少走弯路的同学。1. 这不是选型题而是生存题国内语音 Agent 怎么搭1.1 LiveKit 和 LiveKit Agents 解决了什么问题实时语音对话 Agent 的本质是“听—想—说”的闭环麦克风采集声音语音识别把它变成文本丢给大模型生成回答再通过语音合成放出来。听起来不复杂真正落地时最难的反而不是大模型而是实时通信链路。浏览器和 App 端的音频采集、弱网丢包补偿、音频编码、回声消除、打断处理这些东西如果自己从 WebRTC 裸写代价非常高。LiveKit 就是把这层底座做扎实的开源方案它提供 SFU 媒体服务器、信令服务、房间管理和一整套客户端 SDK几个命令就能拉起一个支持多人实时音视频的服务。LiveKit Agents 是官方在媒体底座之上长出来的 Agent 框架它把“听—想—说”拆成 STT、LLM、TTS 三段并且替你处理了 VAD 语音检测、说话人打断、音频流切换这些脏活。默认插件里带的是 OpenAI Whisper、Deepgram、ElevenLabs 这些海外服务。我承认本地开发和原型验证阶段这些服务开箱即用非常爽。但一旦考虑国内正式上线问题就来了一是海外节点延迟高语音识别讲究首包时延每多 100 毫秒体验都差别很大二是很多企业对数据出境有明确顾虑用户的语音内容不希望绕到海外再回来。于是最自然的思路就是LiveKit 框架完全不动把 STT 和 TTS 换成阿里云智能语音交互。LLM 那一段如果也想整体留在国内可以接兼容 OpenAI 接口的国内大模型甚至接阿里云自己的通义千问。今天要说的“语音交互服务”在阿里云产品矩阵里对应智能语音交互NLS包含实时语音识别、一句话识别、语音合成、长文本音色复刻等能力是给这套迁移方案提供识别与合成底座的。1.2 为什么把 ASR/TTS 换成阿里云这个决策背后的理由非常实际。我最初用 Deepgram 做识别离线测试准确率不错但线上用户集中在华东和华北音频从浏览器进 LiveKit SFU再跨公网送到海外识别整条链路能明显感到延迟网络一抖动还会丢结果。而阿里云智能语音交互的网关在北京、上海、杭州都有节点和 LiveKit 服务器部署在同一地域时Agent 与语音网关之间的网络路径很短实测首包时延比跨洋链路少了一大截。第二个原因是企业采购和运维习惯。海外服务通常要绑信用卡计费按秒做严肃项目时合同、发票、配额样样不方便。阿里云按语音时长计费企业认证后可以签合同、开票并发不够还能提工单扩容。AccessKey、RAM 子账号、操作审计和日志都跟现有运维习惯对齐落地阻力小很多。第三个原因很多人容易忽略音频格式对齐方便。LiveKit Agents 把远端音频交给 STT 时默认就是 16kHz、16bit、单声道 PCM正好落在阿里云实时语音识别支持的最佳参数区间。格式转换几乎不用做少踩一个最容易出 bug 的音频格式坑。这也是我敢在项目里直接替换插件而不是自研音频处理链路的底气。2. 整体链路一个语音对话是怎么跑通的2.1 从麦克风到回答的完整数据流把整条链路拆开看一次正常的语音交互大约是这样的用户打开页面或 App与 LiveKit 建立 WebRTC 连接音频流进入 SFU 转发给 Agent Worker。Agent 框架里的 VAD 先判断什么时候有真人开始说话检测到语音段后把音频帧持续推给 STT。STT 拿到 16kHz PCM 音频流通过 WebSocket 发给阿里云实时语音识别阿里云返回增量识别结果和最终整句文本。整句文本进入 LLM带上系统提示词和对话历史生成回答文本。回答文本交给 TTSTTS 调用阿里云语音合成把文本转成音频帧再通过 LiveKit 的音频轨道推回房间用户端播放出来。如果中途用户开口打断VAD 检测到新语音Agent 会触发 interrupt 机制暂停 TTS、丢弃未播完的音频优先处理新的输入。这里每一个环节都有延迟预算识别首包要快LLM 流式输出要快TTS 首帧也要快。阿里云的服务在三大件里面识别和合成都能压到几百毫秒内返回第一帧主要时间其实花在大模型生成上。所以这套架构能真正用起来的关键就是不能让 ASR/TTS 拖后腿。国内网络环境下用阿里云这一侧几乎是必然选择。2.2 阿里云智能语音交互的关键概念集成阿里云语音交互之前建议先把它的几个核心概念弄清楚不然看到文档会头大。AppKey在阿里云智能语音交互控制台创建项目后生成的应用标识相当于这个语音项目的身份证。不同项目之间资源隔离计费和配额也按项目分。集成时所有 WebSocket 请求里的 header 都要带 AppKey。AccessKey 与 TokenAccessKey ID 和 AccessKey Secret 用于调用阿里云开放 API相当于账号级别的钥匙。但实时语音识别和语音合成通常不直接拿 AccessKey 去连 WebSocket而是先用 AccessKey 换取一个临时 Token再拿 Token 建立语音连接。Token 有过期时间一般有效期在 24 小时左右所以服务端要做定时刷新不能写死在配置里。Region 与网关地址不同地域的语音网关域名不同最常用的是 cn-shanghai、cn-beijing、cn-shenzhen。Agent 服务部署在哪个地域语音网关最好也选同地域延迟最低。网关地址类似wss://nls-gateway-cn-shanghai.aliyuncs.com/ws/v1具体以官方文档为准。音色与参数语音合成有大量音色可选实时识别也有中间结果、标点预测、ITN 数字规整等开关。这些参数都是通过 WebSocket 启动消息里的 payload 字段下发的选型时要根据业务场景决定比如客服场景建议开标点预测和数字规整信息播报场景可能需要调整语速和音量。2.3 这次集成用到的组件与版本我这次实践的运行环境是这样的一台阿里云华东 2上海的 ECS跑 LiveKit Server 和 Agent WorkerAgent 使用 Python 版本livekit-agents 版本是 1.x阿里云语音交互走的是标准 WebSocket 协议直接复用官方 Python SDK 或者自己用 websockets 库对接。LLM 我用的是兼容 OpenAI 接口的国内大模型因为本文重点不在模型选型后文代码里会留一个占位。之所以强调版本是因为 LiveKit Agents 的 API 还在快速迭代不同版本的自定义 STT/TTS 接口签名有差异。网上的旧教程很多已经不能直接跑我在文中给出的代码基于 livekit-agents 1.x 的接口约定你安装新版本后如果发现方法名对不上以你本地的类型声明为准改一下参数即可整体结构是不变的。3. 动手集成把阿里云语音装进 LiveKit Agents3.1 开通服务、创建项目并搞定 Token 鉴权第一步是开通阿里云智能语音交互服务。登录阿里云控制台搜索“智能语音交互”进入产品页按引导开通。开通后在控制台创建项目拿到 AppKey。同时为项目配置一个 RAM 子账号最小权限原则只授语音服务的调用权限避免把主账号 AccessKey 直接暴露到服务器上。接下来是 Token 获取。我建议在 Agent 进程里封装一个 TokenProvider启动时用 AccessKey 获取一次 Token放在内存里快过期前再刷新。用一个简单的 HTTP 请求就能拿到 Token接口地址和签名方式以阿里云官方文档为准大致逻辑是# 伪代码实际签名参数按官方文档填写 curl -X POST https://nls-meta.cn-shanghai.aliyuncs.com/api/v1/token \ -H Content-Type: application/json \ -d {accessKeyId:...,accessKeySecret:...,grantType:...}拿到响应里的 Token 和过期时间点后写进一个异步刷新任务里。Token 刷新周期建议设为过期时间的一半留出缓冲不要在每次语音连接时临时去获取否则高并发下会出现大量鉴权请求打满限额。3.2 给 LiveKit 写一个阿里云 STT 插件LiveKit Agents 的自定义 STT 核心是继承livekit.agents.stt.STT实现流式识别方法。阿里云实时语音识别走 WebSocket客户端先把启动消息发过去之后持续推二进制 PCM 音频服务端返回中间 result 和最终句子的 JSON 事件。整套流程放在一个异步生成器里返回给 LiveKit代码结构大致是import asyncio import json import uuid import websockets from livekit.agents import stt from livekit.agents.types import AudioFrame class AliyunNLSSTT(stt.STT): def __init__(self, appkey: str, token_provider, region: str cn-shanghai): super().__init__(capabilitiesstt.STTCapabilities( languages[zh-CN], sample_rate16000, channels1, )) self._appkey appkey self._token_provider token_provider self._gateway fwss://nls-gateway-{region}.aliyuncs.com/ws/v1 async def recognize(self, buffer, *, languageNone): token await self._token_provider.get_token() async def _stream(): async with websockets.connect(self._gateway) as ws: await ws.send(json.dumps({ header: { message_id: uuid.uuid4().hex, appkey: self._appkey, token: token, action: start-transcription, }, payload: { format: pcm, sample_rate: 16000, enable_intermediate_result: True, enable_punctuation_prediction: True, }, context: {service: livekit-agent}, })) send_task asyncio.create_task(self._push_audio(ws, buffer)) async for raw in ws: # raw 可能是文本事件也可能是二进制帧 # 解析 SentenceBegin / ResultChanged / SentenceEnd 事件 # 对结果封装成 stt.STTSegment 并 yield pass await send_task return stt.STTSegments(stream_stream()) async def _push_audio(self, ws, buffer): async for frame in buffer: # 如果采样率不是 16kLiveKit 框架通常会提前转好 if isinstance(frame, AudioFrame): await ws.send(frame.data.tobytes())这里有个细节LiveKit 的 STT 输入 buffer 不是普通的 async iterator它是有结束标记的。实现_push_audio时要注意对结束信号的处理当说话人停止时把连接发一个 stop 事件告诉阿里云“这句结束了”然后等服务端把最终完整句子返回过来再退出。否则容易出现整句识别结果缺失、只有增量词的尴尬情况。3.3 给 LiveKit 写一个阿里云 TTS 插件TTS 的方向正好反过来LiveKit 把文本给你你调用阿里云语音合成拿到音频帧之后封装成tts.SynthesizedAudio输出。阿里云语音合成同样是 WebSocket 协议客户端发启动消息服务端持续推二进制音频最后发一个结束事件。from livekit.agents import tts class AliyunNLSTTS(tts.TTS): def __init__(self, appkey: str, token_provider, voice: str zhixiaobai, region: str cn-shanghai): super().__init__(capabilitiestts.TTSCapabilities( sample_rate24000, channels1, )) self._appkey appkey self._token_provider token_provider self._voice voice self._gateway fwss://nls-gateway-{region}.aliyuncs.com/ws/v1 async def synthesize(self, text: str, *, languageNone): audio_chunks [] async def _synthesize_and_collect(): token await self._token_provider.get_token() async with websockets.connect(self._gateway) as ws: await ws.send(json.dumps({ header: { message_id: uuid.uuid4().hex, appkey: self._appkey, token: token, action: start-synthesis, }, payload: { text: text, voice: self._voice, format: wav, sample_rate: 24000, volume: 50, speech_rate: 0, pitch_rate: 0, }, })) async for raw in ws: if isinstance(raw, bytes): audio_chunks.append(raw) else: # 服务端会返回结束事件收到后 break pass await _synthesize_and_collect() # 把收到的 PCM/WAV 封装成 livekit.agents.tts.SynthesizedAudio yield tts.SynthesizedAudio( texttext, datab.join(audio_chunks), )关于采样率要特别说一句。阿里云 TTS 输出 24kHz 或 16kHz 都支持但 LiveKit 播放链路对采样率有要求最好在 TTS 输出端就统一不要指望播放端帮你做高质量重采样。如果你在代码里看到奇怪杂音或者音调不对先检查这一段是不是采样率对齐了。3.4 把插件接进 Agent 会话的完整配置STT 和 TTS 都写好之后接进 Agent 就很简单了。在 LiveKit Agents 的入口文件里用自定义类替换默认插件from livekit.agents import AgentSession, JobContext, llm async def entrypoint(ctx: JobContext): await ctx.connect() session AgentSession( sttAliyunNLSSTT(appkeyAPPKEY, token_providertoken_provider), llmcreate_llm(), # 兼容 OpenAI 接口的国内大模型 ttsAliyunNLSTTS(appkeyAPPKEY, token_providertoken_provider), ) await session.start( roomctx.room, agent_namealiyun-voice-agent, )接完之后建议做一轮音频参数校准阿里云实时识别最稳的参数是 16kHz 单声道 PCMTTS 输出如果选 16kHzLiveKit 端不用额外转换。真实项目里我还遇到过 echo 和噪声把 VAD 触发搞乱的场景这种问题不在语音服务商而在客户端采集参数和降噪配置要单独处理。4. 部署到阿里云 ECS 并接入域名与证书4.1 服务器规划与 Docker 部署 LiveKit部署前先规划服务器拓扑我推荐一种简单又清晰的方案一台 ECS 同时跑 LiveKit Server、Agent Worker 和 Nginx初期用户量不大时完全够用。Agent Worker 不要直接暴露给外部只允许访问 LiveKit Server 的 7880 信令端口其他端口一律内网访问。LiveKit Server 用官方 Docker 镜像最省事。先在服务器上装好 Docker 和 docker-compose然后准备livekit.yaml配置文件重点配置信令端口、RTC 端口和 keys。keys 是服务器与客户端通信的共享密钥自己生成一组随机字符串填进去。启动命令大致是docker run --rm -p 7880:7880 \ -p 7881:7881 \ -p 50000-60000:50000-60000/udp \ -v /opt/livekit/livekit.yaml:/etc/livekit.yaml \ livekit/livekit-server --config /etc/livekit.yamlAgent Worker 我建议用 Python 虚拟环境跑代码通过 Git 发布。进程管理用 systemd 或 supervisor 都行关键是崩溃后要能自动拉起日志要轮转。第一次上线时我因为没配置 systemd RestartalwaysAgent 半夜崩了一次直到第二天才发现家里线上服务挂了好几个小时这种低级错误希望大家一次都不要踩。4.2 Agent 服务的进程模型与令牌刷新Agent Worker 本质是一个长期运行的 Python 进程连接 LiveKit Server 并订阅任务。由于实时语音识别和合成都要保持 WebSocket 长连接进程内部会有大量异步任务所以要用 asyncio 模型来写不要混用同步阻塞库否则高并发下整个进程都会卡。TokenProvider 在这个环节要设计成线程安全、且进程内共享。我提供一个实现要点初始化时同步获取一次 Token 并写入变量每隔一段时间由后台任务刷新刷新时加锁防止多个语音请求同时发现 Token 过期、同时去刷新导致重复请求。另外Token 的有效期不要卡着边界用建议在剩余 5 分钟时就开始刷新网络抖动也能覆盖。多 Agent Worker 的场景可以做水平扩容多个 Worker 进程共享同一个 LiveKit Server任务由 LiveKit 分发。这个时候 Token 刷新逻辑必须统一最好抽成一个独立的刷新服务或 Redis 缓存避免每个 Worker 各自拿着一份快要过期的 Token。4.3 安全组、SSL 证书与 WebRTC 端口阿里云 ECS 安全组是一个极易踩坑的地方。LiveKit 默认需要放行这些端口TCP 7880 用于 WebSocket 信令转发TCP 7881 用于 RTC 通信UDP 50000-60000 是 WebRTC 媒体端口。如果只放 7880页面能连上但声音出不来就是因为媒体端口被挡了。HTTPS 和 WSS 证书建议直接使用阿里云 SSL 证书服务。域名解析到 ECS 公网 IP 后申请免费证书并绑定再把 Nginx 配置为反向代理把/路径转发到 LiveKit Server 的 7880 端口。WebRTC 的媒体流走的不是 HTTP而是直接的 UDP 传输所以证书只影响信令连接不影响媒体连通性但现代浏览器要求安全上下文才允许开启麦克风没有 HTTPS 基本没法调试这一步必须做好。我遇到过一个诡异问题页面偶尔能连上偶尔一直转圈后来排查发现是安全组把 UDP 端口范围只放了一半导致部分媒体传输失败。建议部署后用 LiveKit 官方诊段工具或手动测一下 UDP 端口连通性别等到线上用户反馈。5. 高频报错排查与延迟调优实录5.1 高频报错排查速查表集成和上线过程中我遇到最多的问题集中在鉴权、音频格式和网络三个方面整理成速查表供大家直接查。现象大概率原因解决办法连接阿里云返回 401Token 过期或 AppKey 错误检查 TokenProvider 刷新逻辑确认 AppKey 与地域匹配识别结果全空启动消息参数错误或没有发送 stop 事件按官方文档核对 header 和 payload 字段补全结束标志识别乱码、音调异常采样率或音频格式不对确认发给阿里云的音频是 16kHz 16bit 单声道 PCMTTS 只有前半句收到第一个结束事件就关闭连接需要等待服务端完整推完所有二进制音频再关闭页面连不上麦克风HTTPS 未配置或麦克风权限问题配置 SSL 证书浏览器审查下确认安全上下文能连上但没声音安全组未放行 UDP 媒体端口放行 UDP 50000-60000检查服务器防火墙Agent 进程神秘退出asyncio 任务异常未被捕获给所有 task 加 exception handler配置日志输出并发一高就大量超时阿里云并发配额不够控制台查看配额必要时提工单扩容其中 TTS 只生成前半句这个问题我要重点说一下。阿里云语音合成返回二进制音频的同时最后会有一个结束事件程序里必须正确解析到这个结束事件再关闭连接。我第一次实现时想当然地以为收到固定字节数就结束了结果长文本被截断主播播报类场景非常明显。后来改成循环接收直到收到结束标记问题解决。5.2 延迟与并发的实际调优经验先看延迟。实时语音对话里用户能感知的主要是三个时间点说完话到看到转写或听到回响的时间、大模型首 token 的时间、TTS 首音频帧的时间。阿里云这边能优化的主要是第一个和第三个。识别延迟方面开启中间结果后虽然最终整句识别要等说完才返回但增量识别结果可以提前给业务侧做展示或预判。如果业务对首包时延极其敏感可以把enable_intermediate_result打开并适当调短max_sentence_silence这类静音判断参数。不过调得太短会把正常停顿切碎导致一句完整的话被拆成多段反而影响大模型理解我目前一般保持默认值。TTS 延迟方面文本过长时合成耗时明显增加。对长文本可以先做分句逐句调用合成但要注意句子边界要断在语义完整的位置否则合成语调会不自然。另一个技巧是预连接提前把到阿里云语音网关的 WebSocket 连接池化省去每次连接握手的时间高并发下效果非常明显。并发方面阿里云语音交互有并发路数限制超过配额会直接报错或排队。如果你做的是客服机器人这种高峰时段流量集中的业务一定要提前估算并发并测试限制。我曾经在压测时把并发调得太高阿里云侧报了不少 resource exhausted后来紧急提工单临时扩容才稳住。建议上线前先做一次真实的并发压测确定上限后在上游加限流别把压力全怼到语音服务上。另外关于 Agent 本身如果同一时间房间太多单个 Agent Worker 的 CPU 和内存也会吃紧。遇到这类问题不要急着加服务器先检查是不是有音频转码或 VAD 占用过高。LiveKit Agents 默认的 Silero VAD 在 CPU 上跑很快但如果你的机器性能很弱可以降低采样率或减少同时处理的音频路数。最后再说一个小技巧在 Agent 代码里把每一次 ASR 和 TTS 的关键节点耗时都打出来记录首帧时间、结束事件时间、总耗时。排查延迟问题时这些日志比什么监控都管用。我在生产环境里加了结构化日志出了问题直接按 message_id 拉全链路时间线几分钟就能定位是阿里云慢还是大模型慢还是网络慢。写在最后这套方案我从原型验证到现在稳定跑了大半年最深的感受是语音 Agent 不是“大模型接上去就行”的玩具ASR/TTS 这层底座的选择直接决定线上体验。LiveKit 的框架灵活度足够好阿里云语音交互的国内接入质量和稳定性也对得起生产环境的标准。如果你也在做中文语音 Agent照着这条路线走能少走很多弯路。