AutoPipelineBlocks

AutoPipelineBlocks são um tipo de bloco múltiplo que contém blocos que suportam diferentes fluxos de trabalho. Ele seleciona automaticamente quais subblocos executar com base na entrada fornecida em tempo de execução. Isso é tipicamente usado para empacotar vários fluxos de trabalho - texto para imagem, imagem para imagem, inpaint - em um único pipeline para conveniência.

Este guia mostra como criar AutoPipelineBlocks.

Crie três ModularPipelineBlocks para texto-para-imagem, imagem-para-imagem e inpainting. Esses representam os diferentes fluxos de trabalho disponíveis no pipeline.

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

Crie uma classe AutoPipelineBlocks que inclua uma lista das classes de sub-blocos e seus respectivos nomes de bloco.

Você também precisa incluir block_trigger_inputs, uma lista de nomes de entrada que acionam o bloco correspondente. Se uma entrada de acionamento for fornecida em tempo de execução, esse bloco será selecionado para execução. Use None para especificar o bloco padrão a ser executado caso nenhuma entrada de acionamento seja detectada.

Por fim, é importante incluir uma description que explique claramente quais entradas acionam qual fluxo de trabalho. Isso ajuda os usuários a entenderem como executar fluxos de trabalho específicos.

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"
        )

É muito importante incluir uma description para evitar qualquer confusão sobre como executar um bloco e quais são as entradas necessárias. Embora os AutoPipelineBlocks sejam convenientes, sua lógica condicional pode ser difícil de compreender se não for explicada adequadamente.

Crie uma instância de AutoImageBlocks.

auto_blocks = AutoImageBlocks()

Para composições mais complexas, como blocos AutoPipelineBlocks aninhados quando usados como sub-blocos em pipelines maiores, use o método get_execution_blocks() para extrair o bloco que é efetivamente executado com base na sua entrada.

auto_blocks.get_execution_blocks(mask=True)

ConditionalPipelineBlocks

AutoPipelineBlocks é um caso especial de ConditionalPipelineBlocks. Enquanto AutoPipelineBlocks seleciona blocos com base em se uma entrada de gatilho é fornecida ou não, ConditionalPipelineBlocks é capaz de selecionar um bloco com base em uma lógica de seleção personalizada fornecida no método select_block.

Aqui está o mesmo exemplo escrito usando ConditionalPipelineBlocks diretamente:

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")

As entradas listadas em block_trigger_inputs são passadas como argumentos de palavra-chave para select_block(). Quando select_block retorna None, ele recorre a default_block_name. Se default_block_name também for None, o bloco condicional inteiro é ignorado — isso é útil para etapas de processamento opcionais que devem ser executadas apenas quando entradas específicas são fornecidas.

Fluxos de trabalho

Pipelines que contêm blocos condicionais (AutoPipelineBlocks ou ConditionalPipelineBlocks) podem suportar múltiplos fluxos de trabalho — por exemplo, nosso pipeline modular SDXL suporta uma dúzia de fluxos de trabalho em um único pipeline. Mas isso também significa que pode ser confuso para os usuários saberem quais fluxos de trabalho são suportados e como executá-los. Para os criadores de pipelines, é útil poder extrair apenas os blocos relevantes para um fluxo de trabalho específico.

Recomendamos definir um _workflow_map para dar um nome a cada workflow e listar explicitamente as entradas que ele requer.

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},
    }

Todos os nossos pipelines modulares integrados vêm com fluxos de trabalho predefinidos. A propriedade available_workflows lista todos os fluxos de trabalho suportados:

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

Recupere um workflow específico com get_workflow para inspecionar e depurar um bloco específico que executa o workflow.

pipeline_blocks.get_workflow("inpaint")