说:逐词喂 TTS 与按真实时间释放的 RealtimeQueue

30 秒导读: LLM 一个 token 一个 token 地吐文本,TTS 想要的是"整词"、还得读得跟真人语速一样自然。这一章讲 Unmute 怎么把"文本流"变成"同步的语音流":先把碎 token 重新拼成整词逐个喂给 TTS,再用一个按时间戳排序的小堆(RealtimeQueue),把 TTS 抢跑生成出来的音频和文字,掐着它们真该出现的那一刻才放出去。

本章只讲"说"这一侧的时序工程。打断时怎么取消留给 04-quest-lifecycle-interruption,声音克隆/人格留给 05-voices-prompt-personality。上游怎么"听"、怎么决定该说话,见 02-stt-vad-turn-taking;整个回合的编排见 01-orchestration-loop。

1. 这是什么(零基础也能懂)

一句话定义: 把文本大模型输出的文字流,实时转成一段和它自己节奏对齐的语音流。

它要解决的两个直觉难题:

难题 A:LLM 吐的是碎片,不是词。 流式接口按 token 给你 "Hel"、"lo wor"、"ld" 这种碎块。TTS(文字转语音)如果拿到半个词,它不知道词到哪结束,就会读错音。所以喂给 TTS 之前,必须先把碎块重新拼成整词。
难题 B:机器比真人快。 TTS 生成语音往往快于实时——它能在 1 秒内算完 3 秒的语音。如果算出来就立刻播,音频会挤成一坨,而且屏幕上的字幕会和声音对不上。所以要有个"节流阀":掐着每段语音/每个字真该出现的时刻再放出去。

用起来什么样: 从调用方看,喂词只是一个循环——LLM 每吐出一个整词,就 await tts.send(词);词发完了发一个 Eos(end of stream)。剩下的同步全由 TTS 客户端内部搞定:

# 示意,非源码:喂词循环的骨架
async for word in rechunk_to_words(llm.chat_completion(messages)):
    await tts.send(word)      # 逐个整词发给 TTS 服务器
await tts.send(TTSClientEosMessage())  # 告诉 TTS:说完了
# 另一侧:`async for msg in tts` 会按真实时间吐出 音频帧 和 文字

一句话类比: 把它想成同声传译的字幕组 + 配音。翻译(LLM)一个词一个词地报;配音演员(TTS)提前把整句都录好了,但不能一股脑全放——得掐着影片的真实时间轴,让声音和字幕卡在正确的一帧出现。那个"时间轴"就是 RealtimeQueue。

2. 顶层全景(它大概怎么转)

这条"说"的流水线横跨三个文件:LLM 侧做整词切分,unmute_handler 做喂词与取音频,TTS 客户端做协议编解码 + 时序释放。

怎么读下面这张图: 从左到右是数据流。上半条是"文本进 TTS",下半条是"音频/文字出 TTS";中间那个 RealtimeQueue 是全章的心脏——两个方向的输出都要过它这道"按时间释放"的闸。

  LLM 碎 token 流
       │
       ▼
 ┌──────────────────┐   整词  ┌──────────────────────────┐
 │ rechunk_to_words │ ──────▶ │ _generate_response_task  │  逐词
 │ (按空白切整词)   │         │   的喂词循环             │ ─────┐
 └──────────────────┘         └──────────────────────────┘      │
                                                                 ▼
                                                     ┌────────────────────┐
                                                     │ tts.send(word)     │
                                                     │ msgpack ─▶ WS ─▶TTS │
                                                     └────────────────────┘
                                                                 │  (TTS 服务器抢跑生成)
                          ┌──────────────────────────────────────┘
                          ▼
              ┌─────────────────────────┐
              │ TextToSpeech.__aiter__  │   收到 Audio / Text 消息
              │                         │
              │   ┌───────────────────┐ │   put(音频, 收到样本数/采样率 − 4帧)
              │   │  RealtimeQueue    │ │   put(文字, message.start_s)
              │   │ (堆:按时间戳排序)│ │
              │   └───────────────────┘ │   get_nowait(): 只放"到点了"的
              └─────────────────────────┘
                          │  按真实时间 yield
                          ▼
              ┌─────────────────────────┐
              │ _tts_loop:              │
              │  音频→output_queue(播) │
              │  文字→chat_history(字幕)│
              └─────────────────────────┘

各部件一句话职责:

部件	干什么	在哪个文件
`rechunk_to_words`	把碎 token 流重切成整词流(空格归到下一个词)	`unmute/llm/llm_utils.py:65`
`preprocess_messages_for_llm`	喂 LLM 前清洗历史:合并同角色、去打断符、给 Gemma 塞假 user 消息	`unmute/llm/llm_utils.py:16`
`VLLMStream.chat_completion`	调 OpenAI 兼容接口拿流式补全,过滤空/keep-alive 块	`unmute/llm/llm_utils.py:140`
`_generate_response_task`	喂词循环:逐词发给 TTS,发完发 Eos	`unmute/unmute_handler.py:184`
`TextToSpeech.send`	把一个词(或 Voice/Eos)msgpack 打包发给 TTS 服务器	`unmute/tts/text_to_speech.py:181`
`TextToSpeech.__aiter__`	收 TTS 消息,塞进 `RealtimeQueue`,按时释放	`unmute/tts/text_to_speech.py:267`
`prepare_text_for_tts`	去掉不可发音字符(`*_`` 等)、规范引号	`unmute/tts/text_to_speech.py:97`
`RealtimeQueue`	按时间戳排序的堆,到点才放	`unmute/tts/realtime_queue.py:18`
`_tts_loop`	消费 TTS 输出:音频进播放队列,文字进字幕	`unmute/unmute_handler.py:508`

主线走一遍(高层): LLM 流 → rechunk_to_words 攒成整词 → 喂词循环逐个 tts.send → TTS 服务器返回音频帧和带时间戳的文字 → RealtimeQueue 按时间戳掐点释放 → _tts_loop 把音频送去播、把文字送去当字幕。

3. 核心原理(逐个机制,由浅入深)

3.1 整词切分:为什么必须按整词喂

它要解决的小问题: LLM 流式返回的是任意切分的 token,一个词可能被劈成两半("wonder" → "won" + "der")。TTS 逐块拿到就无法定位词边界,会读错音。

思路: 在文本进 TTS 前加一道"重切分"——把流拼进一个 buffer,只在遇到空白时才切出一个完整的词;buffer 尾部那截"可能还没写完的词"留着,等下一块补齐。

一个关键细节:空格归到下一个词。 "foo bar baz" 被切成 "foo"、" bar"、" baz"——前导空格跟着后一个词走。这样拼回去时不用另外补空格,TTS 也能正确处理词间停顿。多个连续空白会被合并成一个空格。

原理演示:

# 示意,非源码:整词切分的核心想法
buffer = ""
prefix = ""              # 除第一个词外,每个词前面带一个空格
async for delta in llm_stream:
    buffer += delta
    while (m := re.search(r"\s+", buffer)):   # 找到一个空白 = 一个词写完了
        word = buffer[:m.start()]
        buffer = buffer[m.end():]             # 空白连同后面留在 buffer
        if word:
            yield prefix + word               # 吐出:前导空格 + 整词
        prefix = " "
# 流结束后,buffer 里剩的最后一个词也要吐出来
if buffer:
    yield prefix + buffer

真实实现: rechunk_to_words 在 unmute/llm/llm_utils.py:65-91。函数 docstring 直接点破动机:"Otherwise the TTS doesn't know where word boundaries are and will mispronounce split words."。注意 llm_utils.py:90-91 那个收尾——流结束后 buffer 里的最后一个词(它后面没有空白触发切分)必须补 yield,否则会丢词。

3.2 喂 LLM 前的消息清洗:三个坑

它要解决的小问题: 对话历史里有三种"脏东西"会让 LLM 犯迷糊,得在发出去前擦干净。preprocess_messages_for_llm(unmute/llm/llm_utils.py:16-62)一次处理这三件事:

坑	现象	处理	源码行
空打断消息	一次打断发生在 LLM 还没说出任何话之前,留下一条只含打断符 `—` 的消息	整条丢弃	`llm_utils.py:27-28`
打断符入上下文	若把结尾的 `—`(`INTERRUPTION_CHAR`, em-dash)留在历史里,LLM 可能有样学样,自己也去复读它	`.removesuffix(INTERRUPTION_CHAR)` 去掉结尾打断符	`llm_utils.py:30-32`
同角色相邻消息	打断/续话会产生两条相邻的同角色消息	合并成一条,中间加空格	`llm_utils.py:34-37`

第四件事——Gemma 需要一条假 user 消息。 有些模型(如 Gemma)如果系统消息后面直接跟 assistant 消息(没有 user 消息在中间)会犯迷糊。所以当第 0 条是 system、第 1 条是 assistant 或不存在时,硬塞一条 {"role": "user", "content": "Hello."} 进去(llm_utils.py:44-47)。

注意有两处在处理"Gemma 需要 user 消息":Chatbot.preprocessed_messages(unmute/llm/chatbot.py:77-92)在历史还很短(≤2 条,只有系统提示)时先补一条 "Hello!";preprocess_messages_for_llm 再针对"system 后面紧跟 assistant"的情况补一条 "Hello."。前者管"对话刚开始",后者管"打断后 assistant 消息被顶到最前"。

第五件事——去静默标记。 用户长时间不说话时,系统会插一条以 ...(USER_SILENCE_MARKER)开头的 user 消息。如果用户在标记插入后、LLM 回应前又开口了,这条消息就变成 "...真正说的话"。系统提示里对 ... 有专门指令,所以要把开头的标记剥掉、只留真正内容,免得混淆 LLM(llm_utils.py:49-60)。

3.3 TTS 客户端协议:发什么、收什么

它要解决的小问题: Unmute 的 TTS 是一个独立的 WebSocket 服务(moshi-server),两边用 msgpack(二进制序列化,比 JSON 小/快)通信。得约定好发什么消息、收什么消息。

发给 TTS(client → server) 有三种消息,靠 type 字段区分(unmute/tts/text_to_speech.py:30-53):

消息	作用	关键字段
`TTSClientTextMessage`	"把这段文字读出来"	`text`
`TTSClientVoiceMessage`	传自定义声音的 embedding(声音克隆,见第 05 章)	`embeddings`, `shape`
`TTSClientEosMessage`	"我发完了"(end of stream)	无

从 TTS 收(server → client) 有四种(text_to_speech.py:56-81):TTSAudioMessage(一帧 PCM 音频,pcm 是浮点样本列表)、TTSTextMessage(一个词,带 start_s/stop_s 时间戳——这个时间戳是同步的关键)、TTSErrorMessage、TTSReadyMessage。

连接参数:cfg_alpha=1.5。 建连时用 TtsStreamingQuery(text_to_speech.py:111-129)把参数拼进 URL,其中 format="PcmMessagePack" 指定音频用 msgpack 打包的 PCM,cfg_alpha=1.5(在 TextToSpeech.__init__ 里硬编码,text_to_speech.py:154-161)是 classifier-free guidance 的强度——控制生成语音多大程度贴合给定声音/文本条件。

发送时的两点讲究(TextToSpeech.send, text_to_speech.py:181-204):

原始字符串会被预处理,TTSClientTextMessage 不会。 传进来的裸 str 会走 prepare_text_for_tts 清洗;但如果你已经构造好一个 TTSClientTextMessage,它会原样发送(见 docstring text_to_speech.py:182-185)。
空文本直接丢弃(text_to_speech.py:198-199),并在发第一段文字时启动"首字延迟"计时器,用于统计 TTS 的 time-to-first-token。

prepare_text_for_tts 去掉不可发音字符(text_to_speech.py:97-108):删掉 markdown 残留的 *、_、反引号(否则 TTS 会试图"读"出这些符号),把花引号 “”‘’ 规范成直引号,并把 " : " 压成空格。

3.4 RealtimeQueue:按真实时间释放的堆(全章心脏)

它要解决的小问题: TTS 服务器抢跑——它比实时快,一股脑把音频和文字都算出来发过来了。如果收到就立刻放,声音会挤成一坨、字幕会跑到声音前头。需要一个东西:攒着这些带时间戳的项,到了它们该出现的真实时刻再放。

思路: 一个按时间戳排序的最小堆。每个项带一个"它该在第几秒被释放"的时间戳;释放时看"从启动到现在过了多久",只放那些"到点了"的项。

为什么用堆而不是队列? 因为项不一定按时间戳顺序到达(FIFO 不够用)。堆保证每次都能 O(log n) 取到"时间戳最小"的那个。看 RealtimeQueue 的 docstring:"Implemented as a heap, so it doesn't have to be FIFO."(unmute/tts/realtime_queue.py:18-22)。数据项是 TimedItem,用 @dataclass(order=True) + item 字段标 compare=False(realtime_queue.py:9-15)——这样堆只按 time 比较,不会因为 payload 不可比较而报错。

原理演示:

# 示意,非源码:RealtimeQueue 的核心
import heapq
queue = []                      # 最小堆,元素是 (时间戳, 项)
start_time = now()

def put(item, t):
    heapq.heappush(queue, (t, item))

def get_nowait():               # 只放"到点了"的,不阻塞
    elapsed = now() - start_time
    while queue and queue[0][0] <= elapsed:
        yield heapq.heappop(queue)

真实实现: RealtimeQueue.put(realtime_queue.py:39-40)只管压堆;get_nowait(realtime_queue.py:59-66)吐出所有"时间戳 ≤ 已过时间"的项、不阻塞;get 和 __aiter__(realtime_queue.py:42-79)则会 await asyncio.sleep(delta) 等到下一个项到点——用于连接关闭后把剩余项按真实时间排空。start_if_not_started(realtime_queue.py:35-37)在第一帧到来时才锁定 start_time,即"时间轴的零点"。

外部时间函数是个巧思。 构造时可传 get_time(realtime_queue.py:24-33),默认用事件循环时间。Unmute 传的是 handler.audio_received_sec(unmute_handler.py:475)——即"已收到多少秒音频",而不是墙钟时间。这样整套时序绑定到音频流的进度,不受真实流速快慢影响(见 audio_received_sec 的说明 unmute_handler.py:273-278)。

3.5 两个时间戳陷阱:音频"提前一点"、文字"来自未来"

TextToSpeech.__aiter__(text_to_speech.py:267-355)对音频和文字用两种不同的时间戳往 RealtimeQueue 里塞——这是本章最精妙的地方。

音频:只提前一点点释放。 音频入队时间戳是 已收样本数/采样率 − AUDIO_BUFFER_SEC(text_to_speech.py:294-296),AUDIO_BUFFER_SEC = FRAME_TIME_SEC * 4(4 帧 ≈ 0.32 秒,text_to_speech.py:94)。减去这个 buffer 意味着音频比"严格实时"早一点放出去,给下游 WebRTC 留一点缓冲余量。注释(text_to_speech.py:88-93)点破权衡:

`AUDIO_BUFFER_SEC`	后果
太小	缓冲不够,音频卡顿(stuttering)
太大	提前太多,难以控制文本和音频的同步(同步由 `emit()` 和 WebRTC 控制)

文字:按 start_s 释放,解决"来自未来的时间戳"。 文字入队用的时间戳是 message.start_s(text_to_speech.py:327),即这个词该开始被念的时刻。代码里两条注释(text_to_speech.py:315-326)解释了为什么文字不能收到就发:

文字消息带的时间戳是**"来自未来"**的——因为音频流被延迟了约 2 秒,而文字带的是它在音频里对应的时间点;
即便如此,音频/文字都比实时快到达。要精确追踪"已经播了多少音频"很难(.emit() 会立刻吃掉输入、内部还有缓冲),所以干脆只在文字真该显示的那一刻才放出去。

为什么用 start_s 而不是 stop_s? 注释明说(text_to_speech.py:325-326):用 stop_s(词结束时刻)会漏掉最后一个词。用 start_s(词开始时刻)更稳妥。

释放动作。 每收到一条消息后,循环末尾调 output_queue.get_nowait()(text_to_speech.py:329-333),把所有"到点了"的音频/文字 yield 出去。连接关闭后,再用 async for ... in output_queue(text_to_speech.py:347-352)按真实时间把堆里剩余的排空。

有个小细节:TTS 服务器总会发一条空文字消息 TTSTextMessage(text="", start_s=0, stop_s=0),直接 continue 跳过(text_to_speech.py:309-313)。

3.6 text/audio 同步与 emit 的关系

它要解决的小问题: __aiter__ 已经按时间释放了音频和文字,那"最终对齐"是谁保证的?答案:RealtimeQueue 只做粗对齐(掐大致时刻),精对齐交给下游的 emit() + WebRTC。

下游怎么消费: _tts_loop(unmute_handler.py:508-555)async for message in tts 拿到已经"按时释放"的消息,分流:

TTSAudioMessage → 转成 numpy 数组 await output_queue.put((SAMPLE_RATE, audio))(unmute_handler.py:535-543)进播放队列;
TTSTextMessage → 发 ResponseTextDelta 给前端当字幕,并 add_chat_message_delta 追加进聊天历史(unmute_handler.py:547-553)。

为什么把精对齐交给 emit/WebRTC? 注释(text_to_speech.py:88-93、320-324)反复强调:emit() 会立即吞掉音频输入且内部有缓冲,难以精确追踪"已播了多少",所以在客户端这层不追求逐样本对齐——只要文字在它该显示的真实时刻进队列,音频比实时早一丢丢进队列,剩下的细粒度同步由 WebRTC 的实时传输去兜。"Precise timing/buffering is less important here."(text_to_speech.py:324)

_tts_loop 里还顺手算了个吞吐比 time_received_yielded / time_since_start(unmute_handler.py:517-530)放进 debug dict——用来观测 TTS 是否真的快于实时(比值 >1 即抢跑)。

4. 巧妙之处(可借鉴的技术)

整词切分把"空格归后词",拼回免补格。 "foo"、" bar"、" baz" 直接拼接就是原句,下游不用关心词间距。unmute/llm/llm_utils.py:78-91。
一个 RealtimeQueue,两种时间戳,解两个不同的病。 音频用"收样本进度 − buffer"治卡顿,文字用 start_s 治来自未来的时间戳。同一个数据结构,喂不同的时间戳就干不同的活。text_to_speech.py:294-296 vs :327。
用堆而非 FIFO,因为项不保证按时间到。 @dataclass(order=True) + compare=False 让堆只按时间比较、payload 随便。realtime_queue.py:9-22。
时间轴绑音频进度而非墙钟。 传 get_time=audio_received_sec,整套时序对"流速快慢"免疫。unmute_handler.py:475 + realtime_queue.py:24-33。
cfg_alpha=1.5 与 PcmMessagePack 这些 moshi-server 专有参数被收进一个 pydantic TtsStreamingQuery 再序列化进 URL,连接配置集中可读。text_to_speech.py:111-161。
start_s 而非 stop_s,只为不丢最后一个词 —— 一句注释背后是踩过的坑。text_to_speech.py:325-326。

5. 边界与局限(诚实)

同步是"够好"而非"逐样本精确"。 客户端只做粗对齐,精对齐甩给 WebRTC;注释自己承认 TTS 慢于实时时仍可能失同步,因为 WebRTC 会自行缓冲(text_to_speech.py:88-93)。
AUDIO_BUFFER_SEC 是个魔法常量(4 帧),需要在"卡顿"和"失同步"之间手调,代码里没有自适应机制。
打断时的取消不在本章。 _generate_response_task 里有 if len(chat_history) > generating_message_i: break(unmute_handler.py:246-247)这类打断检查,以及 CancelledError 处理(unmute_handler.py:260-262),但真正的生命周期/取消语义由 Quest 管理,见 04。
start_up 里有"race condition"防御:建连后读最多 10 条消息,因为可能收到上一个 TTS 客户端的残包(text_to_speech.py:226-238)——说明这套服务发现下有已知的竞态,靠"多读几条直到 Ready"硬扛。
goodbye 检测靠字符串 endswith("bye!")(unmute_handler.py:621),注释自承认用 function calling 更稳,但为了能随意换 LLM 而没这么做。

6. 代码地图(导航索引)

主题	文件路径	符号名
整词切分(核心)	`unmute/llm/llm_utils.py`	`rechunk_to_words`
喂 LLM 前清洗历史	`unmute/llm/llm_utils.py`	`preprocess_messages_for_llm`
打断符 / 静默标记常量	`unmute/llm/llm_utils.py`	`INTERRUPTION_CHAR`, `USER_SILENCE_MARKER`
LLM 流式补全	`unmute/llm/llm_utils.py`	`VLLMStream.chat_completion`
Gemma 用假 user 消息(对话开头)	`unmute/llm/chatbot.py`	`Chatbot.preprocessed_messages`
TTS 客户端消息类型	`unmute/tts/text_to_speech.py`	`TTSClientTextMessage`, `TTSClientVoiceMessage`, `TTSClientEosMessage`
TTS 服务器消息类型	`unmute/tts/text_to_speech.py`	`TTSAudioMessage`, `TTSTextMessage`, `TTSReadyMessage`, `TTSErrorMessage`
连接参数(cfg_alpha/format)	`unmute/tts/text_to_speech.py`	`TtsStreamingQuery`
去不可发音字符	`unmute/tts/text_to_speech.py`	`prepare_text_for_tts`
发消息(msgpack 打包)	`unmute/tts/text_to_speech.py`	`TextToSpeech.send`
收消息 + 按时释放(两种时间戳)	`unmute/tts/text_to_speech.py`	`TextToSpeech.__aiter__`
音频提前量常量	`unmute/tts/text_to_speech.py`	`AUDIO_BUFFER_SEC`
按时间戳排序的堆	`unmute/tts/realtime_queue.py`	`RealtimeQueue`, `TimedItem`
非阻塞取"到点"项	`unmute/tts/realtime_queue.py`	`RealtimeQueue.get_nowait`
喂词循环(逐词 send + Eos)	`unmute/unmute_handler.py`	`_generate_response_task`
消费 TTS 输出(音频→播,文字→字幕)	`unmute/unmute_handler.py`	`_tts_loop`
时间轴绑音频进度	`unmute/unmute_handler.py`	`audio_received_sec`, `start_up_tts`

1. 这是什么(零基础也能懂)​

2. 顶层全景(它大概怎么转)​

3. 核心原理(逐个机制,由浅入深)​

3.1 整词切分:为什么必须按整词喂​

3.2 喂 LLM 前的消息清洗:三个坑​

3.3 TTS 客户端协议:发什么、收什么​

3.4 RealtimeQueue:按真实时间释放的堆(全章心脏)​

3.5 两个时间戳陷阱:音频"提前一点"、文字"来自未来"​

3.6 text/audio 同步与 emit 的关系​

4. 巧妙之处(可借鉴的技术)​

5. 边界与局限(诚实)​

6. 代码地图(导航索引)​