diffusers/v0.43.0/ru/modular_diffusers/auto_pipeline_blocks.md

AutoPipelineBlocks

AutoPipelineBlocks — это многоблочный тип, содержащий блоки, поддерживающие различные рабочие процессы. Он автоматически выбирает, какие подблоки выполнять, на основе входных данных, предоставленных во время выполнения. Обычно это используется для объединения нескольких рабочих процессов — генерация изображения по тексту, преобразование изображения в изображение, inpaint — в единый пайплайн для удобства.

В этом руководстве показано, как создавать 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")