Skip to main content

02 - 流水线与阶段

上一篇列的第一个问题是「请求现在跑到哪一步了，没人知道全局」。这一篇解它。

三十个请求同时在跑，四个进程里散落着它们的碎片。这时候几个很朴素的问题都变得不好回答：

问题难在哪
用户挂了电话，第 17 号请求要撤销，去哪儿撤它可能在四个进程的任意一个里，也可能同时在两个里（前半句在说、后半句在想）
最后一块音频吐完了，谁告诉发起方「结束了」吐音频的那个阶段并不知道文字那边完没完
omni 有文字和声音两个出口，两个都完才算完谁来等这个「都」

最容易想到的写法是让每个阶段各存一份「我手上有哪些请求」再互相广播。三个阶段还行，到五个阶段加上流式边就会失控 —— 状态散在五处，任意两处不一致就是一个 bug。

SGLang-Omni 把这份账集中到一处，叫协调器（Coordinator）；其余阶段退化成纯粹的收发员，叫阶段（Stage）。

一、先看整条流水线长什么样​

文字描述流水线很容易越描越糊。先看一眼 Qwen3-Omni 的语音路径实际被拆成了什么 —— 八个阶段、三个进程、两张卡、两个出口：

粗实线是数据面（张量在阶段之间搬），虚线是控制面（协调器和阶段之间的协调消息）。两条线走的通道完全不同，04 篇专讲这个。

三件事从图上能直接看出来，文字讲半天也不如看一眼：

  • 协调器不在数据通路上。张量从来不经过它 —— 它只发指令、收完成事件。所以它不会成为带宽瓶颈。
  • 两个终点。decode 出文本、code2wav 出音频，各自独立向协调器报完成，协调器等两个都到齐才算这个请求结束。
  • 流式边是额外的一条。thinker → talker_ar 那条标了「流式隐状态」的边，跟 thinker → decode 那条普通边同时存在，不是二选一。thinker 一边往 decode 送 token、一边往 talker 送隐状态。

下面这张把同一条链路按时间摊开，看消息实际的往返顺序：

两处容易忽略的：

  • 文本比音频先到。看第 5 步和第 10 步的间隔 —— thinker 一开口文本就往外流了，而音频要等 talker 攒够一块码本、再等声码器解完才出得来。omni 产品普遍带字幕，是这条链路的自然产物，不是产品设计选择。
  • 中止是广播，不是沿链传递。用户挂电话时，协调器同时给图上每个阶段发 AbortMessage，不是从 preprocessing 一级级往下传。因为这个请求的碎片同时散在好几个阶段里，沿链传会漏掉已经跑到后面的那些。

二、协调器持有全局状态​

阶段是无状态的流水线环节，谁来记住「请求 abc 现在跑到哪了」？协调器。它的职责清单很短：

  • 把新请求投给入口阶段；
  • 跟踪请求状态：待处理、运行中、已完成、失败、已中止；
  • 收集终点阶段的完成事件；
  • 流水线有多个终点时合并结果（Qwen3-Omni 的 decode 和 code2wav）；
  • 向所有阶段广播中止。

一条重要的性质：协调器对阶段的实现完全无感知。张量并行的阶段组里，它只跟 rank 0 说话，其余 rank 是阶段组内部的事。

三、阶段是一个不做计算的 IO 外壳​

协调器管全局，那具体干活的那一层呢？

最自然的设计是让每个阶段自己收发消息、自己判断下一步给谁、自己处理上游没到齐的情况。写三个阶段还行，写到八个就会发现每个阶段里都有一坨几乎一样的收发代码，而且各自都写得略有不同。

SGLang-Omni 的做法是把这坨东西整个抽出来，叫 Stage。它不做任何计算 —— 存在的唯一意义是把「阶段之间」的复杂度全吸收掉，让真正干活的调度器只需要关心「拿到一份输入、算出一份输出」。

入口SubmitMessageDataReadyMessagestream chunk / doneAbortMessage 广播前三条走 ZMQ PUSH/PULL，中止走 PUB/SUBSTAGE 的四项职责全部与模型无关UNPACK解包载荷从 relay 取张量缓冲，按布局表还原成完整的 StagePayloadFAN-IN扇入聚合按 wait_for 攒齐多个上游，再调 merge_fn 合成一份输入ROUTE路由决策按 next 或 route_fn 决定下游，终点阶段则直接回协调器STREAM流路由按 stream_to 把增量送出去，同卡走 CUDA IPC，跨卡走 relay与调度器的全部接触面scheduler.inboxQueue of IncomingMessagetype 取 new_request /stream_chunk / stream_donescheduler.outboxQueue of OutgoingMessagetype 取 result / stream / error出口DataReadyMessage → 下游阶段StreamMessage → 流式下游CompleteMessage 或错误 → 协调器整层没有一处 if 判断调度器类型 —— 三种调度器呈现的是同一个「两个队列 ＋ 四个方法」的接口。这条约束的实际收益在 13 篇：接一个新模型时你写的是调度器和运行器，完全不用碰 Stage，也就不可能把它写坏。代价是调试多一跳 —— 一条消息从上游发出到下游调度器拿到，中间经过解包、扇入、入队三步，出问题要能分清卡在哪一步。框架为此内建了请求级阶段事件剖析：阶段内的排队/构建/prefill/decode 分解，加上阶段之间的跳转延迟。
把「扇入」放进 Stage 而不是调度器，是这套设计里一个容易被低估的决定：扇入意味着要缓存半份输入、要处理某一路失败、要处理中止时清理已到的部分。这些全是 IO 语义，跟算什么完全无关。

四、声明式的拓扑​

第一节那张八阶段的图，是怎么告诉框架的？

不是写代码连线，而是列一张清单 —— 每个阶段一条，说清楚它叫什么、跑什么、下一步给谁。入口阶段、终点集合、GPU 放置、传输方式全部从这张清单推导出来，没有第二处定义。

stages = [
    StageConfig(name="preprocessing",
                factory="...create_preprocessing_executor",
                route_fn="...routing.preprocessing_next"),

    StageConfig(name="image_encoder",
                factory="...create_image_encoder_executor",
                gpu=0, next="mm_aggregate"),

    StageConfig(name="mm_aggregate",
                # 扇入：三个上游都到齐才往下走，merge_fn 决定怎么合
                wait_for=["preprocessing", "image_encoder", "audio_encoder"],
                merge_fn="...merge_for_thinker",
                next="thinker"),

    StageConfig(name="thinker",
                factory="...create_thinker_executor",
                gpu=0,
                next=["decode", "talker_ar"],       # 普通路由：整份结果送下游
                stream_to=["talker_ar", "decode"]), # 流式边：与上一行并行，边生成边送

    StageConfig(name="decode", terminal=True),          # 终点之一：文本
    StageConfig(name="code2wav", gpu=1, terminal=True), # 终点之二：音频
]

字段与它们各自回答的问题：

字段回答什么问题硬约束
next下一步去哪（静态）与 route_fn、terminal 恰好三选一，都不写或写多个都是配置错误
route_fn下一步去哪（由数据决定）返回值必须是 next 里声明过的阶段；禁止返回 None
terminal这是终点，结果回协调器一条流水线可以有多个终点
wait_for等哪些上游到齐才开工必须和 merge_fn 成对出现
merge_fn到齐之后怎么合成一份输入同上，只写一个是配置错误
gpu跑在哪张卡留空表示这是 CPU 阶段；写成列表即张量并行
tp_size张量并行度必须与 gpu 列表长度一致
process跑在哪个进程同名即共置。进程拓扑的唯一真相来源
stream_to边生成边送给谁与 next 并行存在，不是替代关系
relay传输配置几乎不用手写，默认从 gpu 推导

最后两行值得注意：走哪条传输通路不是配置项，它完全由阶段的局部性和放置推导出来，配置只能调槽位大小之类的旋钮。把「走哪条路」从配置面拿掉，是为了让拓扑成为唯一真相来源 —— 否则拓扑和传输配置会各说各话。

五、动态路由的收窄契约​

动态路由看起来是个通用能力，但它的契约被卡得很死：

  • 返回值必须是已经在 next 里声明过的阶段，这样拓扑仍然静态可推导；
  • 返回 None 被禁止 —— 丢弃要走显式的终点汇聚，不能藏在路由函数里；
  • 文档字符串限定它只用于一种场景：隐状态本身带模态标签、下游分支必须由数据决定（Ming 的输出要在视频头和音频头之间二选一）。

Qwen3-Omni 和 Fish S2-Pro 用 next 加 stream_to 就够了，一个 route_fn 都不需要。一个字段、窄契约、等真有第二个使用者时再放宽 —— 这比先做成通用能力再去约束使用方式要靠得住。

还有一条明确排除的拓扑：没有「一个 thinker 扇出到多个 talker」。理由是 08 篇说过的 —— talker 的解码依赖 thinker 隐状态作为前缀，两个独立的 talker 共用一个前缀没有语义。这不是「暂不支持」，是「没有意义所以不做」。

六、扇出时的载荷投影​

next 写成列表就是扇出，但一份载荷投给两个下游有个陷阱：它们会共享同一个 Python 对象。同进程直传时尤其危险，一个下游改了字段，另一个下游看到的就变了。

框架的规则是：扇出只在「每个下游拿到的都是各自独立 data 容器的 StagePayload」时才允许走同进程直传。张量叶子可以刻意共享，但必须当只读用。

投影还有个性能用途。Qwen3-Omni 在 thinker → decode 和 talker_ar → code2wav 两条边上做了载荷投影，把下游用不到的张量裁掉再发 —— 少搬一份就少一份传输开销。配合同进程直传，纯文本路径的平均延迟降了 33%。

顺带一条容易漏的资源规矩：请求对象该留的参数要留，但不该把消费完的大块媒体一路带到后面的阶段。把原始媒体变成流水线状态的那个阶段负责释放这些引用 —— 否则一段 30 秒的音频会跟着请求一路传到声码器。

下一篇：03 - 三种调度器与模型运行器。