Diffusers ModularPipeline 完全指南：从模块化 Blocks 组装、懒加载组件到多工作流推理-二趣网

Diffusers ModularPipeline 完全指南：从模块化 Blocks 组装、懒加载组件到多工作流推理

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

ModularPipeline是 🤗 Diffusers 在传统DiffusionPipeline之外提供的新一代可组合推理接口：它把扩散模型的"编码 → 去噪 → 解码"等计算步骤抽象成可插拔的ModularPipelineBlocks，由ModularPipeline负责把 Blocks 组装成可执行管线。相比DiffusionPipeline，它的核心差异是创建与加载分离（懒加载）、可通过 Blocks 自由组装或从现有仓库一键转换、并且运行时的调用方式与原有 API 保持一致。读完本文，你将掌握：用init_pipeline从零组装自己的管线、用from_pretrained从三类仓库加载管线、按需load_components管理模型权重，以及利用ComponentsManager做跨管线内存管理。

核心概念：Blocks、状态与规格

在深入用法之前，先明确三个贯穿全文的基本概念，它们都定义在 modular_pipeline.py 与 modular_pipeline_utils.py 中：

ModularPipelineBlocks：管线块基类，通过sub_blocks（一个InsertableDict）持有子块，通过expected_components、expected_configs、inputs、intermediate_outputs等属性声明"这个块需要哪些组件、吃什么输入、吐什么输出"。从源码结构看，它派生了四类常用块：ConditionalPipelineBlocks（按输入条件选择分支）、AutoPipelineBlocks（按触发输入自动选择工作流）、SequentialPipelineBlocks（顺序执行子块）、LoopSequentialPipelineBlocks（循环执行子块）。
PipelineState：贯穿整条管线的状态容器，本质是values字典加kwargs_mapping映射。块与块之间通过它传递中间结果（例如文本编码后的prompt_embeds、去噪后的latents）。它支持set/get/get_by_kwargs以及属性式访问（state.denoiser_input_fields）。
ComponentSpec：描述"某个组件从哪里加载、是什么类型"的规格对象，字段包括name、type_hint（如CLIPTextModel）、pretrained_model_name_or_path、subfolder、variant、revision以及default_creation_method（"from_pretrained"或"from_config"）。ComponentSpec.load()走from_pretrained，ComponentSpec.create()走from_config。

一个ModularPipeline实例的组成可以概括为：Blocks（计算逻辑） + ComponentSpecs（组件加载规格） + 可选的 ComponentsManager（设备与内存管理）。

三种开箱即用的完整示例（SDXL）

文档给出了基于 SDXL 的 text-to-image、image-to-image、inpainting 三个完整示例。三者结构完全一致：from_pretrained只读取配置，load_components(dtype=...)才真正加载权重，to(device)移动设备，最后以与DiffusionPipeline相同的方式调用。

import torch from diffusers import ModularPipeline pipeline = ModularPipeline.from_pretrained("stabilityai/stable-diffusion-xl-base-1.0") pipeline.load_components(dtype=torch.float16) pipeline.to("cuda") # or "mps", "xpu", "cpu" image = pipeline(prompt="Astronaut in a jungle, cold color palette, muted colors, detailed, 8k").images[0] image.save("modular_t2i_out.png")

image-to-image 只需多传image与strength：

import torch from diffusers import ModularPipeline from diffusers.utils import load_image pipeline = ModularPipeline.from_pretrained("stabilityai/stable-diffusion-xl-base-1.0") pipeline.load_components(dtype=torch.float16) pipeline.to("cuda") url = "https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/diffusers/sdxl-text2img.png" init_image = load_image(url) prompt = "a dog catching a frisbee in the jungle" image = pipeline(prompt=prompt, image=init_image, strength=0.8).images[0] image.save("modular_i2i_out.png")

inpainting 则额外传入mask_image：

import torch from diffusers import ModularPipeline from diffusers.utils import load_image pipeline = ModularPipeline.from_pretrained("stabilityai/stable-diffusion-xl-base-1.0") pipeline.load_components(dtype=torch.float16) pipeline.to("cuda") img_url = "https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/diffusers/sdxl-text2img.png" mask_url = "https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/diffusers/sdxl-inpaint-mask.png" init_image = load_image(img_url) mask_image = load_image(mask_url) prompt = "A deep sea diver floating" image = pipeline(prompt=prompt, image=init_image, mask_image=mask_image, strength=0.85).images[0] image.save("modular_inpaint_out.png")

注意：调用时返回的其实是一个PipelineState对象，image = pipeline(...)直接取images[0]是因为这些块把最终图像以images键写入了状态（PipelineState支持属性式访问）。如果只想取某个中间值，可以显式传output=参数，见下文"运行管线"一节。

这三个示例之所以能"一个管线跑三种工作流"，正是因为 SDXL 的默认块使用了AutoPipelineBlocks——它在运行时根据"是否传了image、mask_image"等触发输入自动选择执行文本生成、图生图还是重绘分支（对应源码AutoPipelineBlocks.select_block的实现，见 modular_pipeline.py）。

创建管线：init_pipeline 与 from_pretrained

文档明确给出两条创建路径：从 Blocks 组装（ModularPipelineBlocks.init_pipeline）与从仓库加载（ModularPipeline.from_pretrained）。

用 init_pipeline 从 Blocks 组装

首先定义一个最小块。下面的MyBlock声明了一个text_encoder组件（类型为 transformers 的CLIPTextModel，默认从openai/clip-vit-large-patch14加载），__call__目前只是把components与state原样传下去：

from transformers import CLIPTextModel from diffusers.modular_pipelines import ( ComponentSpec, ModularPipelineBlocks, PipelineState, ) class MyBlock(ModularPipelineBlocks): @property def expected_components(self): return [ ComponentSpec( name="text_encoder", type_hint=CLIPTextModel, pretrained_model_name_or_path="openai/clip-vit-large-patch14", ), ] def __call__(self, components, state: PipelineState) -> PipelineState: return components, state

调用init_pipeline()把它转成管线，pipe.blocks保存着创建它的 Blocks——它决定了管线的输入、输出与计算逻辑：

block = MyBlock() pipe = block.init_pipeline() pipe.blocks

打印结果是 Blocks 的配置快照（MyBlock类名与 diffusers 版本号）：

MyBlock { "_class_name": "MyBlock", "_diffusers_version": "0.37.0.dev0" }

[!WARNING] Blocks 是可变的——在创建管线之前，你可以自由增删、替换子块。但一旦管线创建完成，修改pipeline.blocks不会生效，因为blocks属性返回的是深拷贝（源码见ModularPipeline.blocks属性，它执行deepcopy(self._blocks)）。想要不同的块结构，请先改 Blocks 再重新创建管线。

当init_pipeline不传仓库时，它依据块内ComponentSpec的pretrained_model_name_or_path决定每个组件的加载来源。打印pipe可以看到组件加载配置——注意每个组件是一个三元组[library, class, loading_spec]，此时尚未加载任何权重：

ModularPipeline { "_blocks_class_name": "MyBlock", "_class_name": "ModularPipeline", "_diffusers_version": "0.37.0.dev0", "text_encoder": [ null, null, { "pretrained_model_name_or_path": "openai/clip-vit-large-patch14", "revision": null, "subfolder": "", "type_hint": [ "transformers", "CLIPTextModel" ], "variant": null } ] }

如果给init_pipeline传一个仓库，则会用该仓库的管线配置（model_index.json或modular_model_index.json）覆盖加载路径——按组件名与仓库配置匹配。下面的例子中pretrained_model_name_or_path被更新为"stabilityai/stable-diffusion-xl-base-1.0"，且subfolder变成了"text_encoder"（说明该组件在仓库的text_encoder/子目录下）：

pipe = block.init_pipeline("stabilityai/stable-diffusion-xl-base-1.0") pipe ModularPipeline { "_blocks_class_name": "MyBlock", "_class_name": "ModularPipeline", "_diffusers_version": "0.37.0.dev0", "text_encoder": [ null, null, { "pretrained_model_name_or_path": "stabilityai/stable-diffusion-xl-base-1.0", "revision": null, "subfolder": "text_encoder", "type_hint": [ "transformers", "CLIPTextModel" ], "variant": null } ] }

如果块中的某个组件在仓库里不存在，它的条目保持null，并在load_components时被跳过——这与源码中"无有效加载规格的组件会被忽略"的逻辑一致。

用 from_pretrained 从仓库加载

ModularPipeline.from_pretrained免去手写 Blocks 的麻烦，支持三类仓库：

① 普通 diffusers 仓库：传入任意受支持模型的仓库，自动映射到默认管线块。当前仓库的MODULAR_PIPELINE_MAPPING（见 modular_pipeline.py）登记的模型包括 SDXL、Stable Diffusion 3、Wan、Wan-Animate-2、Flux、Flux-Kontext、Flux2、Flux2-Klein、Ideogram4、Krea2、Qwen-Image、Anima、Z-Image、Cosmos3-Omni、Helios、Hunyuan-Video-1.5、LTX、LTX2、MiniMax-H3、MiniMax-Music3、Ernie-Image 等。以 SDXL 为例，同时挂上一个ComponentsManager：

from diffusers import ModularPipeline, ComponentsManager components = ComponentsManager() pipeline = ModularPipeline.from_pretrained( "stabilityai/stable-diffusion-xl-base-1.0", components_manager=components )

② modular 仓库：仓库内含modular_model_index.json，逐组件指明加载来源——各组件可以来自不同仓库，modular 仓库本身甚至可以不含任何权重。例如diffusers/flux2-bnb-4bit-modular从 A 仓库加载量化后的 transformer，其余组件从 B 仓库加载：

from diffusers import ModularPipeline, ComponentsManager components = ComponentsManager() pipeline = ModularPipeline.from_pretrained( "diffusers/flux2-bnb-4bit-modular", components_manager=components )

③ 带自定义代码的 modular 仓库：仓库里除了加载配置还附带自定义的管线块（如diffusers/Florence2-image-Annotator），加载时需显式trust_remote_code=True：

from diffusers import ModularPipeline, ComponentsManager components = ComponentsManager() pipeline = ModularPipeline.from_pretrained( "diffusers/Florence2-image-Annotator", trust_remote_code=True, components_manager=components )

按工作流裁剪（workflow）：当管线块定义了工作流（用pipeline.blocks.available_workflows查看）时，可传workflow=只保留该工作流用到的块——这与ModularPipelineBlocks.get_workflow是同一套剪枝逻辑。裁剪后，管线只声明该工作流需要的组件，其 docstring 也只描述该工作流的输入：

pipeline = ModularPipeline.from_pretrained("Qwen/Qwen-Image", workflow="inpainting")

从源码看，workflow参数会调用blocks.get_workflow(workflow)得到该工作流的执行块序列，再据此收敛组件规格（ModularPipeline.__init__中blocks = blocks.get_workflow(workflow)）。

两种创建方式的内部差异

从源码ModularPipeline.from_pretrained的实现（modular_pipeline.py）可以看出加载的优先级链：

若仓库有modular_model_index.json，直接以其作为配置；
否则回退到model_index.json，先解析出传统管线类名，再通过MODULAR_PIPELINE_MAPPING映射到对应的 Modular 管线类（例如StableDiffusionXLModularPipeline）；
两者都没有时，认为该管线块不需要任何from_pretrained组件，pretrained_model_name_or_path置为None。

此外，from_pretrained还支持标准的 Hub 加载参数（cache_dir、force_download、local_files_only、local_dir、proxies、revision、token等），并会先尝试从仓库读取自定义 Blocks 类（ModularPipelineBlocks.from_pretrained处理auto_map与trust_remote_code）。

加载组件：懒加载与按需加载

ModularPipeline创建后不会自动实例化任何组件——它只持有配置与组件规格。真正的权重加载发生在load_components()。

加载全部组件

不传参数时，加载所有"有有效加载规格"的组件（即default_creation_method == "from_pretrained"且指定了pretrained_model_name_or_path、且当前尚未加载的组件）：

import torch pipeline.load_components(dtype=torch.float16)

按名字加载指定组件

只想加载text_encoder时：

pipeline.load_components(names=["text_encoder"], dtype=torch.float16)

names也接受单个字符串。未知组件名会被警告并忽略。

按工作流加载

对于定义了工作流的管线，workflow=只加载该工作流用到的组件，且管线会保留全部块——这是"一个管线跑多种工作流"的推荐方式：每次调用只会补齐新工作流还缺的组件，已经加载过的组件不会重复加载：

pipeline.load_components(workflow="inpainting", dtype=torch.float16)

注意names与workflow不能同时传入，源码会直接抛出ValueError。

如何确认组件是否已加载

加载完成后再次打印管线，每个组件的三元组前两个字段会从null变成实际的库名与类名：

# text_encoder is loaded - shows library and class "text_encoder": [ "transformers", "CLIPTextModel", { ... } ] # unet is not loaded yet - still null "unet": [ null, null, { ... } ]

加载参数的传递规则

load_components的 kwargs（如dtype、variant、revision、quantization_config）会被透传给每个组件的from_pretrained()。传法有两种：

单值：对所有组件生效，例如dtype=torch.bfloat16；
字典：按组件名分别指定，"default"键作为兜底值，例如：

# apply bfloat16 to all components pipeline.load_components(dtype=torch.bfloat16) # different dtypes per component pipeline.load_components(dtype={"transformer": torch.bfloat16, "default": torch.float32})

源码中对字典的处理逻辑是：如果该组件名在字典里就取它的值，否则取"default"键的值（见load_components中component_load_kwargs的组装逻辑）。

另外两个值得注意的安全与幂等细节：

trust_remote_code只透传给与管线同仓库的组件：如果某个组件来自modular_model_index.json里指向的外部仓库，trust_remote_code会被剥离，以免用户无意中信任外部仓库代码；此时若加载失败，源码会给出提示——请手动用AutoModel.from_pretrained(..., trust_remote_code=True)加载后通过update_components注入。
幂等性：load_components只会加载"尚未加载且有有效加载规格"的组件。如果某个组件已经设置到管线上，再次调用load_components不会重新加载它。

更新组件：换模型、换 Guider、改配置

update_components()用于替换管线上的组件。替换后，组件对应的加载规格也会同步更新进管线配置，后续load_components会跳过它。

从 AutoModel 注入

用AutoModel.from_pretrained()加载的模型对象会被自动打上加载信息标签，可以直接注入：

from diffusers import AutoModel unet = AutoModel.from_pretrained( "RunDiffusion/Juggernaut-XL-v9", subfolder="unet", variant="fp16", dtype=torch.float16 ) pipeline.update_components(unet=unet)

从 ComponentSpec 修改后加载

用get_component_spec()拿一份当前规格的拷贝（注意是深拷贝，修改不影响管线自身），改完再加载、注入：

unet_spec = pipeline.get_component_spec("unet") # modify to load from a different repository unet_spec.pretrained_model_name_or_path = "RunDiffusion/Juggernaut-XL-v9" # load and update unet = unet_spec.load(dtype=torch.float16) pipeline.update_components(unet=unet)

也可以完全从零构造一个ComponentSpec。

从配置创建（from_config 组件）

并非所有组件都从预训练权重加载——有些组件由配置创建，这类组件列在pipeline.config_component_names里。对它们用ComponentSpec.create()而不是load()：

guider_spec = pipeline.get_component_spec("guider") guider_spec.config = {"guidance_scale": 5.0} guider = guider_spec.create() pipeline.update_components(guider=guider)

或者直接传对象：

from diffusers.guiders import ClassifierFreeGuidance guider = ClassifierFreeGuidance(guidance_scale=5.0) pipeline.update_components(guider=guider)

update_components还接受纯配置值的更新（例如pipeline.update_components(requires_safety_checker=False)），并会把新值同步到_config_specs与管线配置字典中。想了解可用的 guider 及其配置，参见 Guiders 指南。

将管线拆分为多阶段：文本编码与去噪解耦

由于 Blocks 是可组合的，你可以把一条管线拆成多条独立子管线，分别执行不同阶段。文档示例展示了如何把"文本编码"从 FLUX.2 管线的其余部分中拆出来：先用独立的 text encoder 管线编码 prompt 得到 embeddings，再把这些 embeddings 喂给主管线做去噪与解码。这样可以在不同设备上分别部署两个阶段，或复用同一份文本编码结果跑多次去噪。

from diffusers import ModularPipeline, ComponentsManager import torch device = "cuda" # or "mps", "xpu", "cpu" dtype = torch.bfloat16 repo_id = "black-forest-labs/FLUX.2-klein-4B" # get the blocks and separate out the text encoder blocks = ModularPipeline.from_pretrained(repo_id).blocks text_block = blocks.sub_blocks.pop("text_encoder") # use ComponentsManager to handle offloading across multiple pipelines manager = ComponentsManager() manager.enable_auto_cpu_offload(device=device) # create separate pipelines for each stage text_encoder_pipeline = text_block.init_pipeline(repo_id, components_manager=manager) pipeline = blocks.init_pipeline(repo_id, components_manager=manager) # encode text text_encoder_pipeline.load_components(dtype=dtype) text_embeddings = text_encoder_pipeline(prompt="a cat").get_by_kwargs("denoiser_input_fields") # denoise and decode pipeline.load_components(dtype=dtype) output = pipeline( **text_embeddings, num_inference_steps=4, ).images[0]

要点拆解：

blocks.sub_blocks.pop("text_encoder")从块字典中取出 text encoder 子块，剩下的块仍由blocks持有，两者分别init_pipeline成两条独立管线；
get_by_kwargs("denoiser_input_fields")是PipelineState的方法，它按kwargs_type取出所有标记为去噪器输入字段的中间值（例如prompt_embeds），并以字典形式返回，可直接用**展开传给主管线；
两条管线共享同一个ComponentsManager，由它统一管理内存与设备。

ComponentsManager：动态的内存调度

与DiffusionPipeline固定顺序的卸载策略不同，ComponentsManager在每次模型前向传播时根据当前内存状况动态做出卸载决策（enable_auto_cpu_offload支持device与memory_reserve_margin等参数，见 components_manager.py）。这意味着无论你创建多少条管线、以什么顺序运行，它都能正常工作。详见 ComponentsManager 指南。

如果多个阶段的管线共享组件（例如同一个 VAE 既用于编码又用于解码），可以用update_components把已加载的组件传给另一条管线，避免重复加载。

运行管线与输出控制

ModularPipeline.__call__的执行流程（源码见 modular_pipeline.py）大致是：创建/深拷贝PipelineState→ 把用户 kwargs 与块的默认输入写入 state → 在torch.no_grad()下依次执行各块 → 根据output参数决定返回值。

调用方式与DiffusionPipeline类似，但返回值更灵活：

# Get complete pipeline state state = pipeline(prompt="A beautiful sunset", num_inference_steps=20) # Get specific output image = pipeline(prompt="A beautiful sunset", output="image") # Get multiple specific outputs results = pipeline(prompt="A beautiful sunset", output=["image", "latents"]) image, latents = results["image"], results["latents"] # Continue from previous state state = pipeline(prompt="A beautiful sunset") new_state = pipeline(state=state, output="image") # Continue processing

不传output：返回完整PipelineState（含所有输入与中间值）；
output="image"：返回指定中间值；
output=["image", "latents"]：返回一个名字到值的字典；
传入state=可以基于已有状态继续执行——这为"分阶段执行、断点续跑"提供了可能。

Modular 仓库：把加载配置与自定义代码发布出去

如果管线块使用了pretrained 组件，就需要一个仓库来提供加载规格与元数据。ModularPipeline开箱即用地兼容普通 diffusers 仓库；而modular 仓库提供更大的灵活性——它包含一个modular_model_index.json，其中有三个关键要素：

library与class：指明组件从哪个库加载、类名是什么；若为null表示该组件尚未加载。
loading_specs_dict：包含加载组件所需的信息，例如来源仓库与子文件夹。

modular 仓库的核心优势是组件可以来自不同仓库。例如diffusers/flux2-bnb-4bit-modular从diffusers/FLUX.2-dev-bnb-4bit加载量化 transformer，其余组件从black-forest-labs/FLUX.2-dev加载。

将普通仓库转换为 modular 仓库

流程非常简单：先用普通仓库创建管线，然后save_pretrained推送到 Hub——保存的仓库会自动生成包含全部加载规格的modular_model_index.json：

from diffusers import ModularPipeline # load from a regular repo pipeline = ModularPipeline.from_pretrained("stabilityai/stable-diffusion-xl-base-1.0") # push as a modular repository pipeline.save_pretrained("local/path", repo_id="my-username/sdxl-modular", push_to_hub=True)

save_pretrained还支持safe_serialization（默认True，用 safetensors 保存）、variant、max_shard_size（权重分片大小，如"5GB"）以及overwrite_modular_index（设为True时，把所有组件引用改写为指向目标仓库repo_id，适用于把散落在不同仓库的组件"收编"到新仓库的场景）。

在 modular 仓库中附带自定义 Blocks

modular 仓库还可以把自定义管线块以 Python 代码的形式一并发布，方便分享 Diffusers 原生没有的特殊块。diffusers/Florence2-image-Annotator的仓库结构如下：

Florence2-image-Annotator/ ├── block.py # Custom pipeline blocks implementation ├── config.json # Pipeline configuration and auto_map ├── mellon_config.json # UI configuration for Mellon └── modular_model_index.json # Component loading specifications

其中config.json的auto_map键告诉ModularPipeline去哪里找自定义块：

{ "_class_name": "Florence2AnnotatorBlocks", "auto_map": { "ModularPipelineBlocks": "block.Florence2AnnotatorBlocks" } }

auto_map的值是"模块文件.类名"格式："block.Florence2AnnotatorBlocks"表示从仓库的block.py中加载Florence2AnnotatorBlocks类。加载这类自定义代码仓库时同样需要trust_remote_code=True（见上文"from_pretrained"一节）。如何编写自己的自定义块，参见 Custom blocks 指南。

实践要点速查

懒加载是默认行为：from_pretrained只读配置；权重必须在load_components()中加载，加载参数（dtype、variant、revision、quantization_config）在这一步传入，支持"单值全组件"与"按组件名字典"两种传法。
一个管线多种工作流：当块的available_workflows非空时，用workflow=剪枝加载；AutoPipelineBlocks会在运行时根据image/mask_image等触发输入自动切换 text-to-image、image-to-image、inpainting。
Blocks 在创建管线前修改：创建后pipeline.blocks返回深拷贝，改它不生效；要换结构就重新init_pipeline。
多管线共享内存：多阶段拆分（如文本编码与去噪分离）时，给各子管线传入同一个ComponentsManager并enable_auto_cpu_offload，由它按当前内存动态卸载；共享组件用update_components复用，避免重复加载。
发布 modular 仓库：save_pretrained(..., push_to_hub=True)自动生成modular_model_index.json；跨仓库组件引用通过loading_specs_dict表达；自定义块用auto_map声明并通过trust_remote_code=True加载。

相关参考：Modular Diffusers 概览、快速上手、Pipeline Block 详解、AutoPipelineBlocks、SequentialPipelineBlocks、ComponentsManager、自定义块。

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

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

企业官网建设流程全解析