ModularPipeline

ModularPipeline преобразует ModularPipelineBlocks в исполняемый конвейер, который загружает модели и выполняет вычислительные шаги, определённые в блоках. Это основной интерфейс для запуска конвейера, и его API очень похож на DiffusionPipeline, но с несколькими ключевыми отличиями.

• Loading is lazy. With DiffusionPipeline, from_pretrained() creates the pipeline and loads all models at the same time. With ModularPipeline, creating and loading are two separate steps: from_pretrained() reads the configuration and knows where to load each component from, but doesn’t actually load the model weights. You load the models later with load_components(), which is where you pass loading arguments like dtype and quantization_config.

• Два способа создания пайплайна. Можно использовать from_pretrained() с существующим репозиторием модели diffusers — он автоматически сопоставляется с блоками пайплайна по умолчанию, а затем преобразуется в ModularPipeline без дополнительной настройки. Список поддерживаемых моделей можно найти в modular_pipelines_directory. Также можно собрать собственный пайплайн из ModularPipelineBlocks и преобразовать его с помощью метода init_pipeline() (подробнее см. Создание пайплайна).

• Running the pipeline is the same. Once loaded, you call the pipeline with the same arguments you’re used to. A single ModularPipeline can support multiple workflows (text-to-image, image-to-image, inpainting, etc.) when the pipeline blocks use AutoPipelineBlocks to automatically select the workflow based on your inputs.

Ниже приведены полные примеры для генерации изображений по тексту, преобразования изображения в изображение и инпейнтинга с SDXL.

import torch
from diffusers import ModularPipeline

pipeline = ModularPipeline.from_pretrained("stabilityai/stable-diffusion-xl-base-1.0")
pipeline.load_components(dtype=torch.float16)
pipeline.to("cuda")  # or "mps", "xpu", "cpu"

image = pipeline(prompt="Astronaut in a jungle, cold color palette, muted colors, detailed, 8k").images[0]
image.save("modular_t2i_out.png")

import torch
from diffusers import ModularPipeline
from diffusers.utils import load_image

pipeline = ModularPipeline.from_pretrained("stabilityai/stable-diffusion-xl-base-1.0")
pipeline.load_components(dtype=torch.float16)
pipeline.to("cuda")  # or "mps", "xpu", "cpu"

url = "https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/diffusers/sdxl-text2img.png"
init_image = load_image(url)
prompt = "a dog catching a frisbee in the jungle"
image = pipeline(prompt=prompt, image=init_image, strength=0.8).images[0]
image.save("modular_i2i_out.png")

import torch
from diffusers import ModularPipeline
from diffusers.utils import load_image

pipeline = ModularPipeline.from_pretrained("stabilityai/stable-diffusion-xl-base-1.0")
pipeline.load_components(dtype=torch.float16)
pipeline.to("cuda")  # or "mps", "xpu", "cpu"

img_url = "https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/diffusers/sdxl-text2img.png"
mask_url = "https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/diffusers/sdxl-inpaint-mask.png"

init_image = load_image(img_url)
mask_image = load_image(mask_url)

prompt = "A deep sea diver floating"
image = pipeline(prompt=prompt, image=init_image, mask_image=mask_image, strength=0.85).images[0]
image.save("modular_inpaint_out.png")

В этом руководстве показано, как создать ModularPipeline, управлять его компонентами и запустить конвейер.

Создание конвейера

There are two ways to create a ModularPipeline. Assemble and create a pipeline from ModularPipelineBlocks with init_pipeline(), or load an existing pipeline with from_pretrained().

Вы также можете инициализировать ComponentsManager для управления размещением на устройствах и управлением памятью. Если автоматическое выгружение не требуется, вы можете пропустить этот шаг и вручную переместить конвейер на ваше устройство с помощью pipeline.to("cuda").

См. документацию ComponentsManager для получения подробной информации о том, как он помогает управлять компонентами в разных рабочих процессах.

init_pipeline

init_pipeline() преобразует любой ModularPipelineBlocks в ModularPipeline.

Определим минимальный блок, чтобы посмотреть, как он работает:

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

Вызовите init_pipeline(), чтобы преобразовать его в пайплайн. Атрибут blocks пайплайна содержит блоки, из которых он был создан, — они определяют ожидаемые входные и выходные данные, а также логику вычислений.

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

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

Блоки изменяемы — вы можете свободно добавлять, удалять или заменять блоки до создания конвейера. Однако, как только конвейер создан, изменение pipeline.blocks не повлияет на конвейер, так как возвращается копия. Если вам нужна другая структура блоков, создайте новый конвейер после изменения блоков.

При вызове init_pipeline() без указания репозитория используется pretrained_model_name_or_path, определённое в ComponentSpec блока, чтобы определить, откуда загружать каждый компонент. Вывод пайплайна показывает конфигурацию загрузки компонентов.

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

Если вы передадите репозиторий в init_pipeline(), он переопределит путь загрузки, сопоставляя компоненты вашего блока с конфигурацией пайплайна в этом репозитории (model_index.json или modular_model_index.json). См. Modular repository, чтобы узнать, как спецификации загрузки записываются и сохраняются.

В примере ниже pretrained_model_name_or_path будет обновлено до "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
    }
  ]
}

Если компонент в вашем блоке отсутствует в репозитории, он остаётся null и пропускается при вызове load_components().

from_pretrained

from_pretrained() — удобный способ создать ModularPipeline, не определяя блоки самостоятельно.

Он работает с тремя типами репозиториев.

Обычный репозиторий diffusers. Передайте любой поддерживаемый репозиторий модели, и он автоматически сопоставится с блоками стандартного пайплайна. В настоящее время поддерживаются модели SDXL, Wan, Qwen, Z-Image, Flux и Flux2.

from diffusers import ModularPipeline, ComponentsManager

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

Модульный репозиторий. Такие репозитории содержат modular_model_index.json, в котором указано, откуда загружать каждый компонент — компоненты могут происходить из разных репозиториев, а сам модульный репозиторий может не содержать весов модели. Например, diffusers/flux2-bnb-4bit-modular загружает квантованный трансформер из одного репозитория, а остальные компоненты — из другого. Подробнее о формате см. Модульный репозиторий.

from diffusers import ModularPipeline, ComponentsManager

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

Модульный репозиторий с собственным кодом. Некоторые репозитории содержат наряду с конфигурацией загрузки собственные блоки конвейера. Добавьте trust_remote_code=True, чтобы загрузить их. О том, как создать собственные блоки, см. в разделе Собственные блоки.

from diffusers import ModularPipeline, ComponentsManager

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

Когда блоки пайплайна определяют рабочие процессы (см. pipeline.blocks.available_workflows), передайте workflow=, чтобы оставить только блоки этого рабочего процесса — это та же обрезка, что и в get_workflow(). Тогда пайплайн объявляет только те компоненты, которые использует данный рабочий процесс, а его docstring описывает именно его входные данные.

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

Загрузка компонентов

ModularPipeline не инициализирует компоненты автоматически. Он загружает только конфигурацию и спецификации компонентов. Загрузить компоненты можно с помощью load_components().

Это загрузит все компоненты, для которых определена корректная спецификация загрузки.

import torch

pipeline.load_components(dtype=torch.float16)

Вы также можете загрузить конкретные компоненты по имени. В примере ниже загружается только text_encoder.

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

В конвейере, блоки которого определяют рабочие процессы, параметр workflow= загружает только те компоненты, которые используются этим рабочим процессом. Конвейер сохраняет все свои блоки, поэтому это удобный способ запускать один конвейер для разных рабочих процессов: каждый вызов добавляет лишь то, чего ещё не хватает новому рабочему процессу.

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

После загрузки вывод пайплайна показывает, какие компоненты загружены — первые два поля меняются с null на библиотеку и класс компонента.

pipeline

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

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

Параметры, передаваемые по имени, такие как dtype, variant, revision и quantization_config, передаются в from_pretrained() для каждого компонента. Можно передать одно значение, чтобы применить его ко всем компонентам, или словарь, чтобы задать значения для каждого компонента отдельно.

# 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() загружает только те компоненты, которые ещё не были загружены и для которых задано корректное описание загрузки. Это означает, что если вы уже установили компонент в конвейере, повторный вызов load_components() не перезагрузит его.

Обновление компонентов

update_components() заменяет компонент в конвейере новым. При обновлении компонента спецификации загрузки также обновляются в конфигурации конвейера, и load_components() будет пропускать его при последующих вызовах.

Из AutoModel

Вы можете передать объект модели, загруженный с помощью AutoModel.from_pretrained(). Модели, загруженные таким образом, автоматически помечаются информацией о загрузке.

from diffusers import AutoModel

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

Из ComponentSpec

Используйте get_component_spec(), чтобы получить копию текущей спецификации компонента, изменить её и загрузить новый компонент.

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)

Вы также можете создать ComponentSpec с нуля.

Не все компоненты загружаются из предобученных весов — некоторые создаются на основе конфигурации (перечислены в pipeline.config_component_names). Для них используйте create() вместо load().

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

Или просто передайте объект напрямую.

from diffusers.guiders import ClassifierFreeGuidance

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

См. руководство Guiders для получения подробной информации о доступных guiders и способах их настройки.

Разделение конвейера на этапы

Поскольку блоки комбинируемы, вы можете разобрать пайплайн и собрать из него отдельные пайплайны для каждого этапа. В примере ниже показано, как можно выделить блок text encoder из остальной части пайплайна, чтобы независимо закодировать промпт и передать эмбеддинги в основной пайплайн.

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 управляет памятью в нескольких конвейерах. В отличие от стратегий выгрузки в DiffusionPipeline, которые следуют фиксированному порядку, ComponentsManager принимает решения о выгрузке динамически при каждом проходе модели вперёд, исходя из текущего состояния памяти. Это означает, что он работает независимо от количества созданных конвейеров и порядка их выполнения. Подробнее см. в руководстве по ComponentsManager.

Если этапы конвейера используют общие компоненты (например, одна и та же VAE для кодирования и декодирования), вы можете использовать update_components(), чтобы передать уже загруженный компонент в другой конвейер, а не загружать его заново.

Модульный репозиторий

Репозиторий требуется, если блоки конвейера используют предобученные компоненты. Репозиторий предоставляет спецификации загрузки и метаданные.

ModularPipeline работает с обычными репозиториями diffusers без дополнительной настройки. Однако для большей гибкости можно создать модульный репозиторий. Модульный репозиторий содержит файл modular_model_index.json с тремя элементами.

• library и class показывают, из какой библиотеки был загружен компонент и его класс. Если null, компонент ещё не был загружен.
• loading_specs_dict содержит информацию, необходимую для загрузки компонента, например, репозиторий и подпапку, из которых он загружается.

Ключевое преимущество модульного репозитория состоит в том, что компоненты можно загружать из разных репозиториев. Например, diffusers/flux2-bnb-4bit-modular загружает квантованный трансформер из diffusers/FLUX.2-dev-bnb-4bit, а остальные компоненты — из black-forest-labs/FLUX.2-dev.

Чтобы преобразовать обычный репозиторий diffusers в модульный, создайте пайплайн с помощью обычного репозитория, а затем отправьте его на Hub. Сохранённый репозиторий будет содержать modular_model_index.json со всеми спецификациями загрузки.

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)

По умолчанию save_pretrained() записывает каждый загруженный в данный момент компонент, который Diffusers может сериализовать. Компоненты, которые не загружены или не могут быть сериализованы, не записываются и сохраняют свои существующие спецификации загрузки. Это даёт вам два способа сохранения в зависимости от ваших задач.

Сохранить автономную копию

Загрузите все компоненты, а затем сохраните. Каждая спецификация указывает на результат, поэтому он полностью перезагружается из одного места, включая офлайн-режим.

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

Путь назначения, записанный в modular_model_index.json, — это то место, куда save_pretrained() записывает данные: save_directory при сохранении локально или repo_id, если вы передаёте push_to_hub=True.

Начиная с huggingface_hub>=1.32.0, идентичные файлы, использующие Xet, могут разделять один кэшированный payload между любыми репозиториями Hub. Это касается отдельных файлов в шардированном чекпоинте: шард с одинаковым Xet-хэшем может быть переиспользован, даже если другие шарды отличаются. Оба процесса загрузки должны использовать один и тот же каталог кэша Hub с поддержкой символьных ссылок. Изменение dtype, способа сериализации или границ шардов может изменить хэши файлов и предотвратить переиспользование. Это экономит трафик загрузки и место на диске, но не память, используемую загруженными компонентами.

Общее кэширование работает в режиме best-effort. Установка переменной HF_HUB_DISABLE_SHARED_BLOBS=1 по-прежнему поддерживает загрузку, но кэширует отдельные копии. Существующие ссылки на компоненты в modular_model_index.json остаются без изменений; сохраняйте внешние ссылки для компонентов, которые не хранятся в модульном репозитории.

Обновление существующего модульного индекса

Миграция необязательна: ссылки на исходный репозиторий уже используют его кэшированные файлы. Обновите ссылку, если она использовалась только для предотвращения дублирования загрузок, и вы хотите, чтобы этот компонент загружался из модульного репозитория.

Сначала убедитесь, что модульный репозиторий содержит ту же конфигурацию компонентов и все необходимые файлы весов, включая индекс весов и каждый шард для шардированного чекпоинта. Совпадение хешей Xet определяет, какие файлы весов можно повторно использовать из общего кэша. Сохраняйте ссылки на компоненты, которые отсутствуют или намеренно загружаются из другого чекпоинта.

Измените только соответствующую спецификацию загрузки в локальной копии modular_model_index.json. Например, чтобы использовать VAE, хранящийся в vae/ в модульном репозитории, с тем же вариантом весов:

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

Это также удаляет устаревшее поле repo. Ревизия из исходного репозитория не определяет ревизию в целевом: используйте None для ветки по умолчанию целевого репозитория или зафиксируйте ревизию целевого репозитория, содержащую компонент. При необходимости скорректируйте subfolder и variant, если имена файлов различаются.

Перед публикацией обновленного индекса загрузите его с помощью ModularPipeline.from_pretrained("path/to/local-copy"), вызовите load_components(names="vae") и убедитесь, что компонент загружается с ожидаемыми параметрами. Загружайте только отредактированный индекс; повторное сохранение весов может изменить их хеши и помешать повторному использованию кэша.

Сохраняйте ссылки на существующие компоненты

Загружайте только новое (или ничего). Сохраняются только загруженные компоненты; остальные остаются указателями на исходный репозиторий. Используйте этот режим, если хотите заменить один компонент, продолжая загружать остальные из исходного репозитория. Например, сохраните собственный transformer, а остальные компоненты продолжат загружаться из базового репозитория.

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)

Передайте overwrite_modular_index=False, чтобы сохранить спецификации загрузки в modular_model_index.json без изменений. Сохранённый компонент с пустой спецификацией загрузки всё равно будет дополнен целевым значением, так как сохранять нечего.

Перемещение файлов любым другим способом не приводит к переписыванию индекса. Копия, загруженная с помощью hf download --local-dir, по-прежнему работает: при загрузке пайплайна из локального каталога каждый компонент, файлы которого присутствуют в этом каталоге, загружается из него, а не из указанного в индексе репозитория. Копия, загруженная с помощью hf upload, по-прежнему указывает на старое расположение, поэтому в этом случае индекс нужно обновить вручную.

Модульный репозиторий также может включать пользовательские блоки конвейера в виде Python-кода. Это позволяет делиться специализированными блоками, которые не являются нативными для Diffusers. Например, diffusers/Florence2-image-Annotator содержит пользовательские блоки наряду с конфигурацией загрузки:

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

Файл config.json содержит ключ auto_map, который указывает ModularPipeline, где искать пользовательские блоки:

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

Загружайте репозитории с пользовательским кодом с помощью trust_remote_code=True, как показано в from_pretrained. О том, как создавать и делиться собственными блоками, см. в разделе Custom blocks.