做 AI 语音助手最怕的一件事就是用户问完一个问题之后对面干等一两秒才听到声音。比这更怕的是用户说错了想打断机器还在自顾自地把上一段话说完。我最近在做语音对话类的产品要求很直接用户说话的尾音刚落合成音频就要开始播用户随时可以插话插话后机器立刻闭嘴等下一轮输入。试过把普通 HTTP 语音合成接口直接改成轮询延迟和体验都过不了关最后换了火山引擎大模型语音合成通过它的双向流式 API 把整个链路搭通。这篇文章就把我从零接入、代码实现、参数优化、断线重连一直到常见问题排查的完整过程整理出来给同样在做实时合成应用的朋友做个参考。先说结论双向流式 API 不是把 HTTP 换成 WebSocket 那么简单的“升级版”它解决的是实时对话场景里最核心的三个问题——首字延迟、长文本等待、以及用户随时打断。下面我会用 Python 从协议原理讲到可运行代码再讲优化和避坑全部是实际跑过的方案。1. 为什么实时对话场景必须上“双向流式”1.1 传统 HTTP 和单向流式的瓶颈传统 TTS 接口的逻辑是客户端把一整段文本 POST 给服务端服务端合成完毕后返回完整音频请求结束。这个模式在批量配音、离线生成音频时没问题但放到实时对话里到处都是坑。第一个坑是首字延迟不可控。服务端必须等整段文本都合成完才能把结果一次性返回。文本越长你等待的时间就越长。哪怕模型一次只合成三秒钟的音频中间如果有几百字客户端的等待时间也会被拉到几秒甚至更长。用户那边感受到的就是“我问完话空气安静了三四秒”这种体验在对话产品里基本属于不可用状态。第二个坑是连接开销太大。每一次 HTTP 请求都要重新鉴权、建立连接然后再断开。连接握手本身就要消耗几十到几百毫秒在实时交互里这几十毫秒就是用户能感知到的卡顿。第三个坑最致命客户端没办法中途“撤回”。用户如果发现自己说错了要打断HTTP 接口根本没有任何机制告诉你“不用再合成了”。你只能等这一整段合成完、播完或者客户端本地强行停掉播放但服务端还在白白消耗算力生成后面的内容。体验就是机器和用户抢话怎么调参数都救不回来。单向流式接口比 HTTP 好一些能让服务端边生成边把音频推下来解决了“等完整结果”的问题但仍解决不了“用户插话”和“服务端状态改写”的问题。因为上行通道仍然很弱客户端没办法及时把控制指令传给服务端。1.2 双向流式到底“双向”在哪里双向流式的本质是在一条 WebSocket 长连接上同时存在两条独立的数据通道一条上行客户端把文本、控制指令像流一样不断推给服务端一条下行服务端把音频事件、合成音频像流一样不断推给客户端。上行通道的价值在于文本不再是一次性提交的。你可以让用户在说话时ASR 识别出一小段就把这一小段推送过去服务端可以做到“边说边合成”。如果你要做用户指令打断也可以通过上行通道发一个专门的控制帧让服务端立刻停止当前合成任务。下行通道的价值在于音频是按块返回的。模型每生成一块音频服务端就通过 WebSocket 推给你一块。客户端拿到第一块音频就可以开始播放不用等全部合成完。理论上做好优化后从文本送入到第一段音频回来是可以在几百毫秒量级内完成的。这里有个容易混淆的点很多人以为“支持流式”就是音频流式但其实“双向”才是最值钱的。音频下行是基础能力控制指令上行才是实时交互产品的命脉。没有上行控制通道所有打断、静默恢复、节奏控制都是空谈。1.3 适合的场景和不太适合的场景这种方案最适合的场景我列一下你可以对照自己的项目判断AI 语音助手尤其是有多轮对话、用户随时会插话的那种实时配音、跟读、口语评测需要听一句反馈一句智能客服外呼或者电话 IVR语音菜单需要及时响应按键或话语字幕配音、现场同传需要边生成边播如果你的需求是批量合成一堆长文本比如把一本小说生成音频那用双向流式 API 属于杀鸡用牛刀直接用普通的离线合成接口会更省事。一句话总结实时交互、需要低延迟、允许打断这三个条件占两个以上就值得上双向流式。2. 接入前准备与核心协议理解2.1 开通服务与获取鉴权信息接入的第一步是在火山引擎控制台完成服务开通和应用创建。登录控制台后在语音技术相关目录下找到大模型语音合成服务点击开通然后创建一个新的应用。创建完成后你会拿到三个关键信息AppID、Access Token 和 Secret Key。这三个信息里Access Token 通常是你请求时的鉴权凭证可以理解成用户名密码体系里的那个长期密码一定要保存在服务端不要直接写进前端页面。Secret Key 一般用于参与签名计算尤其在做 WebSocket 建连或加签请求时会用到具体的加签方式要看你拿到的接入文档不同版本差异不小。AppID 则用来标识你这个应用的身份方便服务端统计和做资源隔离。我第一次接入时的习惯是先把这些凭据写成环境变量而不是硬编码在代码里。比如放到.env文件通过os.getenv读取。这样一方面避免误把密钥提交到代码仓库另一方面在本地测试和线上部署时切换环境也更方便。还有一个小提示拿到服务后先别急着写代码。找到官方文档里关于双向流式接入的部分仔细看两遍。重点看三块WebSocket 地址、鉴权 header 的写法、以及消息帧结构的字段定义。这三块理解透后面写 Python 代码就不会瞎猜。2.2 理解消息帧与一次完整的合成会话这类 WebSocket 接口从抽象层面看通信内容就两块文本帧和二进制帧。文本帧一般承载 JSON 格式的消息用来传指令和事件二进制帧承载音频数据可能是 PCM、Opus 或 MP3 编码。一次典型的双向流式合成会话状态流转大致是这样的客户端通过 WebSocket 地址建立连接连接建立成功后客户端发送一个初始化请求说明要用什么音频格式、采样率、音色、语速等参数服务端收到初始化请求后返回一个类似于“会话准备就绪”的事件客户端开始通过上行通道发送待合成的文本服务端一边合成一边通过下行通道推送音频二进制帧全部文本合成完成后服务端返回“合成结束”事件客户端正常关闭连接一次会话结束如果中间发生了打断流程会多一些客户端发送打断控制帧服务端立刻停止当前合成任务、清理内部状态然后返回一个“已打断”事件。之后客户端可以继续发送新的文本开始新一轮合成。这里有一个每一步都要注意的点所有上行消息尽量保证顺序。尤其当你在一个连接上同时发送多个文本分句时不要并发去发要让它们按顺序排队发出。服务端内部通常会根据消息到达顺序做拼接和合成一旦顺序乱了可能出现音频内容和文本对不上、前一句合成结果覆盖后一句这类诡异问题。2.3 Python 环境和依赖准备代码层面我建议使用 Python 3.8 及以上版本配合websockets库。这个库对 asyncio 支持很好API 简单能处理大部分 WebSocket 细节。如果你已经有项目在用aiohttp它的aiohttp.ClientSession.ws_connect也可以用但我个人更喜欢websockets的独立语义写起来更像标准的 WebSocket 客户端。安装命令就一行pip install websockets运行时还需要 Python 标准库的asyncio、json、struct、uuid等这些不需要额外安装。如果你要保存成 WAV 文件做本地验证再装一个numpy也不是必须的用标准库wave模块就够了。我不建议在这个场景用requests去实现流式会话。requests是同步阻塞的做长连接和持久推送非常别扭而且 WebSocket 的二进制帧、事件推送语义它都没有原生支持硬套会很痛苦。老老实实用 asyncio 模型可以让收发逻辑清晰很多。3. Python 双向流式实现从最小示例到可打断版本3.1 最小可运行示例建立连接并拿到第一段音频先写一个最小示例目的是把“连接 - 发初始化 - 发文本 - 收音频 - 收结束事件”这个最简链路跑通。字段名我下面都用通用写法具体字段以你实际拿到的官方文档为准但整体流程大差不差。import asyncio import json import uuid import websockets # 这些值根据你自己的应用信息填写 WS_URL wss://你的接入地址/websocket TOKEN 你的AccessToken VOICE_TYPE 你的音色ID async def tts_simple(): headers {Authorization: fBearer {TOKEN}} async with websockets.connect( WS_URL, extra_headersheaders, max_size4 * 1024 * 1024, ) as ws: # 1. 发送初始化请求 init_payload { user: {uid: example_user}, audio: { format: pcm, sample_rate: 24000, bits: 16 }, business: { voice_type: VOICE_TYPE, speed_ratio: 1.0, volume_ratio: 1.0, pitch_ratio: 1.0 }, request: { reqid: str(uuid.uuid4()) } } await ws.send(json.dumps(init_payload)) # 2. 发送第一句文本 await ws.send(json.dumps({ event: send_text, data: 欢迎使用大模型语音合成双向流式接口 })) # 3. 循环接收服务端消息 audio_chunks [] async for message in ws: if isinstance(message, str): event json.loads(message) if event.get(event) end: break else: # 二进制帧就是音频数据 audio_chunks.append(message) return b.join(audio_chunks) if __name__ __main__: pcm_data asyncio.run(tts_simple()) print(合成音频大小:, len(pcm_data), bytes)这段代码最重要的一点是区分文本帧和二进制帧。websockets库在收到文本消息时返回str收到二进制消息时返回bytes所以用isinstance判断即可。我见过有人在接收循环里一直按 JSON 去解析所有消息结果把二进制数据也硬解析直接抛异常这是新手最容易踩的坑。把这个 PCM 数据保存成本地 WAV 文件可以快速验证合成效果import wave with wave.open(output.wav, wb) as wf: wf.setnchannels(1) wf.setsampwidth(2) # 16bit 2字节 wf.setframerate(24000) wf.writeframes(pcm_data)播放确认听到了声音说明最基础的一条链路已经通了。但这只是“单向”的用法别急双向才是重头戏。3.2 支持队列输入与打断的完整封装实际产品里文本不会像上面那样一次性全发完。语音助手的场景往往是这样ASR 识别出一句话的前几个字你就希望服务端开始合成后面几个字继续识别出来就继续追加。而且用户随时可能说“停一下”或者直接插话这时候你要立刻打断当前合成。要实现这种灵活的收发控制我用asyncio.Queue做输入侧缓冲用一个专门的 task 负责发送另一个 task 负责接收。打断信号用asyncio.Event来传递。核心代码如下import asyncio import json import uuid import websockets class DualStreamTTS: def __init__(self, ws_url, token, voice_type): self.ws_url ws_url self.token token self.voice_type voice_type self.text_queue asyncio.Queue() self.audio_queue asyncio.Queue() self._interrupt asyncio.Event() self._reqid None async def start(self): headers {Authorization: fBearer {self.token}} async with websockets.connect( self.ws_url, extra_headersheaders, max_sizeNone, ) as ws: self._reqid str(uuid.uuid4()) init_payload { user: {uid: example_user}, audio: { format: pcm, sample_rate: 24000, bits: 16 }, business: { voice_type: self.voice_type, speed_ratio: 1.0, volume_ratio: 1.0, pitch_ratio: 1.0 }, request: {reqid: self._reqid} } await ws.send(json.dumps(init_payload)) sender asyncio.create_task(self._send_loop(ws)) receiver asyncio.create_task(self._recv_loop(ws)) await asyncio.gather(sender, receiver) async def _send_loop(self, ws): while True: # 先处理打断信号 if self._interrupt.is_set(): await ws.send(json.dumps({event: barge_in})) self._interrupt.clear() # 用超时轮询既能及时处理新文本又能响应打断 try: text await asyncio.wait_for(self.text_queue.get(), timeout0.05) except asyncio.TimeoutError: continue if text is None: break await ws.send(json.dumps({ event: send_text, data: text })) async def _recv_loop(self, ws): while True: message await ws.recv() if isinstance(message, str): event json.loads(message) if event.get(event) in (error, end): break else: await self.audio_queue.put(message) def push_text(self, text: str): self.text_queue.put_nowait(text) def interrupt(self): self._interrupt.set()这个封装里有一个非常关键的设计所有上行消息都收敛在_send_loop这一个 task 里发。为什么这么做因为 WebSocket 虽然是全双工协议但如果你开多个 task 同时往同一个连接上发消息底层的发送顺序是很难保证的。一旦多个 task 并发send可能出现控制指令穿插在文本中间、服务端解析错乱的情况。把所有上行收敛到一个循环里发什么、按什么顺序发你说了算。打断信号的响应延迟大约是 50ms因为_send_loop每次最多等 0.05 秒就会检查一次_interrupt。对语音交互来说这个延迟完全感知不到。你没有必要为了“更快”去把超时时间降到一个极端值因为那会让 CPU 空转得更厉害收益却微乎其微。3.3 用 reqid 管理每次合成会话上面代码里每次连接都会生成一个新的reqid。这个字段的作用是让服务端识别“这是哪一次会话”。我建议你把它当成会话 ID 来用每次重新建连都必须换一个新的不要复用。为什么要这么做服务端通常会根据reqid做状态关联和缓存。如果你一直用同一个 ID服务端可能把你当成重复请求直接拒绝或者把上一次的合成状态搬过来导致你拿到的音频和这次发的文本完全对不上。尤其是在做断线重连时最容易犯这个错误——重连觉得“还是同一个用户”于是用同一个reqid结果服务端根本不认账。如果你要在日志里排查问题把reqid、用户的 uid、发送文本的开头几十个字一起打出来基本就能定位大多数线上故障了。4. 参数选择、并发和稳定性优化4.1 音频编码和采样率怎么选接入第一步要确认的就是音频格式。一般服务端会提供 PCM、Opus、MP3 三种编码。别看它们最终都能出声实时场景下的差别非常大。我自己在测试中观察到的对比是这样的编码单路带宽占用客户端解码复杂度适合场景PCM 24kHz 16bit 单声道约 384kbps最低直接当采样数据用带宽充足、对延迟敏感Opus低通常不到 64kbps需要 Opus 解码库移动网络、弱网环境MP3中等需要解码且帧对齐麻烦兼容性测试、非实时场景PCM 的带宽占用怎么算出来的采样率 24000Hz每个采样 16bit 也就是 2 字节单声道每秒数据量就是 24000 × 2 48000 字节换算成 kbps 是 48000 × 8 / 1000 384kbps。这个带宽在纯内网或者稳定的宽带网络下没什么压力但如果你要在用户的移动 App 上跑就要慎重考虑流量成本了。我的建议是实时对话场景优先用 PCM因为少一道解码就能少一次延迟。等到确实出现带宽瓶颈再换成 Opus。不要为了所谓的“兼容性”去选 MP3它在流式场景里帧边界处理特别麻烦很容易出现开头有杂音或者结尾有爆音。4.2 降低首包延迟的几个实操技巧首包延迟就是从发送文本到收到第一个音频帧的时间这个指标直接决定用户“等待开口”的感受。我在实际调优中做过几件事效果比较明显。第一件复用连接。不要每说一句就重建一次 WebSocket。同一路对话里第一条连接建立之后尽量用到底只有当连接断开才重建。省掉建连握手的几十到几百毫秒。第二件边识别边发送。不要等 ASR 完整识别出一整句话才送合成。只要识别出前几个字就先把这句的开头发过去。服务端是流式模型它会边接收文本边合成。相当于你话还没说完机器已经在准备开口了。第三件让播放器提前就绪。客户端在发送文本之前就把音频播放器的缓冲打开不要在收到第一个音频块之后才初始化播放设备。初始化播放设备在某些系统上要几十毫秒甚至更多提前做好能保证音频一到就开始播。另外网络环境允许的话把 WebSocket 的地址选在离你服务端最近的地域节点。跨地域的物理延迟是没办法靠代码优化的。4.3 并发多路连接与 asyncio 的正确姿势如果产品要服务多个用户不能所有用户共用同一个 WebSocket 连接那样状态会完全错乱。正确做法是一路会话对应一个连接然后通过 asyncio 并发跑多个连接。用代码表达就是async def serve_one_user(user_id): client DualStreamTTS(WS_URL, TOKEN, VOICE_TYPE) await client.start() async def main(): user_ids [u001, u002, u003] await asyncio.gather(*[serve_one_user(uid) for uid in user_ids])这里有两个细节要注意。第一个是掌握并发的“度”虽然 asyncio 可以同时开很多协程但服务端对每个账号通常有 QPS 或并发连接数限制开太多连接会被限流。我先在文档里确认配额再用实测去摸上限一般先用文档推荐值的十分之一起步。第二个是不要把阻塞操作放进接收循环。比如拿到音频后你要写文件、发到远端、喂播放器这些操作如果直接在_recv_loop里同步执行会阻塞事件循环导致后续音频帧接收不及时听感上就是卡顿。正确做法是把音频丢进audio_queue由独立的消费者任务去处理。上面的DualStreamTTS里就是这个设计。4.4 心跳保活与断线重连策略长连接最怕的是“静默断开”。服务端网关通常有个空闲超时时间连接一段时间没有任何消息交互就可能被回收掉。你在客户端还傻等下一帧音频实际连接已经断了。处理办法是应用层心跳。WebSocket 本身有 ping/pongwebsockets库也支持自动 ping设置ping_interval20, ping_timeout10就能在底层维持连接活跃。但有些服务端的心跳需要发送自定义的 JSON 帧这时候你得看文档决定是在_send_loop里定时发还是用asyncio.create_task单独起一个心跳任务。断线重连我建议用指数退避第一次失败后等 0.5 秒重试第二次等 1 秒第三次等 2 秒最多重试五六次。重连成功后要重新生成reqid重新发送初始化请求并把本地音频缓冲清空。不要妄想着用同一个连接继续发数据WebSocket 一旦断开之前那个连接上的所有会话状态都作废了。5. 常见问题速查与排障实录5.1 鉴权失败、连接直接返回 401 或 403我在接入初期遇到过几次鉴权失败基本都是这几个原因Token 拼写错误、Token 已经过期、或者把 Token 放错了 header 位置。建议第一步把服务端返回的错误体完整打印出来大多数情况下它会直接告诉你是什么原因产生的 401。另外一个容易忽略的点是如果你的接口要求签名而签名包含时间戳请检查客户端服务器的时间是否准确。服务器时间偏差超过服务端允许的容错范围签名校验会直接失败。我踩过一次这个坑排查了很久才发现是测试机的系统时间慢了五分钟。5.2 连接正常但首包迟迟不来或音频中途静音连接建好了文本也发过去了可就是迟迟收不到第一块音频这种情况我先看是不是首包文本太长了。之前我一次性把 200 个字的文本全送过去服务端要合成完这一整块才开始返回首包自然慢。解决办法就是切分文本让第一句短一些后面再追加。音频中途静音则大概率是播放端缓冲问题不是服务端没回数据。检查音频接收 task 和播放 task 之间的队列是否积压了数据如果队列里有数据但没播出来就说明消费端阻塞了。我在代码里给播放端单独起 task避免被其他 I/O 拖住。5.3 打断之后还有旧音频残留这是双向流式接入里最有代表性的一个问题。用户点了打断界面好像停了但过了一会儿又把上一句话的尾巴播出来了。原因有两个一个是本地播放器没有清空缓冲旧的 PCM 数据还留在播放 queue 里打断后继续往外播另一个是服务端内部其实还缓存着之前文本生成的音频块如果你的控制指令只是“停止接收”而不是真正通知服务端重置合成状态那些音频块还会继续推过来。正确做法是打断动作同时做三件事设置interrupt事件让发送 task 发控制帧、清空本地音频播放队列、把接收循环里收到的后续旧音频块丢弃。等收到服务端“已打断”的确认再允许新的文本进入合成。5.4 并发高了之后事件循环忙不过来了asyncio 在单个事件循环里跑上千个协程没问题但如果每个协程都在高频轮询CPU 就会明显上升。我的_send_loop里用了wait_for(..., timeout0.05)来检查打断信号和队列这个写法本身没问题但如果有几百路连接同时都这样轮询CPU 浪费还是很可观的。优化方案有两个方向。一是把轮询间隔适当放大到 0.1 秒打断响应的 100ms 延迟人耳几乎区分不出来二是用asyncio.Event的wait()来唤醒发送 task而不是固定超时轮询只不过代码逻辑会复杂一点。线上环境我通常会先按第一种方案来简单可靠。5.5 高频问题排查记录汇总把最近反复被问到的几类问题整理成一个速查表排查时直接对着看现象可能原因排查方向与解决建议连接返回 401/403Token 错误或签名过期打印服务端返回错误体检查服务器时钟建立连接后无任何返回初始化请求参数有误确认 JSON 字段名、音频格式是否符合文档首包延迟高首句文本过长/播放器未预热切分首句提前初始化播放设备音频断续卡顿网络抖动/消费者 task 阻塞增加播放缓冲把耗时操作移出接收 task打断后旧音频还在播播放缓冲未清空/未发控制帧清空本地队列发打断指令并等待确认重连后拿到的结果异常复用了旧的 reqid每次重连生成新的 reqid清空合成状态这些看起来细碎的问题其实背后都指向同一个原则双向流式语音合成的链路上客户端和服务端一直处于“互相影响”的状态任何一个环节的缓存、状态、时序没有理清楚都会在用户体验上暴露出来。我自己做下来的体会是不要一上来就追求极致的低延迟先把连接管理、打断处理、断线重连这三大块搞稳再去抠首包的几十毫秒。最后再分享一个小技巧线上日志里把文本发送时间和第一块音频接收时间同时打出来这个差值就是你最需要关注的核心指标所有的优化都围绕它来做就不会跑偏。
