通过并行化 Whisper → LLM → Piper 管道(消除各环节阻塞),在未换硬件和模型情况下将语音响应延迟降低 70%。
我的第一个语音 Agent 回答一句口语需要 1200ms。后来我重写了流水线中的三个关键节点,延迟降到了 340ms。没有换硬件、没有换模型、没有换更小的 LLM。用户说的话没变,Agent 回复的话也没变。变的是等待的形态。
如果你做过一个礼貌但迟钝的语音 Agent,就会知道时间都藏在流水线的哪个环节里。
我的第一版是这样按顺序执行的:
每个阶段单独看都没问题。问题在于流水线是一条单行道。Whisper 要等录音结束才能开始。LLM 要等 Whisper 完成才能开始。Piper 要等 LLM 结束才能开始。用户等待的是所有阶段的时间总和。
用赛车做比喻很快就会过时,所以我直接用真实数据。以下是我的机器上时间线的实际样子:
[record]--[200ms silence]--[whisper 380ms]--[LLM 480ms]--[piper 340ms]--[playback]
^
1,200ms
每个阶段都在阻塞下一个阶段。我搭了一场接力赛,每个选手都要等上一个选手坐下了才能起跑。
第一个修复方案是把用户的语音不再是当做一个文件来处理。在捕获音频的同时,以 20-30ms 为一帧喂给 Whisper。等用户到达尾部静音时,大部分转录工作已经完成了。你只需要等待最后几帧加上一个短暂的刷新。
Pipecat 是参考实现。它的核心模型就是基于帧的:每个阶段处理 20-30ms 的数据块,一旦就绪就立即向下传递。没有批处理、没有完整片段的交接、没有"等待这个阶段完成"的概念。它的官方文档声称,当所有模型都托管在同一 GPU 集群上时,语音到语音的延迟可以低于 500ms。
如果你不想引入完整框架,底层原语是一个 VAD(语音活动检测)——它发送分块数据,加上一个流式 ASR 端点。Deepgram、AssemblyAI,以及像 WhisperX 这种基于 Whisper 的流式封装都支持这种形态。
2026 年的延迟选购参考:Deepgram 的 Nova-3 流式端点目标是将服务端延迟压到 300ms 以下,独立基准测试显示,加上网络传输后,客户端总延迟在 200-500ms 区间。这才是我实际参考的数字,不是官网的宣传语。
仅这一招,时间线就变了形状:
[record + whisper (overlapping)]--[flush 80ms]--[LLM 480ms]--[piper 340ms]
^
900ms
我通过消除一个等待环节拿回了 300ms,而不是让任何东西变快了。
下一个堵点是 LLM 到 TTS 的交接。LLM 是流式输出 token 的,TTS 需要完整文本。我之前的做法是把所有 token 缓冲起来,然后把整个回复一股脑交给 Piper。
技巧在于:缓冲到第一个句子边界,然后就把这部分交给 TTS,然后继续缓冲。TTS 开始合成第一个句子时,LLM 还在生成第二个句子。用户听到语音的时间点提前到第一个句子说完的时刻。
import asyncio
async def stream_to_speech(llm_stream, tts, audio_out):
buffer = ""
async for token in llm_stream:
buffer += token
# 简单的句子检测:句号/问号 + 空格
if buffer and buffer[-1] in ".!?":
asyncio.create_task(tts.synthesize_and_play(buffer, audio_out))
buffer = ""
if buffer:
asyncio.create_task(tts.synthesize_and_play(buffer, audio_out))
这里的 asyncio.create_task 才是关键。它在不离断 LLM 流的情况下交接合成任务。LLM 继续产生 token,而 Piper 已经在把第一个句子转成音频了。
首音频时间现在取决于 LLM 生成一个句子所需的时间加上 Piper 的 TTFB(首字节时间),而不是整个回复的长度。在我实际测量中,第一个句子的 LLM 延迟约 180ms,Piper TTFB 约 120ms。转录完成后 300ms 就能听到第一段音频。
[record + whisper]--[flush]--[LLM sent1 180ms][piper sent1 120ms]--[playback starts]
[LLM sent2 in parallel]
^
580ms to first audio
句子级流水线已经把你带到了亚秒级。要压到 400ms 以下,就得给 TTS 喂比句子更小的块。
naive 做法是每个 token 单独流式传输。不要这样做。Piper(以及大多数神经 TTS)是从上下文预测韵律的。喂一个词它只能看到一个词,它无法看到足够远的下文来选对语调,听起来会像被切碎了的机器人声音。
经验上的最佳点是 k=1-2 个词的向前查看。等待当前词之后的那个 token 到来后,再把当前词发送给 TTS。两个词的上下文对于 Piper 来说已经足够在为第一个词选对语调了。
async def token_stream_to_speech(llm_stream, tts, audio_out, lookahead=2):
words = []
async for token in llm_stream:
words.append(token)
if len(words) > lookahead:
chunk = words.pop(0)
await tts.synthesize_and_play(chunk, audio_out, prev_words=words[:lookahead])
for w in words:
await tts.synthesize_and_play(w, audio_out)
此时我的端到端时间线是:
[record + whisper]--[flush]--[LLM first 2 tokens][piper starts]--[playback]
^
340ms end to end
1,200ms 去哪了?不是哪个单一阶段变快了,而是阶段之间的缝隙被闭上了。
我在上面略过了这一点,但传输层的影响比大多数教程承认的要大。通过 HTTP 请求-响应发送音频,每个阶段光是 TCP 建立和 TLS 握手就要增加 100-200ms。WebSocket 消除了重复建立连接的开销。WebRTC 在此基础上换成了 UDP,所以不用为重传付出代价。
这是我一直放在桌上的权衡表:
| 协议 | 开销 | 复杂度 |
|---|---|---|
| HTTP | 100-200ms/阶段 | 低 |
| WebSocket | 0(复用连接) | 中 |
| WebRTC | 0(UDP,无重传) | 高 |
我花了三天让 WebRTC ICE 协商跑起来。STUN、TURN,整套流程。为了那 50-100ms 的底是值得的,但如果你搭的是服务器到服务器的流水线,用户在你应用的另一端,WebSocket 是个不错的默认选择,一个下午就能上线。
LiveKit 是我最终采用的模式:LiveKit 的 SFU 处理 WebRTC 传输,Pipecat 处理 STT-LLM-TTS 的编排。SFU 做选择性数据包转发而不重新编码,低开销就来自这里。如果想要两句话的 mental model:Pipecat 是厨师,LiveKit 是跑堂的。
这之后就没有魔法了。340ms 大致拆解如下:
LLM 那个数字是不换模型就压缩不了的。所以我停在了 340ms。要进入 300ms 以内,要么换更快的模型、要么用投机解码、要么迁移到 Gemini 2.0 这种原生音频输入输出统一的架构——在那种架构里 STT→LLM→TTS 的边界不存在了。这些都是各自独立的大项目。
如果你在追逐 300ms 以内,那里就是路分叉的地方:先把你有的并行化(这篇文章),然后要么缩小 LLM,要么统一流水线(那是另一篇文章的内容)。
别再优化单个阶段了。单独计时每个阶段当然要做,但真正的收益在阶段之间的缝隙里。一个 naive 流水线中的每个阶段边界都是一个什么都不发生的空档。基于帧的 STT、句子级的 TTS 交接、token 级的 TTS 向前查看——每一个技巧都是在不同粒度上应用同一套思路。闭上缝隙。
本书版本中有完整的时间线图、有可运行的 Pipecat + LiveKit + Whisper + LLM + Piper 流水线代码,以及让你在物理上已经无法再快时依然感觉更快的感知技巧:《Voice AI 300ms UX Guide》。第 7 章是上面这篇文章压缩内容的并行化 playbook。第 8 章是当无法再压缩时还能做什么。