ModularPipeline

ModularPipeline convierte ModularPipelineBlocks en un pipeline ejecutable que carga modelos y realiza las etapas de cálculo definidas en los bloques. Es la interfaz principal para ejecutar un pipeline y su API es muy similar a la de DiffusionPipeline, pero con algunas diferencias clave.

La carga es perezosa. Con DiffusionPipeline, from_pretrained() crea el pipeline y carga todos los modelos al mismo tiempo. Con ModularPipeline, la creación y la carga son dos pasos separados: from_pretrained() lee la configuración y sabe desde dónde cargar cada componente, pero no carga realmente los pesos del modelo. Cargas los modelos más adelante con load_components(), donde pasas argumentos de carga como dtype y quantization_config.
Dos formas de crear un pipeline. Puedes usar from_pretrained() con un repositorio de modelos de diffusers existente: se mapea automáticamente a los bloques de pipeline predeterminados y luego se convierte en un ModularPipeline sin configuración adicional. Puedes consultar el modular_pipelines_directory para ver qué modelos se admiten actualmente. También puedes ensamblar tu propio pipeline a partir de ModularPipelineBlocks y convertirlo con el método init_pipeline() (consulta Creación de un pipeline para más detalles).
La ejecución del pipeline es la misma. Una vez cargado, se llama al pipeline con los mismos argumentos a los que está acostumbrado. Un único ModularPipeline puede soportar múltiples flujos de trabajo (texto a imagen, imagen a imagen, inpainting, etc.) cuando los bloques del pipeline usan AutoPipelineBlocks para seleccionar automáticamente el flujo de trabajo en función de sus entradas.

A continuación se muestran ejemplos completos para texto a imagen, imagen a imagen e inpainting con SDXL.

texto a imagen

image-to-image

inpainting

Esta guía le mostrará cómo crear un ModularPipeline, gestionar sus componentes y ejecutar la pipeline.

Creación de un pipeline

Hay dos formas de crear un ModularPipeline. Ensamblar y crear un pipeline a partir de ModularPipelineBlocks con init_pipeline(), o cargar un pipeline existente con from_pretrained().

También puedes inicializar un ComponentsManager para gestionar la colocación de dispositivos y la memoria. Si no necesitas el descargado automático, puedes omitir esto y mover el pipeline a tu dispositivo manualmente con pipeline.to("cuda").

Consulte la documentación de ComponentsManager para obtener más detalles sobre cómo puede ayudar a gestionar componentes en diferentes flujos de trabajo.

init_pipeline

init_pipeline() convierte cualquier ModularPipelineBlocks en un ModularPipeline.

Definamos un bloque mínimo para ver cómo funciona:

from transformers import CLIPTextModel
from diffusers.modular_pipelines import (
    ComponentSpec,
    ModularPipelineBlocks,
    PipelineState,
)

class MyBlock(ModularPipelineBlocks):
    @property
    def expected_components(self):
        return [
            ComponentSpec(
                name="text_encoder",
                type_hint=CLIPTextModel,
                pretrained_model_name_or_path="openai/clip-vit-large-patch14",
            ),
        ]

    def __call__(self, components, state: PipelineState) -> PipelineState:
        return components, state

Llama a init_pipeline() para convertirlo en un pipeline. El atributo blocks del pipeline son los bloques con los que se creó: determina las entradas, salidas y la lógica de cálculo esperadas.

block = MyBlock()
pipe = block.init_pipeline()
pipe.blocks

MyBlock {
  "_class_name": "MyBlock",
  "_diffusers_version": "0.37.0.dev0"
}

Los bloques son mutables: puedes agregar, eliminar o intercambiar bloques libremente antes de crear una pipeline. Sin embargo, una vez que se crea una pipeline, modificar pipeline.blocks no afectará a la pipeline, ya que devuelve una copia. Si deseas una estructura de bloques diferente, crea una nueva pipeline después de modificar los bloques.

Cuando se llama a init_pipeline() sin un repositorio, utiliza el pretrained_model_name_or_path definido en el ComponentSpec del bloque para determinar desde dónde cargar cada componente. Imprimir el pipeline muestra la configuración de carga de los componentes.

pipe
ModularPipeline {
  "_blocks_class_name": "MyBlock",
  "_class_name": "ModularPipeline",
  "_diffusers_version": "0.37.0.dev0",
  "text_encoder": [
    null,
    null,
    {
      "pretrained_model_name_or_path": "openai/clip-vit-large-patch14",
      "revision": null,
      "subfolder": "",
      "type_hint": [
        "transformers",
        "CLIPTextModel"
      ],
      "variant": null
    }
  ]
}

Si pasas un repositorio a init_pipeline(), este anula la ruta de carga al hacer coincidir los componentes de tu bloque con la configuración de la pipeline en ese repositorio (model_index.json o modular_model_index.json). Consulta Modular repository para ver cómo se registran y guardan las especificaciones de carga.

En el ejemplo siguiente, pretrained_model_name_or_path se actualizará a "stabilityai/stable-diffusion-xl-base-1.0".

pipe = block.init_pipeline("stabilityai/stable-diffusion-xl-base-1.0")
pipe
ModularPipeline {
  "_blocks_class_name": "MyBlock",
  "_class_name": "ModularPipeline",
  "_diffusers_version": "0.37.0.dev0",
  "text_encoder": [
    null,
    null,
    {
      "pretrained_model_name_or_path": "stabilityai/stable-diffusion-xl-base-1.0",
      "revision": null,
      "subfolder": "text_encoder",
      "type_hint": [
        "transformers",
        "CLIPTextModel"
      ],
      "variant": null
    }
  ]
}

Si un componente de tu bloque no existe en el repositorio, permanece como null y se omite durante load_components().

from_pretrained

from_pretrained() es una forma conveniente de crear un ModularPipeline sin tener que definir los bloques usted mismo.

Funciona con tres tipos de repositorios.

Un repositorio de diffusers estándar. Pasa cualquier repositorio de modelo compatible y se mapea automáticamente a los bloques de pipeline predeterminados. Los modelos compatibles actualmente incluyen SDXL, Wan, Qwen, Z-Image, Flux y Flux2.

from diffusers import ModularPipeline, ComponentsManager

components = ComponentsManager()
pipeline = ModularPipeline.from_pretrained(
    "stabilityai/stable-diffusion-xl-base-1.0", components_manager=components
)

Un repositorio modular. Estos repositorios contienen un modular_model_index.json que especifica desde dónde cargar cada componente: los componentes pueden provenir de diferentes repositorios y el propio repositorio modular puede no contener pesos de modelo. Por ejemplo, diffusers/flux2-bnb-4bit-modular carga un transformer cuantizado de un repositorio y los componentes restantes de otro. Consulte Repositorio modular para obtener más detalles sobre el formato.

from diffusers import ModularPipeline, ComponentsManager

components = ComponentsManager()
pipeline = ModularPipeline.from_pretrained(
    "diffusers/flux2-bnb-4bit-modular", components_manager=components
)

Un repositorio modular con código personalizado. Algunos repositorios incluyen bloques de pipeline personalizados junto con la configuración de carga. Añade trust_remote_code=True para cargarlos. Consulta Bloques personalizados para saber cómo crear los tuyos.

from diffusers import ModularPipeline, ComponentsManager

components = ComponentsManager()
pipeline = ModularPipeline.from_pretrained(
    "diffusers/Florence2-image-Annotator", trust_remote_code=True, components_manager=components
)

Cuando los bloques de la pipeline definen flujos de trabajo (consulte pipeline.blocks.available_workflows), pase workflow= para conservar solo los bloques de ese flujo de trabajo, la misma poda que realiza get_workflow(). La pipeline entonces solo declara los componentes que utiliza ese flujo de trabajo, y su docstring describe exactamente las entradas de ese flujo de trabajo.

pipeline = ModularPipeline.from_pretrained("Qwen/Qwen-Image", workflow="inpainting")

Cargando componentes

Un ModularPipeline no se instancia automáticamente con componentes. Solo carga la configuración y las especificaciones de los componentes. Puedes cargar los componentes con load_components().

Esto cargará todos los componentes que tienen una especificación de carga válida.

import torch

pipeline.load_components(dtype=torch.float16)

También puede cargar componentes específicos por nombre. El siguiente ejemplo solo carga el text_encoder.

pipeline.load_components(names=["text_encoder"], dtype=torch.float16)

En un pipeline cuyos bloques definen flujos de trabajo, workflow= carga solo los componentes que utiliza ese flujo de trabajo. El pipeline conserva todos sus bloques, por lo que esta es la forma conveniente de ejecutar un mismo pipeline a través de varios flujos de trabajo: cada llamada añade solo lo que el nuevo flujo de trabajo aún no tiene.

pipeline.load_components(workflow="inpainting", dtype=torch.float16)

Tras la carga, al imprimir el pipeline se muestran los componentes cargados: los dos primeros campos cambian de null a la biblioteca y la clase del componente.

pipeline

# text_encoder is loaded - shows library and class
"text_encoder": [
  "transformers",
  "CLIPTextModel",
  { ... }
]

# unet is not loaded yet - still null
"unet": [
  null,
  null,
  { ... }
]

Los argumentos de palabra clave como dtype, variant, revision y quantization_config se transmiten a from_pretrained() para cada componente. Puede pasar un único valor para aplicarlo a todos los componentes, o un diccionario para establecer valores por componente.

# apply bfloat16 to all components
pipeline.load_components(dtype=torch.bfloat16)

# different dtypes per component
pipeline.load_components(dtype={"transformer": torch.bfloat16, "default": torch.float32})

load_components() solo carga los componentes que aún no se han cargado y que tienen una especificación de carga válida. Esto significa que si ya has establecido un componente en el pipeline, llamar a load_components() de nuevo no lo recargará.

Actualización de componentes

update_components() reemplaza un componente del pipeline por uno nuevo. Cuando se actualiza un componente, las especificaciones de carga también se actualizan en la configuración del pipeline y load_components() lo omitirá en las llamadas posteriores.

Desde AutoModel

Puede pasar un objeto de modelo cargado con AutoModel.from_pretrained(). Los modelos cargados de esta manera se etiquetan automáticamente con su información de carga.

from diffusers import AutoModel

unet = AutoModel.from_pretrained(
    "RunDiffusion/Juggernaut-XL-v9", subfolder="unet", variant="fp16", dtype=torch.float16
)
pipeline.update_components(unet=unet)

Desde ComponentSpec

Use get_component_spec() para obtener una copia de la especificación actual del componente, modificarla y cargar un nuevo componente.

unet_spec = pipeline.get_component_spec("unet")

# modify to load from a different repository
unet_spec.pretrained_model_name_or_path = "RunDiffusion/Juggernaut-XL-v9"

# load and update
unet = unet_spec.load(dtype=torch.float16)
pipeline.update_components(unet=unet)

También puedes crear un ComponentSpec desde cero.

No todos los componentes se cargan desde pesos preentrenados: algunos se crean a partir de una configuración (enumerados en pipeline.config_component_names). Para estos, use create() en lugar de load().

guider_spec = pipeline.get_component_spec("guider")
guider_spec.config = {"guidance_scale": 5.0}
guider = guider_spec.create()
pipeline.update_components(guider=guider)

O simplemente pase el objeto directamente.

from diffusers.guiders import ClassifierFreeGuidance

guider = ClassifierFreeGuidance(guidance_scale=5.0)
pipeline.update_components(guider=guider)

Consulte la guía de Guiders para obtener más detalles sobre los guiders disponibles y cómo configurarlos.

Dividir un pipeline en etapas

Dado que los bloques son componibles, puedes desmontar un pipeline y reconstruirlo en pipelines separados para cada etapa. El siguiente ejemplo muestra cómo podemos separar el bloque del codificador de texto del resto del pipeline, de modo que puedas codificar el prompt de forma independiente y pasar los embeddings al pipeline principal.

from diffusers import ModularPipeline, ComponentsManager
import torch

device = "cuda"  # or "mps", "xpu", "cpu"
dtype = torch.bfloat16
repo_id = "black-forest-labs/FLUX.2-klein-4B"

# get the blocks and separate out the text encoder
blocks = ModularPipeline.from_pretrained(repo_id).blocks
text_block = blocks.sub_blocks.pop("text_encoder")

# use ComponentsManager to handle offloading across multiple pipelines
manager = ComponentsManager()
manager.enable_auto_cpu_offload(device=device)

# create separate pipelines for each stage
text_encoder_pipeline = text_block.init_pipeline(repo_id, components_manager=manager)
pipeline = blocks.init_pipeline(repo_id, components_manager=manager)

# encode text
text_encoder_pipeline.load_components(dtype=dtype)
text_embeddings = text_encoder_pipeline(prompt="a cat").get_by_kwargs("denoiser_input_fields")

# denoise and decode
pipeline.load_components(dtype=dtype)
output = pipeline(
    **text_embeddings,
    num_inference_steps=4,
).images[0]

ComponentsManager gestiona la memoria entre varios pipelines. A diferencia de las estrategias de offloading en DiffusionPipeline que siguen un orden fijo, ComponentsManager toma decisiones de offloading dinámicamente cada vez que se ejecuta una pasada hacia adelante de un modelo, basándose en la situación de memoria actual. Esto significa que funciona sin importar cuántos pipelines crees ni en qué orden los ejecutes. Consulta la guía de ComponentsManager para más detalles.

Si las etapas de la pipeline comparten componentes (por ejemplo, el mismo VAE utilizado para codificar y decodificar), puede usar update_components() para pasar un componente ya cargado a otra pipeline en lugar de volver a cargarlo.

Repositorio modular

Se requiere un repositorio si los bloques de pipeline usan componentes preentrenados. El repositorio proporciona las especificaciones de carga y los metadatos.

ModularPipeline funciona con repositorios de diffusers estándar sin configuración adicional. Sin embargo, también puedes crear un repositorio modular para mayor flexibilidad. Un repositorio modular contiene un archivo modular_model_index.json con los siguientes 3 elementos.

library y class indican desde qué biblioteca se cargó el componente y su clase. Si es null, el componente aún no se ha cargado.
loading_specs_dict contiene la información necesaria para cargar el componente, como el repositorio y el subdirectorio desde el que se carga.

La principal ventaja de un repositorio modular es que los componentes pueden cargarse desde diferentes repositorios. Por ejemplo, diffusers/flux2-bnb-4bit-modular carga un transformer cuantizado desde diffusers/FLUX.2-dev-bnb-4bit mientras carga los componentes restantes desde black-forest-labs/FLUX.2-dev.

Para convertir un repositorio regular de diffusers en uno modular, cree el pipeline usando el repositorio regular y luego publíquelo en el Hub. El repositorio guardado contendrá un modular_model_index.json con todas las especificaciones de carga.

from diffusers import ModularPipeline

# load from a regular repo
pipeline = ModularPipeline.from_pretrained("stabilityai/stable-diffusion-xl-base-1.0")

# push as a modular repository
pipeline.save_pretrained("local/path", repo_id="my-username/sdxl-modular", push_to_hub=True)

Por defecto, save_pretrained() escribe cada componente actualmente cargado que Diffusers puede serializar. Los componentes que no están cargados, o que no se pueden serializar, no se escriben y conservan sus especificaciones de carga existentes. Esto te ofrece dos formas de guardar, dependiendo de lo que necesites.

Guardar una copia autocontenida

Cargue todos los componentes y luego guarde. Cada especificación apunta al resultado, por lo que se recarga por completo desde un solo lugar, incluso sin conexión.

pipe = ModularPipeline.from_pretrained("MiniMaxAI/MiniMax-H3")
pipe.load_components()
pipe.save_pretrained("path/to/local-copy")

El destino registrado en modular_model_index.json es donde save_pretrained() escribe: el save_directory para un guardado local, o el repo_id cuando pasas push_to_hub=True.

Con huggingface_hub>=1.32.0, los archivos idénticos respaldados por Xet pueden compartir un mismo payload en caché entre cualquier repositorio de Hub. Esto incluye archivos individuales en un checkpoint fragmentado: un fragmento con el mismo hash de Xet puede reutilizarse incluso cuando otros fragmentos difieren. Ambas descargas deben utilizar el mismo directorio de caché de Hub con soporte para enlaces simbólicos. Cambiar el dtype, la serialización o los límites de los fragmentos puede alterar los hashes de los archivos y evitar la reutilización. Esto ahorra tráfico de descarga y espacio en disco, no la memoria utilizada por los componentes cargados.

La caché compartida es de mejor esfuerzo. Establecer HF_HUB_DISABLE_SHARED_BLOBS=1 sigue admitiendo la carga, pero almacena copias separadas en la caché. Las referencias de componentes existentes en modular_model_index.json permanecen sin cambios; mantenga las referencias externas para los componentes que no se almacenan en el repositorio modular.

Actualizar un índice modular existente

La migración es opcional: las referencias al repositorio original ya reutilizan sus archivos en caché. Actualice una referencia si solo se usaba para evitar descargas duplicadas y desea que ese componente se cargue desde el repositorio modular.

Primero, compruebe que el repositorio modular contiene la misma configuración de componentes y todos los archivos de pesos requeridos, incluido el índice de pesos y cada fragmento para un punto de control fragmentado. Los hashes Xet coincidentes determinan qué archivos de pesos pueden reutilizar la caché compartida. Mantenga las referencias a los componentes que están ausentes o que se cargan intencionalmente desde otro punto de control.

Edite solo la especificación de carga relevante en una copia local de modular_model_index.json. Por ejemplo, para usar un VAE almacenado en vae/ en el repositorio modular con la misma variante de pesos:

import json
from pathlib import Path

index_path = Path("path/to/local-copy/modular_model_index.json")
index = json.loads(index_path.read_text())
spec = index["vae"][2]
spec["pretrained_model_name_or_path"] = "my-username/my-modular-repo"
spec.pop("repo", None)
spec["subfolder"] = "vae"
spec["revision"] = None
index_path.write_text(json.dumps(index, indent=2) + "\n")

Esto también elimina el campo repo heredado. Una revisión del repositorio original no identifica una revisión en el destino: use None para la rama predeterminada del destino, o fije una revisión de destino que contenga el componente. Ajuste subfolder y variant si sus nombres de archivo difieren.

Antes de publicar el índice actualizado, cárgalo con ModularPipeline.from_pretrained("path/to/local-copy"), llama a load_components(names="vae") y verifica que el componente se cargue con los parámetros esperados. Sube solo el índice editado; volver a guardar los pesos puede cambiar sus hashes y evitar la reutilización de la caché.

Mantener referencias a componentes existentes

Carga solo lo nuevo (o nada en absoluto). Solo se guardan los componentes cargados; todo lo demás permanece como un puntero a su repositorio original. Usa este modo cuando quieras reemplazar un componente mientras continúas cargando los demás desde su repositorio original. Por ejemplo, guarda un transformador personalizado mientras los componentes restantes continúan cargándose desde el repositorio base.

pipe = ModularPipeline.from_pretrained("black-forest-labs/FLUX.2-dev")
pipe.update_components(transformer=my_custom_transformer)  # the only component in memory
pipe.save_pretrained("local/path", repo_id="my-username/flux2-custom-transformer", push_to_hub=True)

Pase overwrite_modular_index=False para mantener las especificaciones de carga en modular_model_index.json tal como están. Un componente guardado cuya especificación de carga esté vacía se completará igualmente con el destino, ya que no hay nada que preservar.

Mover los archivos de cualquier otra forma no reescribe el índice. Una copia descargada con hf download --local-dir sigue funcionando: cuando se carga un pipeline desde un directorio local, cada componente cuyos archivos están presentes en ese directorio se carga desde él en lugar del repositorio registrado. Una copia subida con hf upload sigue apuntando a la ubicación anterior, por lo que en ese caso hay que actualizar el índice manualmente.

Un repositorio modular también puede incluir bloques de pipeline personalizados como código de Python. Esto permite compartir bloques especializados que no son nativos de Diffusers. Por ejemplo, diffusers/Florence2-image-Annotator contiene bloques personalizados junto con la configuración de carga:

Florence2-image-Annotator/
├── block.py                    # Custom pipeline blocks implementation
├── config.json                 # Pipeline configuration and auto_map
├── mellon_config.json          # UI configuration for Mellon
└── modular_model_index.json    # Component loading specifications

El archivo config.json contiene una clave auto_map que indica a ModularPipeline dónde encontrar los bloques personalizados:

{
  "_class_name": "Florence2AnnotatorBlocks",
  "auto_map": {
    "ModularPipelineBlocks": "block.Florence2AnnotatorBlocks"
  }
}

Carga repositorios de código personalizados con trust_remote_code=True como se muestra en from_pretrained. Consulta Bloques personalizados para saber cómo crear y compartir los tuyos.

Update on GitHub