LoopSequentialPipelineBlocks

Los LoopSequentialPipelineBlocks son un tipo de bloque múltiple que compone otros ModularPipelineBlocks en un bucle. Los datos fluyen de forma circular, utilizando inputs y intermediate_outputs, y cada bloque se ejecuta de manera iterativa. Esto se utiliza típicamente para crear un bucle de desruido, que es iterativo por defecto.

Esta guía te muestra cómo crear LoopSequentialPipelineBlocks.

Envoltorio de bucle

LoopSequentialPipelineBlocks, también conocido como loop wrapper, porque define la estructura del bucle, las variables de iteración y la configuración. Dentro del loop wrapper, necesitas las siguientes variables.

• loop_inputs son valores proporcionados por el usuario y equivalen a inputs.
• loop_intermediate_outputs son nuevas variables intermedias creadas por el bloque y añadidas al PipelineState. Es equivalente a intermediate_outputs.
• El método __call__ define la estructura del bucle y la lógica de iteración.

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

El envoltorio del bucle puede pasar argumentos adicionales, como el índice de la iteración actual, a los bloques del bucle.

Bloques de bucle

Un bloque de bucle es un ModularPipelineBlocks, pero el método __call__ se comporta de manera diferente.

• Recibe la variable de iteración del envoltorio del bucle.
• Funciona directamente con BlockState en lugar de PipelineState.
• No requiere recuperar ni actualizar el BlockState.

Los bloques de bucle comparten el mismo BlockState para permitir que los valores se acumulen y cambien en cada iteración del bucle.

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 el método from_blocks_dict() para añadir el bloque de bucle al envoltorio de bucle y crear LoopSequentialPipelineBlocks.

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

Agregue más bloques de bucle para ejecutarlos dentro de cada iteración con from_blocks_dict(). Esto le permite modificar los bloques sin cambiar la lógica del bucle en sí.

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

Para añadir un bloque de bucle al bucle de desruido de un pipeline existente, por ejemplo, uno que ejecute una función callback_on_step_end en cada paso, consulte Pipelines modulares en la guía de callbacks.