Diffusers documentation
ModularPipelineBlocks
ModularPipelineBlocks
ModularPipelineBlocks は ModularPipeline を構築するための基本的なブロックです。パイプライン内の特定のステップにおいて、ブロックが実行すべきコンポーネント、入力/出力、および計算を定義します。ModularPipelineBlocks は state を使用して他のブロックと接続し、ワークフローのモジュール化された構築を可能にします。
単体ではModularPipelineBlocksを実行することはできません。これは、パイプライン内のステップが何をすべきかを示す設計図です。パイプラインを実際に実行するには、ModularPipelineBlocksをModularPipelineに変換する必要があります。
このガイドでは、ModularPipelineBlocksの作成方法について説明します。
入力と出力
Modular Diffusers の state の仕組みに不慣れな場合は、States ガイドを参照してください。
ModularPipelineBlocks には inputs と intermediate_outputs が必要です。
inputsは、ブロックが計算を実行するために PipelineState から読み取る値です。これには、ユーザーが提供する値（プロンプトや画像など）や、以前のブロックが生成した値（エンコードされたimage_latentsなど）が含まれます。InputParamを使用してinputsを定義します。
class ImageEncodeStep(ModularPipelineBlocks):
...
@property
def inputs(self):
return [
InputParam(name="image", type_hint="PIL.Image", required=True, description="raw input image to process"),
]
...intermediate_outputsは、ブロックによって生成され、PipelineState に追加される新しい値です。intermediate_outputsは、後続のブロックのinputsとして利用可能であり、パイプライン実行の最終出力としても利用可能です。OutputParamを使用して、intermediate_outputsを定義します。
class ImageEncodeStep(ModularPipelineBlocks):
...
@property
def intermediate_outputs(self):
return [
OutputParam(name="image_latents", description="latents representing the image"),
]
...中間のインプットとアウトプットはデータを共有してブロックを接続します。これらはいつでもアクセス可能であり、ワークフローの進捗を追跡できます。
コンポーネントと設定
ブロックに必要なコンポーネントとパイプラインレベルの設定は、ComponentSpecとConfigSpecで指定されます。
- ComponentSpecには、ブロックで使用される想定コンポーネントが含まれます。コンポーネントの
nameと、コンポーネントが具体的に何であるかを指定するtype_hintが必要です。 - ConfigSpecには、すべてのブロックにわたる動作を制御するパイプラインレベルの設定が含まれています。
class ImageEncodeStep(ModularPipelineBlocks):
...
@property
def expected_components(self):
return [
ComponentSpec(name="vae", type_hint=AutoencoderKL),
]
@property
def expected_configs(self):
return [
ConfigSpec("force_zeros_for_empty_prompt", True),
]
...ブロックがパイプラインに変換されると、コンポーネントは__call__の最初の引数としてブロックに利用可能になります。
計算ロジック
ブロックが実行する計算は__call__メソッドで定義されており、特定の構造に従っています。
inputsのローカルビューを取得するために、BlockStateを取得します。inputsに対して計算ロジックを実装します。- ローカルのBlockStateからの変更を、グローバルのPipelineStateへ反映するために、PipelineStateを更新します。
- 次のブロックで利用可能になるコンポーネントと状態を返します。
class ImageEncodeStep(ModularPipelineBlocks):
def __call__(self, components, state):
# Get a local view of the state variables this block needs
block_state = self.get_block_state(state)
# Your computation logic here
# block_state contains all your inputs
# Access them like: block_state.image, block_state.processed_image
# Update the pipeline state with your updated block_states
self.set_block_state(state, block_state)
return components, stateすべてを組み合わせる
以下は、すべてのパーツを接続した完全なブロックです。
from diffusers import ComponentSpec, AutoencoderKL
from diffusers.modular_pipelines import InputParam, ModularPipelineBlocks, OutputParam
class ImageEncodeStep(ModularPipelineBlocks):
@property
def description(self):
return "Encode an image into latent space."
@property
def expected_components(self):
return [
ComponentSpec(name="vae", type_hint=AutoencoderKL),
]
@property
def inputs(self):
return [
InputParam(name="image", type_hint="PIL.Image", required=True, description="raw input image to process"),
]
@property
def intermediate_outputs(self):
return [
OutputParam(name="image_latents", type_hint="torch.Tensor", description="latents representing the image"),
]
def __call__(self, components, state):
block_state = self.get_block_state(state)
block_state.image_latents = components.vae.encode(block_state.image)
self.set_block_state(state, block_state)
return components, state各ブロックには、上記で定義したプロパティから自動生成されるdocプロパティがあります。このプロパティは、ブロックの説明、コンポーネント、入力、出力の概要を提供します。
block = ImageEncoderStep()
print(block.doc)
class ImageEncodeStep
Encode an image into latent space.
Components:
vae (`AutoencoderKL`)
Inputs:
image (`PIL.Image`):
raw input image to process
Outputs:
image_latents (`torch.Tensor`):
latents representing the image