# カスタムブロックの構築

[ModularPipelineBlocks](./pipeline_block) は [ModularPipeline](/docs/diffusers/main/ja/api/modular_diffusers/pipeline#diffusers.ModularPipeline) の基本的な構成要素です。入力、出力、および計算ロジックを定義することで、カスタムブロックを作成できます。このガイドでは、カスタムブロックの作成と使用方法を説明します。

> [!TIP]
> 公式のカスタムブロックについては、[Modular Diffusers Custom Blocks](https://huggingface.co/collections/diffusers/modular-diffusers-custom-blocks)コレクションをご覧ください。

## プロジェクトの構成

カスタムブロックプロジェクトには、以下の構造を使用してください。

```shell
.
├── block.py
└── modular_config.json
```

- `block.py`には、カスタムブロックの実装が含まれています。
- `modular_config.json` には、ブロックを読み込むために必要なメタデータが含まれています。

## テンプレートを使ったクイックスタート

カスタムブロックを作成する最も速い方法は、テンプレートから始めることです。テンプレートには、`block.py` と `modular_config.json` ファイルを含む事前設定済みのプロジェクト構造が提供されており、コンポーネント、入力、出力、および `__call__` メソッドの定義方法を示すコメント付きの例が含まれています。これにより、定型のセットアップではなく、カスタムロジックに集中できます。

### テンプレートをダウンロード

```python
from diffusers import ModularPipelineBlocks

model_id = "diffusers/custom-block-template"
local_dir = model_id.split("/")[-1]

blocks = ModularPipelineBlocks.from_pretrained(
    model_id, 
    trust_remote_code=True, 
    local_dir=local_dir
)
```

これにより、テンプレートファイルがローカルの`custom-block-template/`に保存されます。特定の場所に保存するには、`local_dir`を使用することもできます。

### ローカルで編集

`block.py` を開き、カスタムブロックを実装してください。テンプレートには、各プロパティの定義方法を示すコメント付きの例が含まれています。完全な実装については、以下の[Florence-2 の例](#example-florence-2-image-annotator)を参照してください。

### ブロックのテスト

```python
from diffusers import ModularPipelineBlocks

blocks = ModularPipelineBlocks.from_pretrained(local_dir, trust_remote_code=True)
pipeline = blocks.init_pipeline()
output = pipeline(...)  # your inputs here
```

### Hubにアップロード

```python
pipeline.save_pretrained(local_dir, repo_id="your-username/your-block-name", push_to_hub=True)
```

## 例: Florence-2 イメージアノテーター

この例では、[Florence-2](https://huggingface.co/docs/transformers/model_doc/florence2) を使用してカスタムブロックを作成し、入力画像を処理してインペインティング用のマスクを生成します。

### コンポーネントの定義

ブロックに必要なコンポーネントである`Florence2ForConditionalGeneration`とそのプロセッサーを定義します。コンポーネントを定義する際は、`name`（コード内でアクセスするための名称）、`type_hint`（モデルクラス）、`pretrained_model_name_or_path`（重みの読み込み元）を指定します。

```python
# Inside block.py
from diffusers.modular_pipelines import ModularPipelineBlocks, ComponentSpec
from transformers import AutoProcessor, Florence2ForConditionalGeneration

class Florence2ImageAnnotatorBlock(ModularPipelineBlocks):

    @property
    def expected_components(self):
        return [
            ComponentSpec(
                name="image_annotator",
                type_hint=Florence2ForConditionalGeneration,
                pretrained_model_name_or_path="florence-community/Florence-2-base-ft",
            ),
            ComponentSpec(
                name="image_annotator_processor",
                type_hint=AutoProcessor,
                pretrained_model_name_or_path="florence-community/Florence-2-base-ft",
            ),
        ]
```

### 入力と出力を定義する

入力には、画像、アノテーションタスク、プロンプトが含まれます。出力には、生成されたマスクとアノテーションが含まれます。

```python
from typing import List, Union
from PIL import Image
from diffusers.modular_pipelines import InputParam, OutputParam

class Florence2ImageAnnotatorBlock(ModularPipelineBlocks):

    # ... expected_components from above ...

    @property
    def inputs(self) -> List[InputParam]:
        return [
            InputParam(
                "image",
                type_hint=Union[Image.Image, List[Image.Image]],
                required=True,
                description="Image(s) to annotate",
            ),
            InputParam(
                "annotation_task",
                type_hint=str,
                default="<REFERRING_EXPRESSION_SEGMENTATION>",
                description="Annotation task to perform (e.g., <OD>, <CAPTION>, <REFERRING_EXPRESSION_SEGMENTATION>)",
            ),
            InputParam(
                "annotation_prompt",
                type_hint=str,
                required=True,
                description="Prompt to provide context for the annotation task",
            ),
            InputParam(
                "annotation_output_type",
                type_hint=str,
                default="mask_image",
                description="Output type: 'mask_image', 'mask_overlay', or 'bounding_box'",
            ),
        ]

    @property
    def intermediate_outputs(self) -> List[OutputParam]:
        return [
            OutputParam(
                "mask_image",
                type_hint=Image.Image,
                description="Inpainting mask for the input image",
            ),
            OutputParam(
                "annotations",
                type_hint=dict,
                description="Raw annotation predictions",
            ),
            OutputParam(
                "image",
                type_hint=Image.Image,
                description="Annotated image",
            ),
        ]
```

### `__call__`メソッドを実装する

`__call__` メソッドにはブロックのロジックが含まれています。`block_state` を介して入力にアクセスし、計算を実行して、出力を `block_state` に設定します。

```python
import torch
from diffusers.modular_pipelines import PipelineState

class Florence2ImageAnnotatorBlock(ModularPipelineBlocks):

    # ... expected_components, inputs, intermediate_outputs from above ...

    @torch.no_grad()
    def __call__(self, components, state: PipelineState) -> PipelineState:
        block_state = self.get_block_state(state)
        
        images, annotation_task_prompt = self.prepare_inputs(
            block_state.image, block_state.annotation_prompt
        )
        task = block_state.annotation_task
        fill = block_state.fill
        
        annotations = self.get_annotations(
            components, images, annotation_task_prompt, task
        )
        block_state.annotations = annotations
        if block_state.annotation_output_type == "mask_image":
            block_state.mask_image = self.prepare_mask(images, annotations)
        else:
            block_state.mask_image = None

        if block_state.annotation_output_type == "mask_overlay":
            block_state.image = self.prepare_mask(images, annotations, overlay=True, fill=fill)

        elif block_state.annotation_output_type == "bounding_box":
            block_state.image = self.prepare_bounding_boxes(images, annotations)

        self.set_block_state(state, block_state)

        return components, state
    
    # Helper methods for mask/bounding box generation...
```

> [!TIP]
> [diffusers/Florence2-image-Annotator](https://huggingface.co/diffusers/Florence2-image-Annotator) で完全な実装を確認できます。

## カスタムブロックの使用

[from_pretrained()](/docs/diffusers/main/ja/api/modular_diffusers/pipeline#diffusers.ModularPipeline.from_pretrained) でカスタムブロックを読み込み、`trust_remote_code=True` を設定します。

```py
import torch
from diffusers import ModularPipeline
from diffusers.utils import load_image

# Load the Florence-2 annotator pipeline
image_annotator = ModularPipeline.from_pretrained(
    "diffusers/Florence2-image-Annotator",
    trust_remote_code=True
)

# Check the docstring to see inputs/outputs
print(image_annotator.blocks.doc)
```

ブロックを使用してマスクを生成します:

```python
image_annotator.load_components(dtype=torch.bfloat16)
image_annotator.to("cuda")  # or "mps", "xpu", "cpu"

image = load_image("https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/transformers/tasks/car.jpg")
image = image.resize((1024, 1024))
prompt = ["A red car"]
annotation_task = "<REFERRING_EXPRESSION_SEGMENTATION>"
annotation_prompt = ["the car"]

mask_image = image_annotator_node(
    prompt=prompt,
    image=image,
    annotation_task=annotation_task,
    annotation_prompt=annotation_prompt,
    annotation_output_type="mask_image",
).images
mask_image[0].save("car-mask.png")
```

他のブロックと組み合わせることで、新しいパイプラインを作成できます:

```python
# Get the annotator block
annotator_block = image_annotator.blocks

# Get an inpainting workflow and insert the annotator at the beginning
inpaint_blocks = ModularPipeline.from_pretrained("Qwen/Qwen-Image").blocks.get_workflow("inpainting")
inpaint_blocks.sub_blocks.insert("image_annotator", annotator_block, 0)

# Initialize the combined pipeline
pipe = inpaint_blocks.init_pipeline()
pipe.load_components(dtype=torch.float16, device="cuda")  # or "mps", "xpu", "cpu"

# Now the pipeline automatically generates masks from prompts
output = pipe(
    prompt=prompt,
    image=image,
    annotation_task=annotation_task,
    annotation_prompt=annotation_prompt,
    annotation_output_type="mask_image",
    num_inference_steps=35,
    guidance_scale=7.5,
    strength=0.95,
    output="images"
)
output[0].save("florence-inpainting.png")
```

## カスタムブロックの編集

カスタムブロックをローカルにダウンロードして編集します。これは[テンプレートによるクイックスタート](#quick-start-with-template)と同じワークフローですが、テンプレートではなく既存のブロックから開始します。

`local_dir` 引数を使用して、カスタムブロックを特定のフォルダにダウンロードします:

```python
from diffusers import ModularPipelineBlocks

# Download to a local folder for editing
annotator_block = ModularPipelineBlocks.from_pretrained(
    "diffusers/Florence2-image-Annotator",
    trust_remote_code=True,
    local_dir="./my-florence-block"
)
```

このフォルダー内のブロックファイルに行った変更は、ブロックを再度読み込む際に反映されます。変更を共有する準備が整ったら、新しいリポジトリにアップロードしてください:

```python
pipeline = annotator_block.init_pipeline()
pipeline.save_pretrained("./my-florence-block", repo_id="your-username/my-custom-florence", push_to_hub=True)
```

## 次のステップ

このガイドでは、1つのカスタムブロックの作成方法について説明しました。複数のブロックを組み合わせる方法を学びましょう：

- [SequentialPipelineBlocks](./sequential_pipeline_blocks): ブロックを連鎖させて順番に実行します
- [ConditionalPipelineBlocks](./auto_pipeline_blocks): 異なる実行パスを選択する条件付きブロックを作成します
- [LoopSequentialPipelineBlocks](./loop_sequential_pipeline_blocks): デノイジングループのような反復的なワークフローを定義します

カスタムブロックをMellonのビジュアルインターフェースで動作させる方法です。[Mellon Custom Blocks](./mellon)ガイドを参照してください。

インスピレーションやすぐに使えるブロックを探すには、[Modular Diffusers Custom Blocks](https://huggingface.co/collections/diffusers/modular-diffusers-custom-blocks)コレクションをご覧ください。

## 依存関係

カスタムブロック内でパッケージの依存関係を宣言しておくことで、実行時のインポートエラーを防止できます。Diffusers は依存関係を検証し、パッケージが欠落しているか互換性がない場合は警告を返します。

ブロッククラスに`_requirements`属性を設定し、パッケージ名をバージョン指定子にマッピングします。

```py
from diffusers.modular_pipelines import PipelineBlock

class MyCustomBlock(PipelineBlock):
    _requirements = {
        "transformers": ">=4.44.0",
        "sentencepiece": ">=0.2.0"
    }
```

要件が異なるブロックがある場合、Diffusers はそれらの要件をマージします。

```py
from diffusers.modular_pipelines import SequentialPipelineBlocks

class BlockA(PipelineBlock):
    _requirements = {"transformers": ">=4.44.0"}
    # ...

class BlockB(PipelineBlock):
    _requirements = {"sentencepiece": ">=0.2.0"}
    # ...

pipe = SequentialPipelineBlocks.from_blocks_dict({
    "block_a": BlockA,
    "block_b": BlockB,
})
```

このブロックが[save_pretrained()](/docs/diffusers/main/ja/api/modular_diffusers/pipeline#diffusers.ModularPipeline.save_pretrained)で保存されると、要件は`modular_config.json`ファイルに保存されます。このブロックが読み込まれると、Diffusers は各要件を現在の環境と照合します。不一致がある場合やパッケージが見つからない場合、Diffusers は以下の警告を返します。

```md
# missing package
xyz-package was specified in the requirements but wasn't found in the current environment.

# version mismatch
xyz requirement 'specific-version' is not satisfied by the installed version 'actual-version'. Things might work unexpected.
```