Özel Bloklar Oluşturma

ModularPipelineBlocks, bir ModularPipeline'in temel yapı taşlarıdır. Girdilerini, çıktıları ve hesaplama mantığını tanımlayarak özel bloklar oluşturabilirsiniz. Bu kılavuz, özel bir blok oluşturma ve kullanma yöntemini gösterir.

Resmi özel bloklar için Modular Diffusers Custom Blocks koleksiyonunu keşfedin.

Proje Yapısı

Özel blok projeniz aşağıdaki yapıyı kullanmalıdır:

.
├── block.py
└── modular_config.json

block.py özel blok uygulamasını içerir
modular_config.json, bloğu yüklemek için gereken meta verileri içerir

Şablon ile Hızlı Başlangıç

Özel bir blok oluşturmanın en hızlı yolu, şablonumuzdan başlamaktır. Şablon, block.py ve modular_config.json dosyaları ile bileşenlerin, girdilerin, çıktıların ve __call__ metodunun nasıl tanımlanacağını gösteren yorumlu örnekler içeren önceden yapılandırılmış bir proje yapısı sunar; böylece standart kurulum yerine özel mantığınıza odaklanabilirsiniz.

Şablonu indirin

from diffusers import ModularPipelineBlocks

model_id = "diffusers/custom-block-template"
local_dir = model_id.split("/")[-1]

blocks = ModularPipelineBlocks.from_pretrained(
    model_id, 
    trust_remote_code=True, 
    local_dir=local_dir
)

Bu, şablon dosyalarını yerel olarak custom-block-template/ dizinine kaydeder veya belirli bir konuma kaydetmek için local_dir parametresini kullanabilirsiniz.

Yerel olarak düzenle

block.py dosyasını açın ve özel bloğunuzu uygulayın. Şablon, her özelliğin nasıl tanımlanacağını gösteren yorum satırlı örnekler içerir. Tam bir uygulama için aşağıdaki Florence-2 örneğine bakın.

Blokunuzu test edin

from diffusers import ModularPipelineBlocks

blocks = ModularPipelineBlocks.from_pretrained(local_dir, trust_remote_code=True)
pipeline = blocks.init_pipeline()
output = pipeline(...)  # your inputs here

Hub'a Yükle

pipeline.save_pretrained(local_dir, repo_id="your-username/your-block-name", push_to_hub=True)

Örnek: Florence-2 Görsel Anlatıcı

Bu örnek, bir girdi görüntüsünü işleyip inpainting için bir maske üretmek üzere Florence-2 ile özel bir blok oluşturur.

Bileşenleri tanımlayın

Bloğun ihtiyaç duyduğu bileşenleri, Florence2ForConditionalGeneration ve işlemcisini tanımlayın. Bileşenleri tanımlarken name (kodda nasıl erişileceğini), type_hint (model sınıfı) ve pretrained_model_name_or_path (ağırlıkların nereden yükleneceğini) belirtin.

# Inside block.py
from diffusers.modular_pipelines import ModularPipelineBlocks, ComponentSpec
from transformers import AutoProcessor, Florence2ForConditionalGeneration


class Florence2ImageAnnotatorBlock(ModularPipelineBlocks):

    @property
    def expected_components(self):
        return [
            ComponentSpec(
                name="image_annotator",
                type_hint=Florence2ForConditionalGeneration,
                pretrained_model_name_or_path="florence-community/Florence-2-base-ft",
            ),
            ComponentSpec(
                name="image_annotator_processor",
                type_hint=AutoProcessor,
                pretrained_model_name_or_path="florence-community/Florence-2-base-ft",
            ),
        ]

Girdileri ve çıktıları tanımlayın

Girdiler arasında görüntü, etiketleme görevi ve istek (prompt) yer alır. Çıktılar arasında üretilen maske ve etiketler bulunur.

from typing import List, Union
from PIL import Image
from diffusers.modular_pipelines import InputParam, OutputParam


class Florence2ImageAnnotatorBlock(ModularPipelineBlocks):

    # ... expected_components from above ...

    @property
    def inputs(self) -> List[InputParam]:
        return [
            InputParam(
                "image",
                type_hint=Union[Image.Image, List[Image.Image]],
                required=True,
                description="Image(s) to annotate",
            ),
            InputParam(
                "annotation_task",
                type_hint=str,
                default="<REFERRING_EXPRESSION_SEGMENTATION>",
                description="Annotation task to perform (e.g., <OD>, <CAPTION>, <REFERRING_EXPRESSION_SEGMENTATION>)",
            ),
            InputParam(
                "annotation_prompt",
                type_hint=str,
                required=True,
                description="Prompt to provide context for the annotation task",
            ),
            InputParam(
                "annotation_output_type",
                type_hint=str,
                default="mask_image",
                description="Output type: 'mask_image', 'mask_overlay', or 'bounding_box'",
            ),
        ]

    @property
    def intermediate_outputs(self) -> List[OutputParam]:
        return [
            OutputParam(
                "mask_image",
                type_hint=Image.Image,
                description="Inpainting mask for the input image",
            ),
            OutputParam(
                "annotations",
                type_hint=dict,
                description="Raw annotation predictions",
            ),
            OutputParam(
                "image",
                type_hint=Image.Image,
                description="Annotated image",
            ),
        ]

call metodunu uygulayın

__call__ yöntemi bloğun mantığını içerir. Girdilere block_state üzerinden erişin, hesaplamanızı çalıştırın ve çıktıları block_state içine geri ayarlayın.

import torch
from diffusers.modular_pipelines import PipelineState


class Florence2ImageAnnotatorBlock(ModularPipelineBlocks):

    # ... expected_components, inputs, intermediate_outputs from above ...

    @torch.no_grad()
    def __call__(self, components, state: PipelineState) -> PipelineState:
        block_state = self.get_block_state(state)
        
        images, annotation_task_prompt = self.prepare_inputs(
            block_state.image, block_state.annotation_prompt
        )
        task = block_state.annotation_task
        fill = block_state.fill
        
        annotations = self.get_annotations(
            components, images, annotation_task_prompt, task
        )
        block_state.annotations = annotations
        if block_state.annotation_output_type == "mask_image":
            block_state.mask_image = self.prepare_mask(images, annotations)
        else:
            block_state.mask_image = None

        if block_state.annotation_output_type == "mask_overlay":
            block_state.image = self.prepare_mask(images, annotations, overlay=True, fill=fill)

        elif block_state.annotation_output_type == "bounding_box":
            block_state.image = self.prepare_bounding_boxes(images, annotations)

        self.set_block_state(state, block_state)

        return components, state
    
    # Helper methods for mask/bounding box generation...

Tamamını diffusers/Florence2-image-Annotator adresinde inceleyin.

Özel Blokların Kullanımı

Özel bir bloğu from_pretrained() ile yükleyin ve trust_remote_code=True olarak ayarlayın.

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

# Load the Florence-2 annotator pipeline
image_annotator = ModularPipeline.from_pretrained(
    "diffusers/Florence2-image-Annotator",
    trust_remote_code=True
)

# Check the docstring to see inputs/outputs
print(image_annotator.blocks.doc)

Bir maske oluşturmak için bloğu kullanın:

image_annotator.load_components(dtype=torch.bfloat16)
image_annotator.to("cuda")  # or "mps", "xpu", "cpu"

image = load_image("https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/transformers/tasks/car.jpg")
image = image.resize((1024, 1024))
prompt = ["A red car"]
annotation_task = "<REFERRING_EXPRESSION_SEGMENTATION>"
annotation_prompt = ["the car"]

mask_image = image_annotator_node(
    prompt=prompt,
    image=image,
    annotation_task=annotation_task,
    annotation_prompt=annotation_prompt,
    annotation_output_type="mask_image",
).images
mask_image[0].save("car-mask.png")

Yeni bir pipeline oluşturmak için diğer bloklarla birleştirin:

# Get the annotator block
annotator_block = image_annotator.blocks

# Get an inpainting workflow and insert the annotator at the beginning
inpaint_blocks = ModularPipeline.from_pretrained("Qwen/Qwen-Image").blocks.get_workflow("inpainting")
inpaint_blocks.sub_blocks.insert("image_annotator", annotator_block, 0)

# Initialize the combined pipeline
pipe = inpaint_blocks.init_pipeline()
pipe.load_components(dtype=torch.float16, device="cuda")  # or "mps", "xpu", "cpu"

# Now the pipeline automatically generates masks from prompts
output = pipe(
    prompt=prompt,
    image=image,
    annotation_task=annotation_task,
    annotation_prompt=annotation_prompt,
    annotation_output_type="mask_image",
    num_inference_steps=35,
    guidance_scale=7.5,
    strength=0.95,
    output="images"
)
output[0].save("florence-inpainting.png")

Özel blokları düzenleme

Özel blokları düzenlemek için bunları yerel olarak indirin. Bu, Şablonla Hızlı Başlangıç ile aynı iş akışıdır, ancak şablondan değil, mevcut bir bloktan başlar.

Özel bir bloğu belirli bir klasöre indirmek için local_dir argümanını kullanın:

from diffusers import ModularPipelineBlocks

# Download to a local folder for editing
annotator_block = ModularPipelineBlocks.from_pretrained(
    "diffusers/Florence2-image-Annotator",
    trust_remote_code=True,
    local_dir="./my-florence-block"
)

Bu klasördeki blok dosyalarında yapılan değişiklikler, bloğu yeniden yüklediğinizde yansıtılır. Değişikliklerinizi paylaşmaya hazır olduğunuzda, yeni bir depoya yükleyin:

pipeline = annotator_block.init_pipeline()
pipeline.save_pretrained("./my-florence-block", repo_id="your-username/my-custom-florence", push_to_hub=True)

Sonraki Adımlar

Blok türlerini öğrenin

Mellon'da Kullanın

Mevcut blokları keşfedin

Bağımlılıklar

Özel bloklarda paket bağımlılıklarını bildirmek, daha sonraki çalışma zamanı içe aktarma hatalarını önler. Diffusers, bağımlılıkları doğrular ve bir paket eksik veya uyumsuzsa bir uyarı döndürür.

Blok sınıfınızda bir _requirements özniteliği tanımlayın ve paket adlarını sürüm belirleyicilerine eşleyin.

from diffusers.modular_pipelines import PipelineBlock

class MyCustomBlock(PipelineBlock):
    _requirements = {
        "transformers": ">=4.44.0",
        "sentencepiece": ">=0.2.0"
    }

Gereksinimleri farklı olan bloklar olduğunda, Diffusers bu gereksinimleri birleştirir.

from diffusers.modular_pipelines import SequentialPipelineBlocks

class BlockA(PipelineBlock):
    _requirements = {"transformers": ">=4.44.0"}
    # ...

class BlockB(PipelineBlock):
    _requirements = {"sentencepiece": ">=0.2.0"}
    # ...

pipe = SequentialPipelineBlocks.from_blocks_dict({
    "block_a": BlockA,
    "block_b": BlockB,
})

Bu blok save_pretrained() ile kaydedildiğinde, gereksinimler modular_config.json dosyasına kaydedilir. Bu blok yüklendiğinde Diffusers, her bir gereksinimi mevcut ortamla karşılaştırır. Bir uyumsuzluk varsa veya bir paket bulunamazsa Diffusers aşağıdaki uyarıyı döndürür.

# missing package
xyz-package was specified in the requirements but wasn't found in the current environment.

# version mismatch
xyz requirement 'specific-version' is not satisfied by the installed version 'actual-version'. Things might work unexpected.

Update on GitHub

Diffusers

Özel Bloklar Oluşturma

Proje Yapısı

Şablon ile Hızlı Başlangıç

Şablonu indirin

Yerel olarak düzenle

Blokunuzu test edin

Hub'a Yükle

Örnek: Florence-2 Görsel Anlatıcı

Bileşenleri tanımlayın

Girdileri ve çıktıları tanımlayın

__call__ metodunu uygulayın

Özel Blokların Kullanımı

Özel blokları düzenleme

Sonraki Adımlar

Bağımlılıklar

call metodunu uygulayın