Diffusers 模块化流水线入门：使用 ModularPipelineBlocks 构建可复用的 Pipeline 步骤

Diffusers 模块化流水线入门：使用 ModularPipelineBlocks 构建可复用的 Pipeline 步骤

【免费下载链接】diffusers🤗 Diffusers: State-of-the-art diffusion models for image, video, and audio generation in PyTorch.项目地址: https://gitcode.com/GitHub_Trending/di/diffusers

Modular Diffusers 是 🤗 Diffusers 提供的一套统一流水线体系，它将传统DiffusionPipeline中的长流程拆分为一个个可复用、可组合的pipeline blocks（流水线块）。本文将以官方指南 pipeline_block.md 为骨架，系统讲解ModularPipelineBlocks的设计理念、inputs/intermediate_outputs/ComponentSpec/ConfigSpec等核心概念的用法、__call__计算逻辑的编写范式，并结合 src/diffusers/modular_pipelines 目录下的源码与真实示例块（如 SDXL 的编码器块）进行纵深解读。读完本文，你将掌握从零编写一个自定义 pipeline block、将其组装进ModularPipeline并最终运行推理的完整技能。

一、什么是 ModularPipelineBlocks：流水线中的"步骤蓝图"

ModularPipelineBlocks是构建ModularPipeline的基本单元（basic block）。它回答三个问题：一个步骤需要哪些组件、接收什么输入、产出什么输出、执行什么计算。

• 一个ModularPipelineBlocks定义了流水线中某个具体步骤应具备的组件（components）、输入/输出（inputs/outputs）与计算逻辑（computation）。
• 它通过 state（状态）与其他块连接，从而以模块化方式拼装出完整工作流。
• 单个ModularPipelineBlocks本身不能被执行——它只是"这一步该做什么"的蓝图（blueprint）。要真正运行流水线，需要先把这些块转换成可执行的ModularPipeline（转换入口是ModularPipelineBlocks.init_pipeline()或ModularPipeline.from_pretrained()，详见 modular_pipeline.md）。

这种"声明式定义 + 运行时执行"分离的设计，使得同一批块可以按不同顺序组合、被多个流水线复用，甚至只新增那些对当前流水线独特的块即可（参见 overview.md 中对"复用性"的定位）。

在源码中，ModularPipelineBlocks 继承自ConfigMixin与PushToHubMixin，因此它天然具备配置保存/加载（save_pretrained/from_pretrained）与推送 Hub 的能力。基类为以下四个子类提供公共基础设施（类文档原文，见 modular_pipeline.py）：

子类	职责	源码位置
ConditionalPipelineBlocks	根据输入条件选择运行哪个子块	modular_pipeline.py
AutoPipelineBlocks	按触发输入自动选择工作流（txt2img/img2img/inpaint 等）	modular_pipeline.py
SequentialPipelineBlocks	按顺序串联多个块	modular_pipeline.py
LoopSequentialPipelineBlocks	循环执行一组块	modular_pipeline.py 提及

这些组合块对应的官方教程分别是 sequential_pipeline_blocks.md、loop_sequential_pipeline_blocks.md 与 auto_pipeline_blocks.md。

二、inputs 与 intermediate_outputs：定义块的输入输出契约

一个ModularPipelineBlocks必须定义inputs和intermediate_outputs两个属性。

2.1 inputs：块从全局状态中读取的值

inputs是块从PipelineState中读取、用于执行计算的值。这些值可能来自用户（如 prompt、image），也可能来自前一个块（如编码后的image_latents）。使用InputParam声明：

class ImageEncodeStep(ModularPipelineBlocks): ... @property def inputs(self): return [ InputParam(name="image", type_hint="PIL.Image", required=True, description="raw input image to process"), ] ...

InputParam在 modular_pipeline_utils.py 中定义，字段如下：

字段	含义	默认值
name	参数名，同时也是在PipelineState中查找/写入的键	None
type_hint	类型标注（如str、PIL.Image.Image、torch.Tensor）	None
default	可选参数的默认值，运行时在get_block_state中解析	None
required	是否为必需输入，缺失会触发ValueError	False
description	参数说明（会进入自动生成的doc与 model card）	""
kwargs_type	当name为None时使用，表示按分组批量接收输入（如"denoiser_input_fields"）	None
defaults_by_block	条件块中各分支声明不同默认值时由combine_inputs填充	None

此外，InputParam.template(template_name, **overrides)提供常用参数的模板化声明。仓库内置了INPUT_PARAM_TEMPLATES（见 modular_pipeline_utils.py），例如：

• prompt：str类型、required=True、"The prompt or prompts to guide image generation."
• negative_prompt：str类型、可选
• num_inference_steps：int、默认50
• height/width：int，生成图像的像素尺寸
• strength：float、默认0.9，用于 img2img/inpainting
• image：PIL.Image.Image | list[PIL.Image.Image]、required=True
• mask_image：PIL.Image.Image、required=True，用于 inpainting
• control_image：PIL.Image.Image、required=True，用于 ControlNet 条件
• generator：torch.Generator，用于确定性生成
• output_type：str、默认"pil"，可选'pil'/'np'/'pt'
• denoiser_input_fields：特殊模板，name=None、kwargs_type="denoiser_input_fields"，用于把 prompt_embeds 等一组条件输入批量传给去噪器

2.2 intermediate_outputs：块产出并写回全局状态的新值

intermediate_outputs是块创建的新值，会被加入PipelineState，既能作为后续块的inputs，也能作为流水线运行后的最终输出暴露给用户。使用OutputParam声明：

class ImageEncodeStep(ModularPipelineBlocks): ... @property def intermediate_outputs(self): return [ OutputParam(name="image_latents", description="latents representing the image"), ] ...

OutputParam字段（见 modular_pipeline_utils.py）与InputParam类似：name、type_hint、description、kwargs_type、metadata。同样有OUTPUT_PARAM_TEMPLATES模板（同文件 L517-L555），如：

• images：list[PIL.Image.Image]，"Generated images."
• videos：list[PIL.Image.Image]
• latents：torch.Tensor，"Denoised latents."
• prompt_embeds：torch.Tensor、kwargs_type="denoiser_input_fields"
• image_latents：torch.Tensor，"The latent representation of the input image."

intermediate_outputs与inputs共享同一份PipelineState数据，因此在流水线执行的任意时刻都可被读取，方便追踪工作流进度（这是 States 指南中"状态交互"的核心思想：inputs可被修改并回写，intermediate_outputs是新增变量并进入全局values字典）。

三、Components 与 Configs：声明依赖的模型组件与流水线级配置

块运行所需的组件与流水线级配置通过ComponentSpec与ConfigSpec声明：

• ComponentSpec：块使用到的组件及其期望类型。name是必需的，type_hint最好一并给出，以精确说明组件是什么。
• ConfigSpec：跨块生效的流水线级设置（如 SDXL 的force_zeros_for_empty_prompt）。

class ImageEncodeStep(ModularPipelineBlocks): ... @property def expected_components(self): return [ ComponentSpec(name="vae", type_hint=AutoencoderKL), ] @property def expected_configs(self): return [ ConfigSpec("force_zeros_for_empty_prompt", True), ] ...

3.1 ComponentSpec 的完整字段与加载语义

ComponentSpec定义于 modular_pipeline_utils.py，字段如下：

字段	含义
name	组件名，块内通过components.<name>访问
type_hint	组件类型（如UNet2DConditionModel、AutoencoderKL）
description	可选说明
config	通过__init__创建时的配置 dict（配合default_creation_method="from_config"）
pretrained_model_name_or_path	通过from_pretrained加载时的仓库/路径
subfolder	仓库内的子目录
variant	权重变体（如"fp16"）
revision	仓库 revision
default_creation_method	首选创建方式："from_config"或"from_pretrained"（默认）

组件有两种创建路径，分别对应两个方法：

• ComponentSpec.create(config, **kwargs)：使用type_hint.from_config(config)从配置创建（适合无权重或需要现场实例化的组件，如 scheduler、guider、image processor）。参见 modular_pipeline_utils.py。
• ComponentSpec.load(**kwargs)：使用type_hint.from_pretrained(...)从 Hub 加载权重，支持单文件加载（from_single_file）与AutoModel兜底（当type_hint为None时），并支持torch_dtype等加载参数。参见 modular_pipeline_utils.py。

ComponentSpec.from_component(name, component)还可以从已实例化的组件反向生成 spec（仅支持通过load()创建的带_diffusers_load_id的组件，或未继承nn.Module的ConfigMixin对象），用于"组件 -> spec"的往返管理。

3.2 ConfigSpec

ConfigSpec定义于 modular_pipeline_utils.py，仅三个字段：name（配置名）、default（默认值）、description（可选说明）。它描述的是流水线级行为开关——例如 SDXL 的force_zeros_for_empty_prompt=True表示无负 prompt 时用零向量替代无条件嵌入（在 encoders.py 中体现为zero_out_negative_prompt逻辑）。

当这些块被转换成流水线后，expected_components中声明的组件会作为__call__的第一个参数components传入，块内即可通过components.vae、components.text_encoder等形式访问。

四、Computation Logic：__call__方法的标准四步结构

块的计算逻辑写在__call__方法中，遵循固定结构：

1. 取局部视图：调用self.get_block_state(state)获取BlockState，这是当前块所需inputs的本地快照。
2. 执行计算：在block_state上完成计算逻辑（通过属性访问，如block_state.image）。
3. 回写全局状态：调用self.set_block_state(state, block_state)将局部BlockState的变更推回全局PipelineState。
4. 返回：把components和state返回给下一个块。

class ImageEncodeStep(ModularPipelineBlocks): def __call__(self, components, state): # Get a local view of the state variables this block needs block_state = self.get_block_state(state) # Your computation logic here # block_state contains all your inputs # Access them like: block_state.image, block_state.processed_image # Update the pipeline state with your updated block_states self.set_block_state(state, block_state) return components, state

4.1 底层原理：get_block_state 与 set_block_state

从源码看（modular_pipeline.py），这两个方法承担了"契约驱动"的状态搬运：

• get_block_state(state)：遍历该块的inputs，对每个InputParam从state.get(name)取值；值为None时填入声明的default；若required=True且仍为None，抛出ValueError: Required input 'xxx' is missing；若声明了kwargs_type，则通过state.get_by_kwargs(kwargs_type)把该分组下所有非空值一并装入局部字典，最终组装成BlockState返回（返回类型为BlockState(**data)）。
• set_block_state(state, block_state)：先遍历intermediate_outputs，若block_state上缺少某个输出名则抛ValueError，否则state.set(output_param.name, value, output_param.kwargs_type)写入；再遍历inputs，用身份比较（current_value is not param）判断输入对象是否被修改，只有被修改过才回写——这是"块可以就地修改输入并全局传播"机制的实现基础。

4.2 PipelineState 与 BlockState

这两种状态数据结构定义于 modular_pipeline.py：

• PipelineState：全局容器，数据存放在values字典中，是可变的。支持set/get/get_by_kwargs/to_dict，并通过__getattr__支持state.prompt_embeds式属性访问；__repr__会把张量格式化为Tensor(dtype=..., shape=...)，便于调试。
• BlockState：块内局部视图，通过__init__(**kwargs)构造，支持属性访问与block_state["foo"]下标访问，as_dict()可转为字典。

关于两者如何通过inputs/intermediate_outputs协作，参见 States 指南。

五、Putting it all together：完整块示例与自动文档

下面是一个把以上所有要素串联起来的完整块（官方指南中的示例，此处按仓库 API 对齐导入方式）：

from diffusers import ComponentSpec, AutoencoderKL from diffusers.modular_pipelines import InputParam, ModularPipelineBlocks, OutputParam class ImageEncodeStep(ModularPipelineBlocks): @property def description(self): return "Encode an image into latent space." @property def expected_components(self): return [ ComponentSpec(name="vae", type_hint=AutoencoderKL), ] @property def inputs(self): return [ InputParam(name="image", type_hint="PIL.Image", required=True, description="raw input image to process"), ] @property def intermediate_outputs(self): return [ OutputParam(name="image_latents", type_hint="torch.Tensor", description="latents representing the image"), ] def __call__(self, components, state): block_state = self.get_block_state(state) block_state.image_latents = components.vae.encode(block_state.image) self.set_block_state(state, block_state) return components, state

注意：ComponentSpec、ConfigSpec、InputParam、OutputParam、ModularPipelineBlocks等符号均由 src/diffusers/modular_pipelines/init.py 统一导出（其中ModularPipelineBlocks、PipelineState、BlockState、ConditionalPipelineBlocks、SequentialPipelineBlocks、AutoPipelineBlocks、LoopSequentialPipelineBlocks来自modular_pipeline模块；ComponentSpec、ConfigSpec、InputParam、OutputParam来自modular_pipeline_utils模块）。

5.1 每个块都有自动生成的 doc

每个块都有一个doc属性，由 make_doc_string 根据你上面定义的所有属性自动生成——它汇总了块的描述、组件、输入、输出（expected_configs若声明也会包含 Configs 段落）。示例输出（为忠实展示自动生成格式，这里保留了原文档的渲染结果；实例化示例按类名修正为ImageEncodeStep）：

block = ImageEncodeStep() print(block.doc)

输出：

class ImageEncodeStep Encode an image into latent space. Components: vae (`AutoencoderKL`) Inputs: image (`PIL.Image`): raw input image to process Outputs: image_latents (`torch.Tensor`): latents representing the image

在基类实现中，doc属性调用的正是make_doc_string(self.inputs, self.outputs, self.description, class_name=..., expected_components=..., expected_configs=...)（见 modular_pipeline.py）。格式化细节由format_components、format_configs、format_input_params、format_output_params完成（modular_pipeline_utils.py），支持自动换行、可选默认值标注（*optional*, defaults to X）与kwargs_type分组的展示。条件块（如ConditionalPipelineBlocks）还会在__repr__中列出 Trigger Inputs 与 Sub-Blocks 树（见 modular_pipeline.py）。

六、仓库中的真实范例：SDXL 编码器块

为了说明真实项目中块的写法，可以对照 SDXL 模块化流水线 src/diffusers/modular_pipelines/stable_diffusion_xl/encoders.py 中的几个真实块：

StableDiffusionXLTextEncoderStep（encoders.py）：

• expected_components声明text_encoder（CLIPTextModel）、text_encoder_2（CLIPTextModelWithProjection）、tokenizer/tokenizer_2（CLIPTokenizer），以及一个用from_config方式创建的guider（ClassifierFreeGuidance，config=FrozenDict({"guidance_scale": 7.5})）；
• expected_configs声明ConfigSpec("force_zeros_for_empty_prompt", True)——正是官方指南中ConfigSpec示例的出处；
• inputs声明prompt、prompt_2、negative_prompt、negative_prompt_2、cross_attention_kwargs、clip_skip；
• intermediate_outputs产出prompt_embeds、negative_prompt_embeds、pooled_prompt_embeds、negative_pooled_prompt_embeds，且都带kwargs_type="denoiser_input_fields"，表示它们是去噪器的条件输入字段；
• __call__中严格遵循get_block_state→check_inputs/encode_prompt→set_block_state→return components, state的范式，并用@torch.no_grad()装饰。

StableDiffusionXLVaeEncoderStep（encoders.py）：

• 组件为vae（AutoencoderKL）与image_processor（VaeImageProcessor，config=FrozenDict({"vae_scale_factor": 8})，default_creation_method="from_config"）；
• inputs为image（required=True）、height、width、generator、dtype、preprocess_kwargs；
• intermediate_outputs产出image_latents；
• 计算逻辑先image_processor.preprocess(...)预处理，再_encode_vae_image编码出潜在表示，并处理generator为列表时的逐张采样（retrieve_latents）。

StableDiffusionXLInpaintVaeEncoderStep（encoders.py）则展示了更复杂的多输出块：它额外声明mask_processor组件，产出image_latents、mask、masked_image_latents、crops_coords四个中间输出，用于 inpaint 专用 UNet 的通道拼接。

这些真实块的共同模式是：声明式属性（description/expected_components/expected_configs/inputs/intermediate_outputs）+ 命令式__call__，且都带上model_name = "stable-diffusion-xl"以映射到对应的模块化流水线类（映射表MODULAR_PIPELINE_MAPPING见 modular_pipeline.py，当前仓库覆盖 SDXL、SD3、Flux/Flux2、Wan、LTX、QwenImage、Krea2、Z-Image、HunyuanVideo1.5、MiniMax 等模型）。

七、把块组装成可执行的流水线

单个块不可执行，需要转换为ModularPipeline。有三种典型路径：

1. 从已有模型仓库一键转换：ModularPipeline.from_pretrained("stabilityai/stable-diffusion-xl-base-1.0")会自动映射到默认的块集合（由MODULAR_PIPELINE_MAPPING决定），再调用pipeline.load_components(dtype=...)加载模型权重。注意这里加载是惰性的：from_pretrained只读取配置、知道每个组件从哪里加载，权重在load_components时才真正落盘加载。
2. 用init_pipeline()组装自定义块：把你的块通过ModularPipelineBlocks.init_pipeline(...)转换（源码见 modular_pipeline.py），可选传入pretrained_model_name_or_path、components_manager与collection。
3. 用组合块编排：把多个ModularPipelineBlocks放入SequentialPipelineBlocks（顺序执行）、ConditionalPipelineBlocks/AutoPipelineBlocks（按输入自动选择分支，例如"传入mask_image走 inpaint、传入image走 img2img、否则走 txt2img"），或LoopSequentialPipelineBlocks（循环执行）。

组件实例的创建与复用由ComponentsManager统一管理（见 components_manager.md）。执行细节与完整示例（SDXL 的 txt2img/img2img/inpaint）参见 modular_pipeline.md。

八、块的保存、加载与共享

得益于继承ConfigMixin/PushToHubMixin，块本身可以像模型一样序列化：

• save_pretrained(save_directory, push_to_hub=False)：把块的定义写入modular_config.json（config_name = "modular_config.json"，见 modular_pipeline.py），并自动写入auto_map（{基类名: "模块.类名"}）与requirements字段（若块声明了_requirements）。
• from_pretrained(...)：读取配置，解析auto_map找到自定义代码（trust_remote_code相关逻辑），用get_class_from_dynamic_module动态加载块类并实例化。
• 自定义块的创建、校验与分享到 Hub 的完整流程，见 custom_blocks.md。

九、调试与文档生成建议

• 追踪工作流进度：intermediate_outputs与inputs共享PipelineState，任意时刻都可读取中间值；PipelineState.__repr__与BlockState.__repr__会把张量显示为Tensor(dtype=..., shape=...)，非常适合在__call__中断点调试。
• 查看块结构：打印block.doc获得自动生成的规格文档；组合块的__repr__会显示 Sub-Blocks 树与触发输入。
• 常见报错：若get_block_state发现required=True的输入缺失，会抛ValueError: Required input 'xxx' is missing；若set_block_state发现intermediate_outputs中的名字未写入block_state，会抛Intermediate output 'xxx' is missing in block state——这两条错误信息直接来自源码校验（modular_pipeline.py），可作为排查指南。

十、延伸阅读

• States：PipelineState/BlockState与状态交互详解
• SequentialPipelineBlocks、LoopSequentialPipelineBlocks、AutoPipelineBlocks：三类组合块教程
• ModularPipeline：块到可执行流水线的转换与运行
• Custom Blocks：自定义块的分享与 Hub 集成
• ComponentsManager：跨流水线的组件复用
• Overview：Modular Diffusers 文档总览
• 源码参考：modular_pipeline.py（基类与状态）、modular_pipeline_utils.py（Spec/Param 定义与文档生成）、stable_diffusion_xl/encoders.py（真实块示例）

【免费下载链接】diffusers🤗 Diffusers: State-of-the-art diffusion models for image, video, and audio generation in PyTorch.项目地址: https://gitcode.com/GitHub_Trending/di/diffusers