ModularPipeline

ModularPipeline chuyển đổi các ModularPipelineBlocks thành một pipeline có thể thực thi, trong đó tải các mô hình và thực hiện các bước tính toán được định nghĩa trong các khối. Đây là giao diện chính để chạy một pipeline và API rất giống với DiffusionPipeline nhưng có một số khác biệt chính.

Việc tải là theo kiểu lười (lazy). Với DiffusionPipeline, from_pretrained() tạo pipeline và tải tất cả các mô hình cùng một lúc. Với ModularPipeline, việc tạo và tải là hai bước riêng biệt: from_pretrained() đọc cấu hình và biết nơi để tải từng thành phần, nhưng không thực sự tải trọng số mô hình. Bạn tải các mô hình sau đó bằng load_components(), nơi bạn truyền các đối số tải như dtype và quantization_config.
Two ways to create a pipeline. Bạn có thể sử dụng from_pretrained() với một kho lưu trữ mô hình diffusers hiện có — nó sẽ tự động ánh xạ đến các khối pipeline mặc định và sau đó chuyển đổi thành ModularPipeline mà không cần thiết lập thêm. Bạn có thể kiểm tra modular_pipelines_directory để xem những mô hình nào đang được hỗ trợ. Bạn cũng có thể tự lắp ráp pipeline của mình từ các ModularPipelineBlocks và chuyển đổi nó bằng phương thức init_pipeline() (xem Creating a pipeline để biết thêm chi tiết).
Việc chạy pipeline cũng tương tự. Sau khi tải xong, bạn gọi pipeline với các đối số quen thuộc. Một ModularPipeline duy nhất có thể hỗ trợ nhiều luồng công việc (text-to-image, image-to-image, inpainting, v.v.) khi các khối pipeline sử dụng AutoPipelineBlocks để tự động chọn luồng công việc dựa trên đầu vào của bạn.

Dưới đây là các ví dụ hoàn chỉnh cho text-to-image, image-to-image và inpainting với SDXL.

text-to-image

image-to-image

inpainting

Hướng dẫn này sẽ cho bạn thấy cách tạo một ModularPipeline, quản lý các thành phần của nó và chạy pipeline.

Tạo một pipeline

Có hai cách để tạo một ModularPipeline. Lắp ráp và tạo một pipeline từ các ModularPipelineBlocks bằng init_pipeline(), hoặc tải một pipeline có sẵn bằng from_pretrained().

Bạn cũng có thể khởi tạo một ComponentsManager để xử lý việc đặt thiết bị và quản lý bộ nhớ. Nếu bạn không cần offloading tự động, bạn có thể bỏ qua bước này và di chuyển pipeline sang thiết bị của bạn một cách thủ công bằng pipeline.to("cuda").

Tham khảo tài liệu ComponentsManager để biết thêm chi tiết về cách nó có thể hỗ trợ quản lý các thành phần xuyên suốt các quy trình làm việc khác nhau.

init_pipeline

init_pipeline() chuyển đổi bất kỳ ModularPipelineBlocks nào thành ModularPipeline.

Hãy định nghĩa một khối tối thiểu để xem cách nó hoạt động:

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

Gọi init_pipeline() để chuyển đổi nó thành một pipeline. Thuộc tính blocks trên pipeline chính là các khối mà nó được tạo ra từ đó — nó xác định các đầu vào, đầu ra và logic tính toán mong đợi.

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

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

Blocks có thể thay đổi — bạn có thể tự do thêm, xóa hoặc hoán đổi các block trước khi tạo pipeline. Tuy nhiên, một khi pipeline đã được tạo, việc sửa đổi pipeline.blocks sẽ không ảnh hưởng đến pipeline vì nó trả về một bản sao. Nếu bạn muốn cấu trúc block khác, hãy tạo một pipeline mới sau khi đã sửa đổi các block.

Khi bạn gọi init_pipeline() mà không có kho lưu trữ, nó sẽ sử dụng pretrained_model_name_or_path được định nghĩa trong ComponentSpec của khối để xác định vị trí tải mỗi thành phần. In pipeline sẽ hiển thị cấu hình tải thành phần.

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

Nếu bạn truyền một kho lưu trữ vào init_pipeline(), nó sẽ ghi đè đường dẫn tải bằng cách khớp các thành phần của khối của bạn với cấu hình pipeline trong kho lưu trữ đó (model_index.json hoặc modular_model_index.json). Xem Kho lưu trữ Modular để biết cách các thông số tải được ghi lại và lưu trữ.

Trong ví dụ bên dưới, pretrained_model_name_or_path sẽ được cập nhật thành "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
    }
  ]
}

Nếu một thành phần trong khối của bạn không tồn tại trong kho, nó sẽ vẫn là null và bị bỏ qua trong quá trình load_components().

from_pretrained

from_pretrained() là cách tiện lợi để tạo một ModularPipeline mà không cần tự định nghĩa các khối.

Nó hoạt động với ba loại kho lưu trữ.

Một kho diffusers thông thường. Truyền bất kỳ kho model nào được hỗ trợ và nó sẽ tự động ánh xạ sang các khối pipeline mặc định. Các model hiện được hỗ trợ bao gồm SDXL, Wan, Qwen, Z-Image, Flux và Flux2.

from diffusers import ModularPipeline, ComponentsManager

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

Một kho lưu trữ dạng module. Các kho lưu trữ này chứa modular_model_index.json chỉ định vị trí tải mỗi thành phần — các thành phần có thể đến từ những kho lưu trữ khác nhau và chính kho lưu trữ dạng module có thể không chứa bất kỳ trọng số mô hình nào. Ví dụ, diffusers/flux2-bnb-4bit-modular tải một transformer đã lượng tử hóa từ một kho lưu trữ và các thành phần còn lại từ một kho khác. Xem Kho lưu trữ dạng module để biết thêm chi tiết về định dạng.

from diffusers import ModularPipeline, ComponentsManager

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

Một kho lưu trữ dạng module với mã tùy chỉnh. Một số kho lưu trữ bao gồm các khối pipeline tùy chỉnh cùng với cấu hình nạp. Thêm trust_remote_code=True để nạp chúng. Xem Các khối tùy chỉnh để biết cách tạo khối của riêng bạn.

from diffusers import ModularPipeline, ComponentsManager

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

Khi các khối pipeline định nghĩa các luồng công việc (xem pipeline.blocks.available_workflows), hãy truyền workflow= để chỉ giữ lại các khối của luồng công việc đó — cùng một phép cắt tỉa như get_workflow(). Khi đó, pipeline chỉ khai báo các thành phần mà luồng công việc đó sử dụng, và docstring của nó mô tả chính xác các đầu vào của luồng công việc đó.

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

Tải các thành phần

ModularPipeline không tự động khởi tạo với các thành phần. Nó chỉ tải cấu hình và thông số kỹ thuật của các thành phần. Bạn có thể tải các thành phần bằng load_components().

Điều này sẽ tải tất cả các thành phần có thông số tải hợp lệ.

import torch

pipeline.load_components(dtype=torch.float16)

Bạn cũng có thể tải các thành phần cụ thể theo tên. Ví dụ bên dưới chỉ tải text_encoder.

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

Trên một pipeline mà các khối của nó định nghĩa các luồng công việc, workflow= chỉ nạp các thành phần mà luồng công việc đó sử dụng. Pipeline giữ lại tất cả các khối của nó, vì vậy đây là cách tiện lợi để chạy một pipeline xuyên suốt các luồng công việc: mỗi lần gọi chỉ bổ sung những gì mà luồng công việc mới còn thiếu.

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

Sau khi tải, việc in ra pipeline sẽ hiển thị các thành phần đã được tải — hai trường đầu tiên thay đổi từ null thành thư viện và lớp của thành phần đó.

pipeline

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

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

Các đối số từ khóa như dtype, variant, revision và quantization_config được truyền qua from_pretrained() cho từng thành phần. Bạn có thể truyền một giá trị duy nhất để áp dụng cho tất cả các thành phần, hoặc một dict để thiết lập giá trị riêng cho từng thành phần.

# 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() chỉ tải các thành phần chưa được tải và có thông số tải hợp lệ. Điều này có nghĩa là nếu bạn đã thiết lập một thành phần trên pipeline, việc gọi lại load_components() sẽ không tải lại thành phần đó.

Cập nhật các thành phần

update_components() thay thế một thành phần trong pipeline bằng một thành phần mới. Khi một thành phần được cập nhật, các thông số tải cũng được cập nhật trong cấu hình pipeline và load_components() sẽ bỏ qua nó trong các lần gọi sau.

Từ AutoModel

Bạn có thể truyền một đối tượng model được nạp bằng AutoModel.from_pretrained(). Các model được nạp theo cách này sẽ tự động được gắn thẻ với thông tin nạp của chúng.

from diffusers import AutoModel

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

Từ ComponentSpec

Sử dụng get_component_spec() để lấy một bản sao của thông số cấu hình thành phần hiện tại, chỉnh sửa nó và tải một thành phần mới.

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)

Bạn cũng có thể tạo một ComponentSpec từ đầu.

Không phải tất cả các thành phần đều được nạp từ trọng số được huấn luyện trước — một số được tạo từ cấu hình (được liệt kê dưới pipeline.config_component_names). Đối với những thành phần này, hãy sử dụng create() thay vì load().

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

Hoặc đơn giản là truyền trực tiếp đối tượng.

from diffusers.guiders import ClassifierFreeGuidance

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

Xem hướng dẫn Guiders để biết thêm chi tiết về các guider có sẵn và cách cấu hình chúng.

Chia pipeline thành các giai đoạn

Vì các khối có thể kết hợp được, bạn có thể tách một pipeline ra và tái cấu trúc nó thành các pipeline riêng biệt cho từng giai đoạn. Ví dụ bên dưới cho thấy cách chúng ta có thể tách khối text encoder khỏi phần còn lại của pipeline, để bạn có thể mã hóa prompt một cách độc lập và truyền các embeddings vào pipeline chính.

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 quản lý bộ nhớ giữa nhiều pipeline. Khác với các chiến lược offloading trong DiffusionPipeline tuân theo một thứ tự cố định, ComponentsManager đưa ra quyết định offloading một cách động mỗi khi thực hiện một bước forward của mô hình, dựa trên tình trạng bộ nhớ hiện tại. Điều này có nghĩa là nó hoạt động bất kể bạn tạo bao nhiêu pipeline hay chạy chúng theo thứ tự nào. Xem hướng dẫn ComponentsManager để biết thêm chi tiết.

Nếu các giai đoạn pipeline chia sẻ các thành phần (ví dụ: cùng một VAE được dùng cho cả mã hóa và giải mã), bạn có thể sử dụng update_components() để truyền một thành phần đã được tải sang pipeline khác thay vì tải lại.

Kho lưu trữ mô-đun

Một kho lưu trữ là bắt buộc nếu các khối pipeline sử dụng thành phần được huấn luyện trước. Kho lưu trữ cung cấp các thông số cấu hình nạp và dữ liệu mô tả.

ModularPipeline hoạt động với các kho diffusers thông thường ngay khi cài đặt. Tuy nhiên, bạn cũng có thể tạo một kho modular để có nhiều linh hoạt hơn. Một kho modular chứa tệp modular_model_index.json bao gồm 3 thành phần sau.

library và class cho biết thư viện mà thành phần được nạp từ đó và lớp của nó. Nếu là null, thành phần chưa được nạp.
loading_specs_dict chứa thông tin cần thiết để nạp thành phần, chẳng hạn như kho và thư mục con mà nó được nạp từ đó.

Lợi thế chính của một kho lưu trữ dạng module là các thành phần có thể được tải từ các kho lưu trữ khác nhau. Ví dụ, diffusers/flux2-bnb-4bit-modular tải một transformer đã lượng tử hóa từ diffusers/FLUX.2-dev-bnb-4bit trong khi tải các thành phần còn lại từ black-forest-labs/FLUX.2-dev.

Để chuyển đổi một kho diffusers thông thường thành kho dạng module, hãy tạo pipeline bằng kho thông thường, sau đó đẩy lên Hub. Kho đã lưu sẽ chứa modular_model_index.json với tất cả các thông số tải.

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)

Mặc định, save_pretrained() ghi lại từng thành phần đang được nạp mà Diffusers có thể phân tích chuỗi. Các thành phần chưa được nạp, hoặc không thể phân tích chuỗi, sẽ không được ghi và giữ nguyên thông số nạp hiện có. Điều này cho bạn hai cách để lưu, tùy thuộc vào mục đích của bạn.

Lưu bản sao tự chứa

Tải tất cả các thành phần, sau đó lưu lại. Mọi thông số kỹ thuật đều trỏ đến kết quả, do đó chúng được tải lại hoàn toàn từ một vị trí duy nhất, kể cả khi ngoại tuyến.

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

Đích được ghi trong modular_model_index.json là nơi save_pretrained() ghi: save_directory cho việc lưu cục bộ, hoặc repo_id khi bạn truyền push_to_hub=True.

Với huggingface_hub>=1.32.0, các tệp Xet-backed giống hệt nhau có thể chia sẻ một payload được cache chung giữa bất kỳ kho Hub nào. Điều này bao gồm cả các tệp riêng lẻ trong một checkpoint phân mảnh: một shard có cùng Xet hash có thể được tái sử dụng ngay cả khi các shard khác khác biệt. Cả hai lần tải xuống đều phải sử dụng cùng một thư mục cache của Hub với hỗ trợ symlink. Việc thay đổi dtype, phương thức serialize hoặc ranh giới shard có thể làm thay đổi hash của tệp và ngăn cản việc tái sử dụng. Điều này giúp tiết kiệm lưu lượng tải xuống và dung lượng ổ đĩa, chứ không phải bộ nhớ được sử dụng bởi các thành phần đã được nạp.

Caching dùng chung là theo cơ chế best-effort. Việc đặt HF_HUB_DISABLE_SHARED_BLOBS=1 vẫn hỗ trợ việc tải, nhưng sẽ cache các bản sao riêng biệt. Các tham chiếu thành phần hiện có trong modular_model_index.json không thay đổi; hãy giữ các tham chiếu bên ngoài cho các thành phần không được lưu trữ trong kho modular.

Cập nhật chỉ mục modular hiện có

Việc di chuyển là tùy chọn: các tham chiếu đến kho lưu trữ gốc đã tự động tái sử dụng các tệp được lưu trong bộ nhớ đệm của nó. Hãy cập nhật một tham chiếu nếu nó chỉ được dùng để tránh tải xuống trùng lặp và bạn muốn thành phần đó được nạp từ kho lưu trữ modular thay thế.

Đầu tiên, kiểm tra rằng kho modular chứa cùng một cấu hình thành phần và tất cả các tệp trọng số cần thiết, bao gồm chỉ mục trọng số và mọi shard cho checkpoint dạng sharded. Các hash Xet khớp sẽ xác định tệp trọng số nào có thể tái sử dụng bộ nhớ đệm dùng chung. Giữ lại các tham chiếu đến các thành phần vắng mặt hoặc được tải có chủ đích từ một checkpoint khác.

Chỉ chỉnh sửa thông số tải liên quan trong bản sao cục bộ của modular_model_index.json. Ví dụ, để sử dụng VAE được lưu dưới vae/ trong kho modular với cùng biến thể trọng số:

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

Điều này cũng xóa bỏ trường repo cũ. Một revision từ kho gốc không xác định được revision trong kho đích: hãy dùng None cho nhánh mặc định của kho đích, hoặc cố định một revision đích chứa thành phần đó. Điều chỉnh subfolder và variant nếu tên tệp của chúng khác nhau.

Trước khi xuất bản chỉ mục đã cập nhật, hãy tải nó bằng ModularPipeline.from_pretrained("path/to/local-copy"), gọi load_components(names="vae"), và xác minh rằng thành phần được tải với các tham số mong đợi. Chỉ tải lên chỉ mục đã chỉnh sửa; việc lưu lại trọng số có thể làm thay đổi băm của chúng và ngăn chặn việc tái sử dụng bộ nhớ đệm.

Giữ các tham chiếu đến các thành phần hiện có

Chỉ tải những gì mới (hoặc không tải gì cả). Chỉ các thành phần đã tải sẽ được lưu; mọi thứ khác vẫn là con trỏ trỏ về kho lưu trữ gốc. Sử dụng chế độ này khi bạn muốn thay thế một thành phần trong khi tiếp tục tải các thành phần khác từ kho lưu trữ gốc. Ví dụ, lưu một transformer tùy chỉnh trong khi các thành phần còn lại tiếp tục được tải từ kho lưu trữ cơ sở.

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)

Truyền overwrite_modular_index=False để giữ nguyên các thông số nạp trong modular_model_index.json. Một thành phần đã lưu có thông số nạp rỗng vẫn sẽ được điền vào đích, vì không có gì để bảo tồn.

Việc di chuyển các tệp theo cách khác sẽ không viết lại chỉ mục. Bản sao được tải xuống bằng hf download --local-dir vẫn hoạt động: khi một pipeline được nạp từ một thư mục cục bộ, mọi thành phần có tệp hiện diện trong thư mục đó sẽ được nạp từ thư mục đó thay vì từ kho đã ghi nhận. Bản sao được tải lên bằng hf upload vẫn trỏ đến vị trí cũ, vì vậy trong trường hợp đó, hãy cập nhật chỉ mục thủ công.

Một kho lưu trữ dạng module cũng có thể bao gồm các khối pipeline tùy chỉnh dưới dạng mã Python. Điều này cho phép bạn chia sẻ các khối chuyên dụng không có sẵn trong Diffusers. Ví dụ, diffusers/Florence2-image-Annotator chứa các khối tùy chỉnh cùng với cấu hình tải:

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

Tệp config.json chứa một khóa auto_map cho ModularPipeline biết vị trí tìm các khối tùy chỉnh:

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

Tải các kho mã tùy chỉnh với trust_remote_code=True như được minh họa trong from_pretrained. Xem Custom blocks để biết cách tạo và chia sẻ khối của riêng bạn.

Update on GitHub