diffusers/main/ru/modular_diffusers/auto_pipeline_blocks.md

AutoPipelineBlocks

AutoPipelineBlocks are a multi-block type containing blocks that support different workflows. It automatically selects which sub-blocks to run based on the input provided at runtime. This is typically used to package multiple workflows - text-to-image, image-to-image, inpaint - into a single pipeline for convenience.

В этом руководстве показано, как создавать AutoPipelineBlocks.

Создайте три ModularPipelineBlocks для генерации изображений по тексту, преобразования изображения в изображение и инпейнтинга. Они представляют различные рабочие процессы, доступные в конвейере.

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

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

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, содержащий список классов подблоков и соответствующих им имён блоков.

Также необходимо указать block_trigger_inputs — список имен входных параметров, которые активируют соответствующий блок. Если входной параметр-триггер передан во время выполнения, то для запуска выбирается соответствующий блок. Используйте None, чтобы указать блок по умолчанию, который будет выполнен, если входные параметры-триггеры не обнаружены.

Наконец, важно включить description, в котором чётко объясняется, какие входные данные запускают тот или иной workflow. Это помогает пользователям понимать, как запускать конкретные workflow.

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, чтобы избежать путаницы в том, как запускать блок и какие входные данные требуются. Хотя AutoPipelineBlocks удобны, их условная логика может быть трудно понятна, если она не должным образом объяснена.

Создайте экземпляр AutoImageBlocks.

auto_blocks = AutoImageBlocks()

Для более сложных композиций, таких как вложенные блоки AutoPipelineBlocks, когда они используются в качестве подблоков в более крупных конвейерах, используйте метод get_execution_blocks(), чтобы извлечь блок, который фактически выполняется на основе ваших входных данных.

auto_blocks.get_execution_blocks(mask=True)

ConditionalPipelineBlocks

AutoPipelineBlocks является частным случаем ConditionalPipelineBlocks. Если AutoPipelineBlocks выбирает блоки на основе того, предоставлен ли входной триггер, то ConditionalPipelineBlocks позволяет выбирать блоки на основе пользовательской логики выбора, заданной в методе select_block.

Вот тот же пример, написанный с использованием ConditionalPipelineBlocks напрямую:

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 или ConditionalPipelineBlocks), могут поддерживать несколько рабочих процессов — например, наш модульный пайплайн SDXL поддерживает дюжину рабочих процессов в одном пайплайне. Однако это также означает, что пользователям может быть сложно понять, какие рабочие процессы поддерживаются и как их запустить. Для создателей пайплайнов полезно иметь возможность извлекать только те блоки, которые относятся к конкретному рабочему процессу.

Мы рекомендуем определять _workflow_map, чтобы присвоить каждому workflow имя и явно указать необходимые ему входные данные.

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 перечисляет все поддерживаемые рабочие процессы:

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

Используйте get_workflow, чтобы получить конкретный workflow и проверить или отладить блок, который выполняет этот workflow.

pipeline_blocks.get_workflow("inpaint")