LoopSequentialPipelineBlocks

LoopSequentialPipelineBlocks adalah tipe multi-blok yang menyusun ModularPipelineBlocks lainnya dalam sebuah loop. Data mengalir secara sirkular, menggunakan inputs dan intermediate_outputs, dan setiap blok dijalankan secara iteratif. Ini biasanya digunakan untuk membuat loop denoising yang secara default bersifat iteratif.

Panduan ini menunjukkan cara membuat LoopSequentialPipelineBlocks.

Loop pembungkus

LoopSequentialPipelineBlocks, juga dikenal sebagai loop wrapper karena mendefinisikan struktur loop, variabel iterasi, dan konfigurasi. Di dalam loop wrapper, Anda memerlukan variabel berikut.

• loop_inputs adalah nilai yang diberikan pengguna dan setara dengan inputs.
• loop_intermediate_outputs adalah variabel antara baru yang dibuat oleh blok dan ditambahkan ke PipelineState. Ini setara dengan intermediate_outputs.
• Metode __call__ mendefinisikan struktur loop dan logika iterasi.

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

class LoopWrapper(LoopSequentialPipelineBlocks):
    model_name = "test"
    @property
    def description(self):
        return "I'm a loop!!"
    @property
    def loop_inputs(self):
        return [InputParam(name="num_steps")]
    @torch.no_grad()
    def __call__(self, components, state):
        block_state = self.get_block_state(state)
        # Loop structure - can be customized to your needs
        for i in range(block_state.num_steps):
            # loop_step executes all registered blocks in sequence
            components, block_state = self.loop_step(components, block_state, i=i)
        self.set_block_state(state, block_state)
        return components, state

Wrapper loop dapat meneruskan argumen tambahan, seperti indeks iterasi saat ini, ke blok loop.

Blok loop

Blok loop adalah ModularPipelineBlocks, tetapi metode __call__ berperilaku berbeda.

• Itu menerima variabel iterasi dari pembungkus loop.
• Ini bekerja langsung dengan BlockState alih-alih PipelineState.
• Hal ini tidak memerlukan pengambilan atau pembaruan BlockState.

Blok loop berbagi BlockState yang sama untuk memungkinkan nilai terakumulasi dan berubah pada setiap iterasi dalam loop.

class LoopBlock(ModularPipelineBlocks):
    model_name = "test"
    @property
    def inputs(self):
        return [InputParam(name="x")]
    @property
    def intermediate_outputs(self):
        # outputs produced by this block
        return [OutputParam(name="x")]
    @property
    def description(self):
        return "I'm a block used inside the `LoopWrapper` class"
    def __call__(self, components, block_state, i: int):
        block_state.x += 1
        return components, block_state

LoopSequentialPipelineBlocks

Gunakan metode from_blocks_dict() untuk menambahkan blok loop ke wrapper loop guna membuat LoopSequentialPipelineBlocks.

loop = LoopWrapper.from_blocks_dict({"block1": LoopBlock})

Tambahkan lebih banyak blok loop untuk dijalankan di dalam setiap iterasi dengan from_blocks_dict(). Hal ini memungkinkan Anda memodifikasi blok-blok tersebut tanpa mengubah logika loop itu sendiri.

loop = LoopWrapper.from_blocks_dict({"block1": LoopBlock(), "block2": LoopBlock})

Untuk menambahkan blok loop ke dalam loop denoising pipeline yang sudah ada, misalnya yang menjalankan fungsi callback_on_step_end di setiap langkah, lihat Modular pipelines di panduan callback.