ModularPipeline

ModularPipeline converte ModularPipelineBlocks em um pipeline executável que carrega modelos e executa as etapas de computação definidas nos blocos. É a interface principal para executar um pipeline e a API é muito semelhante à do DiffusionPipeline, mas com algumas diferenças importantes.

O carregamento é preguiçoso. Com o DiffusionPipeline, from_pretrained() cria o pipeline e carrega todos os modelos ao mesmo tempo. Com o ModularPipeline, a criação e o carregamento são duas etapas separadas: from_pretrained() lê a configuração e sabe de onde carregar cada componente, mas não carrega efetivamente os pesos do modelo. Você carrega os modelos depois com load_components(), onde você passa argumentos de carregamento como dtype e quantization_config.
Duas formas de criar um pipeline. Você pode usar from_pretrained() com um repositório de modelo diffusers existente — ele mapeia automaticamente para os blocos de pipeline padrão e, em seguida, converte para um ModularPipeline sem configuração adicional. Você pode verificar o modular_pipelines_directory para ver quais modelos são atualmente suportados. Você também pode montar seu próprio pipeline a partir de ModularPipelineBlocks e convertê-lo com o método init_pipeline() (veja Criando um pipeline para mais detalhes).
A execução do pipeline é a mesma. Uma vez carregado, você chama o pipeline com os mesmos argumentos a que está acostumado. Um único ModularPipeline pode suportar múltiplos fluxos de trabalho (texto-para-imagem, imagem-para-imagem, inpainting, etc.) quando os blocos do pipeline usam AutoPipelineBlocks para selecionar automaticamente o fluxo de trabalho com base nas suas entradas.

Abaixo estão exemplos completos para texto-para-imagem, imagem-para-imagem e inpainting com SDXL.

texto-para-imagem

imagem-para-imagem

inpainting

Este guia mostrará como criar um ModularPipeline, gerenciar seus componentes e executar o pipeline.

Criando um pipeline

Existem duas formas de criar um ModularPipeline. Monte e crie um pipeline a partir de ModularPipelineBlocks com init_pipeline(), ou carregue um pipeline existente com from_pretrained().

Você também pode inicializar um ComponentsManager para gerenciar a colocação de dispositivos e a memória. Se você não precisar de offloading automático, pode pular isso e mover o pipeline para o seu dispositivo manualmente com pipeline.to("cuda").

Consulte a documentação do ComponentsManager para mais detalhes sobre como ele pode ajudar a gerenciar componentes em diferentes fluxos de trabalho.

init_pipeline

init_pipeline() converte qualquer ModularPipelineBlocks em um ModularPipeline.

Vamos definir um bloco mínimo para ver como ele 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

Chame init_pipeline() para convertê-lo em um pipeline. O atributo blocks no pipeline são os blocos a partir dos quais ele foi criado — eles determinam as entradas, saídas e a lógica de computação esperadas.

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

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

Os blocos são mutáveis — você pode adicionar, remover ou trocar blocos livremente antes de criar um pipeline. No entanto, uma vez que um pipeline é criado, modificar pipeline.blocks não afetará o pipeline, pois ele retorna uma cópia. Se você deseja uma estrutura de blocos diferente, crie um novo pipeline após modificar os blocos.

Quando você chama init_pipeline() sem um repositório, ele usa o pretrained_model_name_or_path definido no ComponentSpec do bloco para determinar de onde carregar cada componente. Imprimir o pipeline mostra a configuração de carregamento dos 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
    }
  ]
}

Se você passar um repositório para init_pipeline(), ele sobrescreve o caminho de carregamento ao corresponder os componentes do seu bloco à configuração do pipeline nesse repositório (model_index.json ou modular_model_index.json). Consulte Modular repository para saber como as especificações de carregamento são registradas e salvas.

No exemplo abaixo, o pretrained_model_name_or_path será atualizado para "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
    }
  ]
}

Se um componente do seu bloco não existir no repositório, ele permanece como null e é ignorado durante o load_components().

from_pretrained

from_pretrained() é uma maneira conveniente de criar um ModularPipeline sem definir os blocos você mesmo.

Ele funciona com três tipos de repositórios.

Um repositório de diffusers padrão. Passe qualquer repositório de modelo suportado e ele será mapeado automaticamente para os blocos de pipeline padrão. Os modelos atualmente suportados incluem SDXL, Wan, Qwen, Z-Image, Flux e Flux2.

from diffusers import ModularPipeline, ComponentsManager

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

Um repositório modular. Esses repositórios contêm um modular_model_index.json que especifica de onde carregar cada componente — os componentes podem vir de repositórios diferentes e o próprio repositório modular pode não conter nenhum peso de modelo. Por exemplo, diffusers/flux2-bnb-4bit-modular carrega um transformer quantizado de um repositório e os componentes restantes de outro. Veja Repositório modular para mais detalhes sobre o formato.

from diffusers import ModularPipeline, ComponentsManager

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

Um repositório modular com código personalizado. Alguns repositórios incluem blocos de pipeline personalizados junto com a configuração de carregamento. Adicione trust_remote_code=True para carregá-los. Consulte Blocos personalizados para saber como criar os seus próprios.

from diffusers import ModularPipeline, ComponentsManager

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

Quando os blocos do pipeline definem fluxos de trabalho (consulte pipeline.blocks.available_workflows), passe workflow= para manter apenas os blocos daquele fluxo de trabalho — o mesmo recorte de get_workflow(). O pipeline então declara apenas os componentes usados por aquele fluxo de trabalho, e sua docstring descreve exatamente as entradas daquele fluxo de trabalho.

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

Carregando componentes

A ModularPipeline não é instanciada automaticamente com componentes. Ela apenas carrega a configuração e as especificações dos componentes. Você pode carregar os componentes com load_components().

Isso carregará todos os componentes que possuem uma especificação de carregamento válida.

import torch

pipeline.load_components(dtype=torch.float16)

Você também pode carregar componentes específicos pelo nome. O exemplo abaixo carrega apenas o text_encoder.

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

Em um pipeline cujos blocos definem fluxos de trabalho, workflow= carrega apenas os componentes que o fluxo de trabalho utiliza. O pipeline mantém todos os seus blocos, portanto, essa é a maneira conveniente de executar um único pipeline em vários fluxos de trabalho: cada chamada adiciona apenas o que o novo fluxo de trabalho ainda não possui.

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

Após o carregamento, a impressão do pipeline mostra quais componentes foram carregados — os dois primeiros campos mudam de null para a biblioteca e a classe do componente.

pipeline

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

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

Argumentos de palavra-chave como dtype, variant, revision e quantization_config são repassados para from_pretrained() de cada componente. Você pode passar um único valor para aplicar a todos os componentes, ou um dict para definir 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() carrega apenas os componentes que ainda não foram carregados e que possuem uma especificação de carregamento válida. Isso significa que, se você já configurou um componente no pipeline, chamar load_components() novamente não o recarregará.

Atualizando componentes

update_components() substitui um componente no pipeline por um novo. Quando um componente é atualizado, as especificações de carregamento também são atualizadas na configuração do pipeline e load_components() o ignorará nas chamadas subsequentes.

De AutoModel

Você pode passar um objeto de modelo carregado com AutoModel.from_pretrained(). Modelos carregados dessa forma são automaticamente marcados com suas informações de carregamento.

from diffusers import AutoModel

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

De ComponentSpec

Use get_component_spec() para obter uma cópia da especificação atual do componente, modificá-la e carregar um novo 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)

Você também pode criar um ComponentSpec do zero.

Nem todos os componentes são carregados a partir de pesos pré-treinados — alguns são criados a partir de uma configuração (listados em pipeline.config_component_names). Para esses, use create() em vez 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)

Ou simplesmente passe o objeto diretamente.

from diffusers.guiders import ClassifierFreeGuidance

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

Consulte o guia Guiders para mais detalhes sobre os guiders disponíveis e como configurá-los.

Dividir um pipeline em etapas

Como os blocos são composíveis, você pode desmontar um pipeline e reconstruí-lo em pipelines separados para cada etapa. O exemplo abaixo mostra como podemos separar o bloco do codificador de texto do restante do pipeline, para que você possa codificar o prompt de forma independente e passar os embeddings para o 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 gerencia a memória entre vários pipelines. Diferentemente das estratégias de offloading em DiffusionPipeline, que seguem uma ordem fixa, o ComponentsManager toma decisões de offloading dinamicamente a cada vez que uma passagem de modelo para frente é executada, com base na situação atual de memória. Isso significa que ele funciona independentemente de quantos pipelines você cria ou em que ordem você os executa. Consulte o guia de ComponentsManager para mais detalhes.

Se os estágios do pipeline compartilham componentes (por exemplo, o mesmo VAE usado para codificação e decodificação), você pode usar update_components() para passar um componente já carregado para outro pipeline, em vez de carregá-lo novamente.

Repositório modular

Um repositório é obrigatório se os blocos do pipeline usarem componentes pré-treinados. O repositório fornece especificações de carregamento e metadados.

ModularPipeline funciona com repositórios regulares do diffusers sem necessidade de configuração adicional. No entanto, você também pode criar um repositório modular para mais flexibilidade. Um repositório modular contém um arquivo modular_model_index.json com os seguintes 3 elementos.

library e class mostram de qual biblioteca o componente foi carregado e sua classe. Se null, o componente ainda não foi carregado.
loading_specs_dict contém as informações necessárias para carregar o componente, como o repositório e a subpasta de onde ele é carregado.

A principal vantagem de um repositório modular é que os componentes podem ser carregados de repositórios diferentes. Por exemplo, diffusers/flux2-bnb-4bit-modular carrega um transformer quantizado de diffusers/FLUX.2-dev-bnb-4bit enquanto carrega os componentes restantes de black-forest-labs/FLUX.2-dev.

Para converter um repositório regular do diffusers em um repositório modular, crie o pipeline usando o repositório regular e, em seguida, envie para o Hub. O repositório salvo conterá um modular_model_index.json com todas as especificações de carregamento.

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 padrão, save_pretrained() grava cada componente atualmente carregado que o Diffusers pode serializar. Componentes que não estão carregados, ou que não podem ser serializados, não são gravados e mantêm suas especificações de carregamento existentes. Isso oferece duas maneiras de salvar, dependendo do que você deseja.

Salvar uma cópia autossuficiente

Carregue todos os componentes e, em seguida, salve. Cada especificação aponta para o resultado, portanto, ele é recarregado inteiramente de um único local, inclusive offline.

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

O destino registrado em modular_model_index.json é onde save_pretrained() grava: o save_directory para um salvamento local, ou o repo_id quando você passa push_to_hub=True.

Com huggingface_hub>=1.32.0, arquivos idênticos com suporte a Xet podem compartilhar um único payload em cache entre qualquer repositório do Hub. Isso inclui arquivos individuais em um checkpoint fragmentado: um fragmento com o mesmo hash Xet pode ser reutilizado mesmo quando outros fragmentos diferem. Ambos os downloads devem usar o mesmo diretório de cache do Hub com suporte a symlinks. Alterar o dtype, a serialização ou as fronteiras dos fragmentos pode alterar os hashes dos arquivos e impedir a reutilização. Isso economiza tráfego de download e espaço em disco, não a memória usada pelos componentes carregados.

O cache compartilhado é feito da melhor forma possível. Definir HF_HUB_DISABLE_SHARED_BLOBS=1 ainda suporta o carregamento, mas faz o cache de cópias separadas. As referências de componentes existentes em modular_model_index.json permanecem inalteradas; mantenha as referências externas para componentes que não são armazenados no repositório modular.

Atualizar um índice modular existente

A migração é opcional: as referências ao repositório original já reutilizam seus arquivos em cache. Atualize uma referência se ela foi usada apenas para evitar downloads duplicados e você deseja que esse componente seja carregado do repositório modular.

Primeiro, verifique se o repositório modular contém a mesma configuração de componentes e todos os arquivos de pesos necessários, incluindo o índice de pesos e todos os shards para um checkpoint particionado. Os hashes Xet correspondentes determinam quais arquivos de pesos podem reutilizar o cache compartilhado. Mantenha referências a componentes ausentes ou carregados intencionalmente de outro checkpoint.

Edite apenas a especificação de carregamento relevante em uma cópia local de modular_model_index.json. Por exemplo, para usar um VAE armazenado em vae/ no repositório modular com a mesma 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")

Isso também remove o campo legado repo. Uma revisão do repositório original não identifica uma revisão no destino: use None para o branch padrão do destino, ou fixe uma revisão de destino que contenha o componente. Ajuste subfolder e variant se os nomes dos arquivos forem diferentes.

Antes de publicar o índice atualizado, carregue-o com ModularPipeline.from_pretrained("path/to/local-copy"), chame load_components(names="vae") e verifique se o componente é carregado com os parâmetros esperados. Envie apenas o índice editado; salvar novamente os pesos pode alterar seus hashes e impedir a reutilização do cache.

Mantenha referências a componentes existentes

Carregue apenas o que é novo (ou nada). Somente os componentes carregados são salvos; todos os outros permanecem como ponteiros para o repositório original. Use este modo quando você deseja substituir um componente, mas continuar carregando os outros a partir do repositório original. Por exemplo, salve um transformador personalizado enquanto os componentes restantes continuam sendo carregados do repositório 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)

Pass overwrite_modular_index=False para manter as especificações de carregamento em modular_model_index.json como estão. Um componente salvo cuja especificação de carregamento está vazia ainda é preenchido com o destino, pois não há nada a preservar.

Mover os arquivos de qualquer outra forma não reescreve o índice. Uma cópia baixada com hf download --local-dir continua funcionando: quando um pipeline é carregado de um diretório local, cada componente cujos arquivos estão presentes nesse diretório é carregado dele em vez do repositório registrado. Uma cópia enviada com hf upload continua apontando para o local antigo, portanto, atualize o índice manualmente nesse caso.

Um repositório modular também pode incluir blocos de pipeline personalizados como código Python. Isso permite compartilhar blocos especializados que não são nativos do Diffusers. Por exemplo, diffusers/Florence2-image-Annotator contém blocos personalizados junto com a configuração de carregamento:

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

O arquivo config.json contém uma chave auto_map que informa ao ModularPipeline onde encontrar os blocos personalizados:

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

Carregue repositórios de código personalizados com trust_remote_code=True, conforme mostrado em from_pretrained. Consulte Blocos personalizados para saber como criar e compartilhar os seus.

Update on GitHub