diffusers/main/ar/modular_diffusers/custom_blocks.md

بناء كتل مخصصة

تُعدّ ModularPipelineBlocks اللبنات الأساسية لبناء ModularPipeline. يمكنك إنشاء كتل مخصصة عن طريق تحديد مدخلاتها ومخرجاتها ومنطق الحساب الخاص بها. يوضح هذا الدليل كيفية إنشاء كتلة مخصصة واستخدامها.

استكشف مجموعة Modular Diffusers Custom Blocks للحصول على كتل مخصصة رسمية.

هيكل المشروع

يجب أن يستخدم مشروع كتلتك المخصصة البنية التالية:

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

• block.py يحتوي على تنفيذ المخصص للكتلة
• modular_config.json يحتوي على البيانات الوصفية اللازمة لتحميل الكتلة

البدء السريع باستخدام القالب

أسرع طريقة لإنشاء كتلة مخصصة هي البدء من قالبنا. يوفر القالب بنية مشروع مُهيأة مسبقًا تتضمن ملفي block.py وmodular_config.json، بالإضافة إلى أمثلة موثقة تُظهر كيفية تعريف المكونات والمدخلات والمخرجات وطريقة __call__—لذا يمكنك التركيز على منطقك المخصص بدلاً من إعدادات الإعداد الأساسية.

تنزيل القالب

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
)

يحفظ هذا الأمر ملفات القالب محليًا في custom-block-template/ أو يمكنك استخدام local_dir للحفظ في موقع محدد.

التحرير محليًا

افتح block.py ونفّذ كتلتك المخصصة. يتضمن القالب أمثلة معلقة توضّح كيفية تعريف كل خاصية. راجع مثال Florence-2 أدناه للحصول على تنفيذ كامل.

اختبر كتلتك

from diffusers import ModularPipelineBlocks

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

الرفع إلى المستودع

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

مثال: Florence-2 Image Annotator

ينشئ هذا المثال كتلة مخصصة باستخدام Florence-2 لمعالجة صورة إدخال وتوليد قناع لإعادة الرسم.

تعريف المكونات

حدد المكونات التي يحتاجها الكتل، Florence2ForConditionalGeneration ومعالجها. عند تعريف المكونات، حدد name (الطريقة التي ستستخدمها للوصول إليه في الكود)، وtype_hint (فئة النموذج)، وpretrained_model_name_or_path (المكان الذي ستحمّل منه الأوزان).

# 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",
            ),
        ]

تعريف المدخلات والمخرجات

تشمل المدخلات الصورة ومهمة الترميز والموجّه. وتشمل المخرجات القناع المولّد والرموز.

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__

تحتوي طريقة __call__ على منطق الكتلة. اطلع على المدخلات عبر block_state، ونفّذ عملية الحساب، ثم أعد ضبط المخرجات إلى block_state.

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...

يمكنك الاطلاع على التنفيذ الكامل في diffusers/Florence2-image-Annotator.

استخدام كتل مخصصة

حمّل كتلة مخصصة باستخدام from_pretrained() و اضبط trust_remote_code=True.

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)

استخدم block لتوليد قناع:

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

قم بتجميعه مع كتل أخرى لإنشاء خط أنابيب جديد:

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

تعديل الكتل المخصصة

قم بتحرير الكتل المخصصة عن طريق تنزيلها محليًا. هذا هو نفس سير العمل كما في البدء السريع باستخدام القالب، لكن يبدأ من كتلة موجودة مسبقًا بدلاً من القالب.

استخدم الحجة local_dir لتنزيل كتلة مخصصة إلى مجلد محدد:

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

ستُنعكس أي تغييرات تُجرى على ملفات الكتل في هذا المجلد عند إعادة تحميل الكتلة. وعندما تكون مستعدًا لمشاركة تغييراتك، ارفعها إلى مستودع جديد:

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

الخطوات التالية

غطّى هذا الدليل إنشاء كتلة مخصصة واحدة. تعلّم كيفية تركيب عدة كتل معًا:

• SequentialPipelineBlocks: تسلسل الكتل لتنفيذها بالترتيب
• ConditionalPipelineBlocks: إنشاء كتل شرطية تختار مسارات تنفيذ مختلفة
• LoopSequentialPipelineBlocks: تعريف سير عمل تكرارية مثل حلقة إزالة الضوضاء

اجعل كتلتك المخصصة تعمل مع الواجهة البصرية لـ Mellon. راجع دليل Mellon Custom Blocks.

تصفّح مجموعة Modular Diffusers Custom Blocks للحصول على إلهام وكتل جاهزة للاستخدام.

التبعيات

إعلان تبعيات الحزم في الكتل المخصصة يمنع أخطاء الاستيراد وقت التشغيل لاحقًا. يتحقق Diffusers من التبعيات ويعيد تحذيرًا إذا كانت الحزمة مفقودة أو غير متوافقة.

اضبط خاصية _requirements في فئة الكتل لديك، بحيث تُخاطب أسماء الحزم إلى مواصفات الإصدارات.

from diffusers.modular_pipelines import PipelineBlock

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

عند وجود كتل ذات متطلبات مختلفة، يدمج Diffusers متطلباتها.

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,
})

عند حفظ هذا الكتلة باستخدام save_pretrained()، تُحفظ المتطلبات في ملف modular_config.json. عند تحميل هذه الكتلة، يتحقق Diffusers من كل متطلب مقارنةً بالبيئة الحالية. إذا كان هناك عدم تطابق أو لم يتم العثور على حزمة، يعيد Diffusers التحذير التالي.

# 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.