Diffusers 模块化流水线Modular Pipelines之 Pipeline Blocks 完全指南【免费下载链接】diffusers Diffusers: State-of-the-art diffusion models for image, video, and audio generation in PyTorch.项目地址: https://gitcode.com/GitHub_Trending/di/diffusers本文深入剖析 Diffusers 模块化流水线体系中 5 个核心 Pipeline Blocks 类——ModularPipelineBlocks、SequentialPipelineBlocks、LoopSequentialPipelineBlocks、ConditionalPipelineBlocks与AutoPipelineBlocks。你将理解块Block如何作为流水线的可复用积木声明输入/输出、组合执行顺序、按输入条件分流并掌握在src/diffusers/modular_pipelines/modular_pipeline.py与测试用例中沉淀的真实用法从而能够自定义属于自己的流水线块。一、Pipeline Blocks 在模块化流水线中的定位在 Diffusers 的传统流水线中整条__call__是一条写死的顺序代码文本编码 → 图像编码 → 去噪 → 解码。而模块化流水线Modular Pipeline将这条链拆解为若干独立的Pipeline Blocks流水线块每个块只负责一个职责通过共享的PipelineState传递数据。文本编码块 → 图像编码块 → 去噪块含循环→ 解码块 → 后处理块 │ │ │ │ └────────────┴──────────────┴────────────────┘ 共享 PipelineState 传递中间结果从源码结构看这套体系的核心定义全部集中在 modular_pipeline.py约 2960 行中类作用源码位置ModularPipelineBlocks所有块的基类提供加载/保存/状态读写能力modular_pipeline.py#L326SequentialPipelineBlocks将多个块按顺序组合一次调用串行执行modular_pipeline.py#L974LoopSequentialPipelineBlocks将多个块组合进一个 for 循环如去噪迭代modular_pipeline.py#L1334ConditionalPipelineBlocks根据输入是否存在条件性地选择执行哪个子块modular_pipeline.py#L614AutoPipelineBlocksConditionalPipelineBlocks的自动版1:1 触发输入映射modular_pipeline.py#L913此外PipelineState 与 BlockState 是块之间传递数据的载体而 modular_pipeline_utils.py 提供InputParam、OutputParam、ComponentSpec、ConfigSpec等规格定义与文档生成工具。二、理解块的规格声明inputs、outputs 与 expected_components每个块都会通过属性声明自己需要什么、产出什么、依赖哪些组件这些元数据驱动着流水线的组件加载、参数校验与文档生成。2.1 输入输出参数规格InputParam 与 OutputParam 是两个 dataclass字段包括name参数名若为None而设置了kwargs_type则该输入表示一类参数见下文 2.3type_hint类型提示如str、PIL.Image.Image、torch.Tensordefault默认值required是否必填description描述文本kwargs_type输入分组标签例如denoiser_input_fields。为了避免每个块重复手写这些 dataclass工具模块预置了INPUT_PARAM_TEMPLATES与OUTPUT_PARAM_TEMPLATES两套模板字典modular_pipeline_utils.py#L372 与 modular_pipeline_utils.py#L517覆盖了prompt、negative_prompt、height、width、num_inference_steps、generator、image、mask_image、control_image、prompt_embeds、image_latents等常见输入以及images、videos、latents、prompt_embeds等输出。块内通过InputParam.template(prompt)即可快速创建带默认描述的参数例如模板中prompt被标记为必填num_inference_steps默认 50strength默认 0.9。2.2 组件与配置规格ComponentSpec声明块依赖的模型组件如UNet2DConditionModel、VAE、文本编码器字段含name、type_hint、pretrained_model_name_or_path、subfolder、variant、revision以及创建方式default_creation_methodfrom_config用配置原地构建from_pretrained从 Hub 加载。其load_id属性将加载字段拼接为唯一标识pretrained_model_name_or_path|subfolder|variant|revision空段用null占位。ConfigSpec声明流水线级配置项名称 默认值 描述。基类通过expected_components、expected_configs暴露这些规格ModularPipeline在初始化时据此register_components()并写入 configmodular_pipeline.py#L1790-L1805。2.3 kwargs_type一类输入的聚合某些块的输入是一组动态命名的参数。InputParam支持nameNone, kwargs_typedenoiser_input_fields的形式PipelineState.kwargs_mapping记录每个 kwargs_type 下挂载了哪些键modular_pipeline.py#L173-L188get_by_kwargs()按类型批量取回modular_pipeline.py#L205-L216。例如去噪器的输入字段prompt_embeds、negative_prompt_embeds等都会被聚合进denoiser_input_fields分组从而以统一方式传递给去噪块。三、基类ModularPipelineBlocks所有块的共同能力ModularPipelineBlocks继承自ConfigMixin与PushToHubMixinmodular_pipeline.py#L326因此天然具备配置读写与 Hub 推送能力。其类属性与关键方法如下成员说明config_name modular_config.json块定义的配置文件名save_pretrained时序列化为该文件model_name关联的模型标识用于init_pipeline时通过MODULAR_PIPELINE_MAPPING找到对应流水线类sub_blocksInsertableDict容纳嵌套子块modular_pipeline.py#L350inputs/required_inputs/intermediate_outputs/outputs声明式属性子类实现get_block_state(state)从PipelineState汇总输入与中间值生成BlockStatemodular_pipeline.py#L522set_block_state(state, block_state)将块的中间输出写回PipelineState同时回写被修改的输入modular_pipeline.py#L556from_pretrained(...)从 Hub 或本地目录加载块定义支持auto_map远程代码需trust_remote_codeTrue与requirements校验modular_pipeline.py#L423-L479save_pretrained(...)将块类名写入auto_map并保存modular_config.jsonmodular_pipeline.py#L481-L498init_pipeline(...)基于MODULAR_PIPELINE_MAPPING找到流水线类并实例化ModularPipelinemodular_pipeline.py#L500-L520doc通过make_doc_string生成格式化文档字符串MODULAR_PIPELINE_MAPPINGmodular_pipeline.py#L130-L161把模型名映射到具体的模块化流水线类例如stable-diffusion-xl→StableDiffusionXLModularPipeline、flux2→Flux2ModularPipeline。部分映射还依据配置动态选择_wan_map_fn根据boundary_ratio是否存在在Wan22ModularPipeline与WanModularPipeline之间切换_krea2_map_fn根据is_distilled选择 Turbo 变体modular_pipeline.py#L80-L128。四、SequentialPipelineBlocks顺序执行的组合器SequentialPipelineBlocks将多个块类或实例按声明顺序组装调用时逐一执行modular_pipeline.py#L1150-L1163torch.no_grad() def __call__(self, pipeline, state): for block_name, block in self.sub_blocks.items(): pipeline, state block(pipeline, state) # 前一块的输出状态喂给后一块 return pipeline, state子类只需声明两个类属性class MySequentialBlocks(SequentialPipelineBlocks): block_classes [TextEncoderBlock, VaeEncoderBlock, DenoiseBlock, VaeDecoderBlock] block_names [text_encoder, vae_encoder, denoise, vae_decoder]4.1 输入/输出推导逻辑_get_inputs()modular_pipeline.py#L1090-L1112按顺序扫描所有子块某个输入若已被前序块的输出满足就不再对外暴露若子块是无默认值的条件块可能被跳过则其输出不参与满足后续输入的计算。required_inputsmodular_pipeline.py#L1120-L1130取所有子块必填输入的并集。intermediate_outputsmodular_pipeline.py#L1133-L1142只列出新产生的中间变量已被自身输入消费的同名变量会被过滤。4.2get_execution_blocks静态解析实际执行路径get_execution_blocks(**kwargs)modular_pipeline.py#L1189-L1238返回一个只包含会真正执行的块的SequentialPipelineBlocks是调试与 workflow 裁剪的关键 API遍历时维护active_inputs字典遇到条件块调用其get_execution_blocks解析出被选中的分支每解析出一个叶子块就把它的intermediate_outputs以True注入active_inputs使后续依赖该中间值的条件块也能正确解析因此条件判断必须只依赖输入的存在与否None / 非 None而非具体数值——这一点在ConditionalPipelineBlocks.select_block的文档字符串中有明确说明modular_pipeline.py#L765-L768。4.3 Workflow预定义的执行路径SequentialPipelineBlocks是唯一支持 workflow 的块类型。子类可定义_workflow_map将 workflow 名映射到一组触发输入# 来自 flux2 的真实实现简化 # src/diffusers/modular_pipelines/flux2/modular_blocks_flux2.py#L341-L346 block_classes AUTO_BLOCKS.values() block_names AUTO_BLOCKS.keys() _workflow_map { text2image: {prompt: True}, image_conditioned: {image: True, prompt: True}, }available_workflows返回_workflow_map的全部键get_workflow(name)modular_pipeline.py#L1025-L1044先用该 workflow 的触发输入调用get_execution_blocks裁剪出执行块若_workflow_map未定义则抛出NotImplementedError。注意触发输入可写成 tuple同一 workflow 的多种别名拼写如image或last_image实际裁剪时取第一个。ModularPipeline.__init__支持workflow参数一旦指定流水线会blocks.get_workflow(workflow)裁剪块集合使得load_components()只加载该 workflow 真正用到的组件modular_pipeline.py#L1747-L1750可显著减少加载开销。4.4from_blocks_dict从字典构建from_blocks_dict 允许从{块名: 块类或实例}动态组装一个顺序块并可选覆写description。get_execution_blocks内部正是用它来构造裁剪后的结果。五、LoopSequentialPipelineBlocks循环体内的顺序执行LoopSequentialPipelineBlocks用于把一组块包进循环典型场景是多次去噪迭代其__call__必须由子类实现基类直接抛NotImplementedErrormodular_pipeline.py#L1535-L1536循环体单次执行由loop_step(components, state, **kwargs)完成modular_pipeline.py#L1521-L1533。与顺序块的关键差异额外声明一组循环专属规格loop_inputs、loop_required_inputs、loop_intermediate_outputs、loop_expected_components、loop_expected_configsmodular_pipeline.py#L1351-L1380_get_inputs()先合并loop_inputs再合并子块输入modular_pipeline.py#L1408-L1429required_inputs是子块必填 ∪ 循环必填outputs取最后一个子块的中间输出modular_pipeline.py#L1464-L1467约束循环块的子块必须是叶子块不能再嵌套sub_blocks否则__init__抛ValueErrormodular_pipeline.py#L1487-L1493。此外它还内置了进度条工具progress_bar(iterableNone, totalNone)带torch.compiler.disable装饰避免编译期冲突与set_progress_bar_config(**kwargs)modular_pipeline.py#L1606-L1623。六、ConditionalPipelineBlocks按输入条件分流ConditionalPipelineBlocksmodular_pipeline.py#L614根据输入是否存在来选择要运行的子块。四个关键类属性属性含义block_classes候选块类列表长度必须等于block_namesblock_names每个块的名称block_trigger_inputs触发输入的候选名单select_block()依据它们做判断default_block_name无触发匹配时运行的默认块为None时整个条件块可被跳过6.1 实现select_block子类必须实现select_block(**kwargs)modular_pipeline.py#L760-L776返回要执行的块名或NoneNone表示回退到默认块/跳过。测试文件 test_conditional_pipeline_blocks.py 给出了经典写法class ConditionalImageBlocks(ConditionalPipelineBlocks): block_classes [InpaintBlock, ImageToImageBlock, TextToImageBlock] block_names [inpaint, img2img, text2img] block_trigger_inputs [mask, image] default_block_name text2img def select_block(self, maskNone, imageNone): if mask is not None: return inpaint if image is not None: return img2img return None # 回退到 default_block_name - text2img6.2 执行与校验语义__init__会校验block_classes与block_names长度一致且default_block_name必须是block_names之一modular_pipeline.py#L647-L654__call__modular_pipeline.py#L778-L802先收集触发输入 → 调select_block→ 无匹配则用默认块 → 仍为None则记录日志并跳过执行子块出错时附带块名与完整 traceback 重新抛出required_inputsmodular_pipeline.py#L683-L696有默认块时取所有子块必填输入的交集只有所有分支都需要才对外必填无默认块时返回空列表整个块可跳过inputs/intermediate_outputs/outputs通过combine_inputs/combine_outputs合并各子块声明当不同子块对同一输入声明了不同默认值时defaults_by_block会记录按块区分的默认值运行期由真正执行的块在get_block_state中自行解析modular_pipeline.py#L531-L537。6.3 嵌套条件块与递归解析get_execution_blocksmodular_pipeline.py#L804-L835会递归解析嵌套的条件块直到抵达叶子块无sub_blocks或LoopSequentialPipelineBlocks。测试中的NestedImageBlockstest_conditional_pipeline_blocks.py#L307-L319演示了把AutoImageBlocks作为子块的嵌套结构。__repr__还会打印Trigger Inputs列表提醒用户存在运行期分流的块。七、AutoPipelineBlocks自动分流的速成版AutoPipelineBlocksmodular_pipeline.py#L913是ConditionalPipelineBlocks的特化省去手写select_blockblock_classes、block_names、block_trigger_inputs三者长度必须相等构成一一映射用None作为某个块的触发输入表示它是默认块default_block_name会被自动推导若手工设置则抛ValueErrormodular_pipeline.py#L951-L955选择规则是顺序优先返回第一个触发输入非None的块modular_pipeline.py#L966-L971。# 源码 docstring 中的示例 class MyAutoBlock(AutoPipelineBlocks): block_classes [InpaintEncoderBlock, ImageEncoderBlock, TextEncoderBlock] block_names [inpaint, img2img, text2img] block_trigger_inputs [mask_image, image, None] # text2img 是默认块选择语义只要传入了mask_image无论是否传入image都运行 inpaint未传mask_image但传了image运行 img2img其余情况运行默认块 text2img。测试 TestAutoPipelineBlocksSelectBlock 验证了上述优先级行为。八、状态流PipelineState 与 BlockState 如何协作块的执行围绕两个状态对象展开PipelineStatemodular_pipeline.py#L164流水线级共享状态内部持有一个values字典与kwargs_mapping分组索引提供set/get/get_by_kwargs/to_dict并支持属性访问state.foo等价于state.get(foo)__repr__会把张量显示为Tensor(dtype..., shape...)方便调试大模型流水线。BlockStatemodular_pipeline.py#L254单个块的工作台由get_block_state生成、支持属性与下标访问、as_dict()导出__repr__对张量/张量列表/张量字典做了格式化。块内典型的三步模式测试中InpaintBlock的写法见 test_conditional_pipeline_blocks.py#L145-L149def __call__(self, components, state): block_state self.get_block_state(state) # 1. 取本块需要的输入 # ... 在此执行本块逻辑可写 block_state.new_var ... self.set_block_state(state, block_state) # 2. 把中间输出写回流水线状态 return components, stateset_block_state通过同一性比较is判断对象是否被修改过避免无谓回写modular_pipeline.py#L566-L584。九、从零定义一个条件流水线块实战模板综合以上内容一个完整的自定义条件块模板如下结构参照 test_conditional_pipeline_blocks.pyfrom diffusers.modular_pipelines.modular_pipeline import ( ConditionalPipelineBlocks, SequentialPipelineBlocks, ) class MyEncodeBlock(SequentialPipelineBlocks): block_classes [TextEncoderBlock, VaeEncoderBlock] block_names [text, vae] class MyDenoiseBlock(SequentialPipelineBlocks): block_classes [DenoiseBlock, VaeDecoderBlock] block_names [denoise, decode] class MyConditionalBlocks(ConditionalPipelineBlocks): block_classes [MyDenoiseBlock, MyEncodeBlock] block_names [generate, encode_only] block_trigger_inputs [prompt] default_block_name generate def select_block(self, promptNone): return generate if prompt is not None else None自定义块定义好后可以通过save_pretrained序列化为modular_config.json含auto_map与requirements再通过from_pretrained加载复用整个块还可通过init_pipeline接入ModularPipeline运行。相关运行期逻辑load_components、save_pretrained的组件分片与 Hub 推送、model card 生成等位于同一文件后半部分modular_pipeline.py#L1950-L2086。十、验证与测试依据仓库中与 Pipeline Blocks 直接相关的测试tests/modular_pipelines/test_conditional_pipeline_blocks.py覆盖select_block的触发优先级、get_execution_blocks的静态解析默认分支、跳过分支、嵌套结构、AutoPipelineBlocks的自动选择与默认块推导tests/modular_pipelines/cosmos/test_modular_pipeline_cosmos3.py验证多触发输入组合action/image/video下的块选择tests/modular_pipelines/ltx2/test_modular_pipeline_ltx25.py遍历 workflow 并断言裁剪后的子块结构tests/modular_pipelines/test_modular_pipeline_loading.py覆盖from_pretrained/save_pretrained等加载保存链路。真实生产实现可参考 src/diffusers/modular_pipelines/flux2/modular_blocks_flux2.py 中的Flux2AutoBlocksAutoPipelineBlocks子类 _workflow_map以及stable_diffusion_xl、wan、ltx2等模型目录下的同名modular_blocks_*.py文件——它们展示了这套块体系在真实文生图、图生图、视频生成流水线中的落地形态。结语Pipeline Blocks 把 Diffusers 流水线从一段写死的__call__重构为一组可声明、可组合、可复用、可裁剪的积木SequentialPipelineBlocks负责线性编排LoopSequentialPipelineBlocks负责循环迭代ConditionalPipelineBlocks与AutoPipelineBlocks负责按输入分流而ModularPipelineBlocks基类统一提供加载、保存与状态读写能力。理解这 5 个类的职责与协作方式是阅读 Diffusers 模块化流水线源码、乃至自定义流水线块的第一步。【免费下载链接】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），仅供参考