Diffusers documentation
SequentialPipelineBlocks
SequentialPipelineBlocks
SequentialPipelineBlocks são um tipo de bloco múltiplo que compõe outros ModularPipelineBlocks em sequência. Os dados fluem linearmente de um bloco para o próximo usando inputs e intermediate_outputs. Cada bloco em SequentialPipelineBlocks geralmente representa uma etapa no pipeline, e, ao combiná-los, você constrói gradualmente um pipeline.
Este guia mostra como conectar dois blocos em um SequentialPipelineBlocks.
Crie dois ModularPipelineBlocks. O primeiro bloco, InputBlock, produz um valor de batch_size e o segundo bloco, ImageEncoderBlock, usa 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 os dois blocos definindo um SequentialPipelineBlocks. Liste as instâncias dos blocos em block_classes e seus nomes correspondentes em block_names. Os blocos são executados na ordem em que aparecem em block_classes, e os dados fluem de um bloco para o próximo por meio 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."
)Ao criar um SequentialPipelineBlocks, propriedades como inputs, intermediate_outputs e expected_components são agregadas automaticamente dos sub-blocos, portanto não é necessário defini-las novamente.
Existem algumas propriedades que você deve definir:
description: Recomendamos adicionar uma descrição para o bloco montado para explicar o que a etapa combinada faz.model_name: Isso é derivado automaticamente dos sub-blocos, mas nem sempre está correto, então você pode precisar sobrescrevê-lo.outputs: Por padrão, isso é o mesmo queintermediate_outputs, mas você pode configurá-lo manualmente para controlar quais valores aparecem no documento. Isso é útil para mostrar apenas as saídas finais, em vez de todos os valores intermediários.
Essas propriedades, juntamente com os inputs, intermediate_outputs e expected_components agregados, são usadas para gerar automaticamente a propriedade doc.
Imprima o bloco ImageProcessingStep para inspecionar seus sub-blocos e use doc para um resumo completo das entradas, saídas e componentes do bloco.
blocks = ImageProcessingStep()
print(blocks)
print(blocks.doc)