Diffusers documentation
SequentialPipelineBlocks
SequentialPipelineBlocks
SequentialPipelineBlocks es un tipo de bloque múltiple que compone otros ModularPipelineBlocks en secuencia. Los datos fluyen linealmente de un bloque al siguiente mediante inputs y intermediate_outputs. Cada bloque en SequentialPipelineBlocks suele representar un paso en el pipeline, y al combinarlos, se construye gradualmente un pipeline.
Esta guía le muestra cómo conectar dos bloques en un SequentialPipelineBlocks.
Crea dos ModularPipelineBlocks. El primer bloque, InputBlock, genera un valor de batch_size y el segundo bloque, ImageEncoderBlock, utiliza batch_size como inputs.
from diffusers.modular_pipelines import ModularPipelineBlocks, InputParam, OutputParam
class InputBlock(ModularPipelineBlocks):
@property
def inputs(self):
return [
InputParam(name="prompt", type_hint=list, description="list of text prompts"),
InputParam(name="num_images_per_prompt", type_hint=int, description="number of images per prompt"),
]
@property
def intermediate_outputs(self):
return [
OutputParam(name="batch_size", description="calculated batch size"),
]
@property
def description(self):
return "A block that determines batch_size based on the number of prompts and num_images_per_prompt argument."
def __call__(self, components, state):
block_state = self.get_block_state(state)
batch_size = len(block_state.prompt)
block_state.batch_size = batch_size * block_state.num_images_per_prompt
self.set_block_state(state, block_state)
return components, stateimport torch
from diffusers.modular_pipelines import ModularPipelineBlocks, InputParam, OutputParam
class ImageEncoderBlock(ModularPipelineBlocks):
@property
def inputs(self):
return [
InputParam(name="image", type_hint="PIL.Image", description="raw input image to process"),
InputParam(name="batch_size", type_hint=int),
]
@property
def intermediate_outputs(self):
return [
OutputParam(name="image_latents", description="latents representing the image"),
]
@property
def description(self):
return "Encode raw image into its latent presentation"
def __call__(self, components, state):
block_state = self.get_block_state(state)
# Simulate processing the image
# This will change the state of the image from a PIL image to a tensor for all blocks
block_state.image = torch.randn(1, 3, 512, 512)
block_state.batch_size = block_state.batch_size * 2
block_state.image_latents = torch.randn(1, 4, 64, 64)
self.set_block_state(state, block_state)
return components, stateConecte los dos bloques definiendo un SequentialPipelineBlocks. Enumere las instancias de bloques en block_classes y sus nombres correspondientes en block_names. Los bloques se ejecutan en el orden en que aparecen en block_classes, y los datos fluyen de un bloque al siguiente a través de PipelineState.
class ImageProcessingStep(SequentialPipelineBlocks):
"""
# auto_docstring
"""
model_name = "my_model"
block_classes = [InputBlock(), ImageEncoderBlock()]
block_names = ["input", "image_encoder"]
@property
def description(self):
return (
"Process text prompts and images for the pipeline. It:\n"
" - Determines the batch size from the prompts.\n"
" - Encodes the image into latent space."
)Al crear un SequentialPipelineBlocks, propiedades como inputs, intermediate_outputs y expected_components se agregan automáticamente desde los subbloques, por lo que no es necesario definirlas de nuevo.
Hay algunas propiedades que debes configurar:
description: Recomendamos añadir una descripción para el bloque ensamblado para explicar lo que hace el paso combinado.model_name: Se deriva automáticamente de los subbloques, pero no siempre es correcto, por lo que es posible que necesites sobrescribirlo.outputs: Por defecto, es lo mismo queintermediate_outputs, pero puedes configurarlo manualmente para controlar qué valores aparecen en la documentación. Esto es útil para mostrar solo las salidas finales en lugar de todos los valores intermedios.
Estas propiedades, junto con los inputs, intermediate_outputs y expected_components agregados, se utilizan para generar automáticamente la propiedad doc.
Imprima el bloque ImageProcessingStep para inspeccionar sus subbloques y use doc para obtener un resumen completo de las entradas, salidas y componentes del bloque.
blocks = ImageProcessingStep()
print(blocks)
print(blocks.doc)