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/diffusersModular 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.pyAutoPipelineBlocks按触发输入自动选择工作流txt2img/img2img/inpaint 等modular_pipeline.pySequentialPipelineBlocks按顺序串联多个块modular_pipeline.pyLoopSequentialPipelineBlocks循环执行一组块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(nameimage, type_hintPIL.Image, requiredTrue, descriptionraw input image to process), ] ...InputParam在 modular_pipeline_utils.py 中定义字段如下字段含义默认值name参数名同时也是在PipelineState中查找/写入的键Nonetype_hint类型标注如str、PIL.Image.Image、torch.TensorNonedefault可选参数的默认值运行时在get_block_state中解析Nonerequired是否为必需输入缺失会触发ValueErrorFalsedescription参数说明会进入自动生成的doc与 model cardkwargs_type当name为None时使用表示按分组批量接收输入如denoiser_input_fieldsNonedefaults_by_block条件块中各分支声明不同默认值时由combine_inputs填充None此外InputParam.template(template_name, **overrides)提供常用参数的模板化声明。仓库内置了INPUT_PARAM_TEMPLATES见 modular_pipeline_utils.py例如promptstr类型、requiredTrue、The prompt or prompts to guide image generation.negative_promptstr类型、可选num_inference_stepsint、默认50height/widthint生成图像的像素尺寸strengthfloat、默认0.9用于 img2img/inpaintingimagePIL.Image.Image | list[PIL.Image.Image]、requiredTruemask_imagePIL.Image.Image、requiredTrue用于 inpaintingcontrol_imagePIL.Image.Image、requiredTrue用于 ControlNet 条件generatortorch.Generator用于确定性生成output_typestr、默认pil可选pil/np/ptdenoiser_input_fields特殊模板nameNone、kwargs_typedenoiser_input_fields用于把 prompt_embeds 等一组条件输入批量传给去噪器2.2 intermediate_outputs块产出并写回全局状态的新值intermediate_outputs是块创建的新值会被加入PipelineState既能作为后续块的inputs也能作为流水线运行后的最终输出暴露给用户。使用OutputParam声明class ImageEncodeStep(ModularPipelineBlocks): ... property def intermediate_outputs(self): return [ OutputParam(nameimage_latents, descriptionlatents representing the image), ] ...OutputParam字段见 modular_pipeline_utils.py与InputParam类似name、type_hint、description、kwargs_type、metadata。同样有OUTPUT_PARAM_TEMPLATES模板同文件 L517-L555如imageslist[PIL.Image.Image]Generated images.videoslist[PIL.Image.Image]latentstorch.TensorDenoised latents.prompt_embedstorch.Tensor、kwargs_typedenoiser_input_fieldsimage_latentstorch.TensorThe 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(namevae, type_hintAutoencoderKL), ] 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、AutoencoderKLdescription可选说明config通过__init__创建时的配置 dict配合default_creation_methodfrom_configpretrained_model_name_or_path通过from_pretrained加载时的仓库/路径subfolder仓库内的子目录variant权重变体如fp16revision仓库 revisiondefault_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 ConfigSpecConfigSpec定义于 modular_pipeline_utils.py仅三个字段name配置名、default默认值、description可选说明。它描述的是流水线级行为开关——例如 SDXL 的force_zeros_for_empty_promptTrue表示无负 prompt 时用零向量替代无条件嵌入在 encoders.py 中体现为zero_out_negative_prompt逻辑。当这些块被转换成流水线后expected_components中声明的组件会作为__call__的第一个参数components传入块内即可通过components.vae、components.text_encoder等形式访问。四、Computation Logic__call__方法的标准四步结构块的计算逻辑写在__call__方法中遵循固定结构取局部视图调用self.get_block_state(state)获取BlockState这是当前块所需inputs的本地快照。执行计算在block_state上完成计算逻辑通过属性访问如block_state.image。回写全局状态调用self.set_block_state(state, block_state)将局部BlockState的变更推回全局PipelineState。返回把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, state4.1 底层原理get_block_state 与 set_block_state从源码看modular_pipeline.py这两个方法承担了契约驱动的状态搬运get_block_state(state)遍历该块的inputs对每个InputParam从state.get(name)取值值为None时填入声明的default若requiredTrue且仍为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.pyPipelineState全局容器数据存放在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(namevae, type_hintAutoencoderKL), ] property def inputs(self): return [ InputParam(nameimage, type_hintPIL.Image, requiredTrue, descriptionraw input image to process), ] property def intermediate_outputs(self): return [ OutputParam(nameimage_latents, type_hinttorch.Tensor, descriptionlatents 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 段落。示例输出为忠实展示自动生成格式这里保留了原文档的渲染结果实例化示例按类名修正为ImageEncodeStepblock 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 中的几个真实块StableDiffusionXLTextEncoderStepencoders.pyexpected_components声明text_encoderCLIPTextModel、text_encoder_2CLIPTextModelWithProjection、tokenizer/tokenizer_2CLIPTokenizer以及一个用from_config方式创建的guiderClassifierFreeGuidanceconfigFrozenDict({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_skipintermediate_outputs产出prompt_embeds、negative_prompt_embeds、pooled_prompt_embeds、negative_pooled_prompt_embeds且都带kwargs_typedenoiser_input_fields表示它们是去噪器的条件输入字段__call__中严格遵循get_block_state→check_inputs/encode_prompt→set_block_state→return components, state的范式并用torch.no_grad()装饰。StableDiffusionXLVaeEncoderStepencoders.py组件为vaeAutoencoderKL与image_processorVaeImageProcessorconfigFrozenDict({vae_scale_factor: 8})default_creation_methodfrom_configinputs为imagerequiredTrue、height、width、generator、dtype、preprocess_kwargsintermediate_outputs产出image_latents计算逻辑先image_processor.preprocess(...)预处理再_encode_vae_image编码出潜在表示并处理generator为列表时的逐张采样retrieve_latents。StableDiffusionXLInpaintVaeEncoderStepencoders.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。有三种典型路径从已有模型仓库一键转换ModularPipeline.from_pretrained(stabilityai/stable-diffusion-xl-base-1.0)会自动映射到默认的块集合由MODULAR_PIPELINE_MAPPING决定再调用pipeline.load_components(dtype...)加载模型权重。注意这里加载是惰性的from_pretrained只读取配置、知道每个组件从哪里加载权重在load_components时才真正落盘加载。用init_pipeline()组装自定义块把你的块通过ModularPipelineBlocks.init_pipeline(...)转换源码见 modular_pipeline.py可选传入pretrained_model_name_or_path、components_manager与collection。用组合块编排把多个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_hubFalse)把块的定义写入modular_config.jsonconfig_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发现requiredTrue的输入缺失会抛ValueError: Required input xxx is missing若set_block_state发现intermediate_outputs中的名字未写入block_state会抛Intermediate output xxx is missing in block state——这两条错误信息直接来自源码校验modular_pipeline.py可作为排查指南。十、延伸阅读StatesPipelineState/BlockState与状态交互详解SequentialPipelineBlocks、LoopSequentialPipelineBlocks、AutoPipelineBlocks三类组合块教程ModularPipeline块到可执行流水线的转换与运行Custom Blocks自定义块的分享与 Hub 集成ComponentsManager跨流水线的组件复用OverviewModular Diffusers 文档总览源码参考modular_pipeline.py基类与状态、modular_pipeline_utils.pySpec/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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
