ModularPipelineBlocks

ModularPipelineBlocks est le bloc de base pour construire un ModularPipeline. Il définit les composants, les entrées/sorties et le calcul qu'un bloc doit effectuer pour une étape spécifique d'un pipeline. Un ModularPipelineBlocks se connecte à d'autres blocs, en utilisant l'state, pour permettre la construction modulaire de workflows.

Un ModularPipelineBlocks seul ne peut pas être exécuté. Il s'agit d'un plan de ce qu'une étape doit faire dans un pipeline. Pour exécuter réellement un pipeline, il faut convertir le ModularPipelineBlocks en ModularPipeline.

Ce guide vous montre comment créer un ModularPipelineBlocks.

Entrées et sorties

Consultez le guide States si vous n'êtes pas familier avec le fonctionnement de l'état dans Modular Diffusers.

Un ModularPipelineBlocks nécessite inputs et intermediate_outputs.

inputs sont les valeurs qu'un bloc lit depuis l'PipelineState pour effectuer son calcul. Il peut s'agir de valeurs fournies par un utilisateur (comme un prompt ou une image) ou de valeurs produites par un bloc précédent (comme les image_latents encodés).

Utilisez InputParam pour définir les 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 sont de nouvelles valeurs créées par un bloc et ajoutées à l'PipelineState. Les intermediate_outputs sont disponibles en tant que inputs pour les blocs suivants ou en tant que sortie finale de l'exécution du pipeline.

Utilisez OutputParam pour définir intermediate_outputs.

class ImageEncodeStep(ModularPipelineBlocks):
    ...

    @property
    def intermediate_outputs(self):
        return [
            OutputParam(name="image_latents", description="latents representing the image"),
        ]

    ...

Les entrées et sorties intermédiaires partagent des données pour relier les blocs. Elles sont accessibles à tout moment, ce qui vous permet de suivre l’avancement du flux de travail.

Composants et configurations

Les composants et les configurations au niveau du pipeline dont un bloc a besoin sont spécifiés dans ComponentSpec et ConfigSpec.

ComponentSpec contient les composants attendus utilisés par un bloc. Vous avez besoin du name du composant et, idéalement, d'un type_hint qui précise exactement ce qu'est le composant.
ConfigSpec contient les paramètres au niveau du pipeline qui contrôlent le comportement de tous les blocs.

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),
        ]

    ...

Lorsque les blocs sont convertis en pipeline, les composants deviennent disponibles pour le bloc en tant que premier argument de __call__.

Logique de calcul

Le calcul effectué par un bloc est défini dans la méthode __call__ et suit une structure spécifique.

Récupérez le BlockState pour obtenir une vue locale des inputs.
Implémentez la logique de calcul sur les inputs.
Mettre à jour PipelineState pour propager les modifications de l'BlockState local vers l'PipelineState global.
Retourne les composants et l'état qui deviennent disponibles pour le bloc suivant.

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

Tout mettre ensemble

Voici le bloc complet avec toutes les pièces connectées.

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

Chaque bloc possède une propriété doc générée automatiquement à partir des propriétés définies ci-dessus. Elle fournit un résumé de la description du bloc, de ses composants, de ses entrées et de ses sorties.

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

Update on GitHub