LoopSequentialPipelineBlocks

LoopSequentialPipelineBlocks são um tipo de bloco múltiplo que compõe outros ModularPipelineBlocks em um loop. Os dados fluem circularmente, usando inputs e intermediate_outputs, e cada bloco é executado iterativamente. Isso é tipicamente usado para criar um loop de desruidização, que é iterativo por padrão.

Este guia mostra como criar LoopSequentialPipelineBlocks.

Wrapper de loop

LoopSequentialPipelineBlocks, também conhecido como loop wrapper, pois define a estrutura do loop, as variáveis de iteração e a configuração. Dentro do loop wrapper, você precisa das seguintes variáveis.

• loop_inputs são valores fornecidos pelo usuário e equivalentes a inputs.
• loop_intermediate_outputs são novas variáveis intermediárias criadas pelo bloco e adicionadas ao PipelineState. É equivalente a intermediate_outputs.
• O método __call__ define a estrutura do loop e a lógica de iteração.

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

O wrapper de loop pode passar argumentos adicionais, como o índice da iteração atual, para os blocos do loop.

Blocos de loop

Um bloco de loop é um ModularPipelineBlocks, mas o método __call__ se comporta de forma diferente.

• Ele recebe a variável de iteração do wrapper do loop.
• Ele funciona diretamente com o BlockState em vez do PipelineState.
• Não é necessário recuperar ou atualizar o BlockState.

Blocos de loop compartilham o mesmo BlockState para permitir que os valores se acumulem e mudem a cada iteração do 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

Use o método from_blocks_dict() para adicionar o bloco de loop ao wrapper de loop e criar LoopSequentialPipelineBlocks.

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

Adicione mais blocos de loop para executar dentro de cada iteração com from_blocks_dict(). Isso permite modificar os blocos sem alterar a própria lógica do loop.

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

Para adicionar um bloco de loop ao loop de desruidamento de um pipeline existente, por exemplo, um que executa a função callback_on_step_end a cada etapa, consulte Pipelines modulares no guia de callbacks.