AutoPipelineBlocks

AutoPipelineBlocks es un tipo de bloque múltiple que contiene bloques que admiten diferentes flujos de trabajo. Selecciona automáticamente qué subbloques ejecutar en función de la entrada proporcionada en tiempo de ejecución. Generalmente se utiliza para empaquetar varios flujos de trabajo, como texto a imagen, imagen a imagen e inpaint, en un único pipeline para mayor comodidad.

Esta guía muestra cómo crear AutoPipelineBlocks.

Crea tres ModularPipelineBlocks para texto a imagen, imagen a imagen e inpainting. Estos representan los diferentes flujos de trabajo disponibles en el 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

Crea una clase AutoPipelineBlocks que incluya una lista de las clases de subbloques y sus nombres de bloque correspondientes.

También debe incluir block_trigger_inputs, una lista de nombres de entrada que activan el bloque correspondiente. Si se proporciona una entrada de activación en tiempo de ejecución, ese bloque se selecciona para ejecutarse. Use None para especificar el bloque predeterminado que se ejecutará si no se detectan entradas de activación.

Por último, es importante incluir una description que explique claramente qué entradas activan cada flujo de trabajo. Esto ayuda a los usuarios a comprender cómo ejecutar flujos de trabajo 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"
        )

Es muy importante incluir una description para evitar cualquier confusión sobre cómo ejecutar un bloque y qué entradas se requieren. Aunque AutoPipelineBlocks son convenientes, su lógica condicional puede ser difícil de comprender si no se explica adecuadamente.

Cree una instancia de AutoImageBlocks.

auto_blocks = AutoImageBlocks()

Para composiciones más complejas, como bloques AutoPipelineBlocks anidados cuando se usan como subbloques en pipelines más grandes, use el método get_execution_blocks() para extraer el bloque que se ejecuta realmente en función de su entrada.

auto_blocks.get_execution_blocks(mask=True)

ConditionalPipelineBlocks

AutoPipelineBlocks es un caso especial de ConditionalPipelineBlocks. Mientras que AutoPipelineBlocks selecciona bloques en función de si se proporciona una entrada de activación o no, ConditionalPipelineBlocks puede seleccionar un bloque en función de una lógica de selección personalizada proporcionada en el método select_block.

Aquí está el mismo ejemplo escrito usando ConditionalPipelineBlocks directamente:

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

Las entradas enumeradas en block_trigger_inputs se pasan como argumentos de palabra clave a select_block(). Cuando select_block devuelve None, se recurre a default_block_name. Si default_block_name también es None, se omite el bloque condicional completo; esto es útil para pasos de procesamiento opcionales que solo deben ejecutarse cuando se proporcionan entradas específicas.

Flujos de trabajo

Los pipelines que contienen bloques condicionales (AutoPipelineBlocks o ConditionalPipelineBlocks) pueden admitir múltiples flujos de trabajo; por ejemplo, nuestro pipeline modular de SDXL admite una docena de flujos de trabajo en un solo pipeline. Pero esto también significa que puede resultar confuso para los usuarios saber qué flujos de trabajo se admiten y cómo ejecutarlos. Para los creadores de pipelines, es útil poder extraer solo los bloques relevantes para un flujo de trabajo específico.

Recomendamos definir un _workflow_map para asignar un nombre a cada workflow y listar explícitamente las entradas que requiere.

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 nuestros pipelines modulares integrados incluyen flujos de trabajo predefinidos. La propiedad available_workflows enumera todos los flujos de trabajo admitidos:

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

Recupera un workflow específico con get_workflow para inspeccionar y depurar un bloque específico que ejecuta el workflow.

pipeline_blocks.get_workflow("inpaint")