AutoPipelineBlocks

AutoPipelineBlocks adalah tipe multi-blok yang berisi blok-blok yang mendukung berbagai alur kerja. Ia secara otomatis memilih sub-blok mana yang akan dijalankan berdasarkan masukan yang diberikan saat runtime. Hal ini biasanya digunakan untuk mengemas beberapa alur kerja—text-to-image, image-to-image, inpaint—ke dalam satu pipeline demi kemudahan.

Panduan ini menunjukkan cara membuat AutoPipelineBlocks.

Buat tiga ModularPipelineBlocks untuk text-to-image, image-to-image, dan inpainting. Blok-blok ini mewakili berbagai alur kerja yang tersedia di pipeline.

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

Buat kelas AutoPipelineBlocks yang berisi daftar kelas sub-blok dan nama blok yang sesuai.

Anda juga perlu menyertakan block_trigger_inputs, yaitu daftar nama input yang memicu blok yang sesuai. Jika input pemicu diberikan saat runtime, maka blok tersebut akan dipilih untuk dijalankan. Gunakan None untuk menentukan blok default yang akan dijalankan jika tidak ada input pemicu yang terdeteksi.

Terakhir, penting untuk menyertakan description yang menjelaskan dengan jelas input mana yang memicu workflow mana. Hal ini membantu pengguna memahami cara menjalankan workflow tertentu.

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

Sangat penting untuk menyertakan description guna menghindari kebingungan tentang cara menjalankan blok dan input apa yang diperlukan. Meskipun AutoPipelineBlocks praktis, logika kondisional di dalamnya mungkin sulit dipahami jika tidak dijelaskan dengan baik.

Buat sebuah instance dari AutoImageBlocks.

auto_blocks = AutoImageBlocks()

Untuk komposisi yang lebih kompleks, seperti blok AutoPipelineBlocks bersarang ketika digunakan sebagai sub-blok dalam pipeline yang lebih besar, gunakan metode get_execution_blocks() untuk mengekstrak blok yang benar-benar dijalankan berdasarkan input Anda.

auto_blocks.get_execution_blocks(mask=True)

ConditionalPipelineBlocks

AutoPipelineBlocks adalah kasus khusus dari ConditionalPipelineBlocks. Sementara AutoPipelineBlocks memilih blok berdasarkan apakah input pemicu diberikan atau tidak, ConditionalPipelineBlocks mampu memilih blok berdasarkan logika seleksi kustom yang diberikan di metode select_block.

Berikut adalah contoh yang sama yang ditulis menggunakan ConditionalPipelineBlocks secara langsung:

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

Input yang tercantum dalam block_trigger_inputs diteruskan sebagai argumen kata kunci ke select_block(). Ketika select_block mengembalikan None, sistem beralih ke default_block_name. Jika default_block_name juga bernilai None, seluruh blok kondisional akan dilewati — ini berguna untuk langkah pemrosesan opsional yang hanya harus dijalankan ketika input tertentu disediakan.

Alur Kerja

Pipelines yang memuat blok kondisional (AutoPipelineBlocks atau ConditionalPipelineBlocks) dapat mendukung beberapa alur kerja — misalnya, pipeline modular SDXL kami mendukung belasan alur kerja dalam satu pipeline. Namun, hal ini juga berarti pengguna dapat merasa bingung untuk mengetahui alur kerja apa saja yang didukung dan cara menjalankannya. Bagi pembangun pipeline, sangat berguna untuk dapat mengekstrak hanya blok-blok yang relevan dengan alur kerja tertentu.

Kami merekomendasikan untuk mendefinisikan _workflow_map untuk memberikan nama pada setiap workflow dan secara eksplisit mencantumkan input yang dibutuhkan.

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

Semua pipeline modular bawaan kami dilengkapi dengan alur kerja yang sudah didefinisikan sebelumnya. Properti available_workflows mencantumkan semua alur kerja yang didukung:

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

Ambil workflow tertentu dengan get_workflow untuk memeriksa dan men-debug blok tertentu yang menjalankan workflow.

pipeline_blocks.get_workflow("inpaint")