AutoPipelineBlocks

AutoPipelineBlocks, farklı iş akışlarını destekleyen blokları içeren çok blokludur. Çalışma zamanında sağlanan girdiye dayanarak çalıştırılacak alt blokları otomatik olarak seçer. Bu, genellikle metinden görüntüye, görüntüden görüntüye ve inpaint gibi birden fazla iş akışını kullanışlılık için tek bir pipeline içinde paketlemek için kullanılır.

Bu kılavuz, AutoPipelineBlocks oluşturmayı gösterir.

Metinden görüntüye, görüntüden görüntüye ve inpainting için üç ModularPipelineBlocks oluşturun. Bunlar, pipeline'da kullanılabilir farklı iş akışlarını temsil eder.

import torch
from diffusers.modular_pipelines import ModularPipelineBlocks, InputParam, OutputParam

class TextToImageBlock(ModularPipelineBlocks):
    model_name = "text2img"

    @property
    def inputs(self):
        return [InputParam(name="prompt")]

    @property
    def intermediate_outputs(self):
        return []

    @property
    def description(self):
        return "I'm a text-to-image workflow!"

    def __call__(self, components, state):
        block_state = self.get_block_state(state)
        print("running the text-to-image workflow")
        # Add your text-to-image logic here
        # For example: generate image from prompt
        self.set_block_state(state, block_state)
        return components, state

class ImageToImageBlock(ModularPipelineBlocks):
    model_name = "img2img"

    @property
    def inputs(self):
        return [InputParam(name="prompt"), InputParam(name="image")]

    @property
    def intermediate_outputs(self):
        return []

    @property
    def description(self):
        return "I'm an image-to-image workflow!"

    def __call__(self, components, state):
        block_state = self.get_block_state(state)
        print("running the image-to-image workflow")
        # Add your image-to-image logic here
        # For example: transform input image based on prompt
        self.set_block_state(state, block_state)
        return components, state

class InpaintBlock(ModularPipelineBlocks):
    model_name = "inpaint"

    @property
    def inputs(self):
        return [InputParam(name="prompt"), InputParam(name="image"), InputParam(name="mask")]

    @property
    def intermediate_outputs(self):
        return []

    @property
    def description(self):
        return "I'm an inpaint workflow!"

    def __call__(self, components, state):
        block_state = self.get_block_state(state)
        print("running the inpaint workflow")
        # Add your inpainting logic here
        # For example: fill masked areas based on prompt
        self.set_block_state(state, block_state)
        return components, state

Alt blok sınıflarının ve bunlara karşılık gelen blok adlarının bir listesini içeren bir AutoPipelineBlocks sınıfı oluşturun.

Ayrıca, block_trigger_inputs adlı, ilgili bloğu tetikleyen girdi adlarının listesini de dahil etmeniz gerekir. Çalışma zamanında bir tetikleyici girdi sağlanırsa, çalıştırılmak üzere o blok seçilir. Tetikleyici girdiler algılanmazsa çalıştırılacak varsayılan bloğu belirtmek için None kullanın.

Son olarak, hangi girdilerin hangi iş akışını tetiklediğini açıkça belirten bir description eklemek önemlidir. Bu, kullanıcıların belirli iş akışlarını nasıl çalıştıracağını anlamalarına yardımcı olur.

from diffusers.modular_pipelines import AutoPipelineBlocks

class AutoImageBlocks(AutoPipelineBlocks):
    # List of sub-block classes to choose from
    block_classes = [InpaintBlock, ImageToImageBlock, TextToImageBlock]
    # Names for each block in the same order
    block_names = ["inpaint", "img2img", "text2img"]
    # Trigger inputs that determine which block to run
    # - "mask" triggers inpaint workflow
    # - "image" triggers img2img workflow (but only if mask is not provided)
    # - if none of above, runs the text2img workflow (default)
    block_trigger_inputs = ["mask", "image", None]

    @property
    def description(self):
        return (
            "Pipeline generates images given different types of conditions!\n"
            + "This is an auto pipeline block that works for text2img, img2img and inpainting tasks.\n"
            + " - inpaint workflow is run when `mask` is provided.\n"
            + " - img2img workflow is run when `image` is provided (but only when `mask` is not provided).\n"
            + " - text2img workflow is run when neither `image` nor `mask` is provided.\n"
        )

Bir bloğun nasıl çalıştırılacağı ve hangi girdilerin gerekli olduğu konusunda herhangi bir karışıklığın önüne geçmek için bir description eklemek çok önemlidir. AutoPipelineBlocks kullanışlı olsa da, uygun şekilde açıklanmadığı takdirde koşullu mantığını anlamak zor olabilir.

AutoImageBlocks örneği oluşturun.

auto_blocks = AutoImageBlocks()

Daha karmaşık yapılandırmalar için, örneğin daha büyük pipeline'lerde alt blok olarak kullanıldığında iç içe geçmiş AutoPipelineBlocks blokları için, girdinize göre aslında çalıştırılan bloğu çıkarmak üzere get_execution_blocks() yöntemini kullanın.

auto_blocks.get_execution_blocks(mask=True)

KoşulluPipelineBlokları

AutoPipelineBlocks, ConditionalPipelineBlocks'in özel bir durumudur. AutoPipelineBlocks, bir tetikleyici girdinin sağlanıp sağlanmadığına göre blokları seçerken, ConditionalPipelineBlocks, select_block metodunda sağlanan özel seçim mantığına göre bir blok seçebilir.

İşte ConditionalPipelineBlocks kullanılarak doğrudan yazılmış aynı örnek:

from diffusers.modular_pipelines import ConditionalPipelineBlocks

class AutoImageBlocks(ConditionalPipelineBlocks):
    block_classes = [InpaintBlock, ImageToImageBlock, TextToImageBlock]
    block_names = ["inpaint", "img2img", "text2img"]
    block_trigger_inputs = ["mask", "image"]
    default_block_name = "text2img"

    @property
    def description(self):
        return (
            "Pipeline generates images given different types of conditions!\n"
            + "This is an auto pipeline block that works for text2img, img2img and inpainting tasks.\n"
            + " - inpaint workflow is run when `mask` is provided.\n"
            + " - img2img workflow is run when `image` is provided (but only when `mask` is not provided).\n"
            + " - text2img workflow is run when neither `image` nor `mask` is provided.\n"
        )

    def select_block(self, mask=None, image=None) -> str | None:
        if mask is not None:
            return "inpaint"
        if image is not None:
            return "img2img"
        return None  # falls back to default_block_name ("text2img")

block_trigger_inputs içinde listelenen girdiler, select_block() fonksiyonuna anahtar kelime argümanları olarak iletilir. select_block None döndürdüğünde, default_block_name değerine geri düşülür. default_block_name da None ise, tüm koşullu blok atlanır — bu, yalnızca belirli girdiler sağlandığında çalıştırılması gereken isteğe bağlı işleme adımları için kullanışlıdır.

İş Akışları

Koşullu bloklar içeren (AutoPipelineBlocks veya ConditionalPipelineBlocks) pipeline'lar birden çok iş akışını destekleyebilir — örneğin, SDXL modüler pipeline'ımız tek bir pipeline içinde onlarca iş akışını destekler. Ancak bu durum, kullanıcıların hangi iş akışlarının desteklendiğini ve bunların nasıl çalıştırılacağını bilmelerini karmaşık hale getirebilir. Pipeline oluşturucuları için, belirli bir iş akışına yalnızca ilgili blokları çıkarabilmek faydalıdır.

Her iş akışına bir ad vermek ve gerektirdiği girdileri açıkça listelemek için bir _workflow_map tanımlamanızı öneririz.

from diffusers.modular_pipelines import SequentialPipelineBlocks

class MyPipelineBlocks(SequentialPipelineBlocks):
    block_classes = [TextEncoderBlock, AutoImageBlocks, DecodeBlock]
    block_names = ["text_encoder", "auto_image", "decode"]

    _workflow_map = {
        "text2image": {"prompt": True},
        "image2image": {"image": True, "prompt": True},
        "inpaint": {"mask": True, "image": True, "prompt": True},
    }

Tüm yerleşik modüler boru hatlarımız önceden tanımlanmış iş akışlarıyla birlikte gelir. available_workflows özelliği, desteklenen tüm iş akışlarını listeler:

pipeline_blocks = MyPipelineBlocks()
pipeline_blocks.available_workflows
# ['text2image', 'image2image', 'inpaint']

Belirli bir bloğu denetlemek ve hata ayıklamak için get_workflow ile belirli bir iş akışını alın.

pipeline_blocks.get_workflow("inpaint")