diffusers/v0.43.0/zh/modular_diffusers/sequential_pipeline_blocks.md

SequentialPipelineBlocks

SequentialPipelineBlocks 是一种多块类型，它将其他 ModularPipelineBlocks 按顺序组合在一起。数据通过 inputs 和 intermediate_outputs 从一个块线性流向下一个块。SequentialPipelineBlocks 中的每个块通常代表流水线中的一个步骤，通过将它们组合起来，你可以逐步构建出完整的流水线。

本指南将向您展示如何将两个块连接到一个 SequentialPipelineBlocks 中。

创建两个 ModularPipelineBlocks。第一个块 InputBlock 输出一个 batch_size 值，第二个块 ImageEncoderBlock 将 batch_size 用作 inputs。

from diffusers.modular_pipelines import ModularPipelineBlocks, InputParam, OutputParam

class InputBlock(ModularPipelineBlocks):

    @property
    def inputs(self):
        return [
            InputParam(name="prompt", type_hint=list, description="list of text prompts"),
            InputParam(name="num_images_per_prompt", type_hint=int, description="number of images per prompt"),
        ]

    @property
    def intermediate_outputs(self):
        return [
            OutputParam(name="batch_size", description="calculated batch size"),
        ]

    @property
    def description(self):
        return "A block that determines batch_size based on the number of prompts and num_images_per_prompt argument."

    def __call__(self, components, state):
        block_state = self.get_block_state(state)
        batch_size = len(block_state.prompt)
        block_state.batch_size = batch_size * block_state.num_images_per_prompt
        self.set_block_state(state, block_state)
        return components, state

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

class ImageEncoderBlock(ModularPipelineBlocks):

    @property
    def inputs(self):
        return [
            InputParam(name="image", type_hint="PIL.Image", description="raw input image to process"),
            InputParam(name="batch_size", type_hint=int),
        ]

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

    @property
    def description(self):
        return "Encode raw image into its latent presentation"

    def __call__(self, components, state):
        block_state = self.get_block_state(state)
        # Simulate processing the image
        # This will change the state of the image from a PIL image to a tensor for all blocks
        block_state.image = torch.randn(1, 3, 512, 512)
        block_state.batch_size = block_state.batch_size * 2
        block_state.image_latents = torch.randn(1, 4, 64, 64)
        self.set_block_state(state, block_state)
        return components, state

通过定义 SequentialPipelineBlocks 来连接这两个块。在 block_classes 中列出块实例，并在 block_names 中列出它们对应的名称。这些块将按照其在 block_classes 中出现的顺序执行，数据通过 PipelineState 从一个块流向下一个块。

class ImageProcessingStep(SequentialPipelineBlocks):
    """
    # auto_docstring
    """
    model_name = "my_model"
    block_classes = [InputBlock(), ImageEncoderBlock()]
    block_names = ["input", "image_encoder"]

    @property
    def description(self):
        return (
            "Process text prompts and images for the pipeline. It:\n"
            " - Determines the batch size from the prompts.\n"
            " - Encodes the image into latent space."
        )

创建 SequentialPipelineBlocks 时，诸如 inputs、intermediate_outputs 和 expected_components 等属性会自动从子块中聚合，因此无需再次定义。

您应该设置以下几个属性：

• description：我们建议为组装后的块添加描述，以说明该组合步骤的作用。
• model_name：此值自动从子块推导而来，但并不总是正确，因此您可能需要手动覆盖它。
• outputs：默认情况下，它与 intermediate_outputs 相同，但您可以手动设置它以控制文档中显示哪些值。这对于仅显示最终输出而非所有中间值非常有用。

这些属性连同聚合后的 inputs、intermediate_outputs 和 expected_components，用于自动生成 doc 属性。

打印 ImageProcessingStep 块以检查其子块，并使用 doc 查看该块的输入、输出和组件的完整摘要。

blocks = ImageProcessingStep()
print(blocks)
print(blocks.doc)