# PR #33843 完整报告

- 仓库：`sgl-project/sglang`
- 标题：[diffusion] consolidate pipeline core hygiene
- 合并时间：2026-08-06 21:51
- 原文链接：http://prhub.com.cn/sgl-project/sglang/pull/33843

---

# 执行摘要

- 一句话：集中扩散 pipeline core 重复逻辑，清理未用 logger 与冗余字段
- 推荐动作：值得精读。它展示了大型代码库中“去重 + 集中公共逻辑”的安全重构路径：先抽出共享函数（`calculate_linear_shift`），再逐个删除本地副本，最后处理字段声明与 import。重点关注两点：一是 `denoising_dmd.py` 删除本地覆盖后如何保证与共享基类行为一致；二是 `ServerArgs` 字段合并时对 dataclass 序列化顺序的刻意保护。若后续接手扩散模块，建议先确认 Extra CI 失败原因，并考虑为 `calculate_linear_shift` 补充单元测试。

# 功能与动机

PR body 明确指出："Native pipelines, DMD, and shared stages contained duplicated or unused implementation detail that made behavior drift easier and ownership less clear." 即各 native pipeline（FLUX、Qwen-Image、Z-Image）各自维护了一份几乎相同的 `calculate_shift` 实现，DMD 又在 `denoising_dmd.py` 中重复实现了 `_select_and_manage_model` 与 `_handle_boundary_ratio`，这些重复细节使行为容易漂移、归属不清。目标是把公共调度与去噪行为保留在共享 core 中，同时保留 pipeline 特定配置（如 Qwen-Image 的 `max_seq_len=8192, max_shift=0.9`）。

# 实现拆解

1. **集中 shift 计算**：在 `pipelines_core/diffusion_scheduler_utils.py` 新增 `calculate_linear_shift(image_seq_len, *, base_seq_len=256, max_seq_len=4096, base_shift=0.5, max_shift=1.15)`，将原先 `flux.py`、`qwen_image.py`、`zimage_pipeline.py` 中各自重复的 `calculate_shift` 线性插值实现统一起来。三个 pipeline 的 `prepare_mu` 改为直接调用共享函数，其中 `qwen_image.py` 通过关键字参数覆盖 `max_seq_len=8192`、`max_shift=0.9`，其余使用默认值，行为与原先硬编码一致。

2. **清理未使用 logger**：从 `hunyuan_pipeline.py`、`flux_2.py`、`glm_image.py`、`helios_pipeline.py`、`krea2.py`、`sana.py`、`stable_diffusion_3.py` 及多个 wan 系列 pipeline（`wan_pipeline.py`、`wan_dmd_pipeline.py`、`wan_i2v_pipeline.py`、`wan_i2v_dmd_pipeline.py`、`wan_causal_dmd_pipeline.py`）中删除 `init_logger(__name__)` 与 `logger` 变量——这些 pipeline 实际不使用该 logger，移除后减少无意义的模块级状态。

3. **复用共用 DMD 去噪钩子**：`pipelines_core/stages/denoising_dmd.py` 删除对 `_select_and_manage_model` 与 `_handle_boundary_ratio` 的本地覆盖定义（这两个方法已由共享基类提供），`forward` 中调用处改为 `current_model, _ = ...`（忽略返回的 guidance scale，原值也未被使用）；同时将 `prepared_vars["key"]` 的 dict 访问改为 `prepared_vars.key` 属性访问，与之配套的 `latent_preparation.py` 也做了 1 行调整。

4. **精简 `ServerArgs` 字段声明**：`server_args.py` 合并了重复的 `quantization: str | None = None` 字段声明（原先一处注释为“Quantization method for online quantization”，另一处为“Explicit quantization method override”，语义一致），删除注释掉的 `dmd_denoising_steps` 占位字段。合并后 `quantization` 仍保留在原第一处声明的精确位置，避免破坏 dataclass 的 positional/ 序列化排序。

5. **配套杂项清理**：`models/dits/zimage.py` 删除 2 行无用代码；多个 pipeline 的 import 关系同步调整，删除了不再需要的 `init_logger` 与 `logging_utils` 导入。

测试方面：本 PR 未新增或修改测试文件，PR body 说明按仓库 SGLang-Diffusion 开发指南，重构型变更未在本地运行验证。CI 状态显示 PR Test（Base）通过，但 PR Test（Extra）标记为失败，需要关注。

关键文件：
- `python/sglang/multimodal_gen/runtime/pipelines_core/diffusion_scheduler_utils.py`（模块 调度工具；类别 source；类型 core-logic；符号 calculate_linear_shift）: 新增共享函数 `calculate_linear_shift`，作为所有 native pipeline 动态 shift 计算的唯一真源，是本次重构的核心落点。
- `python/sglang/multimodal_gen/runtime/pipelines_core/stages/denoising_dmd.py`（模块 去噪阶段；类别 source；类型 core-logic；符号 _select_and_manage_model, _handle_boundary_ratio）: 删除 DMD 对 `_select_and_manage_model`、`_handle_boundary_ratio` 的本地覆盖，复用共享去噪钩子；同时将 `prepared_vars` 改为属性访问，是行为风险最集中的文件。
- `python/sglang/multimodal_gen/runtime/pipelines/flux.py`（模块 流水线；类别 source；类型 core-logic；符号 calculate_shift, prepare_mu）: 作为 native pipeline 代表，删除本地 `calculate_shift` 与未用 logger，`prepare_mu` 改为调用共享 `calculate_linear_shift`，验证共享函数接入方式。
- `python/sglang/multimodal_gen/runtime/pipelines/qwen_image.py`（模块 流水线；类别 source；类型 core-logic；符号 calculate_shift, prepare_mu）: 展示了共享函数如何通过关键字参数保持 pipeline 特定配置（max_seq_len=8192、max_shift=0.9），是去重时保留差异性的典型样例。
- `python/sglang/multimodal_gen/runtime/server_args/server_args.py`（模块 参数配置；类别 source；类型 core-logic；符号 quantization）: 合并重复的 quantization 字段、删除注释占位，刻意保持 dataclass 字段原位置以维持序列化顺序，是重构中兼容性考量的关键点。

关键符号：calculate_linear_shift, prepare_mu, _select_and_manage_model, _handle_boundary_ratio

## 关键源码片段

### `python/sglang/multimodal_gen/runtime/pipelines_core/diffusion_scheduler_utils.py`

新增共享函数 `calculate_linear_shift`，作为所有 native pipeline 动态 shift 计算的唯一真源，是本次重构的核心落点。

```python
# SPDX-License-Identifier: Apache-2.0

from __future__ import annotations

from copy import deepcopy
from typing import Any

import torch

from sglang.multimodal_gen.runtime.pipelines_core.schedule_batch import Req
from sglang.multimodal_gen.runtime.platforms import current_platform


# native flow scheduler 根据图像序列长度做仿射插值，得到动态 shift 值 mu。
# 该函数是此前分散在 flux.py / qwen_image.py / zimage_pipeline.py 中
# 三份 calculate_shift 的共享版本，参数默认值对应 FLUX 训练配置。
def calculate_linear_shift(
    image_seq_len: int,
    *,
    base_seq_len: int = 256,
    max_seq_len: int = 4096,
    base_shift: float = 0.5,
    max_shift: float = 1.15,
) -> float:
    """return the affine dynamic shift used by native flow schedulers"""
    slope = (max_shift - base_shift) / (max_seq_len - base_seq_len)
    # 等价写法：base_shift + slope * (image_seq_len - base_seq_len)
    return image_seq_len * slope + base_shift - slope * base_seq_len


def clone_scheduler_runtime(scheduler: Any) -> Any:
    """Create an isolated scheduler runtime from a scheduler template or runtime."""
    return deepcopy(scheduler)

```

### `python/sglang/multimodal_gen/runtime/pipelines_core/stages/denoising_dmd.py`

删除 DMD 对 `_select_and_manage_model`、`_handle_boundary_ratio` 的本地覆盖，复用共享去噪钩子；同时将 `prepared_vars` 改为属性访问，是行为风险最集中的文件。

```python
# denoising_dmd.py 的 forward 去噪循环（head 版本）
# 关键变化：不再使用 DMD 本地实现的 _select_and_manage_model /
# _handle_boundary_ratio，而是调用共享基类版本；
# 返回值只取 current_model，guidance scale 由共享逻辑内部处理。
t_int = int(t.item())
if self.transformer_2 is not None:
    # 共享实现依据 boundary_timestep 在高 / 低噪声专家间切换，
    # 并调用 _manage_dit_use_site 管理 DiT 使用位置
    current_model, _ = self._select_and_manage_model(
        t_int=t_int,
        boundary_timestep=self._handle_boundary_ratio(
            server_args, batch, scheduler
        ),
        server_args=server_args,
        batch=batch,
    )
else:
    current_model = self.transformer
    self._manage_dit_use_site(current_model, "transformer", batch)

```

### `python/sglang/multimodal_gen/runtime/pipelines/flux.py`

作为 native pipeline 代表，删除本地 `calculate_shift` 与未用 logger，`prepare_mu` 改为调用共享 `calculate_linear_shift`，验证共享函数接入方式。

```python
# flux.py 的 prepare_mu（head 版本）
# 原先的本地 calculate_shift 已删除，统一走共享工具函数；
# 默认参数 (base_seq_len=256, max_seq_len=4096, base_shift=0.5, max_shift=1.15)
# 与 FLUX 原本硬编码的值完全一致。
def prepare_mu(batch: Req, server_args: ServerArgs):
    height = batch.height
    width = batch.width
    vae_scale_factor = (
        server_args.pipeline_config.vae_config.arch_config.vae_scale_factor
    )
    image_seq_len = (int(height) // (vae_scale_factor * 2)) * (
        int(width) // (vae_scale_factor * 2)
    )
    # 返回 (key, mu) 二元组，供 add_standard_t2i_stages 注入 timestep kwargs
    return "mu", calculate_linear_shift(image_seq_len)

```

# 评论区精华

本 PR 没有 review 评论或审核意见。仅有的两条 issue 评论是作者触发的 `/tag-and-rerun-ci`（CI 重跑命令），没有形成技术讨论。值得注意的是 Extra CI 状态为失败，但缺少失败日志，无法判断是测试还是环境问题。

- 暂无高价值评论线程

# 风险与影响

- 风险：
 1. **DMD 行为依赖共享实现**：`denoising_dmd.py` 删除了本地 `_select_and_manage_model` / `_handle_boundary_ratio` 覆盖，改为使用共享基类版本。若基类实现与原先 DMD 本地版本在 boundary ratio 计算、高 / 低噪声专家切换或 `_manage_dit_use_site` 调用上有细微差异，可能导致 Wan2.2 等 DMD 模型去噪行为改变，且 PR 未附带测试验证。
 2. **`ServerArgs` 字段顺序敏感**：开发者在 body 中明确强调 `quantization` 字段保持原位置以维持 positional 与序列化顺序，说明 dataclass 字段顺序对下游（如命令行解析、序列化）敏感；合并重复字段本身不改变顺序，但任何后续对其位置的调整都可能引入兼容性问题。
 3. **Extra CI 失败未归因**：CI 显示 Extra 运行失败，但没有日志。若与本次重构相关（如某个 pipeline import 遗漏），可能影响未被 Base CI 覆盖的硬件 / 配置组合。
 4. **无测试配套**：此类大范围去重改动通常需要至少一个单元测试守护 `calculate_linear_shift` 的边界参数，或验证 DMD 去噪循环仍能运行；当前没有新增测试，回归风险敞口。
 - 影响：影响范围集中在 `sglang/multimodal_gen` 扩散运行时：所有 native pipeline（FLUX、Qwen-Image、Z-Image、Hunyuan、GLM-Image、Helios 等）与 DMD 去噪链路。对用户无公共 API 变化，CLI 参数与请求协议不变；对系统无性能影响（纯代码清理）。对团队而言，消除了三处重复的 shift 公式和 DMD 钩子覆盖，后续修改调度逻辑只需改 `diffusion_scheduler_utils.py` 或共享 stage，降低行为漂移与维护成本。
 - 风险标记：缺少测试覆盖 , DMD 行为依赖共享实现 , Extra CI 失败 , 字段顺序敏感性

# 关联脉络

- PR #33845 [diffusion] centralize entrypoint API hygiene: 同为扩散模块 hygiene 系列重构：本 PR 集中 pipeline core 逻辑与 DMD 钩子，33845 集中入口 API 公共逻辑与共享游标分页，二者是同一维护方向（消除重复、明确归属）的连续动作。