Speculative Turns 与 Barge-in:边说边改、可打断的全双工
30 秒导读: 全双工语音助手有两件难事:一是用户在助手说话时插嘴,要立刻掐掉助手的声音(barge-in);二是用户话说了一半又接着说,要把"半句话的转写"升级 成"整句话的转写",而不是让助手对着半句话答一遍、又对着整句话再答一遍。这一章讲这个项目怎么用两套独立的版本号分别解决它们,并把"取消 + 丢弃陈旧数据"做成一条贯穿从 VAD 到 WebSocket 全链的暗线。
本章是全项目工程含量最高的一章。它假设你已读过 02-vad-turn-detection.md(VAD 怎么判断"说完了"、怎么发 SpeechStartedEvent / SpeechStoppedEvent / VADAudio);流水线线程模型见 01-pipeline-runtime.md。这里只讲取消与版本这条线,VAD 的机械细节不重复。
1. 这是什么(零基础也能懂)
1.1 两个真实场景
先用两段人机对话把问题讲清楚。
场景 A —— 打断(barge-in):
用户: "帮我订一张明天去上海的高铁票……"
助手: "好的,正在为您查询明天……" ← 助手开始念了
用户: "等等,改成后天。" ← 用户插嘴
助手: (立刻闭嘴)"好的,后天去上海……"
助手正在朗读音频时用户开口。系统必须当场掐断正在播的声音、丢掉队列里还没发出去的音频,不能等这一句念完。这叫 barge-in(抢话)。
场景 B —— 边说边改(speculative turn):
用户: "订张票……" ← 停顿了一下
[STT 转写出 "订张票",系统开始准备回答]
用户: "……去上海的。" ← 用户又接上了
用户中间那个停顿,VAD 差点当成"说完了"。系统乐观地先按"订张票"开始转写、甚至开始生成回复;但当用户接着说下去,系统要能把这一整段当成同一个话轮,用更长音频的新转写("订张票去上海的")替换旧转写,而不是当成两句话、孤立地回两次。这叫 speculative turn(投机话轮):先赌用户说完了,赌错了能回滚。
1.2 一句话直觉
两个场景背后是同一件事:系统总在处理"可能过时"的数据。
- 助手正在播的音频,可能已经因为用户插嘴而过时。
- STT 刚转出的文本,可能已经因为用户继续说话而过时。
解决办法都是给数据贴版本号:处理到一半时回头看一眼"我手上这份还是最新的吗?",不是就默默丢掉。这个项目用了两套版本号,各管一摊:
| 机制 | 版本载体 | 管什么 | 核心文件 |
|---|---|---|---|
| CancelScope | 一个全局递增的 generation 整数 | 助手说话时被打断 → 掐音频、清队列 | pipeline/cancel_scope.py |
| SpeculativeTurnTracker | 每个话轮一个 (turn_id, revision) | 话轮被回答前可"重开",用更长转写替换旧的 | pipeline/speculative_turns.py |
本章把这两套讲透,再串起所有消费方——是谁在什么时候查这些版本号、查到过时了怎么丢。
2. 顶层全景(取消与陈旧丢弃这条暗线)
2.1 两套版本、一条暗线
先看这条暗线怎么贯穿全链。下图从左到右是数据流方向,竖线上标出"谁在查哪套版本号"。
麦克风音频
│
▼
┌─────────┐ speech_started / speech_stopped / VADAudio(turn_id,revision)
│ VAD │───────────────┬──────────────────────────────┐
│ handler │ 管 (B) 话轮: │ │
└─────────┘ 开新轮/重开轮 │ │
│ ▼ ▼
│ ┌─────────────┐ ┌──────────────┐
▼ │ STT │ │ 送 send loop │
VADAudio ──────────▶│ base_stt │ │ 的 text 队列 │
(带 turn_id,rev) │ 输入/输出 │ └──────────────┘
│ 陈旧门控 (B) │ │
└─────────────┘ │ speech_started
│ Transcription(turn_id,rev) │ 到达时
▼ ▼
┌─────────────┐ ┌──────────────┐
│ LLM │ │ CancelScope │
│ 生成回复 │ AssistantText │ .cancel() (A)│
└─────────────┘ (turn_id,rev, │ 掐音频/清队列 │
│ cancel_gen) └──────────────┘
▼ ▲
┌───────────────────────────┐ │
│ service._dispatch / │ 查 (B) │ (A) generation
│ response.on_assistant_text │◀────────┘ 贴在每个音频块上
│ commit_if_latest… (B) │
└───────────────────────────┘
│
▼ response.output_text / audio 回给客户端
怎么读这张图: 标 (A) 的地方在用 CancelScope 的 generation(barge-in);标 (B) 的地方在用 SpeculativeTurnTracker 的 (turn_id, revision)(投机话轮)。两套完全独立,但都在回答同一个问题:"我手上这份数据还算数吗?"
2.2 两套机制的分工
为什么要两套,不能合并成一套? 因为它们保护的东西不同、时机不同:
| 维度 | (A) CancelScope | (B) SpeculativeTurnTracker |
|---|---|---|
| 触发时刻 | 助手已经在说话、用户插嘴 | 助手还没回答、话轮可能变长 |
| 要丢弃的东西 | 已在飞的音频/文本输出 | 过时的转写和待生成请求 |
| 粒度 | 全局一个计数器(整个 pipeline 一个响应) | 每个话轮各一条 revision 线 |
| 结果 | 硬取消,清空队列 | 软回滚,旧 revision 静默作废 |
| 载体 | int generation | dict[turn_id → revision] + 状态机 |
一句话:(A) 是"闭嘴",(B) 是"这句话我改主意了"。 下面两大节分别讲。
3. 机制 A:CancelScope —— 助手说话时被打断
3.1 它要解决的小问题
助手的一次回复不是一个原子操作,而是一串音频块在流水线里排队、逐块发给客户端。用户插嘴那一刻,这串音频块散落在:LLM 还没生成完的、TTS 正在合成的、发送队列里排着的、WebSocket 已经在发的。要"闭嘴",就得让所有这些散落的块都知道"你们过时了,别发了"。
难点是时序:你不能用"发一个短脉冲信号"去通知,因为某个线程可能刚好没在监听那一瞬。需要一个持久、可轮询的标记。
3.2 思路:一个只增不减的世代号
CancelScope 的核心就是一个整数 generation(世代号)。
- 每次开始一个新响应,pipeline 线程(LLM、TTS)在起点抓一份当前的
generation,一路带着它。 - 用户插嘴 → 调
cancel(),generation加一。 - 任何线程回头一比:"我起点抓的世代 ≠ 现在的世代" → 我这份过时了(
is_stale),停手。
因为世代号单调递增、随时可读,没有"错过脉冲"的问题——晚一点读也读得到"变了"。类注释把这点讲得很直白:用世代计数器代替 "brief-pulse timing games"(cancel_scope.py:6-8,CancelScope docstring)。
3.3 原理演 示
# 示意,非源码:世代号如何判定"过时"
scope = CancelScope()
# 响应 1 起点:抓世代
gen = scope.generation # 假设 = 0
# ... LLM/TTS 边跑边检查 ...
if scope.is_stale(gen): # gen(0) != 当前世代 → True 就停手
return
# 用户插嘴:
scope.cancel() # 世代 0 → 1,并打开 discarding 开关
# 响应 1 的线程下一次检查:
scope.is_stale(gen) # 0 != 1 → True,过时,停手
重点看:cancel() 之后,所有还抱着旧世代的线程下一次 is_stale 都会返回 True,自动作废——不需要挨个通知。
3.4 真实实现:cancel_scope.py 全文导读
整个类只有 65 行,五个关键方法。逐个看它们在解决什么。
generation / is_stale —— 版本判定的地基。 线程起点读 generation(cancel_scope.py:18-22),之后用 is_stale(gen) 判断有没有被 cancel 超越:
def is_stale(self, gen: int) -> bool: # cancel_scope.py:52-54
return gen != self._gen
cancel() —— 打断的入口。 做三件事:记下"被丢弃的那一代"、世代加一(带 & 0xFFFFFFFF 防 32 位溢出)、打开 discarding 开关(cancel_scope.py:24-33)。加一是给 is_stale 用的;discarding 是给 send loop 用的(见下)。
discarding —— 给发送循环的"丢弃窗口"。 只有 is_stale 还不够:cancel 之后到"下一个响应正式开始"之间有一段真空,这段时间里漏出来的陈旧输出得有人丢。discarding 就是这段窗口的开关(cancel_scope.py:56-59)。
response_done / new_response —— 关掉丢弃窗口。 窗口不能一直开着,否则会误杀新响应。两条关闭路径:
response_done(generation):pipeline 正常跑完一个响应时调用,清掉discarding。它带一个世代守卫——只有传进来的generation等于"被丢弃那代"或"当前代"才真正清除,防止一个迟到的旧响应把新响应的丢弃窗口误关(cancel_scope.py:35-44)。new_response():客户端显式发response.create开新响应时,直接清窗口(cancel_scope.py:46-50)。
3.5 消费方:send loop 在 SpeechStartedEvent 上执行 barge-in
CancelScope 的实际"扣扳机"在 WebSocket 的每-pipeline 发送循环里。send loop 先处理文本队列(注释明说"Text events first (speech_started cancels active response)",websocket_router.py:466-467),因为 SpeechStartedEvent 一旦到达就意味着用户开口了。
关键逻辑在 websocket_router.py:487-503:只有当当前确实有响应在进行或待生成(was_in_response or was_response_pending)、且这次 speech-start 带 interrupt_response 且会话允许打断时,才执行完整的 barge-in 四连:
# websocket_router.py:492-498(节选)
unit.cancel_scope.cancel() # ① 世代 +1,开丢弃窗口
if session_id:
unit.service._state(session_id).response_pending = False # ② 清"待生成"标记
_flush_queue(unit.output_queue, preserve=_keep_audio_sentinel) # ③ 清音频队列
_flush_queue(unit.text_output_queue, preserve=_keep_user_text_event)
if unit.response_playing.is_set():
unit.response_playing.clear() # ④ 清"正在播放"标记
四步的意义:① 让所有在飞的音频块 is_stale;② 让"排队等生成"的响应也别再触发;③ 把队列里已经堆着的音频/文本当场倒掉(但保留哨兵和用户文本事件);④ 复位播放状态。ResponseCancelEvent(客户端显式取消)走的是同一套四连,只是入口在 websocket_router.py:370-379。
3.6 关键细节:音频和文本要"锁步"丢弃
一个微妙的坑。send loop 丢弃陈旧输出时,音频块和助手文本是两条路,但它们必须用同一套判据,否则会出事。这个判据抽成了 _generation_is_discardable(websocket_router.py:205-220):
def _generation_is_discardable(unit, generation): # websocket_router.py:205
if generation is not None and unit.cancel_scope.is_stale(generation):
return True # 被世代超越 → 丢
if unit.cancel_scope.discarding and generation != unit.cancel_scope.generation:
return True # 在丢弃窗口内、且不是当前活代 → 丢
return False
注释(websocket_router.py:210-215)记了为什么两条路要锁步:如果只用 discarding 去丢文本(不带这个世代检查),会误杀一个新响应的转写——比如某个被超越的投机话轮的 TTS 从没发出 AUDIO_RESPONSE_DONE 哨兵,于是 response_done() 从没被调、discarding 一直挂着,结果把后面新响应的文本也一起吞了。加上 generation != 当前活代 这个条件,就把"陈旧的"和"当前活的"精确分开了。音频侧走 _should_discard_audio(websocket_router.py:223-224)、文本侧在 websocket_router.py:478-480,共用同一函数。
响应正常收尾时怎么关窗口? send loop 收到 AUDIO_RESPONSE_DONE 哨兵,若该音频世代已 stale 就走"陈旧完成"分支、否则走正常完成分支,两条都会调 cancel_scope.response_done(audio_generation) 把 discarding 关掉、重新允许监听(websocket_router.py:524-541)。
4. 机制 B:SpeculativeTurnTracker —— 话轮的重开与提交
4.1 它要解决的小问题
回到场景 B。VAD 因为一个短停顿"软结束"了一个话轮,发出了带 (turn_id, revision=0) 的 VADAudio,STT 开始转写。这时用户又接着说了。系统面临选择:
- 当新话轮:那"订张票"和"去上海的"会变成两条独立记录,助手可能对着半句就抢答了。
- 重开旧话轮:把新说的音频接到旧音频后面,重转写成一个
revision=1,让revision=0静默作废。
后者才对。但"重开"要满足两个约束:
- 只有还没被回答(未 commit)的话轮才能重开——助手已经答过的话不能反悔。
- 重开有时间窗:用户停太久(超过阈值)就不该再算同一句了。
SpeculativeTurnTracker 就是管这套状态的线程安全跟踪器(speculative_turns.py:24-25)。
4.2 核心状态:四张表
Tracker 内部四张表(speculative_turns.py:33-36),都以 turn_id 为键:
| 表 | 类型 | 含义 |
|---|---|---|
_latest_revision | OrderedDict[turn_id → int] | 这个话轮当前最新是第几版(顺序表,便于按 LRU 剪枝) |
_committed_revision | dict[turn_id → int] | 已被回答/落定到第几版(commit 之后不能再改) |
_pending_reopen | dict[turn_id → _PendingReopen] | 有个"重开候选"正在悬着(base 版 → candidate 版) |
_reopen_grace | dict[turn_id → _ReopenGrace] | 一个宽限截止时刻:在此之前先别急着 commit |
is_latest(turn_id, rev) 就是查第一张表:传进来的 rev 等于最新 rev 才算数(speculative_turns.py:50-54)。所有消费方判"陈旧"最终都归到这一问。
4.3 三个阶段:观察 → 重开 → 提交
一个话轮的生命周期,配着三组方法:
observe(turn,0) begin_reopen_candidate → confirm commit_if_latest…
│ │ │
▼ ▼ ▼
┌─────────┐ 用户接着说 ┌──────────────────────┐ 助手要答了 ┌──────────────┐
│ rev=0 │─────────────▶│ pending_reopen 悬 起 │────────────▶│ 落定 committed│
│ 已观察 │ │ candidate=rev+1 │ (grace 到期) │ = 最新 rev │
└─────────┘ │ confirm 后 latest=rev+1 │ └──────────────┘
└──────────────────────┘
│ 用户没接着说 / 另起新轮
▼
cancel_reopen_candidate(丢弃候选)
阶段 1 —— 观察。 VAD 每开一个新轮就 observe(turn_id, revision),把最新版登记进 _latest_revision,并 notify_all 唤醒等在门控上的线程(speculative_turns.py:38-48)。只接受"比现存更大"的 revision,天然幂等。
阶段 2 —— 重开(两段式)。 重开不是一步到位,而是"先挂候选、后确认",避免竞态:
begin_reopen_candidate(turn, rev):若该轮未 commit、且 rev 仍是最新,就登记一个_PendingReopen(base=rev, candidate=rev+1),返回候选版号(speculative_turns.py:250-277)。这一步只挂起、不改 latest——此时任何查_latest_revision的还看到旧 rev,但知道有个重开悬着(_has_pending_reopen_locked,speculative_turns.py:343-345)。confirm_reopen_candidate(turn, base, candidate):真正把_latest_revision提升到 candidate,删掉 pending(speculative_turns.py:279-316)。确认前还会复查"没被 commit、base 仍是最新",否则放弃。cancel_reopen_candidate:候选作废(用户其实没接着说、或开了全新轮),删掉 pending(speculative_turns.py:318-330)。
阶段 3 —— 提交。 助手真的要产出这句话的回复时,调 commit_* 系列把 _committed_revision 定死。commit 之后 is_committed 为真,该轮再也不能重开(speculative_turns.py:241-248)。
4.4 关键武器:pending-reopen 门控与 grace 窗口
这是整套机制最精妙的地方——让下游"等一等",而不是让它读到一个正在变的中间态。
pending-reopen 门控: 下游(STT、dispatch)在判断"我这份是不是最新"时,如果发现该轮有个重开候选正悬着,它不能立刻下结论,得先等候选落定。is_latest_after_pending_reopen(speculative_turns.py:56-61)就是"先 _wait_for_pending_reopen_locked 等,再查 latest";_wait_for_pending_reopen_locked(speculative_turns.py:372-388)用 Condition.wait 阻塞,最长 2 秒(_PENDING_REOPEN_WAIT_TIMEOUT_S,speculative_turns.py:27),超时就强行清掉悬起的候选并告警。
为什么要等? 如果不等,STT 可能在候选"就要变成 rev=1"的前一刻,判定"rev=0 还是最新"、把 rev=0 的转写放行,然后 rev=1 才落定——旧转写就漏出去了。等一下,让候选先落定,再判断,就不会误放。
reopen-grace 窗口: grace 管的是另一个时机——话轮软结束后、助手还没提交前的那一小段"再等等,说不定用户还要接"。VAD 在发出 VADAudio(mode="final") 那一刻,调 start_reopen_grace(turn, rev, speculative_reopen_ms/1000) 开一个宽限截止时刻(vad_handler.py:711-719;start_reopen_grace 定义在 speculative_turns.py:183-201)。在这个窗口内,助手侧的 commit 会被 is_latest_after_reopen_grace 拖住(speculative_turns.py:76-81 → _wait_for_reopen_gate_locked,speculative_turns.py:361-370),给"用户续说 → 触发重开"留出时间。
门控与 grace 的区别一句话: pending-reopen 门控是"已经有候选悬着,等它定";grace 是"可能会有候选,先别急着 commit,预留一点时间"。前者被动等已发生的,后者主动预留可能发生的。
4.5 阻塞版 vs 非阻塞版:try_ 家族
Tracker 里几乎每个门控方法都有一对:阻塞版(is_latest_after_pending_reopen)和非阻塞版(try_is_latest_after_pending_reopen,speculative_turns.py:63-74)。非阻塞版遇到"还有候选悬着"时返回 None,意思是"现在下不了结论,等它定了再来问我"。
这是给异步 send loop 用的:send loop 不能在 asyncio 里同步阻塞 2 秒,所以它用 try_ 版拿到 None 就把事件留到下一轮再 dispatch,而 STT/LM 这种独立线程可以放心用阻塞版。commit 也同理有 commit_if_latest_after_reopen_grace 和 try_commit_if_latest_after_reopen_grace(speculative_turns.py:111-122 / 143-162)。
5. 串起消费方:这条暗线经过谁的手
版本号只是数据,真正的行为在谁在查、查完怎么办。按数据流顺序走一遍。
5.1 VADHandler:重开的发起者
VAD 是唯一产生 revision 变化的地方。它靠三个判断和三个动作(vad_handler.py:211-342):
判定能否重开 —— _should_reopen_current_turn(vad_handler.py:220-248): 满足全部才 True——用了 realtime/speculative 模式、有当前轮元数据、该轮未 committed、且距上次软结束的音频时钟间隔在窗口内。间隔用音频时钟而非墙钟(elapsed_ms = audio_start_ms - self._last_final_audio_ms,vad_handler.py:236),所以推挽式(push-to-talk)的静默间隙会"冻结"这个窗口,不会误判超时。
挂候选 —— _begin_pending_reopen_if_needed(vad_handler.py:250-270): 检测到续说时,调 tracker 的 begin_reopen_candidate 挂起候选,记在 VAD 本地的 _pending_reopen_candidate。
确认/重开 —— _confirm_pending_reopen(vad_handler.py:280-294)与 _reopen_current_turn(vad_handler.py:296-316): 语音正式接上时,_ensure_turn_for_speech_start(vad_handler.py:318-342)优先确认已挂的候选,否则在窗口内直接重开,再不行才 _start_new_turn 开全新轮(开新轮会先 _cancel_pending_reopen 撤掉悬着的候选,vad_handler.py:176-177)。
5.2 BaseSTTHandler:输入/输出双向陈旧门控
STT 在两个方向都拦陈旧数据(base_stt_handler.py 全文):
- 输入门控
should_process_input(base_stt_handler.py:24-61): 处理一条VADAudio前先问三件事——这版是否已"最终转写完"(_is_completed_final_revision,同一(turn,rev)只做一次 final)、progressive 版是否已有对应 final 排队(有就跳过,base_stt_handler.py:32-34)、以及是否仍是最新。判"最新"时,mode=="final"会额外走wait_for_stability(is_latest_after_stability_window,speculative_turns.py:203-223),多等一个 settle 窗口确认没有更新版;非 final 走 pending-reopen 门控。判定陈旧就顺手_drop_stale_queued_inputs把队列里同样陈旧的一起清掉(base_stt_handler.py:103-127)。 - 输出门控
should_emit_output(base_stt_handler.py:63-71): 转写出来了、发出去前再查一次最新性,过时就不发。发出 final 前用before_emit_output→_mark_completed_final_revision登记"这版 final 完成了"(base_stt_handler.py:73-75、200-208),供输入门控去重。
一句话:STT 是双保险——进来时不处理陈旧的,出去时不发陈旧的。
5.3 service:dispatch 层的陈旧判定
助手文本、token 用量等 pipeline 事件回到协议层时,_dispatch_pipeline_event(service.py:326-356)先过 _is_stale_turn_event(service.py:358-377)。这里有个分层:
- 转写类事件用最朴素的
is_latest。 AssistantTextEvent/TokenUsageEvent(即助手的产出)则走is_latest_after_reopen_grace(阻塞)或try_is_latest_after_reopen_grace(非阻塞,返回None表示"重开候选/grace 未定,稍后重试")。为什么助手产出要过 grace?因为这些才是"要不要真的落定这句话"的关键决策点,必须等 grace 窗关了、确认用户没续说,才敢往下走。
非阻塞路径由 dispatch_pipeline_event / try_dispatch_pipeline_event(service.py:305-316)分别对应阻塞/非阻塞消费方;should_defer_pipeline_event(service.py:318-324)让 send loop 能提前判断"这事件现在有 pending/grace,先押后"。
5.4 response.on_assistant_text:提交的临门一脚
真正的 commit 发生在 response.on_assistant_text(response.py:259-335)。助手文本要变成回给客户端的 response.output_text / audio 之前,先调 commit_if_latest_after_reopen_grace(阻塞)或其 try_ 版(response.py:267-283):
- 返回
None(仅try_版):重开候选还悬着,on_assistant_text直接返回None,让 send loop 稍后重试。 - 返回
False:这版已不是最新(用户续说、被更长转写替换),丢掉这段助手文本(response.py:281-283),不回给客户端——这就是"投机赌错了、回滚"的落点。 - 返回
True:落定,继续构造ResponseAudioTranscriptDoneEvent/ResponseTextDeltaEvent等真正发出去(response.py:284-335)。
这一步把两条暗线收口:commit 成功 → 该轮 is_committed,VAD 那边 _should_reopen_current_turn 就再也不会重开它(§5.1),投机窗口正式关闭。
6. 巧妙之处(可借鉴的技术)
世代计数器代替脉冲信号。 用单调递增的 int + is_stale(gen) 轮询,而不是"发一次取消信号"。晚读也读得到"变了",彻底消掉"错过通知"的时序竞态(cancel_scope.py:6-8、52-54)。
丢弃判据在音频与文本间锁步。 _generation_is_discardable 把"stale 或 处于丢弃窗口且非当前活代"抽成一个函数,音频侧和文本侧共用,避免了"只丢一半、吞掉新响应转写"的隐蔽 bug——注释里记着这个真实教训(websocket_router.py:205-220)。
重开做成两段式(begin → confirm)。 先挂候选(不改 latest,但下游能看见"有候选悬着"),再确认(真正改 latest)。中间态可被下游"等待",避免了"读到半变状态"的竞态(speculative_turns.py:250-316)。
门控用 Condition.wait 让下游"等一下"而非"读脏"。 is_latest_after_pending_reopen 宁可阻塞至多 2 秒等候选落定,也不放行一个即将过时的转写(speculative_turns.py:56-61、372-388)。
每个阻塞门控都配一个 try_ 非阻塞版。 独立线程用阻塞版求简单,asyncio send loop 用 try_ 版拿 None 后押后重试,同一套语义两种消费姿势(speculative_turns.py:63-74、124-141)。
音频时钟而非墙钟量重开窗口。 用 audio_start_ms - _last_final_audio_ms 计时,push-to-talk 的静默间隙自动"冻结"窗口,避免网络抖动/手动停顿被误判为超时(vad_handler.py:236)。
有界跟踪 + LRU 剪枝。 _latest_revision 用 OrderedDict 配 _prune_tracked_turns,超过 _MAX_TRACKED_TURNS=2048 就淘汰最老的、且跳过还悬着 pending/grace 的轮,长会话不会内存无限增长(speculative_turns.py:390-404)。
7. unanswered_reopen_ms 与 grace 的取舍
这是本章最值得单独讲的一个设计权衡:两个窗口,量的是不同的东西。
7.1 两个窗口各管一段
软结束(VADAudio final)
│
├──── speculative_reopen_ms (默认 1000ms) ────┐ ← grace 窗:延迟 commit
│ "标准重开宽限:先别急着落定这句话" │
│ │
└──── unanswered_reopen_ms (默认 7000ms) ──────────────┐ ← 重开资格上限
"只要还没被任何助手输出 commit, │
续说仍可重开同一轮(资格上限)" │
▼
超过则开新轮
| 参数 | 默认 | 量什么 | 定义 |
|---|---|---|---|
speculative_reopen_ms | 1000ms | commit 前的宽限期:这段时间内延迟落定,等用户可能续说 | vad_arguments.py:64-69 |
unanswered_reopen_ms | 7000ms | 未被回答的话轮的重开资格上限:更宽松的兜底窗 | vad_arguments.py:70-75 |
7.2 为什么要拆成两个
关键在 _should_reopen_current_turn(vad_handler.py:245-248):有 speculative_turns 时,重开资格用的是 unanswered_reopen_ms(7 秒),不是 grace 的 speculative_reopen_ms(1 秒)。
- grace(1 秒)只延迟 commit,不代表"过了 1 秒就不能重开"。VAD 那边的注释点破了这个分工:"The grace window only delays response commits; reopen eligibility ... is extended separately via unanswered_reopen_ms"(
vad_handler.py:712-714)。 - unanswered_reopen_ms(7 秒)是资格兜底:只要这一轮还没有任何助手输出 commit 它,用户在 7 秒内续说,就仍然重开同一轮,而不是孤立地开新轮让助手对着半句话答一遍(
vad_handler.py:240-248)。这个上限也顺带兜住"空转写"这种永远不会 commit 的退化情况——否则那种轮会永远重开不掉。
7.3 取舍的本质
| 窗口太小 | 窗口太大 |
|---|---|
| grace 太短 → 用户稍一停顿就把半句话落定、抢答 | grace 太长 → 每句回复都白等,延迟升高,牺牲全双工的"快" |
| unanswered 太短 → 慢速说话者/长停顿被拆成多轮,助手重复回应 | unanswered 太大 → 一个真正结束的话轮被"钉"太久不 commit,占资源、也拖后续 |
所以默认值 grace=1s / unanswered=7s 是"常态快、异常宽"的折中:绝大多数流利话轮 1 秒宽限就落定、延迟低;只有明显还没被回答的、可能被拆断的话轮,才享受 7 秒的宽松兜底。构造时还强制 unanswered_reopen_ms = max(speculative_reopen_ms, unanswered_reopen_ms)(vad_handler.py:92),保证兜底窗不会小于宽限窗(参数 help 也写明"Has no effect below speculative_reopen_ms")。
8. 边界与局限
- pending-reopen 阻塞封顶 2 秒。
_PENDING_REOPEN_WAIT_TIMEOUT_S = 2.0(speculative_turns.py:27)。若上游因故 2 秒内没确认候选,门控会强行清掉候选并告警放行(speculative_turns.py:381-385)。极端情况下这可能放过一个本应重开的轮——用超时上限换"绝不永久卡死"。 - 世代号 32 位回绕(理论)。
cancel()用& 0xFFFFFFFF截断(cancel_scope.py:30-32),约 43 亿次取消后回绕;实践中一次会话远达不到,代码注释自己都在打趣("after 4 billion generations, we'll wrap around xD")。 - 跟踪表有界,老轮会被淘汰。 超过 2048 个轮后最老的被 LRU 剪掉(
speculative_turns.py:390-404);对一个轮做极晚的陈旧判定,若它已被淘汰,is_latest会因查不到而默认返回 True(speculative_turns.py:53-54)——即"当作最新放行"。这是有意的降级(宁可放行也不误杀),但意味着超长历史的精确性不保证。 - CancelScope 假设单写多读。 类注释明说依赖 GIL 让 int/bool 读写原子、无锁(
cancel_scope.py:8-11),只有一个写者(asyncio router 线程)。若未来搬去无 GIL 运行时或多写者,这个无锁假设会破。 - 两套机制不互替。 barge-in 掐的是"已在播的音频",投机话轮回滚的是"还没答的转写";一个已 committed 并开始播放的响应,只能靠 CancelScope 打断,SpeculativeTurnTracker 已经管不了它了。
9. 代码地图(导航索引)
| 主题 | 文件 | 关键符号 |
|---|---|---|
| 世代计数器全文 | src/speech_to_speech/pipeline/cancel_scope.py | CancelScope、cancel、is_stale、response_done、new_response、discarding |
| 投机话轮跟踪器全文 | src/speech_to_speech/pipeline/speculative_turns.py | SpeculativeTurnTracker、observe、is_latest、begin_reopen_candidate、confirm_reopen_candidate、commit_if_latest_after_reopen_grace、start_reopen_grace |
| barge-in 扣扳机(send loop) | src/speech_to_speech/api/openai_realtime/websocket_router.py | _send_loop_for(:466-509)、_generation_is_discardable(:205-224)、_should_discard_audio |
| 显式取消入口 | src/speech_to_speech/api/openai_realtime/websocket_router.py | ResponseCancelEvent 分支(:370-379) |
| 重开发起(VAD) | src/speech_to_speech/VAD/vad_handler.py | _should_reopen_current_turn、_begin_pending_reopen_if_needed、_confirm_pending_reopen、_reopen_current_turn、_ensure_turn_for_speech_start(:211-342) |
| grace 窗开启(VAD) | src/speech_to_speech/VAD/vad_handler.py | start_reopen_grace 调用(:711-719) |
| STT 输入/输出陈旧门控 | src/speech_to_speech/STT/base_stt_handler.py | should_process_input、should_emit_output、_is_latest_turn_item、_drop_stale_queued_inputs |
| dispatch 层陈旧判定 | src/speech_to_speech/api/openai_realtime/service.py | _dispatch_pipeline_event、_is_stale_turn_event(:305-377) |
| 提交临门一脚 | src/speech_to_speech/api/openai_realtime/handlers/response.py | on_assistant_text(:259-335) |
| 参数定义 | src/speech_to_speech/arguments_classes/vad_arguments.py | speculative_reopen_ms、unanswered_reopen_ms(:64-75) |
| 单元测试(行为参照) | tests/test_speculative_turns.py、tests/openai_realtime/test_websocket_router.py | — |