Buckets:

hf-doc-build
/

doc

Files

xet

hf-doc-build/doc / diffusers /v0.43.0 /zh /modular_diffusers /auto_pipeline_blocks.md

mishig

2 days ago

preview code

download

raw

9.48 kB

# AutoPipelineBlocks

[AutoPipelineBlocks](/docs/diffusers/v0.43.0/zh/api/modular_diffusers/pipeline_blocks#diffusers.AutoPipelineBlocks) 是一种多块类型，包含支持不同工作流的块。它会根据运行时提供的输入自动选择要运行的子块。这通常用于将多个工作流（如文生图、图生图、局部重绘）打包到单个流水线中，以方便使用。

本指南展示了如何创建 [AutoPipelineBlocks](/docs/diffusers/v0.43.0/zh/api/modular_diffusers/pipeline_blocks#diffusers.AutoPipelineBlocks)。

创建三个 [ModularPipelineBlocks](/docs/diffusers/v0.43.0/zh/api/modular_diffusers/pipeline_blocks#diffusers.ModularPipelineBlocks)，分别用于文本生成图像、图像生成图像和图像修复。这些代表了流水线中可用的不同工作流。

```py
import torch
from diffusers.modular_pipelines import ModularPipelineBlocks, InputParam, OutputParam
class TextToImageBlock(ModularPipelineBlocks):
    model_name = "text2img"
    @property
    def inputs(self):
        return [InputParam(name="prompt")]
    @property
    def intermediate_outputs(self):
        return []
    @property
    def description(self):
        return "I'm a text-to-image workflow!"
    def __call__(self, components, state):
        block_state = self.get_block_state(state)
        print("running the text-to-image workflow")
        # Add your text-to-image logic here
        # For example: generate image from prompt
        self.set_block_state(state, block_state)
        return components, state
```

```py
class ImageToImageBlock(ModularPipelineBlocks):
    model_name = "img2img"
    @property
    def inputs(self):
        return [InputParam(name="prompt"), InputParam(name="image")]
    @property
    def intermediate_outputs(self):
        return []
    @property
    def description(self):
        return "I'm an image-to-image workflow!"
    def __call__(self, components, state):
        block_state = self.get_block_state(state)
        print("running the image-to-image workflow")
        # Add your image-to-image logic here
        # For example: transform input image based on prompt
        self.set_block_state(state, block_state)
        return components, state
```

```py
class InpaintBlock(ModularPipelineBlocks):
    model_name = "inpaint"
    @property
    def inputs(self):
        return [InputParam(name="prompt"), InputParam(name="image"), InputParam(name="mask")]
    @property
    def intermediate_outputs(self):
        return []
    @property
    def description(self):
        return "I'm an inpaint workflow!"
    def __call__(self, components, state):
        block_state = self.get_block_state(state)
        print("running the inpaint workflow")
        # Add your inpainting logic here
        # For example: fill masked areas based on prompt
        self.set_block_state(state, block_state)
        return components, state
```

创建一个 [AutoPipelineBlocks](/docs/diffusers/v0.43.0/zh/api/modular_diffusers/pipeline_blocks#diffusers.AutoPipelineBlocks) 类，其中包含子块类及其对应块名的列表。

您还需要包含 `block_trigger_inputs`，这是一个输入名称列表，用于触发对应的 block。如果在运行时提供了某个触发输入，则会选择该 block 来运行。使用 `None` 来指定在未检测到任何触发输入时要运行的默认 block。

最后，重要的是包含一个 `description`，清楚地说明哪些输入会触发哪些工作流。这有助于用户了解如何运行特定的工作流。

```py
from diffusers.modular_pipelines import AutoPipelineBlocks
class AutoImageBlocks(AutoPipelineBlocks):
    # List of sub-block classes to choose from
    block_classes = [InpaintBlock, ImageToImageBlock, TextToImageBlock]
    # Names for each block in the same order
    block_names = ["inpaint", "img2img", "text2img"]
    # Trigger inputs that determine which block to run
    # - "mask" triggers inpaint workflow
    # - "image" triggers img2img workflow (but only if mask is not provided)
    # - if none of above, runs the text2img workflow (default)
    block_trigger_inputs = ["mask", "image", None]
    @property
    def description(self):
        return (
            "Pipeline generates images given different types of conditions!\n"
            + "This is an auto pipeline block that works for text2img, img2img and inpainting tasks.\n"
            + " - inpaint workflow is run when `mask` is provided.\n"
            + " - img2img workflow is run when `image` is provided (but only when `mask` is not provided).\n"
            + " - text2img workflow is run when neither `image` nor `mask` is provided.\n"
        )
```

包含 `description` 以明确如何运行该 block 以及需要哪些输入，这一点**非常**重要。虽然 [AutoPipelineBlocks](/docs/diffusers/v0.43.0/zh/api/modular_diffusers/pipeline_blocks#diffusers.AutoPipelineBlocks) 用起来很方便，但如果缺乏清晰的说明，其条件逻辑可能难以理解。

创建一个 `AutoImageBlocks` 实例。

```py
auto_blocks = AutoImageBlocks()
```

对于更复杂的组合，例如在大型流水线中作为子块使用的嵌套 [AutoPipelineBlocks](/docs/diffusers/v0.43.0/zh/api/modular_diffusers/pipeline_blocks#diffusers.AutoPipelineBlocks) 块，请使用 [get_execution_blocks()](/docs/diffusers/v0.43.0/zh/api/modular_diffusers/pipeline_blocks#diffusers.SequentialPipelineBlocks.get_execution_blocks) 方法提取根据输入实际执行的块。

```py
auto_blocks.get_execution_blocks(mask=True)
```

## ConditionalPipelineBlocks

[AutoPipelineBlocks](/docs/diffusers/v0.43.0/zh/api/modular_diffusers/pipeline_blocks#diffusers.AutoPipelineBlocks) 是 [ConditionalPipelineBlocks](/docs/diffusers/v0.43.0/zh/api/modular_diffusers/pipeline_blocks#diffusers.ConditionalPipelineBlocks) 的一个特例。[AutoPipelineBlocks](/docs/diffusers/v0.43.0/zh/api/modular_diffusers/pipeline_blocks#diffusers.AutoPipelineBlocks) 根据是否提供了触发输入来选择块，而 [ConditionalPipelineBlocks](/docs/diffusers/v0.43.0/zh/api/modular_diffusers/pipeline_blocks#diffusers.ConditionalPipelineBlocks) 则能够基于在 `select_block` 方法中提供的自定义选择逻辑来选择块。

以下是直接使用 [ConditionalPipelineBlocks](/docs/diffusers/v0.43.0/zh/api/modular_diffusers/pipeline_blocks#diffusers.ConditionalPipelineBlocks) 编写的相同示例：

```py
from diffusers.modular_pipelines import ConditionalPipelineBlocks
class AutoImageBlocks(ConditionalPipelineBlocks):
    block_classes = [InpaintBlock, ImageToImageBlock, TextToImageBlock]
    block_names = ["inpaint", "img2img", "text2img"]
    block_trigger_inputs = ["mask", "image"]
    default_block_name = "text2img"
    @property
    def description(self):
        return (
            "Pipeline generates images given different types of conditions!\n"
            + "This is an auto pipeline block that works for text2img, img2img and inpainting tasks.\n"
            + " - inpaint workflow is run when `mask` is provided.\n"
            + " - img2img workflow is run when `image` is provided (but only when `mask` is not provided).\n"
            + " - text2img workflow is run when neither `image` nor `mask` is provided.\n"
        )
    def select_block(self, mask=None, image=None) -> str | None:
        if mask is not None:
            return "inpaint"
        if image is not None:
            return "img2img"
        return None  # falls back to default_block_name ("text2img")
```

`block_trigger_inputs` 中列出的输入会作为关键字参数传递给 `select_block()`。当 `select_block` 返回 `None` 时，将回退到 `default_block_name`。如果 `default_block_name` 也是 `None`，则整个条件块会被跳过——这对于仅在提供了特定输入时才应执行的可选处理步骤非常有用。

## 工作流程

包含条件块（[AutoPipelineBlocks](/docs/diffusers/v0.43.0/zh/api/modular_diffusers/pipeline_blocks#diffusers.AutoPipelineBlocks) 或 [ConditionalPipelineBlocks](/docs/diffusers/v0.43.0/zh/api/modular_diffusers/pipeline_blocks#diffusers.ConditionalPipelineBlocks)）的 Pipeline 可以支持多种工作流——例如，我们的 SDXL 模块化 Pipeline 在单个 Pipeline 中支持十余种工作流。但这也意味着，用户可能难以了解支持哪些工作流以及如何运行它们。对于 Pipeline 构建者而言，能够仅提取与特定工作流相关的块会非常有用。

我们建议定义一个 `_workflow_map`，为每个工作流指定名称，并明确列出其所需的输入。

```py
from diffusers.modular_pipelines import SequentialPipelineBlocks
class MyPipelineBlocks(SequentialPipelineBlocks):
    block_classes = [TextEncoderBlock, AutoImageBlocks, DecodeBlock]
    block_names = ["text_encoder", "auto_image", "decode"]
    _workflow_map = {
        "text2image": {"prompt": True},
        "image2image": {"image": True, "prompt": True},
        "inpaint": {"mask": True, "image": True, "prompt": True},
    }
```

我们所有内置的模块化流水线都带有预定义的工作流。`available_workflows` 属性列出了所有支持的工作流：

```py
pipeline_blocks = MyPipelineBlocks()
pipeline_blocks.available_workflows
# ['text2image', 'image2image', 'inpaint']
```

使用 `get_workflow` 获取特定工作流，以检查并调试执行该工作流的特定代码块。

```py
pipeline_blocks.get_workflow("inpaint")
```

Xet Storage Details

Size:: 9.48 kB
Xet hash:: 2ad69c2b197d3a69d5767ce8f185f7ece5da442582fd45d8ca3ccbb18adcdd42

Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.

	# AutoPipelineBlocks

	[AutoPipelineBlocks](/docs/diffusers/v0.43.0/zh/api/modular_diffusers/pipeline_blocks#diffusers.AutoPipelineBlocks) 是一种多块类型，包含支持不同工作流的块。它会根据运行时提供的输入自动选择要运行的子块。这通常用于将多个工作流（如文生图、图生图、局部重绘）打包到单个流水线中，以方便使用。

	本指南展示了如何创建 [AutoPipelineBlocks](/docs/diffusers/v0.43.0/zh/api/modular_diffusers/pipeline_blocks#diffusers.AutoPipelineBlocks)。

	创建三个 [ModularPipelineBlocks](/docs/diffusers/v0.43.0/zh/api/modular_diffusers/pipeline_blocks#diffusers.ModularPipelineBlocks)，分别用于文本生成图像、图像生成图像和图像修复。这些代表了流水线中可用的不同工作流。

	```py
	import torch
	from diffusers.modular_pipelines import ModularPipelineBlocks, InputParam, OutputParam

	class TextToImageBlock(ModularPipelineBlocks):
	model_name = "text2img"

	@property
	def inputs(self):
	return [InputParam(name="prompt")]

	@property
	def intermediate_outputs(self):
	return []

	@property
	def description(self):
	return "I'm a text-to-image workflow!"

	def __call__(self, components, state):
	block_state = self.get_block_state(state)
	print("running the text-to-image workflow")
	# Add your text-to-image logic here
	# For example: generate image from prompt
	self.set_block_state(state, block_state)
	return components, state
	```

	```py
	class ImageToImageBlock(ModularPipelineBlocks):
	model_name = "img2img"

	@property
	def inputs(self):
	return [InputParam(name="prompt"), InputParam(name="image")]

	@property
	def intermediate_outputs(self):
	return []

	@property
	def description(self):
	return "I'm an image-to-image workflow!"

	def __call__(self, components, state):
	block_state = self.get_block_state(state)
	print("running the image-to-image workflow")
	# Add your image-to-image logic here
	# For example: transform input image based on prompt
	self.set_block_state(state, block_state)
	return components, state
	```

	```py
	class InpaintBlock(ModularPipelineBlocks):
	model_name = "inpaint"

	@property
	def inputs(self):
	return [InputParam(name="prompt"), InputParam(name="image"), InputParam(name="mask")]

	@property
	def intermediate_outputs(self):
	return []

	@property
	def description(self):
	return "I'm an inpaint workflow!"

	def __call__(self, components, state):
	block_state = self.get_block_state(state)
	print("running the inpaint workflow")
	# Add your inpainting logic here
	# For example: fill masked areas based on prompt
	self.set_block_state(state, block_state)
	return components, state
	```

	创建一个 [AutoPipelineBlocks](/docs/diffusers/v0.43.0/zh/api/modular_diffusers/pipeline_blocks#diffusers.AutoPipelineBlocks) 类，其中包含子块类及其对应块名的列表。

	您还需要包含 `block_trigger_inputs`，这是一个输入名称列表，用于触发对应的 block。如果在运行时提供了某个触发输入，则会选择该 block 来运行。使用 `None` 来指定在未检测到任何触发输入时要运行的默认 block。

	最后，重要的是包含一个 `description`，清楚地说明哪些输入会触发哪些工作流。这有助于用户了解如何运行特定的工作流。

	```py
	from diffusers.modular_pipelines import AutoPipelineBlocks

	class AutoImageBlocks(AutoPipelineBlocks):
	# List of sub-block classes to choose from
	block_classes = [InpaintBlock, ImageToImageBlock, TextToImageBlock]
	# Names for each block in the same order
	block_names = ["inpaint", "img2img", "text2img"]
	# Trigger inputs that determine which block to run
	# - "mask" triggers inpaint workflow
	# - "image" triggers img2img workflow (but only if mask is not provided)
	# - if none of above, runs the text2img workflow (default)
	block_trigger_inputs = ["mask", "image", None]

	@property
	def description(self):
	return (
	"Pipeline generates images given different types of conditions!\n"
	+ "This is an auto pipeline block that works for text2img, img2img and inpainting tasks.\n"
	+ " - inpaint workflow is run when `mask` is provided.\n"
	+ " - img2img workflow is run when `image` is provided (but only when `mask` is not provided).\n"
	+ " - text2img workflow is run when neither `image` nor `mask` is provided.\n"
	)
	```

	包含 `description` 以明确如何运行该 block 以及需要哪些输入，这一点非常重要。虽然 [AutoPipelineBlocks](/docs/diffusers/v0.43.0/zh/api/modular_diffusers/pipeline_blocks#diffusers.AutoPipelineBlocks) 用起来很方便，但如果缺乏清晰的说明，其条件逻辑可能难以理解。

	创建一个 `AutoImageBlocks` 实例。

	```py
	auto_blocks = AutoImageBlocks()
	```

	对于更复杂的组合，例如在大型流水线中作为子块使用的嵌套 [AutoPipelineBlocks](/docs/diffusers/v0.43.0/zh/api/modular_diffusers/pipeline_blocks#diffusers.AutoPipelineBlocks) 块，请使用 [get_execution_blocks()](/docs/diffusers/v0.43.0/zh/api/modular_diffusers/pipeline_blocks#diffusers.SequentialPipelineBlocks.get_execution_blocks) 方法提取根据输入实际执行的块。

	```py
	auto_blocks.get_execution_blocks(mask=True)
	```

	## ConditionalPipelineBlocks

	[AutoPipelineBlocks](/docs/diffusers/v0.43.0/zh/api/modular_diffusers/pipeline_blocks#diffusers.AutoPipelineBlocks) 是 [ConditionalPipelineBlocks](/docs/diffusers/v0.43.0/zh/api/modular_diffusers/pipeline_blocks#diffusers.ConditionalPipelineBlocks) 的一个特例。[AutoPipelineBlocks](/docs/diffusers/v0.43.0/zh/api/modular_diffusers/pipeline_blocks#diffusers.AutoPipelineBlocks) 根据是否提供了触发输入来选择块，而 [ConditionalPipelineBlocks](/docs/diffusers/v0.43.0/zh/api/modular_diffusers/pipeline_blocks#diffusers.ConditionalPipelineBlocks) 则能够基于在 `select_block` 方法中提供的自定义选择逻辑来选择块。

	以下是直接使用 [ConditionalPipelineBlocks](/docs/diffusers/v0.43.0/zh/api/modular_diffusers/pipeline_blocks#diffusers.ConditionalPipelineBlocks) 编写的相同示例：

	```py
	from diffusers.modular_pipelines import ConditionalPipelineBlocks

	class AutoImageBlocks(ConditionalPipelineBlocks):
	block_classes = [InpaintBlock, ImageToImageBlock, TextToImageBlock]
	block_names = ["inpaint", "img2img", "text2img"]
	block_trigger_inputs = ["mask", "image"]
	default_block_name = "text2img"

	@property
	def description(self):
	return (
	"Pipeline generates images given different types of conditions!\n"
	+ "This is an auto pipeline block that works for text2img, img2img and inpainting tasks.\n"
	+ " - inpaint workflow is run when `mask` is provided.\n"
	+ " - img2img workflow is run when `image` is provided (but only when `mask` is not provided).\n"
	+ " - text2img workflow is run when neither `image` nor `mask` is provided.\n"
	)

	def select_block(self, mask=None, image=None) -> str \| None:
	if mask is not None:
	return "inpaint"
	if image is not None:
	return "img2img"
	return None # falls back to default_block_name ("text2img")
	```

	`block_trigger_inputs` 中列出的输入会作为关键字参数传递给 `select_block()`。当 `select_block` 返回 `None` 时，将回退到 `default_block_name`。如果 `default_block_name` 也是 `None`，则整个条件块会被跳过——这对于仅在提供了特定输入时才应执行的可选处理步骤非常有用。

	## 工作流程

	包含条件块（[AutoPipelineBlocks](/docs/diffusers/v0.43.0/zh/api/modular_diffusers/pipeline_blocks#diffusers.AutoPipelineBlocks) 或 [ConditionalPipelineBlocks](/docs/diffusers/v0.43.0/zh/api/modular_diffusers/pipeline_blocks#diffusers.ConditionalPipelineBlocks)）的 Pipeline 可以支持多种工作流——例如，我们的 SDXL 模块化 Pipeline 在单个 Pipeline 中支持十余种工作流。但这也意味着，用户可能难以了解支持哪些工作流以及如何运行它们。对于 Pipeline 构建者而言，能够仅提取与特定工作流相关的块会非常有用。

	我们建议定义一个 `_workflow_map`，为每个工作流指定名称，并明确列出其所需的输入。

	```py
	from diffusers.modular_pipelines import SequentialPipelineBlocks

	class MyPipelineBlocks(SequentialPipelineBlocks):
	block_classes = [TextEncoderBlock, AutoImageBlocks, DecodeBlock]
	block_names = ["text_encoder", "auto_image", "decode"]

	_workflow_map = {
	"text2image": {"prompt": True},
	"image2image": {"image": True, "prompt": True},
	"inpaint": {"mask": True, "image": True, "prompt": True},
	}
	```

	我们所有内置的模块化流水线都带有预定义的工作流。`available_workflows` 属性列出了所有支持的工作流：

	```py
	pipeline_blocks = MyPipelineBlocks()
	pipeline_blocks.available_workflows
	# ['text2image', 'image2image', 'inpaint']
	```

	使用 `get_workflow` 获取特定工作流，以检查并调试执行该工作流的特定代码块。

	```py
	pipeline_blocks.get_workflow("inpaint")
	```