diffusers/main/ja/modular_diffusers/loop_sequential_pipeline_blocks.md

LoopSequentialPipelineBlocks

LoopSequentialPipelineBlocks は、他の ModularPipelineBlocks をループ内で組み合わせるマルチブロック型です。データは inputs と intermediate_outputs を用いて循環的に流れ、各ブロックが反復的に実行されます。これは、デフォルトで反復的なデノイジングループを作成するために一般的に使用されます。

このガイドでは、LoopSequentialPipelineBlocksの作成方法について説明します。

ループラッパー

LoopSequentialPipelineBlocks は、ループ構造、反復変数、および設定を定義するため、loop wrapper（ループラッパー）とも呼ばれます。loop wrapper 内では、以下の変数が必要です。

• loop_inputs はユーザーが指定する値であり、inputs と同等です。
• loop_intermediate_outputs は、ブロックによって新規に作成され、PipelineState に追加される中間変数です。intermediate_outputs と同等です。
• __call__ メソッドは、ループの構造と反復ロジックを定義します。

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

ループラッパーは、現在の反復インデックスなどの追加引数をループブロックに渡すことができます。

ループブロック

ループブロックはModularPipelineBlocksですが、__call__メソッドの動作が異なります。

• ループラッパーからイテレーション変数を受け取ります。
• PipelineStateではなく、BlockStateを直接扱います。
• BlockStateの取得や更新は不要です。

ループブロックは同じ BlockState を共有し、ループの各イテレーションで値が蓄積・変化できるようにします。

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

from_blocks_dict() メソッドを使用して、ループブロックをループラッパーに追加し、LoopSequentialPipelineBlocks を作成します。

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

from_blocks_dict() を使用して、各イテレーション内で実行されるループブロックを追加できます。これにより、ループロジック自体を変更せずにブロックを修正できます。

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

既存のパイプラインのデノイジングループにループブロックを追加するには、例えば各ステップでcallback_on_step_end関数を実行するブロックなど、Modular pipelinesを参照してください。