今年上半年我们团队用六个月时间从零搭起了一套面向语音 AI 的实时响应系统核心目标是降低“用户说完话到 AI 回复出声”的端到端延迟同时保证在高并发对话场景下的稳定性。这段时间踩过的坑和沉淀下来的设计经验不少本文会以这套系统为线索完整拆解实时语音 AI 系统的架构分层、流式链路、延迟优化方法、工程落地要点和常见问题排查思路希望能给正在做语音助手、实时会议转写、智能客服语音通道的同学一些参考。1. 实时语音 AI 系统到底是什么1.1 从“能对话”到“实时对话”的距离很多入门项目里的语音助手是“先录音、再识别、再回答、再播放”的串行流程用户对着设备说完一句话系统等到整段音频结束才开始处理整个过程存在明显停顿体验更像“对讲机”而不是“通话”。实时语音 AI 系统的核心区别在于音频数据是边产生、边传输、边识别、边推理、边合成的。系统不需要等用户说完一整句话再行动而是可以在用户还在说话的时候就开始做语音活动检测VAD、流式语音识别ASR、甚至提前触发部分回复逻辑。这种设计带来的直接好处是响应延迟大幅降低也是语音 AI 产品从“可用”走向“好用”的关键。1.2 系统包含哪些核心模块一个完整的实时语音 AI 系统通常包含以下模块模块作用常见技术方向音频采集从麦克风/电话通道采集音频WebRTC、PCM、Opus语音活动检测 VAD判断什么时候开始说话、什么时候结束WebRTC VAD、Silero VAD流式语音识别 ASR将连续音频流实时转换为文本Whisper、FunASR、Kaldi对话管理/LLM 推理根据文本生成回复内容支持增量输出GPT 系列、开源 LLM、RAG流式语音合成 TTS将文本逐步合成为音频输出CosyVoice、ChatTTS、Edge TTS音频播放在客户端低延迟播放WebRTC、AudioTrack这些模块不是简单串联而是需要通过实时通信链路和流式协议协同工作任何一个环节出现高延迟都会直接影响用户感知。1.3 实时响应系统的关键指标评价这类系统时大家通常关注以下几个指标端到端延迟E2E Latency从用户开始说话到系统开始播放回复音频的时间。首包延迟First Packet Latency用户说完到第一个合成音频包到达客户端的时间。识别延迟ASR Delay)从说话到完整文本输出之间的延迟。字级/词级增量延迟流式引擎每输出一个字的间隔时间。并发会话数单机或集群能同时维持多少路实时会话。在真实项目中我们一般把“用户说完话到 AI 开始说话”的目标定在 800ms 以内把首包延迟控制在 300ms 左右这一个目标倒逼整个链路做流式化改造。2. 环境准备与整体架构设计2.1 项目背景与团队分工这个项目不是从纯零开始团队在算法侧已经有可用的 ASR 和 TTS 模型但都是离线批处理模式。我们的核心任务是把这些算法能力封装成低延迟、可并发的实时在线服务并设计一套能承载多路会话的交付架构。团队总共投入约 8 人角色分布如下2 人负责音频链路和流式协议2 人负责 ASR/TTS 服务封装与并发优化2 人负责 LLM 推理链路和流式输出1 人负责网关、会话管理和部署1 人负责音质测试、延迟测试和线上监控2.2 技术选型说明这里我们不列出厂商强绑定的私有协议重点讲可以自建的部分语言Python 负责算法服务封装Go 负责高并发网关和会话管理。音频传输WebSocket 承载双向二进制音频流。流式协议自定义 JSON 协议约定connect、audio、text、state、error等消息类型。ASR采用可以流式输出的识别方案文本按句返回同时给出中间结果和最终结果。LLM使用支持 SSE 流式输出的对话模型回复按 token 增量返回。TTS采用支持流式合成的方案输入文本可以分句送入输出音频包按序返回。部署Docker 容器化Kubernetes 管理服务节点Nginx 做 WebSocket 代理。环境版本不能直接照搬这里只说明我们的选择Python 3.10、Go 1.21、Redis 7 用于会话状态缓存Kafka 用于异步事件和数据回放。2.3 系统整体链路为了方便理解下面用文字描述整个调用链客户端麦克风 ↓ PCM/Opus 音频流 WebSocket 网关鉴权、会话路由 ↓ 二进制帧 音频前置服务VAD、静音检测、音频重采样 ↓ 有效语音音频片段 流式 ASR 服务输出增量识别文本 ↓ 中间结果/最终结果 对话服务LLM 流式推理可拼接上下文 ↓ SSE token 流 回复分发服务分句、标点恢复 ↓ 文本分段 流式 TTS 服务逐句合成音频 ↓ 音频包 WebSocket 网关 → 客户端播放这个链路中用户感受到的响应时间 VAD 判定说话结束时间 ASR 最终文本时间 LLM 首个 token 时间 TTS 首包时间 网络传输时间。因此每一层都要做流式优化。2.4 项目目录结构为了后续讲解清楚我们把项目简化成以下目录realtime-voice-ai/ ├── gateway/ # WebSocket 网关服务Go │ ├── main.go │ ├── session.go │ └── protocol.go ├── services/ │ ├── vad/ # VAD 服务Python │ ├── asr/ # 流式 ASR 服务Python │ ├── llm/ # LLM 对话服务Python │ └── tts/ # 流式 TTS 服务Python ├── proto/ │ └── message.json # 消息协议定义 ├── deploy/ │ ├── docker-compose.yml │ └── nginx.conf └── tests/ ├── latency_test.py └── audio_test.py这个结构按服务边界拆开方便独立部署和扩容。接下来我们逐层讲解核心实现。3. 核心模块原理与流式实现3.1 音频流协议设计实时语音系统最基础的是音频数据如何在客户端与服务端之间传输。这里我们采用 WebSocket 承载二进制音频帧同时在相同连接上传输 JSON 控制消息。协议消息分为几类connect建立会话带上客户端 ID、采样率、编码格式。audio二进制音频帧通常每帧 20ms 或 40ms。text文本消息包括 ASR 中间结果、最终结果、LLM 回复文本。state状态消息例如 vad_start、vad_end、tts_start、tts_end。error异常信息。以音频帧为例我们的 JSON 头部如下{ type: audio, session_id: session_001, timestamp: 1720000000123, sequence: 123, format: pcm_16k, duration_ms: 40 }音频帧本身不放在 JSON 里而是在 JSON 之后紧跟二进制数据。读取端先按分隔符读一行 JSON再按长度读音频数据。Go 网关里核心逻辑类似// 文件路径realtime-voice-ai/gateway/session.go // 伪代码读取 WebSocket 消息并解析协议 func (s *Session) handleMessage(msgType int, data []byte) { if msgType websocket.BinaryMessage { // 二进制帧第一部分 JSON 头 后续音频数据 headerLen : binary.BigEndian.Uint32(data[:4]) header : parseHeader(data[4 : 4headerLen]) audioData : data[4headerLen:] s.routeAudio(header, audioData) } else if msgType websocket.TextMessage { msg : parseTextMessage(data) s.routeControl(msg) } }实际项目中我们没有在同一个 WebSocket 里混用文本和二进制而是统一采用“JSON 头 二进制体”的格式所有消息都按同样方式解析简化逻辑。3.2 VAD判断“开始说话”和“结束说话”VAD 是实时系统的节拍器。它决定了音频流中哪些片段需要送识别更重要的是决定了用户什么时候说完一句话。常见做法能量阈值短时能量超过阈值视为有声。模型 VAD用 Silero VAD 等模型输出每帧语音概率。惩罚机制连续 N 帧静音才判定结束。我们最初使用 WebRTC VAD优点是快缺点是在噪声环境下误判多。后来换成 Silero VAD准确率明显提升。VAD 具体输出逻辑# 文件路径realtime-voice-ai/services/vad/vad_service.py # 伪代码基于 Silero VAD 的流式检测 import torch model, utils torch.hub.load( repo_or_dirsnakers4/silero-vad, modelsilero_vad, trust_repoTrue ) def detect_speech(audio_chunk, sample_rate16000): # audio_chunk 为 30ms 或 60ms 的 float32 数组 with torch.no_grad(): speech_prob model(torch.from_numpy(audio_chunk), sample_rate).item() return speech_prob工程上不会把每一帧的 VAD 概率直接暴露给上层而是维护一个状态机silence - speech 概率 0.8 连续 5 帧 speech - silence 概率 0.3 连续 10 帧结束判定要足够保守避免用户中间停顿一下就被误判为说完。3.3 流式 ASR 服务封装ASR 是链路里最影响延迟的模块之一。我们要做两件事让 ASR 支持增量输入实时输出中间识别文本。在 VAD 判定结束时输出当前句的最终文本。下面演示服务端如何通过 WebSocket 接收音频并返回结果。这里使用一个简化的流式 ASR 客户端逻辑# 文件路径realtime-voice-ai/services/asr/asr_server.py # 伪代码接收音频帧并回调识别结果 import asyncio import json class RealtimeASR: def __init__(self, recognizer): self.recognizer recognizer self.audio_buffer b async def process_audio(self, audio_chunk: bytes, is_final: bool): self.audio_buffer audio_chunk # 这里调用底层流式识别模型的 decode 接口 partial self.recognizer.decode(self.audio_buffer, is_finalis_final) return { type: asr_result, text: partial[text], is_final: is_final, confidence: partial.get(confidence, 0.0) }关键点is_finalFalse时ASR 返回中间结果前端可以展示但不要播报。is_finalTrue时ASR 返回最终结果后续 LLM 才拿这个文本走回复逻辑。识别缓存要及时清理避免长对话内存增长。3.4 LLM 流式对话服务LLM 是系统里最“智能”也最不稳定的模块。实时语音系统中我们希望 LLM 的回复像打字机一样增量输出而不是等全部生成完再交给 TTS。如果使用 OpenAI 兼容接口通常会配置streamTrue然后逐段读取响应的内容。下面是封装示例# 文件路径realtime-voice-ai/services/llm/llm_stream.py # 伪代码流式请求 LLM按 token 回调 import json import requests def stream_llm_reply(messages, on_token): # 注意这里以常见 SSE 格式为例具体字段依赖模型服务 resp requests.post( https://api.example.com/v1/chat/completions, json{ model: voice-llm, messages: messages, stream: True, }, streamTrue, timeout10 ) for line in resp.iter_lines(): if not line: continue line line.decode(utf-8) if line.startswith(data:): data_str line[5:].strip() if not data_str or data_str [DONE]: continue try: data json.loads(data_str) token data[choices][0][delta].get(content, ) if token: on_token(token) except Exception: # 跳过不完整数据保证流式解析不中断 continue这里有个细节LLM 流式输出可能一次返回多个 token也可能一个 token 分两次返回所以必须用on_token回调把文本累积再交给分句模块。分句策略优先按标点切分遇到句号、问号、感叹号时认为一句已经完整。如果没有标点则累积到 20 个字左右强制切分避免 TTS 等待过长。切分后的句子要保留上下文用于 TTS 合成时保持语义完整。3.5 流式 TTS 服务封装TTS 是响应质量的关键。LLM 输出文本后TTS 还在合成用户已经听到声音了。因此我们要边合成边播放而不是等整句话合成完再播放。现在的开源 TTS 一般都支持 batch 合成但流式效果差异很大。封装思路如下# 文件路径realtime-voice-ai/services/tts/tts_stream.py # 伪代码TTS 增量合成并返回音频块 class StreamTTS: def __init__(self, synthesizer): self.synthesizer synthesizer def synthesize_segment(self, text: str): # 分句后逐句合成 audio_chunks self.synthesizer.synthesize_stream(text) for chunk in audio_chunks: yield { type: audio, audio: chunk, format: pcm_16k }常见问题有些 TTS 引擎有句首延迟也就是不管文本多短都要先跑一个固定时常的预处理。这会严重影响体验。解决方案是在首句还没合成完时先用“开场音/呼吸声”占位或者后端提前把常用回复模板缓存起来。4. 完整的实时链路实战这一节我们把前面模块串起来演示一个最小可运行的服务端链路。这里代码做了简化重点展示各模块如何配合。4.1 创建项目结构先创建目录mkdir -p realtime-voice-ai/{gateway,services,proto,tests} cd realtime-voice-ai4.2 编写 WebSocket 网关网关负责接收客户端音频流转发给后续服务。这里用 Go 写一个最小网关// 文件路径realtime-voice-ai/gateway/main.go package main import ( log net/http github.com/gorilla/websocket ) var upgrader websocket.Upgrader{ CheckOrigin: func(r *http.Request) bool { return true }, } type Client struct { conn *websocket.Conn } func main() { http.HandleFunc(/ws, handleWS) log.Println(gateway listen on :8080) log.Fatal(http.ListenAndServe(:8080, nil)) } func handleWS(w http.ResponseWriter, r *http.Request) { conn, err : upgrader.Upgrade(w, r, nil) if err ! nil { log.Println(upgrade error:, err) return } client : Client{conn: conn} go client.readLoop() } func (c *Client) readLoop() { defer c.conn.Close() for { _, data, err : c.conn.ReadMessage() if err ! nil { log.Println(read error:, err) return } // data 包含 JSON 头和音频数据这里转发给下游服务 // 例如sendToASR(data) log.Printf(recv packet len%d, len(data)) } }这部分是最外层入口实际项目还需要做鉴权、会话绑定、断线重连。4.3 编写音频策略和 VAD网关收到音频后先交给 VAD 判断是否有语音。这里写一个示例策略# 文件路径realtime-voice-ai/services/vad/vad_router.py # 伪代码音频帧路由 VAD 状态机 class AudioRouter: def __init__(self): self.state silence self.speech_frames 0 self.silence_frames 0 def feed(self, frame_prob, audio_chunk): if self.state silence: if frame_prob 0.8: self.speech_frames 1 if self.speech_frames 5: self.state speech self.speech_frames 0 return vad_start else: self.speech_frames 0 else: if frame_prob 0.3: self.silence_frames 1 if self.silence_frames 10: self.state silence self.silence_frames 0 return vad_end else: self.silence_frames 0 return None这里的参数不是固定的需要根据实际麦克风采样率、环境噪声调整。核心是想清楚VAD 要尽可能避免“话没说完就切掉”的场景。4.4 编写流式 ASR 转发逻辑当 VAD 判定 speech 开始后后续音频持续送入 ASR。ASR 返回增量文本系统把文本发回客户端作为中间结果同时在 VAD 结束时把最终文本发给 LLM。简化代码如下# 文件路径realtime-voice-ai/services/asr/asr_worker.py # 伪代码接收音频帧并回调识别结果 class ASRWorker: def __init__(self, recognizer): self.recognizer recognizer self.buffer b def feed(self, audio, is_final): self.buffer audio text self.recognizer.recognize(self.buffer, finalis_final) if is_final: self.buffer b return text实际项目里要注意并不是所有识别器都支持任意长度的缓冲输入很多流式模型有固定窗口需要内部维护状态向量不能简单“拼接音频”。这里只是演示消息流。4.5 编写 LLM 回复与 TTS 串接LLM 输出文本后我们按句子切分逐句交给 TTS 合成。这里用一个简化的事件驱动写法# 文件路径realtime-voice-ai/services/dialog/dialog_service.py # 伪代码LLM token - 分句 - TTS - audio event class DialogService: def __init__(self, llm_stream, tts_stream): self.llm_stream llm_stream self.tts_stream tts_stream self.pending_text def handle_user_text(self, user_text, on_audio): messages [{role: user, content: user_text}] def on_token(token): self.pending_text token if self.should_split(self.pending_text): sentence self.pending_text.strip() self.pending_text if sentence: for audio_chunk in self.tts_stream.synthesize_segment(sentence): on_audio(audio_chunk) self.llm_stream.stream_llm_reply(messages, on_token)这里should_split可以简单判断是否包含句末标点或者缓冲长度是否超过设定阈值。实际系统里还要考虑语气停顿、数字读法、复读等问题。4.6 运行与验证启动网关和 VAD/ASR 服务后可以用 WebSocket 客户端测试# 文件路径realtime-voice-ai/tests/ws_client.py import asyncio import json import websockets async def test(): async with websockets.connect(ws://localhost:8080/ws) as ws: # 发送音频帧这里假设读取本地 pcm 文件 with open(test_audio.pcm, rb) as f: audio_data f.read(3200) # 200ms 16kHz 16bit await ws.send(audio_data) while True: msg await ws.recv() print(recv:, msg[:100]) asyncio.run(test())预期输出应能看到服务端逐步返回识别文本和合成音频事件。4.7 结果说明这个最小链路虽然不能直接上生产但验证了核心问题音频可以从客户端持续流向 ASR识别文本可以增量返回LLM 回复可以边生成边合成。后面所有性能优化都是在这个链路基础上做减法。5. 延迟优化与性能调优5.1 端到端延迟拆解我们在完成基础链路后针对线上会话统计了耗时分布典型情况如下阶段平均耗时优化目标音频上行传输40ms30msVAD 判定结束200ms120msASR 最终结果250ms180msLLM 首 token800ms500msTTS 首包300ms220ms音频下行传输40ms30ms总耗时约 1.6 秒。优化重点在 LLM 和 ASR/TTS 的配合上。5.2 并行化与预判一个很有效的优化是在用户还没说完话之前系统已经可以开始推理。具体做法VAD 检测到 speech 后ASR 持续给出中间文本。当中间文本已经累积成一个完整问句例如“今天天气怎么样”对话服务可以先启动一个“预备请求”。如果用户后面继续补充内容再取消或修正之前的请求。这种“提前响应”会带来准确率风险但可以大幅降低首 token 延迟。我们实际采用的是“半预判”策略只对命中了意图模板的句子提前触发回复不全局开启。5.3 减少 LLM 等待时间LLM 模块是最大的延迟来源优化手段包括使用支持快速首 token 输出的推理服务。减少历史消息数量只保留最近几轮关键信息。对固定开场白、欢迎语、常见问答做模板缓存。将系统提示词压缩减少每轮请求的输入 token。如果企业部署私有模型还可以使用模型量化、动态批处理、KVCache 优化这些都可以显著降低首个 token 输出时间。5.4 TTS 缓存与分句优化TTS 首包延迟受文本长度影响较大工程上可以对常用回复做音频缓存命中缓存直接发送。对长文本优先合成第一句并立刻发送后续句子边合成边发。使用更小的音频 chunk例如 20ms 一包降低播放端缓冲。另外要注意不同 TTS 引擎的音频格式可能不一致要统一转成 16kHz 16bit PCM再由客户端编码成 Opus 传输减少带宽。6. 常见问题与排查思路6.1 声音断句太早现象用户还在思考停顿系统已经判定说话结束并开始回复。可能原因VAD 静音阈值太短。解决思路调大 speech 到 silence 的连续帧数。结合语义判断如果 ASR 中间文本结尾不是标点不触发结束。在静音期间设置“临时结束”状态等待 500ms 后再决定是否结束。6.2 识别结果乱序现象同一会话里后发送的音频帧结果先返回。可能原因ASR 服务并发处理时依赖了请求级状态没有严格按 sequence 排序。解决思路每条音频消息带上递增序号。输出端按序号缓冲乱序包排队等待。ASR 解码节点只接收单路音频流避免并发写入。6.3 声音卡顿或断续现象播放的 AI 回复音频中途卡顿。可能原因网络 jitter 导致音频包到达不均匀客户端播放缓冲太小。解决思路客户端增加 100ms 左右的 jitter buffer。服务端连续发送音频包时做平滑排队。如果使用 WebRTC可以开启 NACK 或 FEC但会增加延迟需权衡。6.4 LLM 响应中断现象模型回复到一半连接断开没有任何音频输出。可能原因上游 LLM 请求超时、连接被服务端关闭、推理节点 OOM。解决思路客户端先播放“请稍等”音频兜底。对话服务设置较短的读超时第一时间重试。解析 SSE 时对异常行做容错不直接中断。6.5 长对话内存增长现象服务运行一段时间后内存持续上升。可能原因ASR 状态缓存、LLM 历史消息、音频 buffer 没有清理。解决思路为每个会话设置最大空闲时间超时自动回收。限制历史消息条数。音频帧处理完立即置空不要持有引用。6.6 排错清单遇到问题可以按以下顺序排查检查项命令/方式网关连接是否正常curl -i http://localhost:8080/healthz音频是否上行在网关打印recv packet lenVAD 是否正确触发观察vad_start/vad_end日志ASR 是否返回文本打印asr_result事件LLM 是否返回 token打印stream_llm_reply回调TTS 是否合成音频打印audio事件长度延迟分布埋点统计各阶段 timestamp7. 最佳实践与工程建议7.1 编码与协议规范所有时间戳使用毫秒级 Unix 时间服务端统一时区。消息类型不要用纯数字使用可读字符串方便日志排查。音频编码格式必须在连接时协商避免服务端猜。协议升级时保留 version 字段便于灰度兼容。7.2 会话状态管理实时语音系统是典型的有状态服务不能像普通 HTTP 一样无脑水平扩容。使用 Redis 保存会话上下文、用户偏好、最近几轮文本。WebSocket 网关节点与 ASR/LLM/TTS 服务通过消息队列或内部 RPC 解耦。会话迁移要谨慎尽量不要在服务重启时丢失 active session。7.3 延迟监控与告警实时系统最重要的可观测性指标是延迟而不是功能可用性。在客户端上报每个事件的 timestamp服务端计算每个阶段耗时。对 p95、p99 延迟设置告警。对“用户说完话 2 秒内没有音频输出”这样的体验级事件进行离线分析。7.4 安全与权限边界WebSocket 连接需要鉴权禁止未授权设备直接连入。音频数据可能包含敏感信息传输层建议使用 WSS。不要在日志里打印完整用户语音内容脱敏后再存。LLM 接入时对输入文本做基本安全过滤避免不适当内容流向用户。7.5 灰度发布与回滚由于语音链路涉及多个服务上线任何一个模块都可能影响整体体验。建议按会话维度灰度内部白名单优先体验新链路。新旧链路并行通过流量比例逐步切换。任何一个服务出现错误率上升立即切回上一版本。模型更新前先在离线音频集上回归测试观察 ASR 准确率和 TTS 自然度。8. 项目复盘与下一步学习方向经过六个月打磨这套实时语音 AI 系统从最初每秒只能处理几十路会话到现在可以稳定支撑上千路并发端到端延迟从最开始的 2.5 秒降到 0.9 秒左右。这个过程中最大的体会是实时语音系统的难点不在单个模型而在把所有模型和工程组件以“流”的方式组织起来。如果你们也准备搭建类似系统建议按以下顺序推进先把最简单的一条串行链路跑通确认每个服务都能通信。再用 VAD 控制语音边界减少无效音频。接着把 ASR 改成流式输出让文本增量可见。然后接 LLM 流式回复解决分句和取消策略。最后把 TTS 变成边合成边播放完成端到端实时化。每一步都做延迟埋点用数据判断瓶颈在哪里。不要一上来就追求全链路流式那样排错会非常困难。下一步可以继续深入研究的方向包括多说话人分离、自动打断处理、误唤醒抑制、实时情感识别、端侧模型推理优化。如果你们团队正在做语音相关产品希望这篇文章能成为一份值得收藏的架构参考。遇到具体问题也欢迎在评论区交流我会根据实际情况补充更细致的踩坑记录。 实时语音AI系统架构与延迟优化实战
今年上半年我们团队用六个月时间从零搭起了一套面向语音 AI 的实时响应系统核心目标是降低“用户说完话到 AI 回复出声”的端到端延迟同时保证在高并发对话场景下的稳定性。这段时间踩过的坑和沉淀下来的设计经验不少本文会以这套系统为线索完整拆解实时语音 AI 系统的架构分层、流式链路、延迟优化方法、工程落地要点和常见问题排查思路希望能给正在做语音助手、实时会议转写、智能客服语音通道的同学一些参考。1. 实时语音 AI 系统到底是什么1.1 从“能对话”到“实时对话”的距离很多入门项目里的语音助手是“先录音、再识别、再回答、再播放”的串行流程用户对着设备说完一句话系统等到整段音频结束才开始处理整个过程存在明显停顿体验更像“对讲机”而不是“通话”。实时语音 AI 系统的核心区别在于音频数据是边产生、边传输、边识别、边推理、边合成的。系统不需要等用户说完一整句话再行动而是可以在用户还在说话的时候就开始做语音活动检测VAD、流式语音识别ASR、甚至提前触发部分回复逻辑。这种设计带来的直接好处是响应延迟大幅降低也是语音 AI 产品从“可用”走向“好用”的关键。1.2 系统包含哪些核心模块一个完整的实时语音 AI 系统通常包含以下模块模块作用常见技术方向音频采集从麦克风/电话通道采集音频WebRTC、PCM、Opus语音活动检测 VAD判断什么时候开始说话、什么时候结束WebRTC VAD、Silero VAD流式语音识别 ASR将连续音频流实时转换为文本Whisper、FunASR、Kaldi对话管理/LLM 推理根据文本生成回复内容支持增量输出GPT 系列、开源 LLM、RAG流式语音合成 TTS将文本逐步合成为音频输出CosyVoice、ChatTTS、Edge TTS音频播放在客户端低延迟播放WebRTC、AudioTrack这些模块不是简单串联而是需要通过实时通信链路和流式协议协同工作任何一个环节出现高延迟都会直接影响用户感知。1.3 实时响应系统的关键指标评价这类系统时大家通常关注以下几个指标端到端延迟E2E Latency从用户开始说话到系统开始播放回复音频的时间。首包延迟First Packet Latency用户说完到第一个合成音频包到达客户端的时间。识别延迟ASR Delay)从说话到完整文本输出之间的延迟。字级/词级增量延迟流式引擎每输出一个字的间隔时间。并发会话数单机或集群能同时维持多少路实时会话。在真实项目中我们一般把“用户说完话到 AI 开始说话”的目标定在 800ms 以内把首包延迟控制在 300ms 左右这一个目标倒逼整个链路做流式化改造。2. 环境准备与整体架构设计2.1 项目背景与团队分工这个项目不是从纯零开始团队在算法侧已经有可用的 ASR 和 TTS 模型但都是离线批处理模式。我们的核心任务是把这些算法能力封装成低延迟、可并发的实时在线服务并设计一套能承载多路会话的交付架构。团队总共投入约 8 人角色分布如下2 人负责音频链路和流式协议2 人负责 ASR/TTS 服务封装与并发优化2 人负责 LLM 推理链路和流式输出1 人负责网关、会话管理和部署1 人负责音质测试、延迟测试和线上监控2.2 技术选型说明这里我们不列出厂商强绑定的私有协议重点讲可以自建的部分语言Python 负责算法服务封装Go 负责高并发网关和会话管理。音频传输WebSocket 承载双向二进制音频流。流式协议自定义 JSON 协议约定connect、audio、text、state、error等消息类型。ASR采用可以流式输出的识别方案文本按句返回同时给出中间结果和最终结果。LLM使用支持 SSE 流式输出的对话模型回复按 token 增量返回。TTS采用支持流式合成的方案输入文本可以分句送入输出音频包按序返回。部署Docker 容器化Kubernetes 管理服务节点Nginx 做 WebSocket 代理。环境版本不能直接照搬这里只说明我们的选择Python 3.10、Go 1.21、Redis 7 用于会话状态缓存Kafka 用于异步事件和数据回放。2.3 系统整体链路为了方便理解下面用文字描述整个调用链客户端麦克风 ↓ PCM/Opus 音频流 WebSocket 网关鉴权、会话路由 ↓ 二进制帧 音频前置服务VAD、静音检测、音频重采样 ↓ 有效语音音频片段 流式 ASR 服务输出增量识别文本 ↓ 中间结果/最终结果 对话服务LLM 流式推理可拼接上下文 ↓ SSE token 流 回复分发服务分句、标点恢复 ↓ 文本分段 流式 TTS 服务逐句合成音频 ↓ 音频包 WebSocket 网关 → 客户端播放这个链路中用户感受到的响应时间 VAD 判定说话结束时间 ASR 最终文本时间 LLM 首个 token 时间 TTS 首包时间 网络传输时间。因此每一层都要做流式优化。2.4 项目目录结构为了后续讲解清楚我们把项目简化成以下目录realtime-voice-ai/ ├── gateway/ # WebSocket 网关服务Go │ ├── main.go │ ├── session.go │ └── protocol.go ├── services/ │ ├── vad/ # VAD 服务Python │ ├── asr/ # 流式 ASR 服务Python │ ├── llm/ # LLM 对话服务Python │ └── tts/ # 流式 TTS 服务Python ├── proto/ │ └── message.json # 消息协议定义 ├── deploy/ │ ├── docker-compose.yml │ └── nginx.conf └── tests/ ├── latency_test.py └── audio_test.py这个结构按服务边界拆开方便独立部署和扩容。接下来我们逐层讲解核心实现。3. 核心模块原理与流式实现3.1 音频流协议设计实时语音系统最基础的是音频数据如何在客户端与服务端之间传输。这里我们采用 WebSocket 承载二进制音频帧同时在相同连接上传输 JSON 控制消息。协议消息分为几类connect建立会话带上客户端 ID、采样率、编码格式。audio二进制音频帧通常每帧 20ms 或 40ms。text文本消息包括 ASR 中间结果、最终结果、LLM 回复文本。state状态消息例如 vad_start、vad_end、tts_start、tts_end。error异常信息。以音频帧为例我们的 JSON 头部如下{ type: audio, session_id: session_001, timestamp: 1720000000123, sequence: 123, format: pcm_16k, duration_ms: 40 }音频帧本身不放在 JSON 里而是在 JSON 之后紧跟二进制数据。读取端先按分隔符读一行 JSON再按长度读音频数据。Go 网关里核心逻辑类似// 文件路径realtime-voice-ai/gateway/session.go // 伪代码读取 WebSocket 消息并解析协议 func (s *Session) handleMessage(msgType int, data []byte) { if msgType websocket.BinaryMessage { // 二进制帧第一部分 JSON 头 后续音频数据 headerLen : binary.BigEndian.Uint32(data[:4]) header : parseHeader(data[4 : 4headerLen]) audioData : data[4headerLen:] s.routeAudio(header, audioData) } else if msgType websocket.TextMessage { msg : parseTextMessage(data) s.routeControl(msg) } }实际项目中我们没有在同一个 WebSocket 里混用文本和二进制而是统一采用“JSON 头 二进制体”的格式所有消息都按同样方式解析简化逻辑。3.2 VAD判断“开始说话”和“结束说话”VAD 是实时系统的节拍器。它决定了音频流中哪些片段需要送识别更重要的是决定了用户什么时候说完一句话。常见做法能量阈值短时能量超过阈值视为有声。模型 VAD用 Silero VAD 等模型输出每帧语音概率。惩罚机制连续 N 帧静音才判定结束。我们最初使用 WebRTC VAD优点是快缺点是在噪声环境下误判多。后来换成 Silero VAD准确率明显提升。VAD 具体输出逻辑# 文件路径realtime-voice-ai/services/vad/vad_service.py # 伪代码基于 Silero VAD 的流式检测 import torch model, utils torch.hub.load( repo_or_dirsnakers4/silero-vad, modelsilero_vad, trust_repoTrue ) def detect_speech(audio_chunk, sample_rate16000): # audio_chunk 为 30ms 或 60ms 的 float32 数组 with torch.no_grad(): speech_prob model(torch.from_numpy(audio_chunk), sample_rate).item() return speech_prob工程上不会把每一帧的 VAD 概率直接暴露给上层而是维护一个状态机silence - speech 概率 0.8 连续 5 帧 speech - silence 概率 0.3 连续 10 帧结束判定要足够保守避免用户中间停顿一下就被误判为说完。3.3 流式 ASR 服务封装ASR 是链路里最影响延迟的模块之一。我们要做两件事让 ASR 支持增量输入实时输出中间识别文本。在 VAD 判定结束时输出当前句的最终文本。下面演示服务端如何通过 WebSocket 接收音频并返回结果。这里使用一个简化的流式 ASR 客户端逻辑# 文件路径realtime-voice-ai/services/asr/asr_server.py # 伪代码接收音频帧并回调识别结果 import asyncio import json class RealtimeASR: def __init__(self, recognizer): self.recognizer recognizer self.audio_buffer b async def process_audio(self, audio_chunk: bytes, is_final: bool): self.audio_buffer audio_chunk # 这里调用底层流式识别模型的 decode 接口 partial self.recognizer.decode(self.audio_buffer, is_finalis_final) return { type: asr_result, text: partial[text], is_final: is_final, confidence: partial.get(confidence, 0.0) }关键点is_finalFalse时ASR 返回中间结果前端可以展示但不要播报。is_finalTrue时ASR 返回最终结果后续 LLM 才拿这个文本走回复逻辑。识别缓存要及时清理避免长对话内存增长。3.4 LLM 流式对话服务LLM 是系统里最“智能”也最不稳定的模块。实时语音系统中我们希望 LLM 的回复像打字机一样增量输出而不是等全部生成完再交给 TTS。如果使用 OpenAI 兼容接口通常会配置streamTrue然后逐段读取响应的内容。下面是封装示例# 文件路径realtime-voice-ai/services/llm/llm_stream.py # 伪代码流式请求 LLM按 token 回调 import json import requests def stream_llm_reply(messages, on_token): # 注意这里以常见 SSE 格式为例具体字段依赖模型服务 resp requests.post( https://api.example.com/v1/chat/completions, json{ model: voice-llm, messages: messages, stream: True, }, streamTrue, timeout10 ) for line in resp.iter_lines(): if not line: continue line line.decode(utf-8) if line.startswith(data:): data_str line[5:].strip() if not data_str or data_str [DONE]: continue try: data json.loads(data_str) token data[choices][0][delta].get(content, ) if token: on_token(token) except Exception: # 跳过不完整数据保证流式解析不中断 continue这里有个细节LLM 流式输出可能一次返回多个 token也可能一个 token 分两次返回所以必须用on_token回调把文本累积再交给分句模块。分句策略优先按标点切分遇到句号、问号、感叹号时认为一句已经完整。如果没有标点则累积到 20 个字左右强制切分避免 TTS 等待过长。切分后的句子要保留上下文用于 TTS 合成时保持语义完整。3.5 流式 TTS 服务封装TTS 是响应质量的关键。LLM 输出文本后TTS 还在合成用户已经听到声音了。因此我们要边合成边播放而不是等整句话合成完再播放。现在的开源 TTS 一般都支持 batch 合成但流式效果差异很大。封装思路如下# 文件路径realtime-voice-ai/services/tts/tts_stream.py # 伪代码TTS 增量合成并返回音频块 class StreamTTS: def __init__(self, synthesizer): self.synthesizer synthesizer def synthesize_segment(self, text: str): # 分句后逐句合成 audio_chunks self.synthesizer.synthesize_stream(text) for chunk in audio_chunks: yield { type: audio, audio: chunk, format: pcm_16k }常见问题有些 TTS 引擎有句首延迟也就是不管文本多短都要先跑一个固定时常的预处理。这会严重影响体验。解决方案是在首句还没合成完时先用“开场音/呼吸声”占位或者后端提前把常用回复模板缓存起来。4. 完整的实时链路实战这一节我们把前面模块串起来演示一个最小可运行的服务端链路。这里代码做了简化重点展示各模块如何配合。4.1 创建项目结构先创建目录mkdir -p realtime-voice-ai/{gateway,services,proto,tests} cd realtime-voice-ai4.2 编写 WebSocket 网关网关负责接收客户端音频流转发给后续服务。这里用 Go 写一个最小网关// 文件路径realtime-voice-ai/gateway/main.go package main import ( log net/http github.com/gorilla/websocket ) var upgrader websocket.Upgrader{ CheckOrigin: func(r *http.Request) bool { return true }, } type Client struct { conn *websocket.Conn } func main() { http.HandleFunc(/ws, handleWS) log.Println(gateway listen on :8080) log.Fatal(http.ListenAndServe(:8080, nil)) } func handleWS(w http.ResponseWriter, r *http.Request) { conn, err : upgrader.Upgrade(w, r, nil) if err ! nil { log.Println(upgrade error:, err) return } client : Client{conn: conn} go client.readLoop() } func (c *Client) readLoop() { defer c.conn.Close() for { _, data, err : c.conn.ReadMessage() if err ! nil { log.Println(read error:, err) return } // data 包含 JSON 头和音频数据这里转发给下游服务 // 例如sendToASR(data) log.Printf(recv packet len%d, len(data)) } }这部分是最外层入口实际项目还需要做鉴权、会话绑定、断线重连。4.3 编写音频策略和 VAD网关收到音频后先交给 VAD 判断是否有语音。这里写一个示例策略# 文件路径realtime-voice-ai/services/vad/vad_router.py # 伪代码音频帧路由 VAD 状态机 class AudioRouter: def __init__(self): self.state silence self.speech_frames 0 self.silence_frames 0 def feed(self, frame_prob, audio_chunk): if self.state silence: if frame_prob 0.8: self.speech_frames 1 if self.speech_frames 5: self.state speech self.speech_frames 0 return vad_start else: self.speech_frames 0 else: if frame_prob 0.3: self.silence_frames 1 if self.silence_frames 10: self.state silence self.silence_frames 0 return vad_end else: self.silence_frames 0 return None这里的参数不是固定的需要根据实际麦克风采样率、环境噪声调整。核心是想清楚VAD 要尽可能避免“话没说完就切掉”的场景。4.4 编写流式 ASR 转发逻辑当 VAD 判定 speech 开始后后续音频持续送入 ASR。ASR 返回增量文本系统把文本发回客户端作为中间结果同时在 VAD 结束时把最终文本发给 LLM。简化代码如下# 文件路径realtime-voice-ai/services/asr/asr_worker.py # 伪代码接收音频帧并回调识别结果 class ASRWorker: def __init__(self, recognizer): self.recognizer recognizer self.buffer b def feed(self, audio, is_final): self.buffer audio text self.recognizer.recognize(self.buffer, finalis_final) if is_final: self.buffer b return text实际项目里要注意并不是所有识别器都支持任意长度的缓冲输入很多流式模型有固定窗口需要内部维护状态向量不能简单“拼接音频”。这里只是演示消息流。4.5 编写 LLM 回复与 TTS 串接LLM 输出文本后我们按句子切分逐句交给 TTS 合成。这里用一个简化的事件驱动写法# 文件路径realtime-voice-ai/services/dialog/dialog_service.py # 伪代码LLM token - 分句 - TTS - audio event class DialogService: def __init__(self, llm_stream, tts_stream): self.llm_stream llm_stream self.tts_stream tts_stream self.pending_text def handle_user_text(self, user_text, on_audio): messages [{role: user, content: user_text}] def on_token(token): self.pending_text token if self.should_split(self.pending_text): sentence self.pending_text.strip() self.pending_text if sentence: for audio_chunk in self.tts_stream.synthesize_segment(sentence): on_audio(audio_chunk) self.llm_stream.stream_llm_reply(messages, on_token)这里should_split可以简单判断是否包含句末标点或者缓冲长度是否超过设定阈值。实际系统里还要考虑语气停顿、数字读法、复读等问题。4.6 运行与验证启动网关和 VAD/ASR 服务后可以用 WebSocket 客户端测试# 文件路径realtime-voice-ai/tests/ws_client.py import asyncio import json import websockets async def test(): async with websockets.connect(ws://localhost:8080/ws) as ws: # 发送音频帧这里假设读取本地 pcm 文件 with open(test_audio.pcm, rb) as f: audio_data f.read(3200) # 200ms 16kHz 16bit await ws.send(audio_data) while True: msg await ws.recv() print(recv:, msg[:100]) asyncio.run(test())预期输出应能看到服务端逐步返回识别文本和合成音频事件。4.7 结果说明这个最小链路虽然不能直接上生产但验证了核心问题音频可以从客户端持续流向 ASR识别文本可以增量返回LLM 回复可以边生成边合成。后面所有性能优化都是在这个链路基础上做减法。5. 延迟优化与性能调优5.1 端到端延迟拆解我们在完成基础链路后针对线上会话统计了耗时分布典型情况如下阶段平均耗时优化目标音频上行传输40ms30msVAD 判定结束200ms120msASR 最终结果250ms180msLLM 首 token800ms500msTTS 首包300ms220ms音频下行传输40ms30ms总耗时约 1.6 秒。优化重点在 LLM 和 ASR/TTS 的配合上。5.2 并行化与预判一个很有效的优化是在用户还没说完话之前系统已经可以开始推理。具体做法VAD 检测到 speech 后ASR 持续给出中间文本。当中间文本已经累积成一个完整问句例如“今天天气怎么样”对话服务可以先启动一个“预备请求”。如果用户后面继续补充内容再取消或修正之前的请求。这种“提前响应”会带来准确率风险但可以大幅降低首 token 延迟。我们实际采用的是“半预判”策略只对命中了意图模板的句子提前触发回复不全局开启。5.3 减少 LLM 等待时间LLM 模块是最大的延迟来源优化手段包括使用支持快速首 token 输出的推理服务。减少历史消息数量只保留最近几轮关键信息。对固定开场白、欢迎语、常见问答做模板缓存。将系统提示词压缩减少每轮请求的输入 token。如果企业部署私有模型还可以使用模型量化、动态批处理、KVCache 优化这些都可以显著降低首个 token 输出时间。5.4 TTS 缓存与分句优化TTS 首包延迟受文本长度影响较大工程上可以对常用回复做音频缓存命中缓存直接发送。对长文本优先合成第一句并立刻发送后续句子边合成边发。使用更小的音频 chunk例如 20ms 一包降低播放端缓冲。另外要注意不同 TTS 引擎的音频格式可能不一致要统一转成 16kHz 16bit PCM再由客户端编码成 Opus 传输减少带宽。6. 常见问题与排查思路6.1 声音断句太早现象用户还在思考停顿系统已经判定说话结束并开始回复。可能原因VAD 静音阈值太短。解决思路调大 speech 到 silence 的连续帧数。结合语义判断如果 ASR 中间文本结尾不是标点不触发结束。在静音期间设置“临时结束”状态等待 500ms 后再决定是否结束。6.2 识别结果乱序现象同一会话里后发送的音频帧结果先返回。可能原因ASR 服务并发处理时依赖了请求级状态没有严格按 sequence 排序。解决思路每条音频消息带上递增序号。输出端按序号缓冲乱序包排队等待。ASR 解码节点只接收单路音频流避免并发写入。6.3 声音卡顿或断续现象播放的 AI 回复音频中途卡顿。可能原因网络 jitter 导致音频包到达不均匀客户端播放缓冲太小。解决思路客户端增加 100ms 左右的 jitter buffer。服务端连续发送音频包时做平滑排队。如果使用 WebRTC可以开启 NACK 或 FEC但会增加延迟需权衡。6.4 LLM 响应中断现象模型回复到一半连接断开没有任何音频输出。可能原因上游 LLM 请求超时、连接被服务端关闭、推理节点 OOM。解决思路客户端先播放“请稍等”音频兜底。对话服务设置较短的读超时第一时间重试。解析 SSE 时对异常行做容错不直接中断。6.5 长对话内存增长现象服务运行一段时间后内存持续上升。可能原因ASR 状态缓存、LLM 历史消息、音频 buffer 没有清理。解决思路为每个会话设置最大空闲时间超时自动回收。限制历史消息条数。音频帧处理完立即置空不要持有引用。6.6 排错清单遇到问题可以按以下顺序排查检查项命令/方式网关连接是否正常curl -i http://localhost:8080/healthz音频是否上行在网关打印recv packet lenVAD 是否正确触发观察vad_start/vad_end日志ASR 是否返回文本打印asr_result事件LLM 是否返回 token打印stream_llm_reply回调TTS 是否合成音频打印audio事件长度延迟分布埋点统计各阶段 timestamp7. 最佳实践与工程建议7.1 编码与协议规范所有时间戳使用毫秒级 Unix 时间服务端统一时区。消息类型不要用纯数字使用可读字符串方便日志排查。音频编码格式必须在连接时协商避免服务端猜。协议升级时保留 version 字段便于灰度兼容。7.2 会话状态管理实时语音系统是典型的有状态服务不能像普通 HTTP 一样无脑水平扩容。使用 Redis 保存会话上下文、用户偏好、最近几轮文本。WebSocket 网关节点与 ASR/LLM/TTS 服务通过消息队列或内部 RPC 解耦。会话迁移要谨慎尽量不要在服务重启时丢失 active session。7.3 延迟监控与告警实时系统最重要的可观测性指标是延迟而不是功能可用性。在客户端上报每个事件的 timestamp服务端计算每个阶段耗时。对 p95、p99 延迟设置告警。对“用户说完话 2 秒内没有音频输出”这样的体验级事件进行离线分析。7.4 安全与权限边界WebSocket 连接需要鉴权禁止未授权设备直接连入。音频数据可能包含敏感信息传输层建议使用 WSS。不要在日志里打印完整用户语音内容脱敏后再存。LLM 接入时对输入文本做基本安全过滤避免不适当内容流向用户。7.5 灰度发布与回滚由于语音链路涉及多个服务上线任何一个模块都可能影响整体体验。建议按会话维度灰度内部白名单优先体验新链路。新旧链路并行通过流量比例逐步切换。任何一个服务出现错误率上升立即切回上一版本。模型更新前先在离线音频集上回归测试观察 ASR 准确率和 TTS 自然度。8. 项目复盘与下一步学习方向经过六个月打磨这套实时语音 AI 系统从最初每秒只能处理几十路会话到现在可以稳定支撑上千路并发端到端延迟从最开始的 2.5 秒降到 0.9 秒左右。这个过程中最大的体会是实时语音系统的难点不在单个模型而在把所有模型和工程组件以“流”的方式组织起来。如果你们也准备搭建类似系统建议按以下顺序推进先把最简单的一条串行链路跑通确认每个服务都能通信。再用 VAD 控制语音边界减少无效音频。接着把 ASR 改成流式输出让文本增量可见。然后接 LLM 流式回复解决分句和取消策略。最后把 TTS 变成边合成边播放完成端到端实时化。每一步都做延迟埋点用数据判断瓶颈在哪里。不要一上来就追求全链路流式那样排错会非常困难。下一步可以继续深入研究的方向包括多说话人分离、自动打断处理、误唤醒抑制、实时情感识别、端侧模型推理优化。如果你们团队正在做语音相关产品希望这篇文章能成为一份值得收藏的架构参考。遇到具体问题也欢迎在评论区交流我会根据实际情况补充更细致的踩坑记录。