Introducción

Modular Diffusers es un marco de trabajo para construir rápidamente pipelines flexibles y personalizables. Estos pipelines pueden ir más allá de lo que pueden hacer los DiffusionPipeline estándar. En el núcleo de Modular Diffusers se encuentran los ModularPipelineBlocks, que pueden combinarse con otros bloques para adaptarse a nuevos flujos de trabajo. Los bloques se convierten en un ModularPipeline, una interfaz amigable para el usuario para ejecutar tareas de generación.

Esta guía te muestra cómo ejecutar un pipeline modular, comprender su estructura y personalizarlo modificando los bloques que lo componen. Si estás familiarizado con DiffusionPipeline, lee primero Coming from DiffusionPipeline para saber qué funciona de manera diferente.

Ejecutar un pipeline

ModularPipeline es la interfaz principal para cargar, ejecutar y gestionar pipelines modulares.

import torch
from diffusers import ModularPipeline, ComponentsManager

# Use ComponentsManager to enable auto CPU offloading for memory efficiency
manager = ComponentsManager()
manager.enable_auto_cpu_offload(device="cuda:0")

pipe = ModularPipeline.from_pretrained("Qwen/Qwen-Image", components_manager=manager)
pipe.load_components(dtype=torch.bfloat16)

image = pipe(
    prompt="cat wizard with red hat, gandalf, lord of the rings, detailed, fantasy, cute, adorable, Pixar, Disney",
).images[0]
image

from_pretrained() utiliza carga diferida: lee la configuración para saber desde dónde cargar cada componente, pero no carga los pesos del modelo hasta que llamas a load_components(). Esto te da control sobre cuándo y cómo se cargan los componentes.

ComponentsManager con enable_auto_cpu_offload mueve automáticamente los modelos entre la CPU y la GPU según sea necesario, reduciendo el uso de memoria para modelos grandes como Qwen-Image. Obtenga más información en la guía de ComponentsManager.

Si no necesita el descargado, elimine el argumento components_manager y mueva el pipeline a su dispositivo manualmente con to("cuda").

Obtenga más información sobre la creación y carga de pipelines en las guías Creating a pipeline y Loading components.

Comprender la estructura

Un ModularPipeline tiene dos partes: una definición (los bloques) y un estado (los componentes y configuraciones cargados).

Imprima el pipeline para ver su estado: los componentes y su estado de carga y configuración.

print(pipe)

QwenImageModularPipeline {
  "_blocks_class_name": "QwenImageAutoBlocks",
  "_class_name": "QwenImageModularPipeline",
  "_diffusers_version": "0.37.0.dev0",
  "transformer": [
    "diffusers",
    "QwenImageTransformer2DModel",
    {
      "pretrained_model_name_or_path": "Qwen/Qwen-Image",
      "revision": null,
      "subfolder": "transformer",
      "type_hint": [
        "diffusers",
        "QwenImageTransformer2DModel"
      ],
      "variant": null
    }
  ],
  ...
}

Acceda a la definición a través de pipe.blocks: este es el ModularPipelineBlocks que define los flujos de trabajo, las entradas, las salidas y la lógica de cálculo del pipeline.

print(pipe.blocks)

QwenImageAutoBlocks(
  Class: SequentialPipelineBlocks

  Description: Auto Modular pipeline for text-to-image, image-to-image, inpainting, and controlnet tasks using QwenImage.
      
      Supported workflows:
        - `text2image`: requires `prompt`
        - `image2image`: requires `prompt`, `image`
        - `inpainting`: requires `prompt`, `mask_image`, `image`
        - `controlnet_text2image`: requires `prompt`, `control_image`
        ...

  Components:
      text_encoder (`Qwen2_5_VLForConditionalGeneration`)
      vae (`AutoencoderKLQwenImage`)
      transformer (`QwenImageTransformer2DModel`)
      ...

  Sub-Blocks:
    [0] text_encoder (QwenImageAutoTextEncoderStep)
    [1] vae_encoder (QwenImageAutoVaeEncoderStep)
    [2] controlnet_vae_encoder (QwenImageOptionalControlNetVaeEncoderStep)
    [3] denoise (QwenImageAutoCoreDenoiseStep)
    [4] decode (QwenImageAutoDecodeStep)
)

La salida devuelve:

• Los flujos de trabajo admitidos (text2image, image2image, inpainting, etc.)
• Los subbloques de los que está compuesto (text_encoder, vae_encoder, denoise, decode)

Flujos de trabajo

Este pipeline admite varios flujos de trabajo y adapta su comportamiento en función de las entradas que proporciones. Por ejemplo, si pasas image al pipeline, ejecuta un flujo de trabajo de imagen a imagen en lugar de texto a imagen. Obtén más información sobre cómo funciona esto por debajo en la guía de AutoPipelineBlocks.

from diffusers.utils import load_image

input_image = load_image("https://github.com/Trgtuan10/Image_storage/blob/main/cute_cat.png?raw=true")

image = pipe(
    prompt="cat wizard with red hat, gandalf, lord of the rings, detailed, fantasy, cute, adorable, Pixar, Disney",
    image=input_image,
).images[0]

Use get_workflow() para extraer los bloques de un workflow específico. Pasa el nombre del workflow (por ejemplo, "image2image", "inpainting", "controlnet_text2image") para obtener solo los bloques relevantes para ese workflow. Esto es útil cuando quieres personalizar o depurar un workflow específico. Puedes consultar pipe.blocks.available_workflows para ver todos los workflows disponibles.

img2img_blocks = pipe.blocks.get_workflow("image2image")

Subbloques

Los bloques pueden contener otros bloques. pipe.blocks te da la definición del bloque de nivel superior (aquí, QwenImageAutoBlocks), mientras que sub_blocks te permite acceder a los bloques más pequeños dentro de él.

QwenImageAutoBlocks está compuesto por: text_encoder, vae_encoder, controlnet_vae_encoder, denoise y decode.

Estos subbloques se ejecutan uno tras otro y los datos fluyen de forma lineal de un bloque al siguiente: los intermediate_outputs de cada bloque están disponibles como inputs para el bloque siguiente. Así es como funcionan los SequentialPipelineBlocks.

Puedes acceder a ellos a través de la propiedad sub_blocks. La propiedad doc es útil para ver la documentación completa de cualquier bloque, incluidos sus entradas, salidas y componentes.

vae_encoder_block = pipe.blocks.sub_blocks["vae_encoder"]
print(vae_encoder_block.doc)

Este bloque se puede convertir en un pipeline para que pueda ejecutarse de forma independiente con init_pipeline().

vae_encoder_pipe = vae_encoder_block.init_pipeline()

# Reuse the VAE we already loaded, we can reuse it with update_components() method
vae_encoder_pipe.update_components(vae=pipe.vae)

# Run just this block
image_latents = vae_encoder_pipe(image=input_image).image_latents
print(image_latents.shape)

Reutiliza el VAE de nuestro pipeline original en lugar de recargarlo, manteniendo un uso de memoria eficiente. Obtenga más información en la guía Carga de componentes.

Dado que los bloques son componibles, puedes modificar la definición del pipeline añadiendo, eliminando o intercambiando bloques para crear nuevos flujos de trabajo. En la siguiente sección, añadiremos un bloque de detección de bordes canny a un pipeline de ControlNet, de modo que puedas pasar una imagen regular en lugar de un mapa de bordes canny preprocesado.

Componer nuevos flujos de trabajo

Agreguemos un bloque de detección de bordes canny a un pipeline de ControlNet. Primero, cargue un bloque canny predefinido desde el Hub (consulte Building Custom Blocks para crear el suyo).

from diffusers.modular_pipelines import ModularPipelineBlocks

# Load a canny block from the Hub
canny_block = ModularPipelineBlocks.from_pretrained(
    "diffusers-internal-dev/canny-filtering",
    trust_remote_code=True,
)

print(canny_block.doc)

class CannyBlock

  Inputs:
      image (`Union[Image, ndarray]`):
          Image to compute canny filter on
      low_threshold (`int`, *optional*, defaults to 50):
          Low threshold for the canny filter.
      high_threshold (`int`, *optional*, defaults to 200):
          High threshold for the canny filter.
      ...

  Outputs:
      control_image (`PIL.Image`):
          Canny map for input image

Use get_workflow para extraer el flujo de trabajo de ControlNet de QwenImageAutoBlocks.

# Get the controlnet workflow that we want to work with
blocks = pipe.blocks.get_workflow("controlnet_text2image")
print(blocks.doc)

class SequentialPipelineBlocks

  Inputs:
      prompt (`str`):
          The prompt or prompts to guide image generation.
      control_image (`Image`):
          Control image for ControlNet conditioning.
      ...

El flujo de trabajo extraído es un SequentialPipelineBlocks y actualmente requiere control_image como entrada. Inserta el bloque canny al principio para que la pipeline acepte una imagen regular en su lugar.

# Insert canny at the beginning
blocks.sub_blocks.insert("canny", canny_block, 0)

# Check the updated structure: CannyBlock is now listed as first sub-block
print(blocks)
# Check the updated doc
print(blocks.doc)

class SequentialPipelineBlocks

  Inputs:
      image (`Union[Image, ndarray]`):
          Image to compute canny filter on
      low_threshold (`int`, *optional*, defaults to 50):
          Low threshold for the canny filter.
      high_threshold (`int`, *optional*, defaults to 200):
          High threshold for the canny filter.
      prompt (`str`):
          The prompt or prompts to guide image generation.
      ...

Ahora el pipeline toma image como entrada en lugar de control_image. Dado que los bloques de una secuencia comparten datos automáticamente, la salida del bloque canny (control_image) fluye hacia el bloque denoise que lo necesita, y la entrada del bloque canny (image) se convierte en una entrada del pipeline porque ningún bloque anterior la proporciona.

Crea un pipeline a partir de los bloques modificados y carga un modelo de ControlNet. ControlNet no forma parte del repositorio original del modelo, por lo que debes cargarlo por separado y añadirlo con update_components().

pipeline = blocks.init_pipeline("Qwen/Qwen-Image", components_manager=manager)

pipeline.load_components(dtype=torch.bfloat16)

# Load the ControlNet model
controlnet_spec = pipeline.get_component_spec("controlnet")
controlnet_spec.pretrained_model_name_or_path = "InstantX/Qwen-Image-ControlNet-Union"
controlnet = controlnet_spec.load(dtype=torch.bfloat16)
pipeline.update_components(controlnet=controlnet)

Ahora ejecuta el pipeline: el bloque canny preprocesa la imagen para ControlNet.

from diffusers.utils import load_image

prompt = "cat wizard with red hat, gandalf, lord of the rings, detailed, fantasy, cute, adorable, Pixar, Disney"
image = load_image("https://github.com/Trgtuan10/Image_storage/blob/main/cute_cat.png?raw=true")

output = pipeline(
    prompt=prompt,
    image=image,
).images[0]
output

Próximos pasos