跳到主要内容

数据截至 (上游 commit c49982eb3aea)

听:流式转写 + 停顿预测决定何时该说话

30 秒导读: 语音助手最难的不是"听懂词",而是"判断你说完了没"。本章讲 Unmute 怎么 用 STT 每帧吐出的一个"停顿概率"数字,经平滑和阈值判断轮次切换(turn-taking);以及用户 插话(barge-in)时怎么打断正在说话的机器人。

本章聚焦一个问题:在一段连续的音频流里,机器什么时候该判定"轮到我说了"? 这在语音 对话里被称作 turn-taking(轮次切换)——人类靠语调、停顿、眼神无缝交接,机器只有一条 音频流可用。判早了会打断用户,判晚了对话就卡顿。

上游怎么把音频喂进来、output_queue 主循环怎么把回复放出去,见 01-orchestration-loop;本章只讲"听"这一侧。

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

一句话定义: 这一层把用户麦克风的音频流,实时转成文字(STT = Speech-To-Text,语音 转文字),并顺带判断用户话说完了没,以此决定何时唤醒 LLM 生成回复。

它要解决的核心难题——轮次切换:

想象你和朋友打电话。你说完一句话会有个自然停顿,朋友就知道"该我接了"。可如果你只是 句中换气,朋友抢话就很尴尬。机器面对的正是这个:

  • 停早了 → 用户还没说完就被打断。
  • 停晚了 → 用户说完半天机器不吭声,体验卡顿。
  • 用户在机器说话时插嘴(barge-in / 抢话)→ 机器得立刻闭嘴让路。

Unmute 的思路,一句话直觉: 不自己去猜停顿,而是信任 STT 模型——Kyutai 的流式 STT 每处理一帧音频(80ms)就回吐一个"现在是不是停顿"的概率值。Unmute 把这串概率 平滑一下、卡个阈值,就得到了轮次判断。可以把它类比成汽车油量表:原始读数会抖, 平滑后才敢据此决策。

用起来什么样(数据流直觉):

麦克风音频 ──► STT 客户端(websocket) ──► Kyutai STT 服务器
                    ▲                             │
                    │      每 80ms 回一个 Step 消息 │
                    │      里面带停顿概率 prs[2]     │
                    └─────────────────────────────┘
                                 │
                    平滑(EMA) + 阈值(0.6)
                                 │
                    ┌───────────┴───────────┐
              "用户说完了"                "用户在抢话"
              → 冲刷 STT → 叫 LLM        → 打断机器人

2. 顶层全景(这一侧怎么转)​

两条独立的异步循环协作完成"听":

循环干什么在哪
receive每收到一帧麦克风音频就调用:灌给 STT、跑停顿/静音检测、处理 flushunmute_handler.py:280 UnmuteHandler.receive
_stt_loop后台消费 STT 回吐的词,拼进 chat_history、必要时触发打断unmute_handler.py:436 UnmuteHandler._stt_loop

中间夹着一个 SpeechToText 客户端(stt/speech_to_text.py),它是对 Kyutai STT 服务器 websocket 连接的封装:send_audio 往上灌音频,__aiter__ 往下吐消息,并在内部维护 那个关键的 pause_prediction(停顿概率的平滑值)。

"用户说话 → 触发 LLM" 主线走一遍(高层):

① receive(): 每帧音频 → stt.send_audio()  ──────────────┐
② STT 服务器实时回吐 Word 消息 ──► _stt_loop() 拼词      │
   拼进 chat_history 后,conversation_state 变 user_speaking
③ receive(): determine_pause() 看平滑后的停顿概率 > 0.6?
                          │是
④ 往 STT 再灌 0.5s 的"静音帧",把管道里残留的词冲刷出来(flush)
⑤ 等 flush 走完(stt_end_of_flush_time 到点)才 _generate_response() 叫 LLM

一个反直觉但关键的点:"用户在不在说话"这个状态,不是靠音量,而是靠 chat_history 的 最后一条消息推断的。 只要 _stt_loop 把用户的词拼进了历史,conversation_state() 就返回 user_speaking(llm/chatbot.py:28-30,最后一条是非空 user 消息即为 user_speaking)。这解释了后面很多"往历史塞一条消息来改状态"的手法。

3. STT 客户端:websocket + msgpack 协议​

它要解决的小问题: 怎么和 Kyutai STT 服务器实时双向通信——一边灌音频,一边收词和 停顿信号。

协议选型: websocket 承载双向流,消息体用 msgpack(一种二进制的 JSON,比 JSON 更紧凑、编解码更快,适合每秒几十帧的音频)编码。发送在 _send 里统一打包:

stt/speech_to_text.py:123 msgpack.packb(data, use_bin_type=True, use_single_float=True) —— use_single_float=True 让 PCM 采样用 32 位浮点而非 64 位,省一半带宽。

发出去的消息只有两种:Audio(一帧 PCM,send_audio)和 Marker(打点标记, send_marker)。

收回来的消息类型是本章的主角,用 Pydantic 联合类型建模(stt/speech_to_text.py:30-69):

消息类型字段含义
Wordtext, start_time识别出一个词(带它在音频里的起始时刻)
EndWordstop_time一个词的结束边界(Unmute 直接忽略)
Stepstep_idx, prs每帧一条,prs 里藏着停顿概率
Markerid之前 send_marker 打的点,原样回来
Ready / Error—握手时用:就绪 / 满载

消费循环 __aiter__(stt/speech_to_text.py:175)是一个 async 生成器:用 match 分派每种消息。注意它对外只 yield Word 和 Marker——Step 被内部消化用于 更新停顿概率,EndWord/Ready 直接 continue 掉:

stt/speech_to_text.py:192-218——case STTWordMessage(): ... yield message; case STTStepMessage(): 分支里更新 current_time 并喂 pause_prediction。

一个细节坑: 服务器在收到结束信号后会关连接,这在 websockets 库里表现为 ConnectionClosedOK,__aiter__ 里专门 catch 掉当正常结束处理 (stt/speech_to_text.py:223-226)。

4. 核心机制:停顿预测(pause prediction)​

这是整章的技术核心。它要解决的小问题: 把 STT 每帧吐出的、抖动的停顿概率,变成一个 稳定、可用于决策的数字。

4.1 概率从哪来:prs[2]​

STT 的每个 Step 消息带一个概率列表 prs。Unmute 只取 prs[2] 当作"当前是停顿" 的概率:

stt/speech_to_text.py:212-214: self.pause_prediction.update(dt=FRAME_TIME_SEC, new_value=message.prs[2])

prs 其余分量的含义在本仓库源码里看不出来(inferred:[2] 是模型约定的某个"停顿/ 静音"类别),Unmute 只关心这一个索引。

头几帧要丢弃。 停顿概率在最初几步"到处乱跳",所以 __aiter__ 里用 n_steps_to_wait = 12(约 12×80ms ≈ 1 秒)先不喂 EMA(stt/speech_to_text.py:184, :209-214)。

4.2 为什么要平滑:带 attack/release 的 EMA​

原始的 prs[2] 逐帧抖动,直接卡阈值会误判。Unmute 用一个指数移动平均 (ExponentialMovingAverage,EMA) 平滑它——但这个 EMA 有个巧妙设计:上行和下行用 不同的时间常数(音频工程里叫 attack / release)。

直觉: 想象油量表指针——你希望它对某些变化反应快、对另一些反应慢。这里两个方向的 时间常数都设成了 0.01 秒(stt/speech_to_text.py:87-89),即上下行同速:

# 示意,非源码:EMA 每帧朝新值靠拢的核心
# attack = 值往上走(越来越像停顿),release = 值往下走(用户又开口)
alpha = 1 - exp(-dt / time_const * ln2)   # time_const 越小,alpha 越大,跟得越紧
value = (1 - alpha) * value + alpha * new_value
# 重点看:0.01s 的时间常数意味着每帧几乎立刻跟上,只滤掉最高频的抖动

真实实现区分两个方向选用哪个时间常数:

stt/exponential_moving_average.py:25-30:if new_value > self.value: 用 attack_time,否则用 release_time;alpha = 1 - np.exp(-dt/τ * np.log(2))。

注释点出了语义映射(stt/speech_to_text.py:85-86):attack = 从"说话"到"不说话", release = 从"不说话"到"说话"。 初值设 1.0(initial_value=1.0),即默认假设"当前 是停顿",这样对话开始时不会误触发。

4.3 阈值:0.6 判停顿,<0.4 允许打断​

平滑后的 pause_prediction.value 落在 [0,1]。Unmute 用两个不对称的阈值做两件事:

阈值判断用途代码
> 0.6用户说完了(停顿)触发轮次切换 → 叫 LLMunmute_handler.py:387 determine_pause
< 0.4用户明显在说话允许打断正在说话的机器人unmute_handler.py:354

中间 0.4~0.6 是滞回区(hysteresis):既不算停顿、也不敢据此打断,避免在临界点反复 横跳。这是控制系统里防抖的经典手法。

4.4 determine_pause:判"该说话了"的完整逻辑​

determine_pause(unmute_handler.py:372)在每帧被调用,它有两道前置门:

# 示意,非源码:determine_pause 的判断顺序
if stt is None:                                  return False   # STT 没起来
if conversation_state() != "user_speaking":      return False   # 用户根本没在说
return stt.pause_prediction.value > 0.6                        # 平滑停顿概率过阈

第二道门是关键:只有当 chat_history 最后一条是"非空 user 消息"时,状态才是 user_speaking(见 §2)。也就是说——必须先有词被 _stt_loop 拼进历史,才可能判停顿。 用户还没开口时永远不会误触发。

对应的真实实现:

unmute_handler.py:376-391——前两道 return False 门 + if stt.pause_prediction.value > 0.6: return True。

一个防"刚说一个词就被判停"的保护在 _stt_loop 里:每当拼进的是一条新用户消息 (用户刚开口),就把 pause_prediction.value 硬置 0.0,给 VAD 反应时间:

unmute_handler.py:462-466:if is_new_message: stt.pause_prediction.value = 0.0。

5. 检测到停顿后:灌静音做 flush,等冲刷完才叫 LLM​

它要解决的小问题: STT 是流式且有延迟的——你灌进去的音频,要过一会儿才吐出对应 的词。判定停顿的那一刻,管道里可能还有几百毫秒音频"在路上",对应的最后几个词还没吐出来。 如果立刻叫 LLM,会漏掉用户最后半句话。

思路: 主动往 STT 灌一段静音帧,把管道"冲干净"(flush),等残留的词都吐完了、 flush 确认完成了,才触发 LLM。这就是 stt_end_of_flush_time 机制。

怎么算冲刷要多久: STT 有固定延迟 STT_DELAY_SEC = 0.5 秒 (kyutai_constants.py:40)。检测到停顿时,记下"冲刷完成的目标时刻",并按这个延迟 换算成静音帧数灌进去:

unmute_handler.py:340-351:

  • self.stt_end_of_flush_time = stt.current_time + stt.delay_sec——目标时刻。
  • num_frames = int(math.ceil(stt.delay_sec / FRAME_TIME_SEC)) + 1——多灌 1 帧留安全余量。
  • 循环 stt.send_audio(zero) 灌 SAMPLES_PER_FRAME 长的全零帧。

flush 期间的"闸门":receive 每帧开头都会看 stt_end_of_flush_time 是否为 None (unmute_handler.py:337)。一旦进入 flush(非 None):

  • 不再检测停顿/打断——因为此刻 STT 处理的是一堆 0,停顿概率没有参考价值 (unmute_handler.py:360-362 的注释点明了这点)。
  • 只等 stt.current_time(STT 侧走过的时间)越过 stt_end_of_flush_time,就: 清空标志、记录 flush 耗时/RTF、_generate_response() 叫 LLM。

unmute_handler.py:363-370:if stt.current_time > self.stt_end_of_flush_time: ... await self._generate_response()。

整个停顿→flush→叫 LLM 的时序图:

 t0  停顿概率越过 0.6 (determine_pause 返回 True)
  │  发 InputAudioBufferSpeechStopped(告诉前端"我判你说完了")
  │  记 stt_end_of_flush_time = current_time + 0.5s
  │  灌 ~7 帧静音(0.5s / 0.08s + 1)
  ▼
 flush 期间:receive 每帧只检查 current_time 有没有越过目标 —— 期间不检测打断
  │
  ▼
 t1  current_time > stt_end_of_flush_time
     → 残留词已全部吐出并拼进历史
     → _generate_response() 叫 LLM(见 01 章的回复生成)

6. 两条打断路径(barge-in)​

用户在机器人说话时插嘴,得让机器立刻闭嘴。Unmute 有两条独立的打断路径,分别对应 "听到词"和"听到有人在说但还没成词"。

6.1 STT-based 打断(听到实词)​

只要 _stt_loop 从 STT 收到一个非空的词,而当前 conversation_state 是 bot_speaking,立刻打断:

unmute_handler.py:456-458:if conversation_state() == "bot_speaking": logger.info("STT-based interruption"); await self.interrupt_bot()。

这条路径最可靠——都识别出词了,用户肯定在说话。

6.2 VAD 打断(听到"有人在说"但还没成词)​

词的识别有延迟。为了更快让路,receive 里还有一条基于停顿概率的快速通道:当机器人正在 说话、且平滑停顿概率 < 0.4(明显在说话),就打断:

unmute_handler.py:352-359: elif conversation_state() == "bot_speaking" and stt.pause_prediction.value < 0.4 and audio_received_sec() > UNINTERRUPTIBLE_BY_VAD_TIME_SEC: interrupt_bot()。

打断后紧接着 add_chat_message_delta("", "user"),往历史塞一条空 user 消息把回合交还给 用户(unmute_handler.py:359)。

6.3 回声消除考量:UNINTERRUPTIBLE_BY_VAD_TIME_SEC = 3​

注意上面 VAD 打断的第三个条件:audio_received_sec() > 3。这是一个**回声消除(echo cancellation)**的补丁。

问题背景(源码注释直说了,unmute_handler.py:60-65):至少在 Mac 上,回声消除刚开始 时要一会儿才生效。开局机器人一说话,麦克风就把 TTS 的声音录进去,STT 误以为"用户在说话", 把机器人自己打断了——只在第一句话时发生。

对策: 头 3 秒(UNINTERRUPTIBLE_BY_VAD_TIME_SEC = 3)不允许 VAD 打断。但注意这个 豁免只挡 VAD 那条路——注释特意说明"STT 识别出的词仍然可以打断机器人" (unmute_handler.py:64),因为真人说的词不会被回声误判成实词。两条路径的分工在这里体现 得很清楚。

两条打断路径对比:

STT-basedVAD-based
触发信号收到一个非空词平滑停顿概率 < 0.4
在哪_stt_loop (:456)receive (:352)
延迟较高(要等成词)较低(不等成词)
头 3 秒仍可打断被回声保护禁用

打断做了什么(概览): interrupt_bot(unmute_handler.py:583)清空 output_queue、发打断事件、并移除 tts/llm 两个 Quest。Quest 移除时怎么优雅收尾 (那个 async 版 RAII 的清理细节),见 04-quest-lifecycle-interruption。

7. 长静音:注入 '...' 让机器人主动开口​

它要解决的小问题: 用户半天不说话,机器人干等着——体验很怪。应该让机器人主动说点 什么("还在吗?")。

巧妙的做法: Unmute 不给"主动开口"写一套特殊逻辑,而是复用现有的轮次机制——往 chat_history 塞一条内容为 '...'(USER_SILENCE_MARKER,llm/llm_utils.py:13)的 user 消息,伪装成"用户说了句话":

unmute_handler.py:626-638 detect_long_silence:当 conversation_state() == "waiting_for_user" 且静默超过 USER_SILENCE_TIMEOUT = 7.0 秒 (unmute_handler.py:57),add_chat_message_delta(USER_SILENCE_MARKER, "user")。

为什么这样就能触发机器人开口? 塞进这条非空 user 消息后,conversation_state() 立刻 变成 user_speaking(见 §2 的推断规则),于是下一帧 determine_pause 的第二道门通过、 停顿一旦成立就叫 LLM——机器人就"回应"了这个 '...'。注释里也点明了这条连锁反应 (unmute_handler.py:633-636)。

LLM 侧怎么理解 '...': 系统提示里有专门规则告诉模型如何处理这些 '...' 消息 (注释 unmute_handler.py:635)。人格/系统提示的细节见 05-voices-prompt-personality。

防误触发的配合: 机器人正在说话(bot_speaking)时,receive 每帧都会把 waiting_for_user_start_time 重置为当前时刻(unmute_handler.py:304-306),这样机器人 说话的那段时间不会被算进"用户静默"里,避免刚说完就立刻误判长静音。

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

  • 信任模型、别自己造 VAD。 轮次判断不写一套音量/能量启发式,而是直接吃 STT 每帧吐出 的停顿概率 prs[2],自己只做平滑+阈值。上游模型升级,轮次判断自动受益。 (stt/speech_to_text.py:212-214)

  • 不对称阈值 = 滞回防抖。 0.6 判停顿、0.4 才允许打断,中间留缓冲区,避免临界点横跳。 (unmute_handler.py:387, :354)

  • flush 机制解决"流式延迟丢尾巴"。 灌 0.5s 静音把管道冲干净、用 stt_end_of_flush_time 等确认,而不是判停就立刻叫 LLM——保证用户最后半句不丢。(unmute_handler.py:344-370)

  • 两条打断路径 + 定向豁免。 STT-word 打断可靠但慢、VAD 打断快但会被回声骗;于是只对 VAD 路径加 3 秒回声保护,word 路径不受影响。分层而非一刀切。(unmute_handler.py:60-65)

  • 用"塞消息改状态"统一轮次逻辑。 conversation_state 完全由 chat_history 尾巴推断, 于是"长静音让机器人开口""打断后交还回合"都退化成"往历史塞一条消息",不需要额外状态机。 (llm/chatbot.py:21-37)

9. 边界与局限​

  • 停顿判断完全依赖 prs[2] 单个数字。 若 STT 模型对某类语音(口音、背景噪声)的停顿 概率不准,轮次切换会跟着失灵——Unmute 侧无从纠正。

  • 回声保护是硬编码 3 秒 + 平台经验值。 注释明说是针对 Mac 观察到的现象 (unmute_handler.py:60-63);别的设备/回声消除实现下这个常数未必合适。

  • flush 期间完全放弃打断检测。 那 ~0.5 秒里若用户抢话,要等 flush 走完才可能响应 (unmute_handler.py:360-362)。

  • prs 其余分量语义在本仓库看不出来。 只知道用了索引 2;其它索引代表什么、模型 侧如何定义,源码里无从得知(inferred)。

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

主题文件符号名
STT websocket 客户端 / msgpack 收发unmute/stt/speech_to_text.pySpeechToText, _send, send_audio
STT 消息类型建模unmute/stt/speech_to_text.pySTTWordMessage, STTStepMessage, STTEndWordMessage, STTMarkerMessage
消费循环 / 喂停顿概率unmute/stt/speech_to_text.pySpeechToText.__aiter__
停顿概率平滑(attack/release EMA)unmute/stt/exponential_moving_average.pyExponentialMovingAverage.update
每帧主处理:灌音频/停顿/flush/打断分支unmute/unmute_handler.pyUnmuteHandler.receive
停顿判定(阈值 0.6)unmute/unmute_handler.pyUnmuteHandler.determine_pause
消费 STT 词 / STT-based 打断unmute/unmute_handler.pyUnmuteHandler._stt_loop
长静音注入 '...'unmute/unmute_handler.pyUnmuteHandler.detect_long_silence
打断机器人(清队列/移除 Quest)unmute/unmute_handler.pyUnmuteHandler.interrupt_bot
关键常量unmute/kyutai_constants.pySTT_DELAY_SEC, FRAME_TIME_SEC, SAMPLES_PER_FRAME
回声保护 / 静默超时常量unmute/unmute_handler.pyUNINTERRUPTIBLE_BY_VAD_TIME_SEC, USER_SILENCE_TIMEOUT
静音标记 / 打断字符unmute/llm/llm_utils.pyUSER_SILENCE_MARKER, INTERRUPTION_CHAR
轮次状态推断unmute/llm/chatbot.pyChatbot.conversation_state