Diffusers documentation
SequentialPipelineBlocks
SequentialPipelineBlocks
SequentialPipelineBlocks は、他の ModularPipelineBlocks をシーケンスとして組み合わせるマルチブロック型です。inputs と intermediate_outputs を用いて、データは1つのブロックから次のブロックへ直線的に流れます。SequentialPipelineBlocks 内の各ブロックは通常、パイプラインの1ステップを表しており、これらを組み合わせることでパイプラインを段階的に構築していきます。
このガイドでは、2つのブロックをSequentialPipelineBlocksに接続する方法を示します。
2つのModularPipelineBlocksを作成します。最初のブロックInputBlockはbatch_sizeの値を出力し、2番目のブロック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 を定義して、2 つのブロックを接続します。ブロックのインスタンスを block_classes に、それに対応する名前を block_names に列挙します。ブロックは block_classes に記載された順序で実行され、データは PipelineState を介して 1 つのブロックから次のブロックへ流れます。
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)