ワークフローブロックを作成する

ワークフローブロック作成の完全な開発者ガイド：マニフェスト、入力、出力、バッチ処理、フロー制御、次元。

これは、Python で Block クラスを作成するための詳細ガイドです。Workflow 内で少しだけカスタムロジックが必要なだけなら、 動的な Python ブロック を Workflow Definition 内でインライン定義するだけで十分な場合がほとんどです。詳しくは カスタムブロック.

Workflows のブロック開発には、Workflow エコシステムへの理解が必要です。詳細に入る前に、必要な知識を要約しましょう:

理解 Workflow 実行、特に:

• Workflow 定義における Workflow ブロックとステップの関係

• Workflow ブロックとその manifest がどのように Workflows Compiler

• 何であるか 次元レベル Workflow を流れるバッチ指向データの

• どのように Execution Engine が、入力と出力に関してステップとやり取りするのか

• の性質と役割は何か Workflow の種類

• どのように pydantic が動作するかを理解すること

環境のセットアップ

すぐに分かるように、Workflow ブロックの作成は、特定のインターフェースを実装する Python クラスを定義するだけです。この設計により、他の Python コードと同様に、Python インタプリタでブロックを実行できます。ただし、Workflow 実行中には通常他のブロックから提供される、必要な入力をすべて組み立てる際に難しさが生じることがあります。そのため、スムーズに作業するには開発環境を適切にセットアップすることが重要です。標準的な開発プロセスの一部として、次の手順に従うことを推奨します（最初の手順は、後続の貢献では省略できます）:

1. をセットアップし、 conda 環境 をインストールして、 推論の主要依存関係を 推論 コントリビューターガイド.

2. に記載のとおりに行ってください。Workflows コードベースの構成に慣れておきましょう。

3. 最小限のブロックを作成する - これを行う方法は次のセクションで学びます。まずは、シンプルなブロック manifest と基本ロジックを実装して、ブロックが期待どおりに動作することを確認してください。

4. ブロックをプラグインに追加する - ブロックを作成したら、プラグインからエクスポートされるブロック一覧に追加します。Roboflow Core プラグインにブロックを追加する場合は、 loader.py. にあなたのブロックのエントリを追加してください。これを忘れると、ブロックは表示されません!

5. ブロックを反復的に改善する - 結果に満足できるまで、ブロックの開発と実行を続けてください。以下のセクションでは、さまざまなシナリオでブロックを反復改善する方法を説明します。

Workflows UI を使ってブロックを実行する

マウント済みボリュームを使って inference サーバーを実行することを推奨します（変更のたびにサーバーを再ビルドするよりもはるかに高速です）: 推論 サーバー

inference_repo$ docker run -p 9001:9001 \
   -v ./inference:/app/inference \
   roboflow/roboflow-inference-server-cpu:latest

を実行し、ローカルサーバーを Roboflow UI に接続します:

プレビューをすばやく実行するには:

私のブロックには追加の依存関係が必要です - 事前ビルド済みの `inference` サーバーは使えません

ブロックによっては追加の依存関係が必要になることは自然です。依存関係を追加するには、関連する Docker イメージにインストールされている requirements ファイルのいずれかに含めるだけです（通常は CPU ビルド の 推論 サーバー）。

その後、次を実行します:

inference_repo$ docker build \
   -t roboflow/roboflow-inference-server-cpu:test \ 
   -f docker/dockerfiles/Dockerfile.onnx.cpu .

次に、今作成したテストタグを指定してローカルビルドを実行できます:

inference_repo$ inference_repo$ docker run -p 9001:9001 \
   -v ./inference:/app/inference \
   roboflow/roboflow-inference-server-cpu:test

Workflows UI を使わずにブロックを実行する

Roboflow プラットフォームにアクセスできないコントリビューターには、上記セクションで述べたようにサーバーを実行することを推奨します。ただし、UI エディターの代わりに、シンプルな Workflow 定義を作成してサーバーにリクエストを送る必要があります。

UI なしで Workflow を実行する

次のコードスニペットは、Workflow を実行するために 推論 サーバーにリクエストを送る方法を示しています。 inference_sdk は 推論 パッケージに含まれており、サーバー向けの軽量クライアントライブラリとして提供されています。

from inference_sdk import InferenceHTTPClient

YOUR_WORKFLOW_DEFINITION = ...

client = InferenceHTTPClient(
    api_url=object_detection_service_url,
    api_key="XXX",  # optional, only required if Workflow uses Roboflow Platform
)
result = client.run_workflow(
    specification=YOUR_WORKFLOW_DEFINITION,
    images={
        "image": your_image_np,   # これは入力例です。適宜調整してください
    },
    parameters={
        "my_parameter": 37,   # これは入力例です。適宜調整してください
    },
)

通常のコントリビューターに推奨される方法

で統合テストを作成することは、 tests/workflows/integration_tests/execution ディレクトリにおいて、開発の反復プロセスの自然な一部です。この方法では、開発とテストを同時に進められ、コードを洗練させる間に貴重なフィードバックが得られます。多少の経験は必要ですが、長期的なコード保守性を大きく高めます。

手順は簡単です:

1. 新しいテストモジュールを作成する: たとえば、次のように名前を付けます test_workflows_with_my_custom_block.py.

2. サンプル Workflow を作成する: 1つ以上のサンプル Workflow を作成します。ブロックがエコシステム内の他のブロックとうまく連携するのが理想です。

3. サンプルデータでテストを実行する: テスト内でこれらの Workflow をサンプルデータを使って実行します（通常使用しているサンプルデータを見つけるには、 fixture を確認してください）。

4. 期待される結果をアサートする: 結果が期待どおりであることを検証します。

テストを開発フローに組み込むことで、ブロックの安定性を長期的に保ち、既存ブロックと効果的に連携させることができ、あなたの成果物の表現力を高められます!

次のコマンドでテストを実行できます:

pytest tests/workflows/integration_tests/execution/test_workflows_with_my_custom_block

他のテストを例として参照しても、次のテンプレートを使っても構いません:

統合テストテンプレート

def test_detection_plus_classification_workflow_when_XXX(
    model_manager: ModelManager,
    dogs_image: np.ndarray, 
    roboflow_api_key: str,
) -> None:
    # given
    workflow_init_parameters = {
        "workflows_core.model_manager": model_manager,
        "workflows_core.api_key": roboflow_api_key,
        "workflows_core.step_execution_mode": StepExecutionMode.LOCAL,
    }
    execution_engine = ExecutionEngine.init(
        workflow_definition=<YOUR-EXAMPLE-WORKLFOW>,
        init_parameters=workflow_init_parameters,
        max_concurrent_steps=WORKFLOWS_MAX_CONCURRENT_STEPS,
    )

    # when
    result = execution_engine.run(
        runtime_parameters={
            "image": dogs_image,
        }
    )

    # then
    assert isinstance(result, list), "list が返されることを期待しています"
    assert len(result) == 1, "1つの入力画像に対して出力の要素が1つであることを期待しています"
    assert set(result[0].keys()) == {
        "predictions",
    }, "宣言されたすべての出力が返されることを期待しています"
    assert (
        len(result[0]["predictions"]) == 2
    ), "入力画像に2匹の犬のクロップがあるため、2つのネストされた分類結果があることを期待しています"
    assert [result[0]["predictions"][0]["top"], result[0]["predictions"][1]["top"]] == [
        "116.Parson_russell_terrier",
        "131.Wirehaired_pointing_griffon",
    ] , "予測結果が参照実行で測定されたとおりであることを期待しています"

• 行 2で、 model_manager fixture が見つかります。これは通常、モデルブロックに必要です。この fixture は ModelManager の抽象化を 推論から提供し、モデルのロードとアンロードに使用されます。

• 行 3 では、2匹の犬の画像を含む fixture を定義しています（他の fixture を見て、さらに例となる画像を探してください）。

• 行 4 は任意の fixture で、テスト対象の Workflow 内のいずれかのブロックが Roboflow API キーを必要とする場合に使用できます。その場合は、テストを実行する前に有効なキーを使って ROBOFLOW_API_KEY 環境変数をエクスポートしてください。

• 行 7-11 では、Workflow 定義に基づいて Execution Engine が実行時に作成するブロックの初期化パラメータのセットアップを行います。

• 行 19-23 では、入力パラメータを注入して Workflow を実行する方法を示しています。runtime_parameters のキーが Workflow 定義で宣言された入力と一致していることを確認してください。

• 行 26以降に、テスト内の例示的なアサーションがあります。

プロトタイプ

Workflow ブロックを作成するには、Workflows ライブラリのコアからいくつかのインポートが必要です。ブロック作成時に役立つインポートの一覧は次のとおりです:

from inference.core.workflows.execution_engine.entities.base import (
    Batch,  # バッチデータは Batch[X] コンテナで渡されます
    OutputDefinition,  # manifest で出力を宣言するためのクラス
    WorkflowImageData,  # 画像の内部表現
    # - 入力の種類が image のときに使用します
)

from inference.core.workflows.prototypes.block import (
    BlockResult,  # `run(...)` メソッドの結果の型エイリアス
    WorkflowBlock,  # あなたのブロックの基底クラス
    WorkflowBlockManifest,  # ブロック manifest の基底クラス
)

from inference.core.workflows.execution_engine.entities.types import *  
# コアライブラリの `kinds` を含むモジュール

最も重要なのは次のものです:

• WorkflowBlock - あなたのブロックの基底クラス

• WorkflowBlockManifest - ブロック manifest の基底クラス

内部データ表現の理解

お気づきかもしれませんが、 Batch と WorkflowImageData クラスのインポートを推奨しています。これらは、私たちのシステムでビルディングブロックを構築する際に使われる基本コンポーネントです。これらのクラスが全体のアーキテクチャの中でどのように位置づくかをより深く理解するには、 Data Representations ページを参照して、さらに詳しい情報を確認することをお勧めします。

ブロック manifest

manifest は、Workflow ブロックの重要なコンポーネントであり、ブロックを使うために Workflow 定義に配置できるステップ宣言のプロトタイプを定義します。特に、次のことを行います:

• 使用します pydantic Workflow 定義の構文解析を支援するために: 次を継承します pydantic BaseModel の機能を使って、Workflow 定義を解析および検証します。このスキーマは、 pydantic の OpenAPI 標準との統合により、Workflows UI と互換性のある形式へ自動的にエクスポートすることもできます。

• データバインディングを定義する: 実行中に workflow を流れるデータのセレクタとなる manifest 内のフィールドと、その種類を指定します。

• ブロック出力を記述する: ブロックが生成する出力を概説します。

• 次元性を指定する: 入力と出力の次元性に関するプロパティを詳述します。

• バッチ入力と空値を示す: このステップがバッチ入力と空値を受け入れるかどうかを Execution Engine に伝えます。

• 互換性を確保する: 安定性を保つため、異なる Execution Engine バージョンとの互換性を規定します。詳細は バージョニング.

manifest のスキャフォールディング

manifest の動作を理解するために、ステップごとに定義してみましょう。ここで作成するサンプルブロックは、画像の類似度を計算するものです。まずは import とクラスのスキャフォールディングから始めます:

from typing import Literal
from inference.core.workflows.prototypes.block import (
    WorkflowBlockManifest,
)

class ImagesSimilarityManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/images_similarity@v1"] 
    name: str

これは manifest の最小表現です。Compiler と Execution engine にとって重要な 2 つの特別なフィールドを定義しています:

• type - 動的なブロックプールに基づく Workflow 定義の構文を解析するために必要です。これは pydantic type discriminator であり、Workflow 定義内の特定のステップを解析する際に、どのブロック manifest を検証すべきかを Compiler が理解できるようにします

• name - このプロパティはステップに一意の名前を付け、他のステップがセレクタを通じてそれを選択できるようにするために使われます

入力の追加

比較対象の画像を含む 2 つの入力をステップに受け取らせたいと考えています。

入力の追加

それらの入力定義を manifest に追加する方法を見てみましょう:

from typing import Literal, Union
from pydantic import Field
from inference.core.workflows.prototypes.block import (
    WorkflowBlockManifest,
)
from inference.core.workflows.execution_engine.entities.types import (
    Selector,
    IMAGE_KIND,
)

class ImagesSimilarityManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/images_similarity@v1"] 
    name: str
    # `type` と `name` 以外のすべてのプロパティは、 
    # ハードコードされたパラメータかデータセレクタとして扱われます。データセレクタは文字列で、 
    # `$steps.` または `$inputs.` から始まり、ランタイムで利用可能なデータへの参照を示します。 
    # この場合は通常、データの種類を指定して
    # Compiler にデータの見た目を知らせます。
    image_1: Selector(kind=[IMAGE_KIND]) = Field(
        description="類似度を計算する最初の画像",
    )
    image_2: Selector(kind=[IMAGE_KIND]) = Field(
        description="類似度を計算する2番目の画像",
    )

• の行で 2-9、必要なものがすべて揃うように、いくつかの import を追加しました

• 行 20 は image_1 パラメータを定義します。manifest は Workflow Definition のプロトタイプなので、ステップが使用する画像を伝える唯一の方法はセレクタを提供することです。コアライブラリには専用の型として Selectorがあります。コードベースをさらに深く見ると、これは型エイリアスのコンストラクタ関数であり、 pydantic に一致する文字列を期待するよう指示しています $inputs.{name} と $steps.{name}.* のパターンに対応し、さらに追加のスキーマフィールドメタデータを提供して、Workflows エコシステムのコンポーネントに、 kind セレクタの背後にあるデータの 画像. 重要な注意: を list として表します。特定の kinds のリストは kind kinds の union として解釈されます Execution Engine によって。

• を示す pydantic Field(...) 属性は、行の後半にあります 20 は任意ですが、特に Workflows UI と連携することを意図したブロックでは推奨されます

• 行の 23から、 image_2 パラメータの定義を見つけることができます。これは image_1.

このような manifest 定義により、Workflow 定義内で次のステップ宣言を扱えます:

{
  "type": "my_plugin/images_similarity@v1",
  "name": "my_step",
  "image_1": "$inputs.my_image",
  "image_2": "$steps.image_transformation.image"
}

この定義により、Compiler と Execution Engine は次のことを行います:

• ステップを、type を宣言する Workflow ブロックから初期化する my_plugin/images_similarity@v1

• ステップの run メソッドに 2 つのパラメータを供給する:

  • input_1 という型の WorkflowImageData で渡される画像で埋められます my_image.

  • imput_2 という型の WorkflowImageData は実行時に、別のステップ image_transformation

manifest にパラメータを追加する

では、ステップの実行に影響するパラメータを追加しましょう。

manifest にパラメータを追加する

from typing import Literal, Union
from pydantic import Field
from inference.core.workflows.prototypes.block import (
    WorkflowBlockManifest,
)
from inference.core.workflows.execution_engine.entities.types import (
    Selector,
    IMAGE_KIND,
    FLOAT_ZERO_TO_ONE_KIND,
)

class ImagesSimilarityManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/images_similarity@v1"] 
    name: str
    # `type` と `name` 以外のすべてのプロパティは、 
    # ハードコードされたパラメータかデータセレクタとして扱われます。データセレクタは文字列で、 
    # `$steps.` または `$inputs.` から始まり、ランタイムで利用可能なデータへの参照を示します。 
    # この場合は通常、データの種類を指定して
    # Compiler にデータの見た目を知らせます。
    image_1: Selector(kind=[IMAGE_KIND]) = Field(
        description="類似度を計算する最初の画像",
    )
    image_2: Selector(kind=[IMAGE_KIND]) = Field(
        description="類似度を計算する2番目の画像",
    )
    similarity_threshold: Union[
        float,
        Selector(kind=[FLOAT_ZERO_TO_ONE_KIND]),
    ] = Field(
        default=0.4,
        description="画像が類似しているとみなすしきい値",
    )

• 行 9 imports float_zero_to_one kind を定義するために使用される定義。

• 行で 27 、次の名前のパラメータの定義を開始します similarity_threshold。manifest は float 値または、 kind float_zero_to_oneでインポートした Workflow 入力へのセレクタのどちらかを受け入れます 9.

このような manifest 定義により、Workflow 定義内で次のステップ宣言を扱えます:

{
  "type": "my_plugin/images_similarity@v1",
  "name": "my_step",
  "image_1": "$inputs.my_image",
  "image_2": "$steps.image_transformation.image",
  "similarity_threshold": "$inputs.my_similarity_threshold"
}

または、代わりに:

{
  "type": "my_plugin/images_similarity@v1",
  "name": "my_step",
  "image_1": "$inputs.my_image",
  "image_2": "$steps.image_transformation.image",
  "similarity_threshold": 0.5
}

ブロック出力の宣言

ブロックの入力定義は成功しましたが、ブロックを正常に実行するために必要な要素がまだいくつか足りません。ブロックの出力を定義しましょう。

ブロック出力の宣言

必要な情報の最小セットは、出力の説明です。さらに、ブロックの安定性を高めるために、実行エンジンの互換性に関する情報を提供することを推奨します。

from typing import Literal, Union
from pydantic import Field
from inference.core.workflows.prototypes.block import (
    WorkflowBlockManifest,
    OutputDefinition,
)
from inference.core.workflows.execution_engine.entities.types import (
    Selector,
    IMAGE_KIND,
    FLOAT_ZERO_TO_ONE_KIND,
    BOOLEAN_KIND,
)

class ImagesSimilarityManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/images_similarity@v1"] 
    name: str
    image_1: Selector(kind=[IMAGE_KIND]) = Field(
        description="類似度を計算する最初の画像",
    )
    image_2: Selector(kind=[IMAGE_KIND]) = Field(
        description="類似度を計算する2番目の画像",
    )
    similarity_threshold: Union[
        float,
        Selector(kind=[FLOAT_ZERO_TO_ONE_KIND]),
    ] = Field(
        default=0.4,
        description="画像が類似しているとみなすしきい値",
    )

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [
          OutputDefinition(
            name="images_match", 
            kind=[BOOLEAN_KIND],
          )
        ]

    @classmethod
    def get_execution_engine_compatibility(cls) -> Optional[str]:
        return ">=1.3.0,<2.0.0"

• 行 5 ステップ出力を記述するために使用されるクラスをインポートします

• 行 11 imports ブール値 kind 出力定義で使用するための

• 行 32-39 ブロックからの出力を指定するクラスメソッドを宣言します。リスト内の各エントリは、各バッチ要素ごとに1つの戻り値プロパティとその kind。私たちのブロックはブールフラグを返します images_match 各画像ペアについて。

• 行 41-43 ブロックが Execution Engine と互換性があることを宣言します - 参照: バージョン管理ページ 詳細について

これらの変更の結果:

• Execution Engine は、このブロックから作成されたステップが指定された出力を提供するはずであることを理解し、他のステップは入力でそれらの出力を参照できるようになります

• Execution Engine がバージョンにないため、ブロックの読み込み機構はそのブロックを読み込みません v1

詳細を見る: 動的な出力

一部のブロックは、パース後に利用可能なステップマニフェストの内容に関係なく、classmethod を使って出力を任意に定義できない場合があります。これをサポートするために、次の規約を導入しました:

• classmethod describe_outputs(...) name を持つ要素を1つ含むリストを返すものとします * および kind * (別名 WILDCARD_KIND)

• さらに、ブロックマニフェストはインスタンスメソッド get_actual_outputs(...) を実装し、入力済みのマニフェストデータに基づいて生成可能な実際の出力のリストを提供する必要があります

from typing import Literal, Union, List, Optional
from pydantic import Field
from inference.core.workflows.prototypes.block import (
    WorkflowBlockManifest,
    OutputDefinition,
)
from inference.core.workflows.execution_engine.entities.types import (
    Selector,
    IMAGE_KIND,
    FloatZeroToOne,
    FLOAT_ZERO_TO_ONE_KIND,
    BOOLEAN_KIND,
    WILDCARD_KIND,
)

class ImagesSimilarityManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/images_similarity@v1"] 
    name: str
    image_1: Selector(kind=[IMAGE_KIND]) = Field(
        description="類似度を計算する最初の画像",
    )
    image_2: Selector(kind=[IMAGE_KIND]) = Field(
        description="類似度を計算する2番目の画像",
    )
    similarity_threshold: Union[
        float,
        Selector(kind=[FLOAT_ZERO_TO_ONE_KIND]),
    ] = Field(
        default=0.4,
        description="画像が類似しているとみなすしきい値",
    )
    outputs: List[str]

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [
          OutputDefinition(
            name="*", 
            kind=[WILDCARD_KIND],
          ),
        ]

    def get_actual_outputs(self) -> List[OutputDefinition]:
        # ここでは `self` にアクセスできます:
        return [
          OutputDefinition(name=e, kind=[BOOLEAN_KIND])
          for e in self.outputs
        ]

ブロッククラスの定義

ここまでで、シンプルなブロックのマニフェストは準備できました。引き続き例を見ていきましょう。 応用トピック のセクションで、今は脱線しすぎない程度のより詳細な内容を確認できます。

基本実装

マニフェストが整ったので、ブロックのベースライン実装を準備できます。

ブロックのひな形

from typing import Literal, Union, List, Optional, Type
from pydantic import Field
from inference.core.workflows.prototypes.block import (
    WorkflowBlockManifest,
    WorkflowBlock,
    BlockResult,
)
from inference.core.workflows.execution_engine.entities.base import (
    OutputDefinition,
    WorkflowImageData,
)
from inference.core.workflows.execution_engine.entities.types import (
    Selector,
    IMAGE_KIND,
    FloatZeroToOne,
    FLOAT_ZERO_TO_ONE_KIND,
    BOOLEAN_KIND,
)

class ImagesSimilarityManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/images_similarity@v1"] 
    name: str
    image_1: Selector(kind=[IMAGE_KIND]) = Field(
        description="類似度を計算する最初の画像",
    )
    image_2: Selector(kind=[IMAGE_KIND]) = Field(
        description="類似度を計算する2番目の画像",
    )
    similarity_threshold: Union[
        FloatZeroToOne,
        Selector(kind=[FLOAT_ZERO_TO_ONE_KIND]),
    ] = Field(
        default=0.4,
        description="画像が類似しているとみなすしきい値",
    )

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [
          OutputDefinition(
            name="images_match", 
            kind=[BOOLEAN_KIND],
          ),
        ]

    @classmethod
    def get_execution_engine_compatibility(cls) -> Optional[str]:
        return ">=1.3.0,<2.0.0"

class ImagesSimilarityBlock(WorkflowBlock):

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return ImagesSimilarityManifest

    def run(
        self,
        image_1: WorkflowImageData,
        image_2: WorkflowImageData,
        similarity_threshold: float,
    ) -> BlockResult:
        pass

• 行 1, 5-6 と 8-11 ブロッククラスとそのすべてのメソッドシグネチャを正しく定義するために必要な追加シンボルを提供するよう、インポート構造に変更を加えました

• 行 53-55 クラスメソッドを定義します get_manifest(...) 先ほど作成したマニフェストクラスを単純に返すようにします

• 行 57-63 定義する run(...) メソッドを定義します。Execution Engine はこれをデータとともに呼び出して、望ましい結果を取得します。なお、入力を定義するマニフェストフィールドは 画像 kind が WorkflowImageData としてマークされており、これは 画像 で説明されている kind の ドキュメント.

ブロックロジックの実装を提供する

それでは、次にブロックへ run(...) メソッドの実装例を追加し、意味のある結果を生成できるようにしましょう。

このセクションの内容は、堅牢なブロック実装を提供するというより、ブロック作成者として Workflow エコシステムとどのようにやり取りするかの例を示すことを目的としています。

`run(...)` メソッドの実装

from typing import Literal, Union, List, Optional, Type
from pydantic import Field
import cv2

from inference.core.workflows.prototypes.block import (
    WorkflowBlockManifest,
    WorkflowBlock,
    BlockResult,
)
from inference.core.workflows.execution_engine.entities.base import (
    OutputDefinition,
    WorkflowImageData,
)
from inference.core.workflows.execution_engine.entities.types import (
    Selector,
    IMAGE_KIND,
    FloatZeroToOne,
    FLOAT_ZERO_TO_ONE_KIND,
    BOOLEAN_KIND,
)

class ImagesSimilarityManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/images_similarity@v1"] 
    name: str
    image_1: Selector(kind=[IMAGE_KIND]) = Field(
        description="類似度を計算する最初の画像",
    )
    image_2: Selector(kind=[IMAGE_KIND]) = Field(
        description="類似度を計算する2番目の画像",
    )
    similarity_threshold: Union[
        FloatZeroToOne,
        Selector(kind=[FLOAT_ZERO_TO_ONE_KIND]),
    ] = Field(
        default=0.4,
        description="画像が類似しているとみなすしきい値",
    )

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [
          OutputDefinition(
            name="images_match", 
            kind=[BOOLEAN_KIND],
          ),
        ]

    @classmethod
    def get_execution_engine_compatibility(cls) -> Optional[str]:
        return ">=1.3.0,<2.0.0"

class ImagesSimilarityBlock(WorkflowBlock):

    def __init__(self):
        self._sift = cv2.SIFT_create()
        self._matcher = cv2.FlannBasedMatcher(dict(algorithm=1, trees=5), dict(checks=50))

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return ImagesSimilarityManifest

    def run(
        self,
        image_1: WorkflowImageData,
        image_2: WorkflowImageData,
        similarity_threshold: float,
    ) -> BlockResult:
        image_1_gray = cv2.cvtColor(image_1.numpy_image, cv2.COLOR_BGR2GRAY)
        image_2_gray = cv2.cvtColor(image_2.numpy_image, cv2.COLOR_BGR2GRAY)
        kp_1, des_1 = self._sift.detectAndCompute(image_1_gray, None)
        kp_2, des_2 = self._sift.detectAndCompute(image_2_gray, None)
        matches = self._matcher.knnMatch(des_1, des_2, k=2)
        good_matches = []
        for m, n in matches:
            if m.distance < similarity_threshold * n.distance:
                good_matches.append(m)
        return {
            "images_match": len(good_matches) > 0,
        }

• 行で 3 OpenCV をインポートします

• 行 55-57 ブロックのコンストラクタを定義します。これにより、ブロックの状態は一度だけ初期化され、連続した run(...) メソッドの呼び出しを通して維持されます。たとえば Execution Engine が動画の連続フレームで実行される場合です

• 行 69-80 ブロック機能の実装を提供します。Workflows エコシステムに関して詳細自体は本質的ではありませんが、注目すべき点がいくつかあります:

  • 行 69 と 70 〜を活用します WorkflowImageData 抽象化を用いて、どのように numpy_image プロパティを使って np.ndarray Workflows における画像の内部表現から取得できることを示しています。残りの WorkflowImageData のプロパティも確認して、さらに発見してみることをお勧めします。

  • ワークフローブロック実行の結果。行 78-80 の場合は単なる辞書です キーはマニフェストで宣言された出力名です、行 43。宣言したすべての出力を必ず提供してください。そうしないと Execution Engine がエラーを送出します。

でブロックを公開する プラグイン

これでブロックは使用可能な状態になりましたが、Execution Engine はその存在を認識していません。これは、登録済みのプラグインのどれも、あなたが作成したブロックをエクスポートしていないためです。ブロックのバンドル化の詳細は 別ページで説明されていますが、残る作業は、プラグインの load_blocks(...) 関数が返すリストにブロッククラスを追加することです:

# あなたのプラグインの __init__.py（または `inference` に直接貢献する場合は roboflow_core プラグイン）

from my_plugin.images_similarity.v1 import  ImagesSimilarityBlock  
# これは例示のインポートです！調整が必要です

def load_blocks():
    return [ImagesSimilarityBlock]

応用トピック

入力バッチを処理するブロック

場合によっては、すべての入力データをバッチとして一度に処理すると、ブロックの性能が向上することがあります。これは GPU 上で動作するモデルで起こり得ます。こうした動作モードは Workflows ブロックでサポートされています。ここでは、そのブロックでの使い方の例を示します。

バッチを受け付けるブロックの実装

from typing import Literal, Union, List, Optional, Type
from pydantic import Field
import cv2

from inference.core.workflows.prototypes.block import (
    WorkflowBlockManifest,
    WorkflowBlock,
    BlockResult,
)
from inference.core.workflows.execution_engine.entities.base import (
    OutputDefinition,
    WorkflowImageData,
    Batch,
)
from inference.core.workflows.execution_engine.entities.types import (
    Selector,
    IMAGE_KIND,
    FloatZeroToOne,
    FLOAT_ZERO_TO_ONE_KIND,
    BOOLEAN_KIND,
)

class ImagesSimilarityManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/images_similarity@v1"] 
    name: str
    image_1: Selector(kind=[IMAGE_KIND]) = Field(
        description="類似度を計算する最初の画像",
    )
    image_2: Selector(kind=[IMAGE_KIND]) = Field(
        description="類似度を計算する2番目の画像",
    )
    similarity_threshold: Union[
        FloatZeroToOne,
        Selector(kind=[FLOAT_ZERO_TO_ONE_KIND]),
    ] = Field(
        default=0.4,
        description="画像が類似しているとみなすしきい値",
    )

    @classmethod
    def get_parameters_accepting_batches(cls) -> bool:
        return ["image_1", "image_2"]

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [
          OutputDefinition(
            name="images_match", 
            kind=[BOOLEAN_KIND],
          ),
        ]

    @classmethod
    def get_execution_engine_compatibility(cls) -> Optional[str]:
        return ">=1.3.0,<2.0.0"

class ImagesSimilarityBlock(WorkflowBlock):

    def __init__(self):
        self._sift = cv2.SIFT_create()
        self._matcher = cv2.FlannBasedMatcher(dict(algorithm=1, trees=5), dict(checks=50))

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return ImagesSimilarityManifest

    def run(
        self,
        image_1: Batch[WorkflowImageData],
        image_2: Batch[WorkflowImageData],
        similarity_threshold: float,
    ) -> BlockResult:
        results = []
        for image_1_element, image_2_element in zip(image_1, image_2): 
          image_1_gray = cv2.cvtColor(image_1_element.numpy_image, cv2.COLOR_BGR2GRAY)
          image_2_gray = cv2.cvtColor(image_2_element.numpy_image, cv2.COLOR_BGR2GRAY)
          kp_1, des_1 = self._sift.detectAndCompute(image_1_gray, None)
          kp_2, des_2 = self._sift.detectAndCompute(image_2_gray, None)
          matches = self._matcher.knnMatch(des_1, des_2, k=2)
          good_matches = []
          for m, n in matches:
              if m.distance < similarity_threshold * n.distance:
                  good_matches.append(m)
          results.append({"images_match": len(good_matches) > 0})
        return results

• 行 13 imports Batch ワークフローライブラリのコアにあるクラスで、バッチ要素を保持するための、リストによく似た（ただし読み取り専用の）コンテナを表します

• 行 40-42 ブロックのデフォルト動作を変更し、バッチを処理できるようにするクラスメソッドを定義します。メソッドがバッチ指向として認識する各パラメータをマークしています run(...) メソッド はバッチ指向として認識します.

• 上記の変更により run(...) メソッドのシグネチャが変わり、今では image_1 と image_2 は WorkflowImageDataのインスタンスではなく、そうではなくこの型の要素のバッチです。 重要な注意: 複数のバッチ指向パラメータがある場合、それらのバッチの対応する位置に相互に関連する要素があることを期待します。つまり、私たちのブロックが image_1[1] を image_2[1] と比較することが、実際に論理的に意味のある操作になるようにします。

• 行 74-77, 85-86 必要に応じてバッチ要素全体で処理を実行するために導入すべき変更を示し、バッチ要素を反復処理する方法を紹介します

• 出力がどのように構築されるかに注意することが重要です。行 85 では、バッチの各要素に対して、 run(...) メソッドから返されるリスト内にそれぞれのエントリが与えられます。順序はバッチ要素の順序と一致していなければなりません。各出力辞書は、ブロック出力で宣言されたすべてのキーを提供する必要があります。

バッチとスカラーの両方を受け付ける入力

それは 比較的まれですが、単一の入力パラメータでバッチ指向データとスカラーの両方を受け付ける必要があるブロックもあり得ます。Execution Engine は、ブロックマニフェストの get_parameters_accepting_batches_and_scalars(...) メソッドを使うことでそれを認識します。以下の例をご覧ください:

from typing import Literal, Union, List, Optional, Type, Any, Dict
from pydantic import Field

from inference.core.workflows.prototypes.block import (
    WorkflowBlockManifest,
    WorkflowBlock,
    BlockResult,
)
from inference.core.workflows.execution_engine.entities.base import (
    OutputDefinition,
    Batch,
)
from inference.core.workflows.execution_engine.entities.types import (
    Selector,
)

class ExampleManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/example@v1"] 
    name: str
    param_1: Selector()
    param_2: List[Selector()]
    param_3: Dict[str, Selector()]

    @classmethod
    def get_parameters_accepting_batches_and_scalars(cls) -> bool:
        return ["param_1", "param_2", "param_3"]

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [OutputDefinition(name="dummy")]

    @classmethod
    def get_execution_engine_compatibility(cls) -> Optional[str]:
        return ">=1.3.0,<2.0.0"

class ExampleBlock(WorkflowBlock):

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return ExampleManifest

    def run(
        self,
        param_1: Any,
        param_2: List[Any],
        param_3: Dict[str, Any],
    ) -> BlockResult:
        batch_size = None
        if isinstance(param_1, Batch):
            param_1_result = ...  # バッチ指向パラメータに対して何かを行う
            batch_size = len(param_1)
        else:
            param_1_result = ... # スカラーパラメータに対して何かを行う
        for element in param_2:
           if isinstance(element, Batch):
              ...
           else:
              ...
        for key, value in param_3.items():
           if isinstance(element, value):
              ...
           else:
              ...
        if batch_size is None:
           return {"dummy": "some_result"}
        result = []
        for _ in range(batch_size):
           result.append({"dummy": "some_result"})
        return result

• 行 20-22 混在した（スカラーとバッチ指向の両方の）入力データを受け付けることが想定されるマニフェストパラメータを指定します。この段階では、定義は前の例と比べて違いがないことに注意してください。

• 行 24-26 指定する get_parameters_accepting_batches_and_scalars(...) メソッドを指定して、ブロック run(...) メソッドが、指定されたパラメータについてスカラー入力とバッチ指向入力の両方を処理できることを Execution Engine に伝えます。

• 行 45-47 混在した性質のパラメータを run(...) メソッドシグネチャに示します。

• 行 49 ブロックロジック内で、想定される出力サイズを把握しておく必要があることを示しています ブロックロジックの中で。そのため、混在入力を持つブロックの実装はかなり難しいのです。通常、ブロック run(...) メソッドがスカラーで動作する場合、多くの場合（例外は下で説明します）、メソッドは単一の出力辞書を構築します。同様に、バッチ指向入力が受け付けられる場合、それらの入力が想定される出力サイズを定義します。ただしこの場合は、バッチを手動で検出し、そのサイズを取得する必要があります。

• 行 50-54 バッチ指向データが検出されたときに異なるロジックを適用する、混在パラメータの通常の扱い方を示します

• 前述のとおり、出力の構築も混在入力の性質に合わせて調整する必要があります。これは行で示されています 65-70

フロー制御ブロックの実装

フロー制御ブロックは、単にデータを処理する他のブロックとはかなり異なります。ここではフロー制御ブロックの作成方法を示しますが、その前に少し理論的な説明をします:

• フロー制御ブロックとは、マニフェスト内でステップセレクタとの互換性を宣言するブロックです（ステップへのセレクタは $steps.{step_name} で定義されます。ステップ出力セレクタに似ていますが、出力名の指定がありません）

• フロー制御ブロックは出力を登録できず、返すものは FlowControl オブジェクト

• FlowControl オブジェクトです。これは、次に選択すべき次のステップ（ステップマニフェストで提供されたセレクタ）を、特定のバッチ要素について（SIMD フロー制御）またはワークフロー実行全体について（非 SIMD フロー制御）指定します

フロー制御の実装

ランダム継続ブロックの実装を例示し、コメントアウトしています

from typing import List, Literal, Optional, Type, Union
import random

from pydantic import Field
from inference.core.workflows.execution_engine.entities.base import (
  OutputDefinition,
  WorkflowImageData,
)
from inference.core.workflows.execution_engine.entities.types import (
    StepSelector,
    Selector,
    IMAGE_KIND,
)
from inference.core.workflows.execution_engine.v1.entities import FlowControl
from inference.core.workflows.prototypes.block import (
    BlockResult,
    WorkflowBlock,
    WorkflowBlockManifest,
)

class BlockManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/random_continue@v1"]
    name: str
    image: Selector(kind=[IMAGE_KIND]) = ImageInputField
    probability: float
    next_steps: List[StepSelector] = Field(
        description="式が true と評価された場合に実行されるステップへの参照",
        examples=[["$steps.on_true"]],
    )

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return []

    @classmethod
    def get_execution_engine_compatibility(cls) -> Optional[str]:
        return ">=1.2.0,<2.0.0"

class RandomContinueBlockV1(WorkflowBlock):

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return BlockManifest

    def run(
        self,
        image: WorkflowImageData,
        probability: float,
        next_steps: List[str],
    ) -> BlockResult:
        if not next_steps or random.random() > probability:
            return FlowControl()
        return FlowControl(context=next_steps)

• 行 10 ブロックがフローを制御していることを Execution Engine に通知するために使用されるステップセレクタの型注釈をインポートします

• 行 14 imports FlowControl フロー制御ブロックから返すことができる唯一の有効な応答であるクラス

• 行 28 ステップセレクタのリストを定義します これによってブロックは実質的にフロー制御ブロックになります

• 行 55 と 56 出力の構築方法を示します - FlowControl オブジェクトはコンテキストとして None, 文字列 または 文字列のリスト - None バッチ要素のフロー終了を表します。文字列は入力で渡された次のステップのセレクタであることが期待されます。

フロー制御の実装 - バッチ版

ランダム継続ブロックの実装を例示し、コメントアウトしています

from typing import List, Literal, Optional, Type, Union
import random

from pydantic import Field
from inference.core.workflows.execution_engine.entities.base import (
  OutputDefinition,
  WorkflowImageData,
  Batch,
)
from inference.core.workflows.execution_engine.entities.types import (
    StepSelector,
    Selector,
    IMAGE_KIND,
)
from inference.core.workflows.execution_engine.v1.entities import FlowControl
from inference.core.workflows.prototypes.block import (
    BlockResult,
    WorkflowBlock,
    WorkflowBlockManifest,
)

class BlockManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/random_continue@v1"]
    name: str
    image: Selector(kind=[IMAGE_KIND]) = ImageInputField
    probability: float
    next_steps: List[StepSelector] = Field(
        description="式が true と評価された場合に実行されるステップへの参照",
        examples=[["$steps.on_true"]],
    )

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return []

    @classmethod
    def get_parameters_accepting_batches(cls) -> List[str]:
        return ["image"]

    @classmethod
    def get_execution_engine_compatibility(cls) -> Optional[str]:
        return ">=1.3.0,<2.0.0"

class RandomContinueBlockV1(WorkflowBlock):

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return BlockManifest

    def run(
        self,
        image: Batch[WorkflowImageData],
        probability: float,
        next_steps: List[str],
    ) -> BlockResult:
        result = []
        for _ in image:
           if not next_steps or random.random() > probability:
               result.append(FlowControl())
           result.append(FlowControl(context=next_steps))
        return result

• 行 11 ブロックがフローを制御していることを Execution Engine に通知するために使用されるステップセレクタの型注釈をインポートします

• 行 15 imports FlowControl フロー制御ブロックから返すことができる唯一の有効な応答であるクラス

• 行 29-32 ステップセレクタのリストを定義します これによってブロックは実質的にフロー制御ブロックになります

• 行 38-40 の定義を含みます get_parameters_accepting_batches(...) メソッドは、ブロック run(...) メソッドはバッチ指向の 画像 パラメータを想定していることを Execution Engine に伝えます。

• 行 59 バッチの各要素ごとにフロー制御ガイドを返す必要があることを示しています。 画像 バッチ。

• その目的を達成するために、行 60 ではバッチの内容を反復処理します。

• 行 61-63 出力の構築方法を示します - FlowControl オブジェクトはコンテキストとして None, 文字列 または 文字列のリスト - None バッチ要素のフロー終了を表します。文字列は入力で渡された次のステップのセレクタであることが期待されます。

ネストされたセレクタ

一部のブロックでは、ブロックマニフェストのフィールドとしてセレクタのリストやセレクタの辞書を提供する必要があります。Execution Engine のバージョン v1 はネストを1階層しかサポートしていないため、セレクタのリストのリストや、セレクタのリストを含む辞書は正しく認識されません。

ネストされたセレクタの実用例を以下に示します。

可変数のモデルからの予測の融合

複数の分類器の予測に対して多数決を行うブロックを作りたいとしましょう。その場合、run メソッドは次のようになるはずです:

# ここに疑似コード
def run(self, predictions: List[dict]) -> BlockResult:
    predicted_classes = [p["class"] for p in predictions]
    counts = Counter(predicted_classes)
    return {"top_class": counts.most_common(1)[0]}

ネストされたセレクタ - モデルのアンサンブル

from typing import List, Literal, Optional, Type

from pydantic import Field
import supervision as sv
from inference.core.workflows.execution_engine.entities.base import (
  OutputDefinition,
)
from inference.core.workflows.execution_engine.entities.types import (
    Selector,
    OBJECT_DETECTION_PREDICTION_KIND,
)
from inference.core.workflows.prototypes.block import (
    BlockResult,
    WorkflowBlock,
    WorkflowBlockManifest,
)

class BlockManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/fusion_of_predictions@v1"]
    name: str
    predictions: List[Selector(kind=[OBJECT_DETECTION_PREDICTION_KIND])] = Field(
        description="ステップ出力へのセレクタ",
        examples=[["$steps.model_1.predictions", "$steps.model_2.predictions"]],
    )

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [
          OutputDefinition(
            name="predictions", 
            kind=[OBJECT_DETECTION_PREDICTION_KIND],
          )
        ]

    @classmethod
    def get_execution_engine_compatibility(cls) -> Optional[str]:
        return ">=1.3.0,<2.0.0"

class FusionBlockV1(WorkflowBlock):

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return BlockManifest

    def run(
        self,
        predictions: List[sv.Detections],
    ) -> BlockResult:
        merged = sv.Detections.merge(predictions)
        return {"predictions": merged}

• 行 23-26 セレクタのリストを受け付けられるマニフェストフィールドの定義方法を示します

• 行 50 ブロックの run(...) メソッドへの入力として何を期待すべきかを示します。特定の kind を表すオブジェクトのリストです。ブロックがバッチを受け付ける場合、 predictions フィールドの入力型は List[Batch[sv.Detections]

このようなブロックは、次のステップ宣言と互換性があります:

{
  "type": "my_plugin/fusion_of_predictions@v1",
  "name": "my_step",
  "predictions": [
    "$steps.model_1.predictions",
    "$steps.model_2.predictions"  
  ]
}

動的パラメータを可能にするデータ変換付きブロック

場合によっては、ブロックが「名前付き」セレクターのグループを受け入れる必要があります。これらの名前と値は、Workflow 定義の作成者によって定義されます。このような場合、ブロックのマニフェストはセレクターの辞書を受け入れる必要があり、キーがそれらのセレクターの名前として機能します。

入れ子のセレクター - 名前付きセレクター

from typing import List, Literal, Optional, Type, Any

from pydantic import Field
import supervision as sv
from inference.core.workflows.execution_engine.entities.base import (
  OutputDefinition,
)
from inference.core.workflows.execution_engine.entities.types import (
    Selector
)
from inference.core.workflows.prototypes.block import (
    BlockResult,
    WorkflowBlock,
    WorkflowBlockManifest,
)

class BlockManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/named_selectors_example@v1"]
    name: str
    data: Dict[str, Selector()] = Field(
        description="ステップ出力へのセレクタ",
        examples=[{"a": $steps.model_1.predictions", "b": "$Inputs.data"}],
    )

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [
          OutputDefinition(name="my_output", kind=[]),
        ]

    @classmethod
    def get_execution_engine_compatibility(cls) -> Optional[str]:
        return ">=1.3.0,<2.0.0"

class BlockWithNamedSelectorsV1(WorkflowBlock):

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return BlockManifest

    def run(
        self,
        data: Dict[str, Any],
    ) -> BlockResult:
        ...
        return {"my_output": ...}

• 行 22-25 セレクターの辞書を受け入れられるマニフェストフィールドを定義する方法を示します。これは、セレクター名と値の対応付けを提供します

• 行 46 ブロックの run(...) メソッド - セレクターで参照されるオブジェクトの辞書。ブロックがバッチを受け入れる場合、入力型の data フィールドの入力型は Dict[str, Union[Batch[Any], Any]]。バッチではない場合、セレクターで参照される非バッチ指向のデータは自動的にブロードキャストされます。一方、バッチを受け入れるブロックでは、 Batch コンテナはバッチ指向の入力のみをラップし、その他の入力は単一値として渡されます。

このようなブロックは、次のステップ宣言と互換性があります:

{
  "type": "my_plugin/named_selectors_example@v1",
  "name": "my_step",
  "data": {
    "a": "$steps.model_1.predictions",
    "b": "$inputs.my_parameter"  
  }
}

実際の影響は次のとおりです：

• 下に data["a"] 内で run(...) モデルの予測を見つけることができます。例えば sv.Detections もし model_1 が物体検出モデルであれば

• 下に data["b"] 内で run(...)、という名前の入力パラメータの値が見つかります my_parameter

入力と出力の次元数 vs run(...) メソッド

ブロック入力の次元数は、形成において重要な役割を果たします run(...) メソッドのシグネチャに。だからこそ、システムは入力間の次元レベルの差に厳格な上限を課しています（許容される最大差は 1）です。この制約は、ブロックを記述する際の一貫性と予測可能性を確保するために不可欠です。

次元の差が制御されていなければ、の構造を予測するのが難しくなります run(...) メソッド。そうなると開発はより難しく、信頼性も低くなります。そのため、このプロパティの検証は Workflow のコンパイル処理中に厳格に実施されます。

同様に、出力の次元数もメソッドのシグネチャと期待される出力形式に影響します。エコシステムは次のシナリオをサポートしています：

• すべての入力が 同じ次元数を持ち 、出力も 変化しない 次元数 - 基本ケース

• すべての入力が 同じ次元数を持ち および出力が 減少する 次元数

• すべての入力が 同じ次元数を持ち および出力が 増加する 次元数

• 入力が 異なる次元数を持ち 、出力は〜の次元数を維持することが許可される 参照入力

一貫性を確保し、メソッドシグネチャの曖昧さを防ぐため、入力／出力の次元数の他の組み合わせは許可されません。

次元数が `run(...)` メソッドに与える影響 - バッチ無効

出力の次元数の増加

この例では、予測に基づいて画像の動的クロップを行います。

from typing import Dict, List, Literal, Optional, Type, Union
from uuid import uuid4

from inference.core.workflows.execution_engine.constants import DETECTION_ID_KEY
from inference.core.workflows.execution_engine.entities.base import (
    OutputDefinition,
    WorkflowImageData,
    ImageParentMetadata,
)
from inference.core.workflows.execution_engine.entities.types import (
    IMAGE_KIND,
    OBJECT_DETECTION_PREDICTION_KIND,
    Selector,
)
from inference.core.workflows.prototypes.block import (
    BlockResult,
    WorkflowBlock,
    WorkflowBlockManifest,
)

class BlockManifest(WorkflowBlockManifest):
    type: Literal["my_block/dynamic_crop@v1"]
    image: Selector(kind=[IMAGE_KIND])
    predictions: Selector(
        kind=[OBJECT_DETECTION_PREDICTION_KIND],
    )

    @classmethod
    def get_output_dimensionality_offset(cls) -> int:
        return 1

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [
            OutputDefinition(name="crops", kind=[IMAGE_KIND]),
        ]

    @classmethod
    def get_execution_engine_compatibility(cls) -> Optional[str]:
        return ">=1.3.0,<2.0.0"

class DynamicCropBlockV1(WorkflowBlock):

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return BlockManifest

    def run(
        self,
        image: WorkflowImageData,
        predictions: sv.Detections,
    ) -> BlockResult:
        crops = []
        for (x_min, y_min, x_max, y_max) in predictions.xyxy.round().astype(dtype=int):
            cropped_image = image.numpy_image[y_min:y_max, x_min:x_max]
            parent_metadata = ImageParentMetadata(parent_id=f"{uuid4()}")
            もし
                result = WorkflowImageData(
                    parent_metadata=parent_metadata,
                    numpy_image=cropped_image,
                )
            else:
                result = None
            crops.append({"crops": result})
        return crops

• 行で 28-30 マニフェストクラスは出力の次元オフセットを宣言します - 値 1 は、追加するものと理解されるべきです 1 次元レベルに

• 行であることに注意してください 63、ブロックは空の画像を以降の処理から除外しますが、 None 出力を持つ辞書の代わりにそれを配置します。これは条件付き実行で使われるものと同じ Execution Engine の動作を利用し、データポイントは下流の処理から除外されます（下流に空の入力を要求するステップがある場合を除く）。

• 行で 64-65 単一入力の結果 画像 と predictions が収集されます。これは、登録されたすべての出力をキーとして含む辞書のリストを意味します。Execution Engine は、このステップが各入力要素に対して要素のバッチを返すと理解し、下流ステップの実行中に追跡するための入れ子のインデックス構造を作成します。

出力の次元数の減少

この例では、ブロックがクロップの予測を可視化し、すべてのクロップ予測を1枚の出力画像にまとめたタイルを作成します。

from typing import List, Literal, Type, Union

import supervision as sv

from inference.core.utils.drawing import create_tiles
from inference.core.workflows.execution_engine.entities.base import (
    Batch,
    OutputDefinition,
    WorkflowImageData,
)
from inference.core.workflows.execution_engine.entities.types import (
    IMAGE_KIND,
    OBJECT_DETECTION_PREDICTION_KIND,
    Selector,
)
from inference.core.workflows.prototypes.block import (
    BlockResult,
    WorkflowBlock,
    WorkflowBlockManifest,
)

class BlockManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/tile_detections@v1"]
    crops: Selector(kind=[IMAGE_KIND])
    crops_predictions: Selector(
        kind=[OBJECT_DETECTION_PREDICTION_KIND]
    )
    scalar_parameter: Union[float, Selector()]

    @classmethod
    def get_output_dimensionality_offset(cls) -> int:
        return -1

    @classmethod
    def get_parameters_enforcing_auto_batch_casting(cls) -> List[str]:
        return ["crops", "crops_predictions"]

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [
            OutputDefinition(name="visualisations", kind=[IMAGE_KIND]),
        ]

class TileDetectionsBlock(WorkflowBlock):

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return BlockManifest

    def run(
        self,
        crops: Batch[WorkflowImageData],
        crops_predictions: Batch[sv.Detections],
        scalar_parameter: float,
    ) -> BlockResult:
        annotator = sv.BoxAnnotator()
        visualisations = []
        for image, prediction in zip(crops, crops_predictions):
            annotated_image = annotator.annotate(
                image.numpy_image.copy(),
                prediction,
            )
            visualisations.append(annotated_image)
        tile = create_tiles(visualisations)
        return {"visualisations": tile}

• 行で 30-32 マニフェストクラスは出力の次元オフセットを宣言します - 値 -1 は、次元レベルを〜だけ減少させるものと理解されるべきです 1

• 行で 34-36 マニフェストクラスは run(...) シグネチャが常に安定するよう、自動バッチキャスティングの対象となるメソッド入力を宣言します。自動バッチキャスティングは Execution Engine で導入されました v0.1.6.0

• 参照してください 変更履歴 詳細について。

• 行で 53-55 出力の次元数の減少がメソッドシグネチャに与える影響がわかります。最初の2つの入力（行で宣言されている 36）は人工的に〜でラップされます Batch[] コンテナに、一方で scalar_parameter はプリミティブ型のままです。これは、すべての入力が同じ次元数を持つ場合に、最終次元レベルを占めるすべての要素へアクセスできるようにするため、出力の次元数が減少したときに Execution Engine によって自動的に行われます。もちろん、トップレベルのバッチ内で同じ要素に関連する要素だけがまとめられます。たとえば、2つの入力画像をクロップした場合、それぞれ異なる画像からのクロップは別々にグループ化されます。

• 行 65-66 出力がどのように構築されるかを示します - 単一の値が返され、その値は次元数が削減された出力バッチ内で Execution Engine によりインデックス付けされます

異なる入力次元数

この例では、ブロックが元画像のクロップに基づいて予測された検出結果をマージします。結果として、すべての部分的な検出を統合した単一の検出結果を提供します。

from copy import deepcopy
from typing import Dict, List, Literal, Optional, Type, Union

import numpy as np
import supervision as sv

from inference.core.workflows.execution_engine.entities.base import (
    Batch,
    OutputDefinition,
    WorkflowImageData,
)
from inference.core.workflows.execution_engine.entities.types import (
    OBJECT_DETECTION_PREDICTION_KIND,
    Selector,
    IMAGE_KIND,
)
from inference.core.workflows.prototypes.block import (
    BlockResult,
    WorkflowBlock,
    WorkflowBlockManifest,
)

class BlockManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/stitch@v1"]
    image: Selector(kind=[IMAGE_KIND])
    image_predictions: Selector(
        kind=[OBJECT_DETECTION_PREDICTION_KIND],
    )

    @classmethod
    def get_input_dimensionality_offsets(cls) -> Dict[str, int]:
        return {
            "image": 0,
            "image_predictions": 1,
        }

    @classmethod
    def get_dimensionality_reference_property(cls) -> Optional[str]:
        return "image"

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [
            OutputDefinition(
                name="predictions",
                kind=[
                    OBJECT_DETECTION_PREDICTION_KIND,
                ],
            ),
        ]

class StitchDetectionsNonBatchBlock(WorkflowBlock):

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return BlockManifest

    def run(
        self,
        image: WorkflowImageData,
        image_predictions: Batch[sv.Detections],
    ) -> BlockResult:
        image_predictions = [deepcopy(p) for p in image_predictions if len(p)]
        for p in image_predictions:
            coords = p["parent_coordinates"][0]
            p.xyxy += np.concatenate((coords, coords))
        return {"predictions": sv.Detections.merge(image_predictions)}

• 行で 31-36 マニフェストクラスは入力の次元オフセットを宣言し、 画像 パラメータがトップレベルであり、 image_predictions 予測の入れ子バッチであることを示します

• 異なる入力次元数が宣言される場合は常に、次元参照プロパティを指定する必要があります（行を参照 38-40） - この次元レベルは出力次元数の計算に使用されます - この特定のケースでは、 画像を指定します。この選択は結果の期待フォーマットに影響します。選択したシナリオでは、登録済みの出力キーをすべて含む単一の辞書を返す必要があります。もし選択が image_predictionsなら、辞書のリスト（ image_predictions バッチの長さに等しいサイズ）を返します。言い換えれば、 get_dimensionality_reference_property(...) どの次元レベルを出力に関連付けるかを示します。

• 行 63-64 行で指定された次元オフセットの影響を示します 31-36。明らかに image_predictions は、〜に関する入れ子バッチです 画像。明らかに、特定の 画像に関連する入れ子の予測のみが バッチとしてまとめられ、実行時にメソッドへ渡されます。

• 先に述べたように、行 69 は単一の辞書となる出力を構築します。これは、出力を〜の次元レベルで登録するためです 画像 （これも単一要素として渡されました）

次元数が `run(...)` メソッドに与える影響 - バッチ有効

出力の次元数の増加

この例では、予測に基づいて画像の動的クロップを行います。

from typing import Dict, List, Literal, Optional, Type, Union
from uuid import uuid4

from inference.core.workflows.execution_engine.constants import DETECTION_ID_KEY
from inference.core.workflows.execution_engine.entities.base import (
    OutputDefinition,
    WorkflowImageData,
    ImageParentMetadata,
    Batch,
)
from inference.core.workflows.execution_engine.entities.types import (
    IMAGE_KIND,
    OBJECT_DETECTION_PREDICTION_KIND,
    Selector,
)
from inference.core.workflows.prototypes.block import (
    BlockResult,
    WorkflowBlock,
    WorkflowBlockManifest,
)

class BlockManifest(WorkflowBlockManifest):
    type: Literal["my_block/dynamic_crop@v1"]
    image: Selector(kind=[IMAGE_KIND])
    predictions: Selector(
        kind=[OBJECT_DETECTION_PREDICTION_KIND],
    )

    @classmethod
    def get_parameters_accepting_batches(cls) -> bool:
        return ["image", "predictions"]

    @classmethod
    def get_output_dimensionality_offset(cls) -> int:
        return 1

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [
            OutputDefinition(name="crops", kind=[IMAGE_KIND]),
        ]

    @classmethod
    def get_execution_engine_compatibility(cls) -> Optional[str]:
        return ">=1.3.0,<2.0.0"

class DynamicCropBlockV1(WorkflowBlock):

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return BlockManifest

    def run(
        self,
        image: Batch[WorkflowImageData],
        predictions: Batch[sv.Detections],
    ) -> BlockResult:
        results = []
        for single_image, detections in zip(image, predictions):
            crops = []
            for (x_min, y_min, x_max, y_max) in detections.xyxy.round().astype(dtype=int):
                cropped_image = single_image.numpy_image[y_min:y_max, x_min:x_max]
                parent_metadata = ImageParentMetadata(parent_id=f"{uuid4()}")
                もし
                    result = WorkflowImageData(
                        parent_metadata=parent_metadata,
                        numpy_image=cropped_image,
                    )
                else:
                    result = None
                crops.append({"crops": result})
            results.append(crops)
        return results

• 行で 29-31 マニフェストは、ブロックが入力のバッチを受け入れることを宣言しています

• 行で 33-35 マニフェストクラスは出力の次元オフセットを宣言します - 値 1 は、追加するものと理解されるべきです 1 次元レベルに

• 行で 55-66。入力パラメータのシグネチャは、それが run(...) メソッドが同じ次元数の入力に対して実行され、それらの入力がバッチで提供されることを反映しています

• 行であることに注意してください 70、ブロックは空の画像を以降の処理から除外しますが、 None 出力を持つ辞書の代わりにそれを配置します。これは条件付き実行で使われるものと同じ Execution Engine の動作を利用し、データポイントは下流の処理から除外されます（下流に空の入力を要求するステップがある場合を除く）。

• 行で示される出力の構築は 71-73 は、入れ子が2段階であることを示しています。まず、ブロックはバッチ上で動作するため、入力バッチ要素ごとに1つの出力を持つ出力のリストを返すことが期待されます。さらに、各入力バッチ要素のこの出力要素は入れ子バッチであることがわかります。したがって、各入力画像と予測に対して、ブロックは出力のリストを生成します。リストの各要素は、各宣言済み出力の値を提供する辞書です。

出力の次元数の減少

この例では、ブロックがクロップの予測を可視化し、すべてのクロップ予測を1枚の出力画像にまとめたタイルを作成します。

from typing import List, Literal, Type, Union

import supervision as sv

from inference.core.utils.drawing import create_tiles
from inference.core.workflows.execution_engine.entities.base import (
    Batch,
    OutputDefinition,
    WorkflowImageData,
)
from inference.core.workflows.execution_engine.entities.types import (
    IMAGE_KIND,
    OBJECT_DETECTION_PREDICTION_KIND,
    Selector,
)
from inference.core.workflows.prototypes.block import (
    BlockResult,
    WorkflowBlock,
    WorkflowBlockManifest,
)

class BlockManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/tile_detections@v1"]
    images_crops: Selector(kind=[IMAGE_KIND])
    crops_predictions: Selector(
        kind=[OBJECT_DETECTION_PREDICTION_KIND]
    )

    @classmethod
    def get_parameters_accepting_batches(cls) -> bool:
        return ["images_crops", "crops_predictions"]

    @classmethod
    def get_output_dimensionality_offset(cls) -> int:
        return -1

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [
            OutputDefinition(name="visualisations", kind=[IMAGE_KIND]),
        ]

class TileDetectionsBlock(WorkflowBlock):

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return BlockManifest

    def run(
        self,
        images_crops: Batch[Batch[WorkflowImageData]],
        crops_predictions: Batch[Batch[sv.Detections]],
    ) -> BlockResult:
        annotator = sv.BoxAnnotator()
        visualisations = []
        for image_crops, crop_predictions in zip(images_crops, crops_predictions):
            visualisations_batch_element = []
            for image, prediction in zip(image_crops, crop_predictions):
                annotated_image = annotator.annotate(
                    image.numpy_image.copy(),
                    prediction,
                )
                visualisations_batch_element.append(annotated_image)
            tile = create_tiles(visualisations_batch_element)
            visualisations.append({"visualisations": tile})
        return visualisations

• 行 29-31 ブロックが入力としてバッチを受け取ることが期待されることを示すマニフェスト

• 行で 33-35 マニフェストクラスは出力の次元オフセットを宣言します - 値 -1 は、次元レベルを〜だけ減少させるものと理解されるべきです 1

• 行で 52-53 出力次元数の減少とバッチ処理がメソッドシグネチャに与える影響がわかります。最初の「層」は Batch[] マニフェストがブロックが入力のバッチを受け入れると宣言していることの副作用です。2つ目の「層」は出力次元数の減少によるものです。Execution Engine は、削減される次元を追加の Batch[] コンテナにまとめて入力として提供します。これにより、プログラマーは特定のトップレベルバッチ要素に属するすべての入れ子バッチ要素を収集できます。

• 行 66-67 出力がどのように構築されるかを示します - 各トップレベルバッチ要素について、ブロックはすべてのクロップと予測を集約して1つのタイルを作成します。ブロックは入力のバッチを受け入れるため、この処理の結果、トップレベルバッチ要素ごとに1つのタイルが生成されます。したがって、辞書のリストを返すことが期待されます。

異なる入力次元数

この例では、ブロックが元画像のクロップに基づいて予測された検出結果をマージします。結果として、すべての部分的な検出を統合した単一の検出結果を提供します。

from copy import deepcopy
from typing import Dict, List, Literal, Optional, Type, Union

import numpy as np
import supervision as sv

from inference.core.workflows.execution_engine.entities.base import (
    Batch,
    OutputDefinition,
    WorkflowImageData,
)
from inference.core.workflows.execution_engine.entities.types import (
    OBJECT_DETECTION_PREDICTION_KIND,
    Selector,
    IMAGE_KIND,
)
from inference.core.workflows.prototypes.block import (
    BlockResult,
    WorkflowBlock,
    WorkflowBlockManifest,
)

class BlockManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/stitch@v1"]
    images: Selector(kind=[IMAGE_KIND])
    images_predictions: Selector(
        kind=[OBJECT_DETECTION_PREDICTION_KIND],
    )

    @classmethod
    def get_parameters_accepting_batches(cls) -> bool:
        return ["images", "images_predictions"]

    @classmethod
    def get_input_dimensionality_offsets(cls) -> Dict[str, int]:
        return {
            "image": 0,
            "image_predictions": 1,
        }

    @classmethod
    def get_dimensionality_reference_property(cls) -> Optional[str]:
        return "image"

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [
            OutputDefinition(
                name="predictions",
                kind=[
                    OBJECT_DETECTION_PREDICTION_KIND,
                ],
            ),
        ]

class StitchDetectionsBatchBlock(WorkflowBlock):

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return BlockManifest

    def run(
        self,
        images: Batch[WorkflowImageData],
        images_predictions: Batch[Batch[sv.Detections]],
    ) -> BlockResult:
        result = []
        for image, image_predictions in zip(images, images_predictions):
            image_predictions = [deepcopy(p) for p in image_predictions if len(p)]
            for p in image_predictions:
                coords = p["parent_coordinates"][0]
                p.xyxy += np.concatenate((coords, coords))
            merged_prediction = sv.Detections.merge(image_predictions)
            result.append({"predictions": merged_prediction})
        return result

• 行 31-33 ブロックが入力としてバッチを受け取ることが期待されることを示すマニフェスト

• 行で 35-40 マニフェストクラスは入力の次元オフセットを宣言し、 画像 パラメータがトップレベルであり、 image_predictions 予測の入れ子バッチであることを示します

• 異なる入力次元数が宣言される場合は常に、次元参照プロパティを指定する必要があります（行を参照 42-44） - この次元レベルは出力次元数の計算に使用されます - この特定のケースでは、 画像。この選択は結果の期待フォーマットに影響します。選択したシナリオでは、〜の各要素に対して単一の辞書を返すことが想定されます 画像 バッチです。もし選択が image_predictionsなら、辞書のリスト（入れ子の image_predictions バッチの長さに等しいサイズ）を各入力 画像 バッチ要素ごとに返します。

• 行 66-67 行で指定された次元オフセットの影響を示します 35-40 ならびに行でのバッチ処理の宣言も 32-34。最初の「層」は Batch[] コンテナは後者の、入れ子の Batch[Batch[]] 〜に対して images_predictions は入力次元オフセットの定義に由来します。明らかに image_predictions は、〜の特定要素に関連する予測のバッチを保持します 画像 バッチ。

• 先に述べたように、行 76-77 は、〜の各要素ごとに単一の辞書となる出力を構築します 画像 バッチ

空の入力を受け入れるブロック

前述のとおり、Workflow の実行中に一部のバッチ要素が「空」になることがあります。これはいくつかの要因で起こりえます：

• フロー制御の仕組み： 特定の実行分岐が、特定のバッチ要素を隠し、後続ステップで処理されるのを防ぐことがあります。

• データ処理ブロックでは： 場合によっては、ブロックが特定のデータポイントに対して意味のある出力を生成できないことがあります。たとえば、Dynamic Crop ブロックはバウンディングボックスのサイズがゼロであればクロップ画像を生成できません。

一部のブロックは、これらの空の入力を扱うように設計されており、たとえば欠落した出力をデフォルト値で置き換えられるブロックがあります。このブロックは、Workflow で構造化された出力を構築する際に特に有用で、一部の要素が空であっても出力に欠落要素がなく、解析しにくくなるのを防ぎます。

空の入力を受け入れるブロック

from typing import Any, List, Literal, Optional, Type

from inference.core.workflows.execution_engine.entities.base import (
    Batch,
    OutputDefinition,
)
from inference.core.workflows.execution_engine.entities.types import Selector
from inference.core.workflows.prototypes.block import (
    BlockResult,
    WorkflowBlock,
    WorkflowBlockManifest,
)

class BlockManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/first_non_empty_or_default@v1"]
    data: List[Selector()]
    default: Any

    @classmethod
    def accepts_empty_values(cls) -> bool:
        return True

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [OutputDefinition(name="output")]

    @classmethod
    def get_execution_engine_compatibility(cls) -> Optional[str]:
        return ">=1.3.0,<2.0.0"

class FirstNonEmptyOrDefaultBlockV1(WorkflowBlock):

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return BlockManifest

    def run(
        self,
        data: Batch[Optional[Any]],
        default: Any,
    ) -> BlockResult:
        result = default
        for data_element in data:
            if data_element is not None:
                return {"output": data_element}
        return {"output": result}

• 行で 20-22 ブロックが空の入力を受け入れることを示す宣言を見つけるかもしれません

• 行の結果として 20-22 は行に見られます 41、シグネチャが入力が Batch 処理する必要のある空要素を含む可能性があります。実際には、このブロックは空の値を置き換える「人工的」な出力を生成するため、このブロックの出力を参照する、空の入力を受け入れないブロックからそれらの出力を「見える」状態にできます。Execution Engine が実行時に生成したデータで置き換える各入力は、オプション要素を提供する可能性があると考えるべきです。

カスタムコンストラクタ引数を持つブロック

一部のブロックは、外部で構築されたオブジェクトを必要とする場合があります。そのような場合、Workflows Execution Engine の役割はそれらのエンティティをブロックへ渡し、利用可能にすることです。この仕組みは Workflows Compiler を紹介するページでで説明されています。これはブロッククラスからステップを動的に構築する責任を持つコンポーネントだからです。

コンストラクタ引数は次の条件を満たす必要があります：

• ブロックによって要求される - クラスメソッドを使用して WorkflowBlock.get_init_parameters(...)

• Workflows Execution Engine を実行している環境で提供される：

  • 次のように直接 この 例

  • デフォルトを使用して Workflow プラグインに登録された

ブロックを定義しながら init パラメータを要求する方法を見てみましょう。

コンストラクタ引数を要求するブロック

from typing import Any, List, Literal, Optional, Type

from inference.core.workflows.execution_engine.entities.base import (
    Batch,
    OutputDefinition,
)
from inference.core.workflows.execution_engine.entities.types import Selector
from inference.core.workflows.prototypes.block import (
    BlockResult,
    WorkflowBlock,
    WorkflowBlockManifest,
)

class BlockManifest(WorkflowBlockManifest):
    type: Literal["my_plugin/example@v1"]
    data: List[Selector()]

    @classmethod
    def describe_outputs(cls) -> List[OutputDefinition]:
        return [OutputDefinition(name="output")]

    @classmethod
    def get_execution_engine_compatibility(cls) -> Optional[str]:
        return ">=1.0.0,<2.0.0"

class ExampleBlock(WorkflowBlock):

    def __init__(my_parameter: int):
        self._my_parameter = my_parameter

    @classmethod
    def get_init_parameters(cls) -> List[str]:
        return ["my_parameter"]

    @classmethod
    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
        return BlockManifest

    def run(
        self,
        data: Batch[Any],
    ) -> BlockResult:
        pass

• 行 30-31 パラメータなしではないクラスコンストラクタを宣言する

• ブロックがカスタム初期化を必要とすることを Execution Engine に知らせるため、 get_init_parameters(...) 行のメソッド 33-35 提供されなければならないすべてのパラメータ名を列挙します

エアギャップ／オフライン利用可否メタデータ

インターネット接続なしで動作する可能性がある環境（エアギャップ展開）向けにブロックを作成する場合、 WorkflowBlockManifest 3つのオプションのクラスメソッドを公開します。これらをオーバーライドすると、エアギャップ対応のワークフロービルダーがどのブロックをオフラインで使用できるか判断する助けになります。

3つとも妥当なデフォルトを持つため、既存のブロックには 変更は不要です.

get_air_gapped_availability()

ブロックがインターネットなしで動作できるかを宣言します。クラウド API（OpenAI、Anthropic など）を呼び出すブロックでは、これをオーバーライドしてください。

from inference.core.workflows.prototypes.block import (
    AirGappedAvailability,
    WorkflowBlockManifest,
)

class BlockManifest(WorkflowBlockManifest):
    # ...

    @classmethod
    def get_air_gapped_availability(cls) -> AirGappedAvailability:
        return AirGappedAvailability(
            available=False,
            reason="requires_internet",
        )

デフォルトでは AirGappedAvailability(available=True) を返します。- 純ロジックのブロック、ローカルネットワークのブロック、および外部接続を必要としないすべてのブロックに適しています。

get_supported_model_variants()

重みをローカルに事前キャッシュできる基盤モデルブロックでは、モデルバリアントIDのリストを返します。ブロックは次の場合にオフラインで利用可能と見なされます いずれかの バリアントにキャッシュ済みアーティファクトがある。

class BlockManifest(WorkflowBlockManifest):
    # ...

    @classmethod
    def get_supported_model_variants(cls) -> Optional[List[str]]:
        return [
            "sam2/hiera_large",
            "sam2/hiera_small",
            "sam2/hiera_tiny",
            "sam2/hiera_b_plus",
        ]

デフォルトでは None。つまり、このブロックはローカルにキャッシュされたモデル重みに依存しません。

get_compatible_task_types()

ユーザーが学習したモデルを受け付ける Roboflow モデルブロックについて、このブロックが扱えるタスク種別を返します。エアギャップ構成のビルダーは、これを使ってキャッシュされたユーザーモデルを互換性のあるブロックに対応付けます。

class BlockManifest(WorkflowBlockManifest):
    # ...

    @classmethod
    def get_compatible_task_types(cls) -> Optional[List[str]]:
        return ["object-detection"]

デフォルトでは None - Roboflow モデルでパラメータ化されないブロック（基盤モデル、ロジックブロック、シンクなど）に適しています。

実行時の制約

一部のブロックは、配備されるランタイム（ホスト型サーバーレス、専用デプロイ、セルフホスト、推論パイプライン）、ステップ実行モード（ローカル vs. リモート）、入力モード（画像 vs. 動画）によって挙動が異なったり、完全に失敗したりします。これを上書きする get_restrictions() on WorkflowBlockManifest これらの注意点をブロック内に一度だけ宣言し、実行エンジン、スキーマエンドポイント、自動生成されるブロックギャラリーがすべて一貫して表示できるようにします。

デフォルトでは []、そのため既存のブロックは 変更は不要です.

重大度: 軽度 対 重大

各制約には 重大度:

• Severity.SOFT - ブロックは最後まで実行され、正しい出力形状を返しますが、値が劣化していたり意味をなさなかったりします（例: トラッカーIDがリクエスト間でリセットされる、クールダウンがスロットリングしない、ファイルが一時ディスクに書き込まれる）。ワークフロー自体は動き続けますが、結果はユーザーの期待通りではありません。

• Severity.HARD - このランタイムではブロックが実行されない／例外を投げる／利用可能な出力を生成できません。エンジンはコンパイルを拒否するか、即時失敗すべきです。

制約の適用範囲を定める

ある RuntimeRestriction は3つの軸の任意の組み合わせに沿って適用範囲を限定できます。ある軸が Noneのままなら、その制約はその軸のすべての値に適用されます。

• applies_to_runtimes: Runtime.HOSTED_SERVERLESS, Runtime.DEDICATED_DEPLOYMENT, Runtime.SELF_HOSTED_CPU, Runtime.SELF_HOSTED_GPU, Runtime.INFERENCE_PIPELINE.

• applies_to_step_execution_modes: StepExecutionMode.LOCAL, StepExecutionMode.REMOTE.

• applies_to_input_modes: RuntimeInputMode.IMAGE, RuntimeInputMode.VIDEO.

この 注記 フィールドは、失敗モードまたは劣化した挙動を1行で人間が読める形で説明するものです。何が起きるか（例: "track_ids がリクエスト間でリセットされる", "一時的な /tmp に書き込まれる"）を述べ、抽象的な前提条件は書かないでください。

共有プリセット

ほとんどの注意点は少数の共通パターンに収まるため、再利用可能なプリセットを dataclass とともにエクスポートし、コードベース全体で表現を一貫させています。まずはこれらを使ってください:

• STATEFUL_VIDEO_HTTP_SOFT_RESTRICTION - 動画トラッキング／カウント／集計ブロック向け。動画ごとの状態がプロセス内メモリにあり、ステートレスな HTTP リクエストの間でリセットされます。

• COOLDOWN_HTTP_SOFT_RESTRICTION - クールダウン／レート制限タイマーがプロセス内メモリに保持されるブロック向けで、そのため複数レプリカの HTTP ランタイムではスロットリングしません。

• STILL_IMAGE_INPUT_SOFT_RESTRICTION - 時間的文脈（動画または繰り返しフレーム）に依存し、静止画像ではほとんどまたはまったく効果がないブロック向け。

例

動画ごとの状態を保持し、静止画像では意味をなさないラインカウンターブロックは、両方の制約を宣言します:

from typing import List

from inference.core.workflows.prototypes.block import (
    STATEFUL_VIDEO_HTTP_SOFT_RESTRICTION,
    STILL_IMAGE_INPUT_SOFT_RESTRICTION,
    RuntimeRestriction,
    WorkflowBlockManifest,
)

class BlockManifest(WorkflowBlockManifest):
    # ...

    @classmethod
    def get_restrictions(cls) -> List[RuntimeRestriction]:
        return [
            STATEFUL_VIDEO_HTTP_SOFT_RESTRICTION,
            STILL_IMAGE_INPUT_SOFT_RESTRICTION,
        ]

カスタム制約（例: ホスト型サーバーレスランタイムに存在しない GPU ハードウェアを必要とするブロック）は、以下のようにインラインで宣言します: Severity.HARD ブロックには CUDA GPU が必要です。CPU のみのワーカーでは例外を投げます。

from inference.core.workflows.prototypes.block import (
    Runtime,
    RuntimeRestriction,
    Severity,
    WorkflowBlockManifest,
)

class BlockManifest(WorkflowBlockManifest):
    # ...

    @classmethod
    def get_restrictions(cls) -> List[RuntimeRestriction]:
        return [
            RuntimeRestriction(
                severity=Severity.HARD,
                note="ブロックには CUDA GPU が必要です。CPU のみのワーカーでは例外を投げます。",
                applies_to_runtimes=[
                    Runtime.HOSTED_SERVERLESS,
                    Runtime.SELF_HOSTED_CPU,
                ],
            ),
        ]

このように宣言された制約は、次の3か所で表示されます:

1. この describe_interface HTTP ペイロード（ RuntimeRestriction.to_dict()）、ワークフローの実行前にワークフロークライアントやビルダーがユーザーに警告できます。

2. ブロックの自動生成されるブロックギャラリーページの「Runtime compatibility」セクションにあり、直後の Properties.

3. 実行エンジン。ここでは Severity.HARD 現在のランタイムに対する制約で即時失敗を選べます。

依存リソースの宣言

多くのブロックは実行時に外部リソースを必要とします。Roboflow で学習したモデルや基盤モデルの重み、Roboflow プロジェクト（例: アクティブラーニングの対象、データセットアップロード先）、またはサードパーティのホスト型モデル（OpenAI、Anthropic、OpenRouter など）です。ワークフローが実際に実行されるまで、システム内のどこにも、どのリソースが必要になるかは分かりません。

上書きする discover_dependent_resources() をブロックマニフェストで上書きすると、そのギャップを埋められます。このメソッドは 解析済みのマニフェストインスタンスに対して呼び出されるため、特定ステップの具体的なフィールド値から宣言を計算できます。これにより呼び出し側は、何も実行せずにコンパイル時点でワークフロー全体のリソースを静的に列挙でき、予測可能な実行時間のためにモデル重みを事前読み込みしたり、API キーが参照されているすべてのモデルとプロジェクトにアクセスできることを事前に確認したりといった用途が可能になります。

def discover_dependent_resources(self) -> Optional[List[DependentResource]]:
    ...

デフォルトでは None。これは、ブロックが 宣言しない 依存関係を――呼び出し側はそれを 不明として扱わなければなりません。これは意図的に []、これはブロックが外部リソースを必要としないことを明示的に宣言します。既存のブロックに変更は不要です。リソースを使用するブロックはオーバーライドすべきです。

リソースのエンベロープ

ブロックはリソースの形状を独自に定義しません。エンベロープは実行エンジンによって管理されます。A DependentResource は resource_type を、その型に登録された型付き（pydantic）のメタデータエンティティと組み合わせます:

• DependentResourceType.ROBOFLOW_PLATFORM_MODEL → RoboflowPlatformModelMetadata(model_id, required_action, execution_location) — Roboflow プラットフォーム経由で提供されるモデル。合成された ID を持つ基盤／コアモデルも含みます（例: clip/ViT-B-32). required_action 利用形態を示します: ModelRequiredAction.EXECUTION （重みが取得される、または推論が要求される）か ModelRequiredAction.ACCESS （モデルエンティティがプラットフォーム上で到達可能であればよいだけの場合。例: メタデータを付与する監視シンク）。実行の場合、 execution_location は LOCAL, REMOTE、 ENVIRONMENT_DEFINED ランタイムで WORKFLOWS_STEP_EXECUTION_MODE によって決定され、コンパイル時には判定できない場合です（ステップ実行モードで分岐するモデルブロックのデフォルト）。

• DependentResourceType.ROBOFLOW_PLATFORM_PROJECT → RoboflowPlatformProjectMetadata(project_url) — ブロックが読み取りまたは書き込みを行う Roboflow プロジェクトです。

• DependentResourceType.THIRD_PARTY_MODEL → ThirdPartyModelMetadata(provider, model_id) — 外部プロバイダーによって実行されるモデル。定義上、リモート実行です。

ファクトリヘルパーにより、実装は1行で済みます: roboflow_platform_model(), roboflow_platform_project(), third_party_model().

セレクター値とランタイム解決

マニフェストのフィールドには、具体的な値の代わりにワークフローセレクターが入ることがあります。宣言はそのような値を そのまま — ブロックはセレクターを解決しません。呼び出し側は $inputs.<name> の参照を、ランタイムパラメータが分かり次第置き換えられます; $steps.<name>.<property> の参照は静的にはまったく解決できません。すべてのメタデータエンティティは requires_runtime_resolution() を公開しており、呼び出し側は具体的な識別子と、まだ解決が必要な参照を区別できます。

When the final identifier is a 関数 の field value（ clip/<version>、カタログ検索など）、置換された入力値だけでは実行される ID になりません。そのような宣言には model_id_resolver — 必要なものをすべてクロージャに保持した呼び出し可能オブジェクト — を付与し、置換された値を最終 ID に変換します。この resolver はプロセス内の補助にすぎず、シリアライズ、JSON スキーマ、等価性の対象外です。プロセス内の呼び出し側（例: エンジンの初回実行時の事前読み込み）は、入力値を置換した後にこれを呼び出します。resolver は None 値が静的に解決不能であることを宣言するために返すことができます（最終 ID がその1つの値だけに依存しない場合）。その場合、呼び出し側はその依存関係をスキップし、実行時に解決します。例外を投げる場合、その値は本当に無効です。

に合わせてください run() 読み込む

宣言された識別子は、ちょうど run() が要求するものと完全に一致していなければなりません。バージョンフィールドから合成された ID も含みます。ブロックが f"my_family/{self.version}" という形を load_core_model(...) の呼び出し箇所で組み立てているなら、宣言も同じ ID を組み立てる必要があります（そして version フィールドにセレクターが入っている場合は、そのままのセレクターにフォールバックします）。

例

モデルブロックが、自身のモデルと任意のアクティブラーニング対象プロジェクトを宣言する例:

from inference.core.workflows.prototypes.block import (
    DependentResource,
    WorkflowBlockManifest,
    roboflow_platform_model,
    roboflow_platform_project,
)

class BlockManifest(WorkflowBlockManifest):
    # ...

    def discover_dependent_resources(self) -> Optional[List[DependentResource]]:
        resources = [roboflow_platform_model(model_id=self.model_id)]
        if self.active_learning_target_dataset is not None:
            resources.append(
                roboflow_platform_project(
                    project_url=self.active_learning_target_dataset
                )
            )
        return resources

version フィールドからモデル ID を合成するブロックの例:

from inference.core.workflows.prototypes.block import (
    DependentResource,
    WorkflowBlockManifest,
    is_workflow_selector,
    roboflow_platform_model,
)

class BlockManifest(WorkflowBlockManifest):
    # ...

    def discover_dependent_resources(self) -> Optional[List[DependentResource]]:
        if is_workflow_selector(self.version):
            # セレクターはそのまま返される。resolver が最終 ID を生成する
            # 入力値が置換された後に。
            return [
                roboflow_platform_model(
                    model_id=self.version,
                    model_id_resolver=lambda version: f"clip/{version}",
                )
            ]
        return [roboflow_platform_model(model_id=f"clip/{self.version}")]

実行せずにモデルエンティティのみを参照するシンクの例:

from inference.core.workflows.prototypes.block import (
    DependentResource,
    ModelRequiredAction,
    WorkflowBlockManifest,
    roboflow_platform_model,
)

class BlockManifest(WorkflowBlockManifest):
    # ...

    def discover_dependent_resources(self) -> Optional[List[DependentResource]]:
        return [
            roboflow_platform_model(
                model_id=self.model_id,
                required_action=ModelRequiredAction.ACCESS,
            )
        ]

宣言すべきでない場合

単に 保持する モデル ID や project-kind の値を一般的なペイロードとして運ぶだけのフィールド（例: 値をクエリパラメータとして転送する webhook シンク）はリソース依存ではありません。そうしたブロックは意図的にデフォルトを維持します。

コアリポジトリの貢献者は、単体テスト（tests/workflows/unit_tests/core_steps/test_dependent_resources.py）が境界を守っていることを認識してください。マニフェストで roboflow_model_id または roboflow_project kind のフィールドを宣言するすべてのコアブロックは、オーバーライドするか、carry-only として明示的に許可リストに載っていなければなりません。 discover_dependent_resources() kind のフィールドを宣言するすべてのコアブロックは、オーバーライドするか、carry-only として明示的に許可リストに載っていなければなりません。

モデル重みを読み込むブロックは モデルマネージャーの外で （ run() 実際に何をするかを確認し、何が model_manager 初期化パラメータに現れるかは確認しない）場合、当面このメソッドを実装すべきではありません。依存関係は未宣言のままです（None).

カスタム Python ブロックは常に Noneを返します。コードは静的解析に対して不透明なため、 不明 が唯一の正直な答えです。