Diffusers documentation
SequentialPipelineBlocks
SequentialPipelineBlocks
SequentialPipelineBlocks는 여러 ModularPipelineBlocks를 순서대로 조합하는 멀티 블록 유형입니다. 데이터는 inputs와 intermediate_outputs를 사용하여 한 블록에서 다음 블록으로 선형적으로 흐릅니다. SequentialPipelineBlocks의 각 블록은 일반적으로 파이프라인의 한 단계를 나타내며, 이를 조합함으로써 파이프라인을 점진적으로 구축할 수 있습니다.
이 가이드에서는 두 개의 블록을 SequentialPipelineBlocks에 연결하는 방법을 보여줍니다.
두 개의 ModularPipelineBlocks를 생성합니다. 첫 번째 블록인 InputBlock은 batch_size 값을 출력하고, 두 번째 블록인 ImageEncoderBlock은 batch_size를 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, stateSequentialPipelineBlocks를 정의하여 두 블록을 연결합니다. block_classes에 블록 인스턴스를, block_names에 해당 이름을 나열합니다. 블록은 block_classes에 나타나는 순서대로 실행되며, 데이터는 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."
)SequentialPipelineBlocks를 생성하면 inputs, intermediate_outputs, expected_components와 같은 속성이 하위 블록에서 자동으로 집계되므로 다시 정의할 필요가 없습니다.
설정해야 할 몇 가지 속성이 있습니다:
description: 조립된 블록에 대해 결합된 단계가 수행하는 내용을 설명하는 설명을 추가하는 것을 권장합니다.model_name: 서브 블록에서 자동으로 파생되지만 항상 정확하지는 않으므로, 필요에 따라 직접 재정의해야 할 수 있습니다.outputs: 기본적으로intermediate_outputs와 동일하지만, 문서에 표시될 값을 제어하기 위해 수동으로 설정할 수 있습니다. 이는 모든 중간 값 대신 최종 출력만 표시하는 데 유용합니다.
이러한 속성과 집계된 inputs, intermediate_outputs, expected_components는 doc 속성을 자동으로 생성하는 데 사용됩니다.
ImageProcessingStep 블록을 출력하여 하위 블록을 확인하고, 블록의 입력, 출력 및 구성 요소에 대한 전체 요약을 보려면 doc을 사용하세요.
blocks = ImageProcessingStep()
print(blocks)
print(blocks.doc)