流水线运行时:BaseHandler、线程模型与消息流
30 秒导读: 这个语音对话系统被拆成一串处理段(听声音→转文字→喊大模型→合成语音), 每一段都是同一个抽象
BaseHandler:自己占一条线程,循环地从"入队"拿东西、处理、把结果丢 进"出队"。 段与段之间只靠队列和几种类型化消息通信,谁也不直接调用谁。本章讲清这个统一的运行时: 一个段长什么样、怎么被驱动、消息怎么流、怎么优雅地停下来。它是读懂后面 VAD、话轮、LLM、 Realtime 服务器各章的词汇表。
1. 这是什么(零基础也能懂)
一句话定义: 这是一套多段流水线运行时——把"语音进、语音出"这件事切成若干独立工序, 每道工序是一个处理单元(handler),各跑各的线程,靠队列串成一条流水线。
它解决什么问题: 语音对话是流式的——你还在说,系统就得开始转写;大模型还在吐字, 就得开始合成播放。如果把这些步骤写成一个大函数顺序调用,任何一步卡住(比如大模型慢)整条链就冻住了。 把每段拆成独立线程 + 队列,各段就能同时干活、互不阻塞:STT 在转上一句时,VAD 已经在听下一句。
一句话直觉/类比: 想象一条工厂流水线。原料(音频块)从头进,依次经过几个工位, 每个工位只做一件事、做完把半成品放到传送带上(队列)给下一个工位。工人(线程)彼此不说话, 只认传送带上的箱 子(消息)。传送带尽头放一个特殊的"收工箱"(哨兵),谁收到就下班。
流水线长这样(共享链,6 段):
recv 音频 ─▶ [VAD] ─▶ [STT] ─▶ [Notifier] ─▶ [LLM] ─▶ [LMProcessor] ─▶ [TTS] ─▶ 送音频
听说完 转文字 通知/组prompt 吐字 分句/成TTS输入 合成
这条链在 s2s_pipeline.py:380 的 _build_pipeline_handlers docstring 里写得明明白白:
VAD → STT → TranscriptionNotifier → LM → LMOutputProcessor → TTS。本章不讲每段的算法(那是后面各章),
只讲它们共同的骨架。
2. 顶层全景(它大概怎么转)
2.1 三个角色
整个运行时只有三类东西,先记住它们:
| 角色 | 是什么 | 在哪 |
|---|---|---|
| BaseHandler | 一个处 理段的统一基类:一线程、一循环、setup/process/run | baseHandler.py:23 |
| 队列 + 事件 | 段之间传数据的 Queue,和跨段协调用的 Event | s2s_pipeline.py:334 initialize_queues_and_events |
| ThreadManager | 给每个 handler 起一条线程,负责 start/wait/stop | utils/thread_manager.py:9 |
2.2 一张图:段怎么被串起来
怎么读这张图: 每个 [H] 是一个 handler(一条线程),中间的 ║ queue ║ 是它们之间的队列。
handler 只碰自己两侧的队列,别的一概不知。ThreadManager 在旁边给每个 [H] 各起一条线程。
┌──────────────── ThreadManager ────────────────┐
│ 给每个 handler 起一条线程,start/wait/stop │
└───────────────────────────────────────────────┘
每个 [H] = 一条线程,跑 handler.run()
║ 入队 ║ ──▶ [ Handler A ] ──▶ ║ 中间队列 ║ ──▶ [ Handler B ] ──▶ ║ 出队 ║
queue_in run 循环: queue_out run 循环: …
get → process (= B 的 queue_in) get → process
→ put 到 queue_out
关键点:A 的 queue_out 就是 B 的 queue_in。装配代码(_build_pipeline_handlers,
s2s_pipeline.py:351)做的核心事情,就是把上一段的出队接成下一段的入队——例如 VAD 的
queue_out=spoken_prompt_queue,而 STT 的 queue_in=spoken_prompt_queue(s2s_pipeline.py:391、775)。
2.3 主线走一遍(高层)
一次完整对话在运行时层面是这样流动的(不进算法):
- IO 段(mic/socket/websocket)把原始音频块放进
recv_audio_chunks_queue。 - VAD 攒够一段完整语音,吐出一个
VADAudio消息 →spoken_prompt_queue。 - STT 转写,吐出
Transcription→stt_output_queue。 - TranscriptionNotifier 把转写变成 LLM 的输入请求 →
text_prompt_queue。 - LLM 流式生成,一句一句吐
LLMResponseChunk→lm_response_queue。 - LMOutputProcessor 把 chunk 整成
TTSInput(分句、加语言码等)→lm_processed_queue。 - TTS 合成音频块 →
send_audio_chunks_queue,IO 段播出去。
每一跳都是"上一段 put、下一段 get"。§6 会把一条音频块从头到尾贯穿走一遍。
3. 核心机制一:BaseHandler 的契约
这是全章的核心。理解了 BaseHandler,就理解了 6 个段里的 5 个半(IO 段稍有不同,见 §5)。
3.1 它要解决的小问题
我们想让每段处理逻辑只关心"给我一个输入,我产出零到多个输出",而不用操心:线程怎么起、
怎么从队列拿数据、怎么定期检查该不该停、结果怎么放回队列、一个输入卡了会不会拖垮整条链。
BaseHandler 把这些公共的脏活一次写好,子类只填 setup 和 process 两个空。
3.2 子类只需填两个方法
BaseHandler 用 Generic[InT, OutT] 标注"我吃什么、吐什么"(baseHandler.py:23)。子类的全部职责:
setup(*args, **kwargs)—— 构造时调一次,加载模型、读配置(baseHandler.py:48)。 注意它在__init__里就被调用(baseHandler.py:45),所以建 handler 时模型就已加载。process(input) -> Iterator[OutT]—— 核心,一个生成器:吃一个输入,yield零个或多个输出 (baseHandler.py:51)。用生成器而非return,正是为了流式——LLM 可以边生成边一句句yield, 下游 TTS 立刻就能开始合成,不必等整段说完。
教学示意(非源码),感受一个最小 handler:
# 示意,非源码:一个把文本转大写的处理段
class UpperHandler(BaseHandler[str, str]):
def setup(self, suffix="!"): # 构造时调一次
self.suffix = suffix
def process(self, text): # 生成器:吃一个,吐若干
for word in text.split():
yield word.upper() + self.suffix # 每个词单独产出,下游立 刻能拿
重点看:process 是 yield 不是 return——每 yield 一次,run 循环就把它推给下一段一次。
3.3 run 循环:驱动引擎
真正驱动一切的是 run(baseHandler.py:84),它是每条线程执行的函数。骨架如下:
# 简化自 baseHandler.py:84 run()
while not self.stop_event.is_set(): # ① 有人喊停就退出
try:
item = self.queue_in.get(timeout=0.1) # ② 带超时地拿,拿不到就轮询
except Empty:
continue # 空了?回去再看一眼 stop_event
# ③ 先辨认"控制/哨兵"消息(见 3.4)……
for output in self.process(item): # ④ 跑子类生成器
self.queue_out.put(output) # ⑤ 每个产出推给下一段
self.cleanup(); self.queue_out.put(PIPELINE_END) # ⑥ 收尾 + 传哨兵
四个设计要点:
- 超时轮询(
timeout=0.1)。 为什么不直接阻塞等队列?因为要周期性回头看stop_event(baseHandler.py:88-93)。阻塞get()会让线程永远卡在空队列上、无法响应停机; 0.1 秒超时让它每 100ms 醒一次检查是否该退出。 - 异常隔离。
process抛异常会被try/except兜住、记日志、继续循环下一个输入 (baseHandler.py:136-137),不会杀死线程。一句转写崩了,不影响下一句。 - 计时。 每个输出的耗时被记进
self._times(baseHandler.py:126),debug 级别下打印 (baseHandler.py:127-128),用于观测各段延迟。 - 收尾。 循环退出后,
cleanup()(子类可覆写释放资源)再往出队塞一个PIPELINE_END哨兵 (baseHandler.py:139-140)——把"我下班了"顺着传送带告诉下一段。
3.4 三种"非数据"消息:哨兵与控制
队列里流的不全是数据消息,还有两类"信令"。run 循 环在跑 process 之前先把它们辨认出来:
| 消息 | 类型 | 语义 | run 里怎么处理 |
|---|---|---|---|
PIPELINE_END | bytes(b"END") | 硬停:这条链收工 | 收到就 break 退出循环(baseHandler.py:107-110) |
SESSION_END | PipelineControlMessage | 软重置:结束本次会话,但线程继续活 | 调 on_session_end(),再把它透传给下一段(baseHandler.py:95-105) |
其他 PipelineControlMessage | 控制消息 | 没预期到的种类 | 只 warning、丢弃(baseHandler.py:112-114) |
两种哨兵的分工是本章最关键的对比之一:
PIPELINE_END(硬停) 定义在messages.py:168(Final[bytes] = b"END")。它像"关厂"信号: 一段收到就break、cleanup、再把PIPELINE_END传给下家——于是哨兵沿链一路传下去, 每段依次干净退出。它是bytes而非富对象,是为了在类型化队列里当"带内(in-band)"标记简单地比较相等 (queue_types.py:20-22解释了为什么用裸bytes)。docstring(baseHandler.py:27)特别提醒: 停 handler 时既要setstop_event、又要往入队放PIPELINE_END,否则线程可能卡在空队列的超时轮询里。SESSION_END(软重置) 定义在control.py:18,是一个frozen的PipelineControlMessage。 它像"换客人":realtime 模式下一个会话结束、下一个会话要复用同一条线程时,用它清掉每会话状态 (on_session_end(),baseHandler.py:160-161默认空实现,子类如qwen3_tts_handler.py:643会响应), 但不停线程。处理完它会被透传到queue_out(baseHandler.py:104),这样重置信号也能沿链走一遍。 会话池怎么用它来"回收/复用"一条流水线,是 05 章 的内容。
3.5 四个钩子:在 process 前后插手
除了 process,run 循环还在几个位置留了可覆写的钩子,让子类微调"要不要处理/要不要发/发什么":
| 钩子 | 触发时机 | 默认行为 | 用途 |
|---|---|---|---|
should_process_input(item) | 拿到输入、跑 process 前 | 若 cancel_scope 判定该输入已过期则丢弃 | 打断:丢弃属于已取消那一轮的输入 |
should_emit_output(output) | 每个产出、放队列前 | 恒 True | 过滤某些产出不外发 |
before_emit_output(output) | 确定要发之后 | 空 | 发前副作用(埋点等) |
output_for_queue(output, src) | 真正 put 前 | 若源输入带 cancel_generation 且产出是音频,包成 AudioOutput | 给音频盖上"哪一轮生成"的戳 |
should_process_input(baseHandler.py:54)是全双工打断的关键一环。 它检查该输入携带的
cancel_generation 号,拿去问 cancel_scope.is_stale(...)(baseHandler.py:61):如果这轮已被用户打断,
就直接跳过、不处理(baseHandler.py:117-118)。EndOfResponse 是例外(始终放行,好让"结束"信号收尾)。
cancel_scope 是构造时通过 setup_kwargs 注进各 handler 的(见 s2s_pipeline.py:514),
handler 用 getattr(self, "cancel_scope", None) 软取,没有就不打断。打断的完整机制留给 03 章。
output_for_queue(baseHandler.py:77)做一件小而关键的事:如果源输入带 cancel_generation
且产出是音频(bytes/ndarray),就把它裹进 AudioOutput(audio=..., cancel_generation=...)
(baseHandler.py:79-82)。这样音频块自带"我属于第几轮回答"的标签,下游播放端能凭它丢弃过期音频——
同样服务打断。
4. 核心机制二:类型化消息与队列
第二个要建立的直觉:段之间不是传裸元组,而是传带类型的消息对象。这让每段"该拿到什么、该吐什么"
在类型上是清楚的,也便于携带 turn_id、cancel_generation 等元数据。
4.1 消息:PipelineMessage 家族
所有跨段数据消息都继承 PipelineMessage(一个 pydantic BaseModel,messages.py:24),
用 tag 字段做判别器(discriminator),方便从原始 dict 校验成具体子类。messages.py 顶部注释
(messages.py:1-7)明说:这些取代了从前在段之间乱飞的临时元组。
按"从哪流到哪"列出主要消息(注释里的分段箭头就是设计者标的数据流):
| 消息 | tag | 从 → 到 | 关键字段 | 定义 |
|---|---|---|---|---|
VADAudio | vad_audio | VAD → STT | audio、mode(progressive/final)、turn_id | messages.py:39 |
Transcription | transcription | STT → Notifier | text、language_code、speech_stopped_at_s | messages.py:62 |
LLMResponseChunk | llm_response_chunk | LLM → LMProcessor | text、tools、runtime_config、cancel_generation | messages.py:76 |
TTSInput | tts_input | LMProcessor → TTS | text、language_code、cancel_generation | messages.py:120 |
EndOfResponse | end_of_response | 标记一次回答结束的 哨兵消息 | turn_id、cancel_generation、error | messages.py:101 |
GenerateResponseRequest | generate_response | Realtime 服务 → LLM | runtime_config、response、turn_id | messages.py:145 |
两个消息值得单独点一句:
EndOfResponse(messages.py:101) 是一个"结束标记"消息(不是bytes哨兵,是富对象): 它顺着LLM → LMProcessor → TTS流,告诉下游"这一轮回答说完了"。它带error字段:生成没起来时置上, 下游据此产出response.done(status="failed")却仍正常收尾(messages.py:104-108)。前面提过,should_process_input对它网开一面,保证结束信号不被打断丢掉。GenerateResponseRequest(messages.py:145) 是触发 LLM 生成的请求。它的设计哲学写在 docstring(messages.py:146-154):携带 LM 需要的一切(runtime_config含本连接会话配置 + 会话 chat), 这样 LM handler 不必回头去摸共享对象——纯消息驱动。这条在多会话并发(会话池)下尤其重要,是 05 章 的伏笔。
4.2 两个 bytes 哨兵
不是所有信令都值得做成富对象。两个带内哨兵用裸 bytes 常量(messages.py:167-168):
PIPELINE_END = b"END"—— §3.4 讲过的硬停哨兵。AUDIO_RESPONSE_DONE = b"__RESPONSE_DONE__"—— 音频队列上的"这段回答的音频放完了"标记, TTSyield它(如kokoro_handler.py:248)、IO 发送端识别它来切分回答边界 (如socket_sender.py:53)。本章不展开音频侧协议。
4.3 队列类型别名:给每根传送带命名
queue_types.py 不定义新逻辑,只给每根队列起类型名,让代码里 Queue[STTOutItem] 一眼看出流的是什么
(queue_types.py:1-5 的注释)。每个别名都是"该段业务消息 | 内部信令"的联合:
| 队列别名 | 装什么 | 定义 |
|---|---|---|
AudioInItem | 进 VAD 的音频(VADIn + 控制) | queue_types.py:25 |
VADOutItem | VAD → STT(VADOut + 内部项) | queue_types.py:28 |
STTOutItem | STT → Notifier | queue_types.py:31 |
TextPromptItem | → LLM | queue_types.py:34 |
LMOutItem | LLM → LMProcessor | queue_types.py:37 |
TTSInItem | → TTS | queue_types.py:40 |
AudioOutItem | → 扬声器/客户端(含 bytes 哨兵) | queue_types.py:43 |
TextEventItem | 侧信道文本事件(给 websocket) | queue_types.py:46 |
其中 PipelineInternalItem = PipelineControlMessage | bytes(queue_types.py:22)就是"任何队列里都可能夹带的
控制消息或哨兵"——注释还解释了为何哨兵用裸 bytes:Queue 是**不变(invariant)**的,Literal[b"END"]
不能塞进要求 bytes 的位置(queue_types.py:20-21)。
4.4 侧信道事件:PipelineEvent
除了主链上的数据消息,还有一条侧信道 text_output_queue,流的是 PipelineEvent
(events.py:17)——如 SpeechStartedEvent、PartialTranscriptionEvent、AssistantTextEvent
(events.py:30/50/69)。这些是给 realtime websocket 客户端看的协议事件(转写进度、助手文本等),
不影响音频主链。它们的用途属于 05 章;这里只需知道:主链走音频,侧信道走给前端的文本事件。
5. 核心机制三:线程模型与优雅关闭
第三个直觉:一 handler 一线程,ThreadManager 管生死。ThreadManager 全文只有 40 行
(utils/thread_manager.py:9),简单到可以整个讲透。
5.1 一 handler 一线程
ThreadManager 拿一串 handler,start() 时给每个 handler 起一条线程,线程体就是 handler.run
(thread_manager.py:18-23):
# thread_manager.py:18 start()
for handler in self.handlers:
thread = threading.Thread(target=handler.run)
thread.daemon = False # 非守护:关闭时必须等它
self.threads.append(thread)
thread.start()
daemon = False 是刻意的(thread_manager.py:21 注释:"Ensure threads are waited for on shutdown"):
非守护线程在主程序退出时不会被强杀,主线程会等它们收尾,保证 cleanup() 跑到、资源释放干净。
5.2 三个方法:start / wait / stop
| 方法 | 干什么 | 代码 |
|---|---|---|
start() | 每个 handler 起一条非守护线程并启动 | thread_manager.py:18 |
wait() | join 所有线程,阻塞直到全部结束 | thread_manager.py:25 |
stop() | 优雅关闭(见下) | thread_manager.py:29 |
stop() 的优雅关闭分两步(thread_manager.py:29-39):
- 广播停机意图: 给每个 handler
stop_event.set()——所有run循环下次轮询到stop_event.is_set()就会退出(回看baseHandler.py:88)。 - 限时等待: 逐个
thread.join(timeout=5.0);5 秒还没停的线程,只打一条 warning 记下"没在超时内退出"(thread_manager.py:36-39),不硬杀。
注意一个真实的坑: 光 set stop_event 不一定能让每段都醒——如果某段正阻塞在
queue_in.get(timeout=0.1) 的轮询上还好(0.1 秒会醒),但若某段在等一个永远不来的输入,
就需要配合 §3.4 说的 PIPELINE_END。baseHandler.py:27 的 docstring 正是为此提醒:
"要停 handler,set stop_event 并往入队放 PIPELINE_END 以避免队列死锁"。
5.3 IO 段:同一契约的变体
流水线两端的 IO 段(mic、socket、websocket)也被 ThreadManager 当普通 handler 起线程,但它们不都继承
BaseHandler 的标准 process 循环——比如 LocalAudioStreamer 是独立实现,SocketReceiver 在结束时也主动
往出队放 PIPELINE_END(connections/socket_receiver.py:68)以触发下游收尾。装配时它们被拼在业务段前后:
ThreadManager([*comms_handlers, *pipeline_handlers])(s2s_pipeline.py:747)。要点:IO 段是"把外部世界的
字节接进/送出队列"的适配器,但仍服从同一套"线程 + 队列 + 哨兵"的运行时契约。
6. 装配主线:s2s_pipeline.py 怎么把这一切搭起来
前面讲的是"零件",这节讲"组装"。入口是 main()(s2s_pipeline.py:984),它的流程是一条清晰的直线。
6.1 main 的五步
parse_arguments() ─▶ prepare_all_args() ─▶ initialize_queues_and_events()
① ② ③
─▶ build_pipeline(...) ─▶ 装信号处理 ─▶ manager.start() + wait()
④ ⑤
- ①
parse_arguments(s2s_pipeline.py:129)——带一个"LM 后端预解析"的巧思。 多个 LM 参数类(responses-api / chat-completions / transformers…)共享同一个基类,字段名会重复, 一股脑注册给HfArgumentParser会冲突。所以它先偷看一眼--llm_backend(JSON 配置就读文件, 命令行就用一个只认--llm_backend的临时 parser,s2s_pipeline.py:138-145),据此只注册那一个 LM 参数类 (_lm_class,s2s_pipeline.py:147-162),避开重复字段。解析完用{type: instance}字典按类型取回各段配置 (s2s_pipeline.py:177),避免依赖返回顺序。 - ②
prepare_all_args(s2s_pipeline.py:289) 统一 Mac 优化、设备覆盖、参数改名(rename_args)。 - ③
initialize_queues_and_events(s2s_pipeline.py:334) 一次性造出所有队列和跨段事件:stop_event、should_listen、response_playing、cancel_scope,以及 §4.3 那一排队列。它们是各段之间的"传送带"和"信号灯"。 - ④
build_pipeline(s2s_pipeline.py:581) 按mode选 IO 段并组链(下节)。 - ⑤ 信号处理 + 起停: 注册
SIGINT/SIGTERM到signal_handler(s2s_pipeline.py:1056-1064), 该 handler 调pipeline_manager.stop()做优雅关闭;然后start()起所有线程、wait()阻塞等结束 (s2s_pipeline.py:1067-1068)。KeyboardInterrupt也兜底走一遍 stop(s2s_pipeline.py:1069-1073)。
6.2 组链:_build_pipeline_handlers
真正"把出队接成下一段入队"的地方是 _build_pipeline_handlers(s2s_pipeline.py:351)。它按固定顺序
new 出 6 个 handler,共享的队列在参数里显式传入,于是链就自然接上了。摘几处看接法(注意队列同名即相连):
| handler | queue_in | queue_out | 代码 |
|---|---|---|---|
VADHandler | recv_audio_chunks_queue | spoken_prompt_queue | s2s_pipeline.py:388 |
STT(get_stt_handler) | spoken_prompt_queue | stt_output_queue | s2s_pipeline.py:403、775 |
TranscriptionNotifier | stt_output_queue | text_prompt_queue | s2s_pipeline.py:396 |
LLM(get_llm_handler) | text_prompt_queue | lm_response_queue | s2s_pipeline.py:416、863 |
LMOutputProcessor | lm_response_queue | lm_processed_queue | s2s_pipeline.py:425 |
TTS(get_tts_handler) | lm_processed_queue | send_audio_chunks_queue | s2s_pipeline.py:432、927 |
get_stt_handler / get_llm_handler / get_tts_handler(s2s_pipeline.py:750/850/907)是工厂:
按 module_kwargs.stt/llm_backend/tts 选具体实现类,但都遵循同一 BaseHandler 契约、接同一对队列。
这就是本章反复强调的统一性——换 STT/LLM/TTS 的具体模型,运行时骨架一行都不用改。
6.3 两种装配路径
build_pipeline(s2s_pipeline.py:581)按 mode 分叉,但共享链的组装函数是同一个:
- 非 realtime(local / websocket / socket): 造好对应 IO 段,注入一个
RuntimeConfig(s2s_pipeline.py:707-717),调_build_pipeline_handlers组出共享链,最后ThreadManager([*comms_handlers, *pipeline_handlers])(s2s_pipeline.py:747)。 - realtime: 造一个流水线池——对
num_pipelines循环调_build_realtime_pipeline_unit(s2s_pipeline.py:642-663),每个 unit 有自己独立的队列、事件、cancel_scope、speculative_turns(s2s_pipeline.py:490-501,用deepcopy隔离配置,s2s_pipeline.py:476-488)。再把一个RealtimeServer和所有 unit 的 handler 一起交给ThreadManager(s2s_pipeline.py:665-675)。会话池 = 多条一模一样的独立流水线, 由服务器按连接分派。细节全在 05 章。
6.4 贯穿路径:一条音频块的一生
把前面所有概念串起来,追一个 16ms 的音频块从入队到产出:
① IO 段(mic/socket)读到一块 bytes,put 进 recv_audio_chunks_queue
│
② VAD 线程 run 循环 get 到它 → process 累积;判定"说完了"才 yield 一个 VADAudio
│ (没说完就不 yield,块被吞进内部缓冲)
▼ spoken_prompt_queue
③ STT 线程 get VADAudio → process 转写 → yield Transcription(text=...)
│
▼ stt_output_queue
④ TranscriptionNotifier get Transcription → 组装 → yield GenerateResponseRequest
│ (顺便把 PartialTranscription/事件发去侧信道 text_output_queue)
▼ text_prompt_queue
⑤ LLM 线程 get 请求 → 流式生成 → 逐句 yield LLMResponseChunk,末尾 yield EndOfResponse
│
▼ lm_response_queue
⑥ LMOutputProcessor get chunk → 分句/加语言码 → yield TTSInput
│
▼ lm_processed_queue
⑦ TTS 线程 get TTSInput → 合成 → yield 音频(bytes/ndarray)
│ output_for_queue 若带 cancel_generation 则包成 AudioOutput
▼ send_audio_chunks_queue
⑧ IO 发送段 get 音频 → 播放/回传;回答结束时的 AUDIO_RESPONSE_DONE 标记边界
贯穿这条路要记住的三件事:
- 每一跳都是
put/get,没有任何段直接调用另一段——完全消息驱动。 process是生成器,所以第 ⑤ 步 LLM 的每一句 chunk 一yield就流向 TTS,实现流式、边说边合成。- 打断时,
cancel_scope让 ②–⑦ 各段的should_process_input把"过期那一轮"的消息沿途丢弃,AudioOutput上的cancel_generation戳让播放端也丢过期音频——这条打断暗线贯穿全链,展开在 03 章。
7. 巧妙之处(可借鉴的技术)
- 一个基类统一异构工序。 VAD(信号处理)、STT(ASR)、LLM(生成)、TTS(合成)算法天差地别,
却共用同一个
BaseHandler[InT, OutT]契约(baseHandler.py:23)。好处:运行时代码只写一遍, 换模型不动骨架;新增一段只需实现setup/process。 process用生成器换来流式。 不是"处理完返回结果"而是"边处理边yield"(baseHandler.py:51/122), 这是整个系统"能边听边说、低延迟"的运行时基础。- 两级哨兵,粗细分明。 富对象消息(
EndOfResponse)负责业务语义级的"一轮结束";裸bytes哨兵(PIPELINE_END/AUDIO_RESPONSE_DONE)负责运行时级的"收工/音频段结束"。 软的SESSION_END重置会话不停线程,硬的PIPELINE_END停线程——同一根传送带上分层传递不同粒度的信令。 - 超时轮询而非阻塞。
get(timeout=0.1)(baseHandler.py:91)让"响应停机"和"等数据"两件事共存, 是简单可靠的可停机循环范式。 - LM 后端预解析绕过参数冲突。 共享基类导致字段重名时,先窥探
--llm_backend只注册需要的那个类 (s2s_pipeline.py:132-162)——一个务实的 argparse 技巧。 - 消息自带上下文,handler 不摸全局。
GenerateResponseRequest把 LM 所需一切打包随消息走 (messages.py:146-154),使同一段代码能被多条并发流水线安全复用——为会话池铺路。
8. 边界与局限(诚实)
- 段间强顺序、线性链。 运行时是一条线性流水线(6 段固定顺序,
s2s_pipeline.py:380); 没有 DAG / 分支路由的通用调度。要加旁路(如侧信道事件)得单独拉队列(text_output_queue)。 - 无背压(back-pressure)。 用的是无界
Queue。若某段持续比上游慢,队列会一直涨—— 代码里对text_output_queue尤其警惕:非 websocket/realtime 模式故意置 None 避免无界增长 (s2s_pipeline.py:610-612)。这是设计取舍,不是 bug。 - 停机靠约定,不完全自动。 单
setstop_event 未必唤醒等输入的线程,需配合PIPELINE_END(baseHandler.py:27);stop()的join超时 5 秒后只 warning、不强杀(thread_manager.py:37-39), 卡死的线程会被留下(非守护)。 - 每段一条 OS 线程。 realtime 池
num_pipelines条流水线 × 每条 6+ handler,线程数随并发线性增长; Python GIL 下 CPU 密集段(STT/LLM/TTS 多在 C/GPU 里放锁)才真正并行。Apple Silicon 上 MLX 还有全局锁, 多流水线时会主动关掉进度式转写以免争锁刷屏(s2s_pipeline.py:1017-1028)。
9. 代码地图(导航索引)
| 主题 | 文件 | 符号 |
|---|---|---|
| handler 契约:setup/process/run 循环 | src/speech_to_speech/baseHandler.py | BaseHandler、run、process、setup |
| 打断钩子:丢弃过期输入 | src/speech_to_speech/baseHandler.py | should_process_input、output_for_queue、should_emit_output |
| 会话软重置 | src/speech_to_speech/baseHandler.py | on_session_end |
| 线程管理与优雅关闭 | src/speech_to_speech/utils/thread_manager.py | ThreadManager、start、wait、stop |
| 类型化消息定义 | src/speech_to_speech/pipeline/messages.py | PipelineMessage、VADAudio、Transcription、LLMResponseChunk、TTSInput、EndOfResponse、GenerateResponseRequest |
| bytes 哨兵 | src/speech_to_speech/pipeline/messages.py | PIPELINE_END、AUDIO_RESPONSE_DONE |
| 控制消息与软哨兵 | src/speech_to_speech/pipeline/control.py | PipelineControlMessage、SESSION_END、is_control_message |
| 队列类型别名 | src/speech_to_speech/pipeline/queue_types.py | PipelineInternalItem、VADOutItem、STTOutItem、TTSInItem、AudioOutItem |
| handler 泛型别名 | src/speech_to_speech/pipeline/handler_types.py | STTIn、STTOut、LLMIn、LLMOut、TTSIn、TTSOut |
| 侧信道事件 | src/speech_to_speech/pipeline/events.py | PipelineEvent、SpeechStartedEvent、AssistantTextEvent |
| 装配:参数解析(含 LM 预解析) | src/speech_to_speech/s2s_pipeline.py | parse_arguments、ParsedArguments |
| 装配:队列/事件初始化 | src/speech_to_speech/s2s_pipeline.py | initialize_queues_and_events |
| 装配:组共享链 | src/speech_to_speech/s2s_pipeline.py | _build_pipeline_handlers、get_stt_handler、get_llm_handler、get_tts_handler |
| 装配:realtime 池与总入口 | src/speech_to_speech/s2s_pipeline.py | build_pipeline、_build_realtime_pipeline_unit、main |
| 每流水线日志前缀 | src/speech_to_speech/pipeline/log_context.py | pipeline_log_ctx、PipelineLogFilter |
下一步该读哪章: 想知道 VAD 怎么判"说完了"→ 02; 打断与边说边改 → 03;LLM 流式分句/工具调用 → 04; realtime 会话池与协议 → 05。全局视角见 index。