> For the complete documentation index, see [llms.txt](https://docs.roboflow.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.roboflow.com/workflows/ja/gaido/developer-guide/create-a-block.md). # Workflow ブロックを作成 {% hint style="info" %} これは Python で Block クラスを作成するための詳細ガイドです。1 つの Workflow 内で少しだけカスタムロジックが必要なだけなら、 [動的 Python ブロック](/workflows/ja/gaido/developer-guide/dynamic-python-blocks.md) を Workflow Definition 内にインラインで定義するだけで通常は十分です。参照: [Custom Blocks](/workflows/ja/burokku/blocks/custom-blocks.md). {% endhint %} Workflows のブロック開発には、Workflow エコシステムの理解が必要です。詳細に入る前に、必要な知識をまとめましょう。 理解 [Workflow 実行](/workflows/ja/gaido/developer-guide/workflow-execution.md)、特に: * Workflow Definition における Workflow ブロックと step の関係 * Workflow ブロックとその manifest がどのように使用されるか [Workflows Compiler](/workflows/ja/gaido/developer-guide/compiler.md) * 何が `次元レベル` で、Workflow を流れるバッチ指向データの * どのように [Execution Engine](/workflows/ja/gaido/developer-guide/execution-engine.md) が step と、その入力および出力についてどのようにやり取りするか * の性質と役割 [Workflow `種類`](/workflows/ja/gaido/developer-guide/kinds.md) * どのように理解するか [`pydantic`](https://docs.pydantic.dev/latest/) が動作するか ## 環境設定 すぐにわかるように、Workflow ブロックの作成は、特定のインターフェースを実装する Python クラスを定義するだけです。この設計により、他の Python コードと同様に Python インタープリターでそのブロックを実行できます。ただし、Workflow 実行中に他のブロックから通常提供される、必要な入力をすべて組み立てる際に困難が生じることがあります。そのため、スムーズに作業できるよう、開発環境を適切に整えることが重要です。標準的な開発プロセスの一環として、次の手順に従うことを推奨します（初回以降の貢献では最初の手順は省略可能です）。 1. **次を `conda` 環境** を設定し、次の主要依存関係をインストールします: `inference`。詳細は以下を参照: [`inference` contributor guide](https://github.com/roboflow/inference/blob/main/CONTRIBUTING.md). 2. **Workflows のコードベースの構成に慣れておきましょう。** 3. **最小限のブロックを作成する** - これについては次のセクションで説明します。まずはシンプルな block manifest と基本ロジックを実装して、ブロックが期待どおりに動作することを確認しましょう。 4. **ブロックをプラグインに追加する** - ブロックを作成したら、プラグインからエクスポートされるブロック一覧に追加します。Roboflow Core プラグインに追加する場合は、次のファイルにブロックのエントリを追加してください: [loader.py](https://github.com/roboflow/inference/blob/main/inference/core/workflows/core_steps/loader.py). **この手順を忘れると、ブロックは表示されません！** 5. **ブロックを反復的に改善する** - 結果に満足できるまで、ブロックの開発と実行を続けてください。以下のセクションでは、さまざまなシナリオでブロックを反復改善する方法を説明します。 ### Workflows UI を使ってブロックを実行する マウントしたボリュームを使って inference サーバーを実行することを推奨します（毎回再ビルドするよりはるかに高速です `inference` 。変更のたびにサーバーを再ビルドする必要がありません）: ```bash inference_repo$ docker run -p 9001:9001 \ -v ./inference:/app/inference \ roboflow/roboflow-inference-server-cpu:latest ``` そして、ローカルサーバーを Roboflow UI に接続します:
Connecting a local Inference Server to the Roboflow Workflows UI
プレビューをすばやく実行するには:
Workflow preview in the Roboflow UI
私のブロックには追加の依存関係が必要です。事前構築済みの `inference` サーバーは使えません ブロックが追加の依存関係を必要とすることは自然です。依存関係を追加するには、次のいずれかの [requirements ファイル](https://github.com/roboflow/inference/tree/main/requirements)で、該当する Docker イメージにインストールされます（通常は [CPU ビルド](https://github.com/roboflow/inference/blob/main/docker/dockerfiles/Dockerfile.onnx.cpu) の `inference` サーバー）です。 その後、次を実行します: ```bash inference_repo$ docker build \ -t roboflow/roboflow-inference-server-cpu:test \ -f docker/dockerfiles/Dockerfile.onnx.cpu . ``` その後、先ほど作成したテストタグを指定してローカルビルドを実行できます: ```bash 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 を実行する 以下のコードスニペットは、次にリクエストを送信する方法を示しています: `inference` Workflow を実行するためのサーバーです。次の `inference_sdk` は次の `inference` パッケージに含まれており、サーバー用の軽量クライアントライブラリとして使えます。 ```python from inference_sdk import InferenceHTTPClient, InferenceConfiguration YOUR_WORKFLOW_DEFINITION = ... client = InferenceHTTPClient( api_url=object_detection_service_url, api_key="XXX", # 省略可。Workflow が Roboflow Platform を使用する場合のみ必要 ).configure(InferenceConfiguration(api_key_transport="header")) 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 をサンプルデータを使って実行します（ [fixtures](https://github.com/roboflow/inference/blob/main/tests/workflows/integration_tests/execution/conftest.py) に、通常使用するサンプルデータがあります）。 4. **期待する結果をアサートする:** 結果が期待どおりであることを確認します。 テストを開発フローに組み込むことで、ブロックが長期的に安定し、既存のブロックと効果的に連携することを保証でき、作業の表現力が高まります！ 次のコマンドでテストを実行できます: ```bash pytest tests/workflows/integration_tests/execution/test_workflows_with_my_custom_block ``` 他のテストを例として参照してもよいですし、次のテンプレートを使っても構いません:
統合テストのテンプレート ```python def test_detection_plus_classification_workflow_when_XXX( model_manager: ModelManager, dogs_image: np.ndarray, roboflow_api_key: str, ) -> None: # 前提 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=, init_parameters=workflow_init_parameters, max_concurrent_steps=WORKFLOWS_MAX_CONCURRENT_STEPS, ) # 実行 result = execution_engine.run( runtime_parameters={ "image": dogs_image, } ) # 検証 assert isinstance(result, 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。これは通常、model block に必要です。この fixture は次を提供します: `ModelManager` の抽象化 `inference`。モデルの読み込みと解放に使用されます。 * 行 `3` は、2 匹の犬の画像を含む fixture を定義しています（他の fixture を参照すると、さらに多くのサンプル画像が見つかります）。 * 行 `4` はオプションの fixture で、テスト対象の workflow 内のいずれかのブロックが Roboflow API キーを必要とする場合に使用できます。その場合は、次を export してください: `ROBOFLOW_API_KEY` 有効なキーを持つ環境変数を、テストを実行する前に設定してください。 * 次の行では `7-11` Workflow 定義に基づいて、Execution Engine が実行時に生成するブロックの初期化パラメータのセットアップを提供しています。 * 次の行では `19-23` 入力パラメータを注入して Workflow を実行する方法を示しています。runtime\_parameters のキーが、Workflow 定義で宣言された入力と一致していることを確認してください。 * 次の行から `26`、テスト内にサンプルのアサーションがあります。
## プロトタイプ Workflow ブロックを作成するには、Workflows ライブラリのコアからいくつかのインポートが必要です。ブロック作成時に役立つインポート一覧は次のとおりです: ```python from inference.core.workflows.execution_engine.entities.base import ( Batch, # データのバッチは Batch[X] コンテナで渡されます OutputDefinition, # manifest で出力を宣言するために使用するクラス WorkflowImageData, # 画像の内部表現 # - 入力 kind が image の場合に使用 ) from inference.core.workflows.prototypes.block import ( BlockResult, # `run(...)` メソッドの結果の型エイリアス WorkflowBlock, # ブロックの基底クラス WorkflowBlockManifest, # block manifest の基底クラス ) from inference.core.workflows.execution_engine.entities.types import * # コアライブラリの `kinds` を含むモジュール ``` 特に重要なのは次のとおりです: * `WorkflowBlock` - ブロックの基底クラス * `WorkflowBlockManifest` - block manifest の基底クラス {% hint style="warning" %} **内部データ表現の理解** 次のものをインポートすることを推奨していることにお気づきかもしれません: `Batch` と `WorkflowImageData` これらのクラスは、私たちのシステムでビルディングブロックを構築する際に使われる基本コンポーネントです。これらのクラスが全体アーキテクチャの中でどのように位置づけられるかをより深く理解するには、次のページを参照してください: [Data Representations](/workflows/ja/gaido/developer-guide/data-representations.md) に、より詳しい情報があります。 {% endhint %} ## Block manifest manifest は Workflow ブロックの重要なコンポーネントであり、ブロックを使用するために Workflow 定義内に配置できる step 宣言のプロトタイプを定義します。特に、次のことを行います: * **使用します `pydantic` Workflows 定義の構文解析を支えます:** これは次を継承します: [`pydantic BaseModel`](https://docs.pydantic.dev/latest/api/base_model/) の機能を使って Workflow 定義を解析・検証します。このスキーマは、次のおかげで Workflows UI と互換性のある形式へ自動的にエクスポートすることもできます: `pydantic の` OpenAPI 標準との統合により実現されています。 * **データバインディングを定義します:** 実行中に workflow を流れるデータの selector となる manifest のフィールドと、その kind を指定します。 * **ブロックの出力を記述します:** ブロックが生成する出力を概説します。 * **次元を指定します:** 入力と出力の次元に関連するプロパティを詳しく説明します。 * **バッチ入力と空の値を示します:** step がバッチ入力と空の値を受け入れるかどうかを Execution Engine に知らせます。 * **互換性を確保します:** 安定性を維持するために、異なる Execution Engine バージョンとの互換性を規定します。詳しくは次を参照してください: [versioning](/workflows/ja/gaido/developer-guide/versioning.md). ### manifest のスキャフォールディング manifest の仕組みを理解するために、ステップごとに定義してみましょう。ここで作成するサンプルブロックは画像の類似度を計算するものです。まずはインポートとクラスのスキャフォールディングから始めます: ```python 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` - 動的なブロック群に基づく Workflows 定義の構文解析に必要です。これが [`pydantic` type discriminator](https://docs.pydantic.dev/latest/concepts/unions/#discriminated-unions) Workflow 定義内の特定の step を解析する際に、どの block manifest を検証すべきかを Compiler が判断できるようにします * `name` - このプロパティは step に一意の名前を付け、他の step が selector 経由でそれを選択できるようにするために使用されます ### 入力を追加する この step では、比較するための画像入力を 2 つ受け取りたいとします。
入力を追加する それらの入力定義を manifest に追加する方法を見てみましょう: ```python 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` 以外のすべてのプロパティは、 # ハードコードされたパラメータかデータ selector のいずれかとして扱われます。データ selector は文字列で、 # `$steps.` または `$inputs.` で始まり、データ参照を示します。 # これは実行時に利用可能なデータです。この場合、通常はデータの kind を指定して、 # コンパイラに期待するデータの形を知らせます。 image_1: Selector(kind=[IMAGE_KIND]) = Field( description="類似度を計算する 1 枚目の画像", ) image_2: Selector(kind=[IMAGE_KIND]) = Field( description="類似度を計算する 2 枚目の画像", ) ``` * 次の行では `2-9`、必要なものがすべて揃うように、いくつかのインポートを追加しました * 行 `20` 定義します `image_1` パラメータ。manifest は Workflow Definition のプロトタイプであるため、step で使用する画像を伝える唯一の方法は selector を提供することです。コアライブラリには、それに使える専用の型があります: `Selector`。コードベースを詳しく見ると、これは型エイリアスのコンストラクタ関数であり、 `pydantic` 次の文字列に一致するものを期待するように指示しています: `$inputs.{name}` と `$steps.{name}.*` というパターンをそれぞれ期待します。さらに、追加のスキーマフィールドメタデータを提供し、Workflows エコシステムの各コンポーネントに対して、 `kind` selector の背後にあるデータの種類が [image](/workflows/ja/gaido/developer-guide/kinds/image.md). **重要な注意:** 次のように表します *kind* をリストとして - 特定の kind のリストは次のように解釈されます: *kind の union* Execution Engine によって。 * を表し `pydantic` `Field(...)` 行末の属性 `20` は任意ですが、特に Workflows UI と連携することを想定したブロックでは指定すると望ましいです * 次の行で `23`、次の定義があります: `image_2` パラメータ。これは次に非常によく似ています: `image_1`.
このような manifest 定義により、Workflow 定義内の次の step 宣言を扱えます: ```json { "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 ブロックから step を初期化する `my_plugin/images_similarity@v1` * step の run メソッドに 2 つのパラメータを渡す: * `input_1` の型 `WorkflowImageData` Workflow 実行入力として送信された画像が格納され、名前は次のとおりです: `my_image`. * `input_2` の型 `WorkflowImageData` これは実行時に、次の別の step によって生成されます: `image_transformation` ### manifest にパラメータを追加する では、step の実行に影響するパラメータを追加しましょう。
manifest にパラメータを追加する ```python 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` 以外のすべてのプロパティは、 # ハードコードされたパラメータかデータ selector のいずれかとして扱われます。データ selector は文字列で、 # `$steps.` または `$inputs.` で始まり、データ参照を示します。 # これは実行時に利用可能なデータです。この場合、通常はデータの kind を指定して、 # コンパイラに期待するデータの形を知らせます。 image_1: Selector(kind=[IMAGE_KIND]) = Field( description="類似度を計算する 1 枚目の画像", ) 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` インポート [`float_zero_to_one`](/workflows/ja/gaido/developer-guide/kinds/float-zero-to-one.md) `kind` 定義。この定義はパラメータの定義に使用されます。 * 次の行で `27` 次のパラメータの定義を開始します: `similarity_threshold`。manifest は、float 値または次の Workflow 入力への selector のいずれかを受け入れます: `kind` [`float_zero_to_one`](/workflows/ja/gaido/developer-guide/kinds/float-zero-to-one.md)。次の行でインポートされた `9`.
このような manifest 定義により、Workflow 定義内の次の step 宣言を扱えます: ```json { "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" } ``` あるいは: ```json { "type": "my_plugin/images_similarity@v1", "name": "my_step", "image_1": "$inputs.my_image", "image_2": "$steps.image_transformation.image", "similarity_threshold": 0.5 } ``` ### ブロック出力を宣言する ブロックの入力はうまく定義できましたが、ブロックを正常に実行するために必要な要素がまだいくつか足りません。ブロックの出力を定義しましょう。
ブロック出力を宣言する 必要最小限の情報は出力の説明です。さらに、ブロックの安定性を高めるため、実行エンジンとの互換性に関する情報も提供することを推奨します。 ```python 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="類似度を計算する 1 枚目の画像", ) 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` インポート [`ブール値`](/workflows/ja/gaido/developer-guide/kinds/boolean.md) `kind` 出力定義で使用される * 行 `32-39` ブロックからの出力を指定するクラスメソッドを宣言します - リスト内の各エントリは、各バッチ要素に対する1つの戻りプロパティとその `kind`。私たちのブロックはブールフラグを返します `images_match` 各画像のペアごとに。 * 行 `41-43` Execution Engine とのブロックの互換性を宣言します - を参照してください [バージョニングページ](/workflows/ja/gaido/developer-guide/versioning.md) 詳細については
これらの変更の結果として: * Execution Engine は、このブロックに基づいて作成されたステップが指定された出力を提供することを理解し、他のステップは入力でそれらの出力を参照できるようになります * Execution Engine のバージョンが `v1`
詳細はこちら: 動的出力 一部のブロックは、パース後に利用可能な step manifest の内容に関係なく、classmethod を使って出力を任意に定義できない場合があります。これをサポートするために、次の規約を導入しました: * classmethod `describe_outputs(...)` という名前の1つの要素を持つリストを返す必要があります `*` および kind を `*` (別名 `WILDCARD_KIND`) * さらに、block manifest はインスタンスメソッド `get_actual_outputs(...)` を実装する必要があり、filled manifest データに基づいて生成可能な実際の出力のリストを提供します ```python 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="類似度を計算する 1 枚目の画像", ) 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 ] ```
## ブロッククラスの定義 この段階で、シンプルなブロックの manifest は準備できました。例を続けていきます。今は気が散るだけなので、 [高度なトピック](#advanced-topics) セクションを参照すると、さらに詳細を確認できます。 ### 基本実装 manifest の準備ができたので、ブロックのベース実装を準備できます。
ブロックの骨組み ```python 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="類似度を計算する 1 枚目の画像", ) 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(...)` 以前作成した manifest クラスを単純に返すように * 行 `57-63` 定義する `run(...)` 関数。Execution Engine はこれをデータとともに呼び出して、目的の結果を取得します。なお、入力を定義する manifest フィールドの [image](/workflows/ja/gaido/developer-guide/kinds/image.md) kind は `WorkflowImageData` としてマークされています - これは `image` で説明されている kind の内部データ表現と整合しています [kind ドキュメント](/workflows/ja/gaido/developer-guide/kinds/image.md).
### ブロックロジックの実装を提供する それでは、ブロックに `run(...)` メソッドの実装例を追加して、意味のある結果を生成できるようにしましょう。 {% hint style="info" %} このセクションの内容は、ブロック作成者として Workflow エコシステムとどう連携するかの例を示すことを目的としており、堅牢なブロック実装を提供するものではありません。 {% endhint %}
`run(...)` メソッドの実装 ```python 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="類似度を計算する 1 枚目の画像", ) 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` ブロック機能の実装を提供します。詳細自体は Workflow エコシステムにとって本質的に重要ではありませんが、注目すべき点がいくつかあります: * 行 `69` と `70` を利用します `WorkflowImageData` 抽象化を利用して、 `numpy_image` プロパティを使って `np.ndarray` Workflows における画像の内部表現から取得できることを示しています。残りの `WorkflowImageData` プロパティも調べることをお勧めします。 * ワークフローブロック実行の結果。行で宣言されています `78-80` は、この場合は単なる辞書です **キーは manifest で宣言された出力名です**、行 `43`。宣言したすべての出力を必ず提供してください - そうしないと Execution Engine がエラーを発生させます。
## ブロックを `プラグインに公開する` これでブロックは使用可能になりましたが、Execution Engine はその存在を認識していません。これは、登録済みのプラグインがあなたが作成したブロックをエクスポートしていないためです。ブロックのバンドル方法の詳細は [別ページ](/workflows/ja/gaido/developer-guide/block-bundling.md)で扱いますが、残っている作業は、プラグインの \`load\_blocks(...)\` `関数から返されるリストにブロッククラスを追加することです:` 関数: ```python # あなたのプラグインの __init__.py (または `inference` に直接貢献する場合は roboflow_core プラグイン) from my_plugin.images_similarity.v1 import ImagesSimilarityBlock # これは import の例です! 調整が必要です def load_blocks(): return [ImagesSimilarityBlock] ``` ## 高度なトピック ### 入力バッチを処理するブロック 場合によっては、すべての入力データをバッチとして一度に処理したほうが、ブロックの性能が向上することがあります。これは GPU 上で動作するモデルで起こりえます。このような動作モードは Workflows のブロックでサポートされています - ここでは、あなたのブロックでそれを使う方法の例を示します。
バッチを受け取るブロックの実装 ```python 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="類似度を計算する 1 枚目の画像", ) 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` インポート `Batch` Workflows ライブラリの core から - このクラスは、バッチ要素を保持するための list に非常によく似た（ただし読み取り専用の）コンテナを表します * 行 `40-42` ブロックのデフォルト動作を変更し、バッチ処理を可能にするクラスメソッドを定義します - このメソッドが `run(...)` メソッド **として認識する各パラメータをバッチ指向としてマークしています**. * 上記の変更により、 `run(...)` メソッドのシグネチャが変更され、今では `image_1` と `image_2` は `WorkflowImageData`のインスタンスではなく、この型の要素のバッチになります。 **重要な注意:** 複数のバッチ指向パラメータがある場合、それらのバッチは対応する位置にある要素どうしが関連していると想定します - つまり、私たちのブロックが `image_1[1]` を `image_2[1]` と比較することが、実際に論理的に意味のある操作になるようにします。 * 行 `74-77`, `85-86` 必要に応じてバッチ要素を反復処理する方法を示しながら、すべてのバッチ要素に対して run 処理を実施するために導入が必要だった変更を示します * 出力が行でどのように構築されるかに注意することが重要です `85` - バッチの各要素には、 `run(...)` メソッドから返されるリスト内でそれぞれのエントリが与えられます。順序はバッチ要素の順序と一致していなければなりません。各出力辞書は、ブロック出力で宣言されたすべてのキーを提供する必要があります。
バッチとスカラーの両方を受け付ける入力 これは **比較的ありそうにないことですが**、1つの入力パラメータでバッチ指向データとスカラーの両方を受け取る必要があるブロックがあるかもしれません。Execution Engine は `get_parameters_accepting_batches_and_scalars(...)` メソッドを使ってそれを認識します。以下の例を見てください: ```python 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 = ... # スカラーパラメータに対して何かを行う param_2 の各要素について: if isinstance(element, Batch): ... else: ... param_3 の各 key, value について: 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` 混合（スカラーとバッチ指向の両方）の入力データを受け取ることが期待される manifest パラメータを指定します - この段階では、定義において前の例との違いはないことに注意してください。 * 行 `24-26` 指定する `get_parameters_accepting_batches_and_scalars(...)` メソッドにより、ブロック `run(...)` メソッドが指定されたパラメータについてスカラー入力とバッチ指向入力の両方を扱えることを Execution Engine に伝えます。 * 行 `45-47` 混合的な性質を持つパラメータを `run(...)` メソッドシグネチャで示します。 * 行 `49` 期待される出力サイズを追跡する必要があることを示しています **ブロックロジック内で**。そのため、混合入力を持つブロックの実装はかなり厄介です。通常、ブロックの `run(...)` メソッドがスカラーを扱う場合 - 大半のケースでは（例外は以下で説明します）- メソッドは単一の出力辞書を構築します。同様に、バッチ指向入力が受け付けられる場合、それらの入力が期待される出力サイズを定義します。しかしこの場合は、バッチを手動で検出し、そのサイズを把握する必要があります。 * 行 `50-54` バッチ指向データが検出されたときに異なるロジックを適用する、混合パラメータの通常の扱い方を示します * 前述のとおり、出力構築も混合入力の性質に合わせて調整する必要があります - これは行で示されています `65-70`
### フロー制御ブロックの実装 フロー制御ブロックは、単にデータを処理する他のブロックとはかなり異なります。ここではフロー制御ブロックの作成方法を示しますが、その前に少し理論を: * フロー制御ブロックとは、manifest でステップセレクタとの互換性を宣言するブロックです（ステップへの selector は `$steps.{step_name}` により定義されます - ステップ出力 selector と似ていますが、出力名の指定がありません） * フロー制御ブロックは出力を登録できません。返すことを意図しており `FlowControl` オブジェクト * `FlowControl` オブジェクトは、指定されたバッチ要素（SIMD フロー制御）またはワークフロー全体の実行（非 SIMD フロー制御）において、どの次のステップを次に選択するべきかを、（step manifest で提供された selector から）指定します
フロー制御の実装 ランダム継続ブロックの実装を示し、コメントアウトしています ```python 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` インポート `FlowControl` フロー制御ブロックからの唯一の有効な応答であるクラス * 行 `28` ステップセレクタのリストを定義します **これにより、ブロックは実質的にフロー制御ブロックになります** * 行 `55` と `56` 出力の構築方法を示します - `FlowControl` オブジェクトはコンテキストとして `None`, `文字列` または `文字列のリスト` - `None` バッチ要素のフロー終了を表します。文字列は次のステップの selector であることが期待され、入力として渡されます。
フロー制御の実装 - バッチ版 ランダム継続ブロックの実装を示し、コメントアウトしています ```python 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` インポート `FlowControl` フロー制御ブロックからの唯一の有効な応答であるクラス * 行 `29-32` ステップセレクタのリストを定義します **これにより、ブロックは実質的にフロー制御ブロックになります** * 行 `38-40` を含みます `get_parameters_accepting_batches(...)` ブロックが `run(...)` メソッドがバッチ指向の `image` パラメータを期待していることを Execution Engine に伝えます。 * 行 `59` これは、バッチの各要素ごとにフロー制御ガイドを返す必要があることを示しています。 `image` バッチ。 * そのために、行では `60` バッチの内容を反復処理します。 * 行 `61-63` 出力の構築方法を示します - `FlowControl` オブジェクトはコンテキストとして `None`, `文字列` または `文字列のリスト` - `None` バッチ要素のフロー終了を表します。文字列は次のステップの selector であることが期待され、入力として渡されます。
### ネストされた selector 一部のブロックでは、ブロック manifest フィールドに selector のリストまたは selector の辞書を提供する必要があります。Execution Engine のバージョン `v1` は1段階のネストのみをサポートしているため、selector のリストのリストや selector のリストを持つ辞書は正しく認識されません。 ネストされた selector の使用例を示す実用的なユースケースを以下に示します。 #### 可変数のモデルからの予測の融合 複数の分類器の予測に対して多数決を行うブロックを作成したいとしましょう - その場合、run メソッドは次のようになるでしょう: ```python # 擬似コード এখানে 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]} ```
ネストされた selector - モデルアンサンブル ```python 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="ステップ出力への selector", 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` selector のリストを受け取れる manifest フィールドの定義方法を示します * 行 `50` ブロックの `run(...)` メソッドへの入力として何を期待すべきかを示します - 特定の kind を表すオブジェクトのリストです。ブロックがバッチを受け付ける場合、 `predictions` フィールドの入力型は `List[Batch[sv.Detections]`
このようなブロックは次のステップ宣言と互換性があります: ```json { "type": "my_plugin/fusion_of_predictions@v1", "name": "my_step", "predictions": [ "$steps.model_1.predictions", "$steps.model_2.predictions" ] } ``` #### 動的パラメータを可能にするデータ変換を持つブロック 場合によっては、ブロックが Workflow 定義の作成者によって定義される名前と値を持つ「名前付き」selector のグループを受け取る必要があるかもしれません。そのような場合、ブロック manifest は selector の辞書を受け入れる必要があり、キーはそれらの selector の名前として機能します。
ネストされた selector - 名前付き selector ```python 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="ステップ出力への selector", 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` セレクタの辞書を受け取れる manifest フィールドの定義方法を示します。これはセレクタ名と値の対応付けを提供します * 行 `46` ブロックの `run(...)` メソッド - セレクタで参照されるオブジェクトの辞書。ブロックがバッチを受け付ける場合、入力型は `data` フィールドの入力型は `Dict[str, Union[Batch[Any], Any]]`。バッチなしの場合、セレクタで参照される非バッチ向けデータは自動的にブロードキャストされますが、バッチを受け付けるブロックでは `Batch` コンテナはバッチ指向の入力のみを包み、その他の入力は単一値として渡されます。
このようなブロックは次のステップ宣言と互換性があります: ```json { "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` if `model_1` が物体検出モデルである場合 * 下で `data["b"]` 内部で `run(...)`には、という名前の入力パラメータの値が見つかります `my_parameter` ### 入力および出力の次元数 vs `run(...)` メソッド ブロック入力の次元数は、形成において重要な役割を果たします `run(...)` メソッドのシグネチャに影響するため、システムは入力間の次元レベル差に厳格な上限を課しています（許容される最大差は `1`）。この制約は、ブロック記述時の一貫性と予測可能性を確保するうえで重要です。 次元差が制御されていなければ、メソッドの構造を予測するのは困難でしょう `run(...)` メソッドの構造を予測するのが難しくなり、開発はより困難で信頼性も低下します。そのため、このプロパティの検証は Workflow のコンパイル処理中に厳格に行われます。 同様に、出力の次元数もメソッドのシグネチャと期待される出力形式に影響します。エコシステムは次のシナリオをサポートしています: * すべての入力が **同じ次元数を持ち** 、出力も **変化しない** 次元数 - 基本ケース * すべての入力が **同じ次元数を持ち** 、出力が **減少する** 次元数 * すべての入力が **同じ次元数を持ち** 、出力が **増加する** 次元数 * 入力が **異なる次元数を持ち** 、出力は次の次元数を維持することが許可される **参照入力** 一貫性を確保し、メソッドシグネチャの曖昧さを防ぐため、他の入力/出力次元数の組み合わせは許可されません。
`run(...)` メソッドへの次元数の影響 - バッチ無効 **出力の次元数増加** この例では、予測結果に基づいて画像の動的クロップを行います。 ```python 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()}") if cropped_image.size: result = WorkflowImageData( parent_metadata=parent_metadata, numpy_image=cropped_image, ) else: result = None crops.append({"crops": result}) return crops ``` * 行で `28-30` manifest クラスは出力の次元数オフセットを宣言しています。値 `1` は追加するものと理解してください `1` 次元レベルに * なお、行では `63`、ブロックは空の画像を以降の処理から除外しますが、 `None` 出力を含む辞書の代わりに配置します。これにより、条件付き実行で使われるのと同じ Execution Engine の挙動を利用できます。つまり、そのデータポイントは下流の処理から除外されます（ただし、途中で空入力を要求するステップがあれば別です）。 * 行で `64-65` 単一入力に対する結果 `image` と `predictions` が収集されます。これは、登録済みのすべての出力をキーとして含む辞書のリストを意味します。Execution Engine は、このステップが各入力要素に対して要素のバッチを返すと理解し、下流ステップの実行中に追跡するための入れ子構造のインデックスを作成します。 **出力の次元数減少** この例では、ブロックはクロップ予測を可視化し、すべてのクロップ予測を 1 枚の出力画像にまとめたタイルを作成します。 ```python 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` manifest クラスは出力の次元数オフセットを宣言しています。値 `-1` は次元レベルを減少させるものと理解してください `1` * 行で `34-36` manifest クラスは `run(...)` シグネチャが常に安定するよう自動バッチキャストの対象となるメソッド入力を宣言しています。自動バッチキャストは Execution Engine に導入されました `v0.1.6.0` * 参照してください [変更履歴](/workflows/ja/gaido/developer-guide/execution-engine-changelog.md) で詳細を確認できます。 * 行で `53-55` 出力次元数の減少がメソッドシグネチャに与える影響を確認できます。最初の 2 つの入力（行で宣言されている `36`）は、人工的に `Batch[]` コンテナで包まれます。一方で `scalar_parameter` はプリミティブ型のままです。これは、すべての入力が同じ次元数を持つ場合に、最終次元レベルに属するすべての要素へアクセスできるようにするため、出力次元数の減少時に Execution Engine によって自動的に行われます。もちろん、最上位バッチの同じ要素に関連する要素だけがグループ化されます。たとえば、2 枚の入力画像をクロップした場合、それぞれ別の画像から得られたクロップは別々にグループ化されます。 * 行 `65-66` 出力がどのように構築されるかを示します。単一の値が返され、その値は次元数が減少した出力バッチ内で Execution Engine によってインデックス付けされます **異なる入力次元数** この例では、ブロックは元画像のクロップに基づいて予測された検出結果をマージします。結果として、個々の部分結果がすべて統合された単一の検出結果を提供します。 ```python 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` manifest クラスは入力の次元数オフセットを宣言しており、これは `image` パラメータが最上位であり `image_predictions` が入れ子の予測バッチであることを示しています * 異なる入力次元数が宣言される場合、次元参照プロパティを指定しなければなりません（行を参照 `38-40`）- この次元レベルが出力次元数の計算に使われます - この特定のケースでは、 `image`を指定しています。この選択は、期待される結果形式に影響します。選択したシナリオでは、登録済みのすべての出力キーを含む単一の辞書を返すことになります。もし選択が `image_predictions`であれば、辞書のリスト（ `image_predictions` バッチの長さに等しいサイズ）を返します。言い換えれば、 `get_dimensionality_reference_property(...)` 出力にどの次元レベルを対応付けるかを示します。 * 行 `63-64` 行で指定された次元数オフセットの影響を示します `31-36`。明らかに、 `image_predictions` はに関する入れ子バッチです `image`。もちろん、特定の `画像` に関連する入れ子の予測のみがバッチとしてまとめられ、実行時にメソッドへ渡されます。 * 前述のとおり、行では `69` 出力を単一の辞書として構築します。これは、出力を次元レベル `image` （これは単一要素としても渡されていました）
`run(...)` メソッドへの次元数の影響 - バッチ有効 **出力の次元数増加** この例では、予測結果に基づいて画像の動的クロップを行います。 ```python 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()}") if cropped_image.size: result = WorkflowImageData( parent_metadata=parent_metadata, numpy_image=cropped_image, ) else: result = None crops.append({"crops": result}) results.append(crops) return results ``` * 行で `29-31` manifest は、ブロックが入力のバッチを受け付けることを宣言しています * 行で `33-35` manifest クラスは出力の次元数オフセットを宣言しています。値 `1` は追加するものと理解してください `1` 次元レベルに * 行で `55-66`。入力パラメータのシグネチャは、 `run(...)` メソッドが同じ次元数の入力に対して実行され、それらの入力がバッチとして提供されることを反映しています * なお、行では `70`、ブロックは空の画像を以降の処理から除外しますが、 `None` 出力を含む辞書の代わりに配置します。これにより、条件付き実行で使われるのと同じ Execution Engine の挙動を利用できます。つまり、そのデータポイントは下流の処理から除外されます（ただし、途中で空入力を要求するステップがあれば別です）。 * 出力の構築は、行で示されている `71-73` 2 層のネストを示しています。まず、ブロックはバッチで動作するため、入力バッチ要素ごとに 1 つの出力を持つ出力リストを返すことが期待されます。さらに、各入力バッチ要素に対するこの出力要素は入れ子バッチであることがわかります。そのため、各入力画像と予測に対して、ブロックは出力のリストを生成し、そのリストの要素は各宣言済み出力の値を提供する辞書です。 **出力の次元数減少** この例では、ブロックはクロップ予測を可視化し、すべてのクロップ予測を 1 枚の出力画像にまとめたタイルを作成します。 ```python 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` ブロックが入力としてバッチを受け取ることが期待される manifest * 行で `33-35` manifest クラスは出力の次元数オフセットを宣言しています。値 `-1` は次元レベルを減少させるものと理解してください `1` * 行で `52-53` 出力次元数の減少とバッチ処理がメソッドシグネチャに与える影響を確認できます。最初の「層」は `Batch[]` 、manifest がブロックに入力バッチの受け付けを宣言していることの副作用です。2 番目の「層」は出力次元数の減少によるものです。Execution Engine は、減らす対象の次元を追加の `Batch[]` コンテナで入力内に包み込み、プログラマが特定の最上位バッチ要素に属するすべての入れ子バッチ要素を収集できるようにします。 * 行 `66-67` 出力がどのように構築されるかを示します。最上位バッチ要素ごとに、ブロックはすべてのクロップと予測を集約して 1 つのタイルを作成します。ブロックは入力のバッチを受け付けるため、この処理の結果、最上位バッチ要素ごとに 1 つのタイルが生成されます。したがって、辞書のリストを返すことが期待されます。 **異なる入力次元数** この例では、ブロックは元画像のクロップに基づいて予測された検出結果をマージします。結果として、個々の部分結果がすべて統合された単一の検出結果を提供します。 ```python 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` ブロックが入力としてバッチを受け取ることが期待される manifest * 行で `35-40` manifest クラスは入力の次元数オフセットを宣言しており、これは `image` パラメータが最上位であり `image_predictions` が入れ子の予測バッチであることを示しています * 異なる入力次元数が宣言される場合、次元参照プロパティを指定しなければなりません（行を参照 `42-44`）- この次元レベルが出力次元数の計算に使われます - この特定のケースでは、 `image`。この選択は期待される結果形式に影響します。選択したシナリオでは、 `image` バッチの各要素に対して単一の辞書を返すことになります。もし選択が `image_predictions`であれば、辞書のリスト（入れ子の `image_predictions` バッチの長さに等しいサイズ）を各入力 `image` バッチ要素ごとに返します。 * 行 `66-67` 行で指定された次元数オフセットの影響を示します `35-40` および、行でのバッチ処理の宣言も `32-34`。最初の「層」は `Batch[]` コンテナは後者の入れ子の `Batch[Batch[]]` によるものです `images_predictions` は入力次元数オフセットの定義に由来します。明らかに、 `image_predictions` は、特定の要素に関連する予測のバッチを保持しています `image` バッチ。 * 前述のとおり、行では `76-77` 各要素ごとに単一の辞書として出力を構築します `image` バッチ
### 空の入力を受け付けるブロック 前述のとおり、Workflow の実行中に一部のバッチ要素が「空」になることがあります。これはいくつかの要因で発生します: * **フロー制御の仕組み:** 実行の特定の分岐が特定のバッチ要素を隠し、後続ステップで処理されないようにすることがあります。 * **データ処理ブロックでは:** 場合によっては、ブロックが特定のデータポイントに対して意味のある出力を生成できないことがあります。たとえば、Dynamic Crop ブロックはバウンディングボックスのサイズが 0 の場合、クロップ画像を生成できません。 一部のブロックは、欠落した出力をデフォルト値で置き換えられるブロックのように、こうした空の入力を処理するよう設計されています。このブロックは、Workflow で構造化された出力を構築する際に特に有用で、いくつかの要素が空でも、欠落要素のない出力を確実に作れるため、解析しにくくなるのを防げます。
空の入力を受け付けるブロック ```python 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 を紹介するページ](/workflows/ja/gaido/developer-guide/compiler.md)に説明されています。これは、ブロッククラスからステップを動的に構築する責任を持つコンポーネントだからです。 コンストラクタ引数は次の条件を満たす必要があります: * ブロックによって要求される - クラスメソッドを使用して `WorkflowBlock.get_init_parameters(...)` * Workflows Execution Engine を実行する環境で提供される: * 直接、 [この](/workflows/ja/depuroi/deploy-a-workflow.md#workflows-in-python-package) 例に示すように * デフォルトを使用して [Workflow プラグインに登録されている](/workflows/ja/gaido/developer-guide/block-bundling.md) ブロック定義時に init パラメータを要求する方法を見てみましょう。
コンストラクタ引数を要求するブロック ```python 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 つのオプションのクラスメソッドを公開します。これらをオーバーライドすると、エアギャップ対応の workflow ビルダーがどのブロックをオフラインで使用できるかを判断するのに役立ちます。 3 つとも妥当なデフォルトがあるため、既存のブロックは **変更不要です**. #### `get_air_gapped_availability()` ブロックがインターネットなしで動作できるかを宣言します。OpenAI、Anthropic などのクラウド API を呼び出すブロックではこれをオーバーライドしてください。 ```python 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 のリストを返します。次の条件を満たせば、そのブロックはオフラインで利用可能と見なされます **いずれかの** バリアントにキャッシュ済みアーティファクトがあること。 ```python 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 モデルブロックでは、このブロックが扱えるタスクタイプを返します。エアギャップ対応ビルダーはこれを使って、キャッシュされたユーザーモデルを互換性のあるブロックに対応付けます。 ```python class BlockManifest(WorkflowBlockManifest): # ... @classmethod def get_compatible_task_types(cls) -> Optional[List[str]]: return ["object-detection"] ``` デフォルトでは次を返します `None` — Roboflow モデルでパラメータ化されないブロックに適しています（基盤モデル、ロジックブロック、シンクなど）。 ### 実行時制約 一部のブロックは、デプロイされる実行環境（ホスト型サーバーレス、専用デプロイ、セルフホスト、推論パイプライン）、ステップ実行モード（ローカル vs. リモート）、入力モード（画像 vs. 動画）によって、挙動が異なったり、完全に失敗したりします。オーバーライド `get_restrictions()` に `WorkflowBlockManifest` それらの注意点をブロック内に一度だけ宣言し、実行エンジン、schema エンドポイント、そして自動生成されるブロックギャラリーがそれらを一貫して表示できるようにします。 デフォルトでは次を返します `[]`、そのため既存のブロックは **変更不要です**. #### 重大度: `ソフト` 対 `ハード` 各制約には `重大度`: * **`Severity.SOFT`** - ブロックは最後まで実行され、正しい出力形状を返しますが、値が劣化していたり、意味をなさなかったりします（例: トラッカー ID がリクエスト間でリセットされる、クールダウンがスロットリングしない、ファイルが一時ディスクに書き込まれる）。ワークフローは引き続き実行されますが、結果はユーザーの期待どおりではありません。 * **`Severity.HARD`** - ブロックは実行されない / 例外を送出する / この実行環境では使用可能な出力を生成できません。エンジンはコンパイルを拒否するか、即時失敗すべきです。 #### 制約のスコープ設定 ある `RuntimeRestriction` 3 つの軸の任意の組み合わせに沿ってスコープを指定できます。ある軸を `None`のままにすると、その制約はその軸のすべての値に適用されます。 * `適用対象ランタイム`: `Runtime.HOSTED_SERVERLESS`, `Runtime.DEDICATED_DEPLOYMENT`, `Runtime.SELF_HOSTED_CPU`, `Runtime.SELF_HOSTED_GPU`, `Runtime.INFERENCE_PIPELINE`. * `適用対象ステップ実行モード`: `StepExecutionMode.LOCAL`, `StepExecutionMode.REMOTE`. * `適用対象入力モード`: `RuntimeInputMode.IMAGE`, `RuntimeInputMode.VIDEO`. その `注記` この field は、失敗モードまたは劣化した挙動を一行で人間が読めるように説明するものです。何が起きるのか（例: "track\_ids reset between requests", "writes to ephemeral /tmp"）を述べ、抽象的な前提条件は書かないでください。 #### 共有プリセット ほとんどの注意点は共通パターンのいくつかに収まるため、コードベース全体で文言を統一するために、再利用可能なプリセットが dataclass と併せてエクスポートされています。まずはこれらを使ってください: * `STATEFUL_VIDEO_HTTP_SOFT_RESTRICTION` - 各動画の状態がプロセスメモリに保存され、ステートレスな HTTP リクエスト間でリセットされる、動画トラッキング / カウント / 集計ブロック向け。 * `COOLDOWN_HTTP_SOFT_RESTRICTION` - クールダウン / レート制限タイマーがプロセスメモリに保持されるため、複数レプリカの HTTP ランタイムではスロットリングしないブロック向け。 * `STILL_IMAGE_INPUT_SOFT_RESTRICTION` - 時間的文脈（動画や繰り返しフレーム）に依存し、静止画像ではほとんど、またはまったく効果がないブロック向け。 #### 例 動画ごとの状態を保持し、静止画像でも意味をなさない行カウンターブロックは、両方の制約を宣言します: ```python 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, ] ``` カスタム制約（たとえば、 `Severity.HARD` ホスト型サーバーレス runtime に存在しない GPU ハードウェアを必要とする block）は、インラインで宣言します: ```python 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」セクション内、プロパティの直後に **プロパティ**. 3. 実行エンジン。必要に応じて `Severity.HARD` 現在の runtime に対する制約で fail-fast することを選べます。 ### 依存リソースの宣言 多くのブロックは実行時に外部リソースを必要とします。Roboflow で学習したモデルや基盤モデルの重み、Roboflow プロジェクト（例: アクティブラーニングのターゲット、データセットアップロード先）、あるいはサードパーティのホスト型モデル（OpenAI、Anthropic、OpenRouter など）です。ワークフローが実際に実行されるまで、システムはどのリソースが必要になるのか分かりません。 オーバーライド `discover_dependent_resources()` をブロックの manifest でオーバーライドすると、そのギャップが埋まります。このメソッドは **パース済み manifest インスタンス**に対して呼び出されるため、特定の step の具体的な field 値から宣言を計算できます。これにより、呼び出し側はワークフロー全体のリソースを静的に、つまりコンパイル時に何も実行せず列挙できるようになります。これにより、予測可能な実行時間のためにモデルの重みを事前読み込みしたり、API キーが参照されたすべてのモデルとプロジェクトにアクセスできることを事前に検証したりといった用途が可能になります。 ```python def discover_dependent_resources(self) -> Optional[List[DependentResource]]: ... ``` デフォルトでは次を返します `None`、これはブロックが **宣言しない** その依存関係を—呼び出し側はそれを *不明*。これは意図的に次とは区別されています: `[]`、これはブロックが外部リソースを必要としないことを明示的に宣言します。既存のブロックに変更は不要で、リソースを使うブロックはオーバーライドしてください。 #### リソースの枠組み ブロックがリソースの形を勝手に定義することはありません。枠組みは Execution Engine によって管理されます。ある `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` （モデルエンティティはプラットフォーム上で到達可能であるだけでよく、例えばメタデータを付与する監視 sink など）。実行時には、 `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()`. #### セレクタ値とランタイム解決 manifest の field には、具体値の代わりに workflow selector を含めることができます。宣言はそのような値を **そのまま** 返します—ブロックは selector を解決しません。呼び出し側は `$inputs.` 参照を、runtime パラメータが分かり次第置き換えられます; `$steps..` 参照は静的にはまったく解決できません。すべてのメタデータエンティティは `requires_runtime_resolution()` を公開しているため、呼び出し側は具体的な識別子と、まだ解決が必要な参照を区別できます。 最終的な識別子が *関数* である場合（たとえば `clip/`のような family prefix、カタログ検索など）、置換された入力値だけでは実行される id にはなりません。そのような宣言には **`model_id_resolver`** — 置換後の値を最終 id に変換する、必要なものをすべてクロージャに保持した callable です。resolver はプロセス内でのみ使う補助機能であり、シリアライズ、JSON schema、等価性からは除外されます。プロセス内の呼び出し側（例: エンジンの初回実行時プリロード）は、入力値を置換した後にこれを呼び出します。resolver は `None` 値が静的には解決不能であること（最終 id がその 1 つの値だけに依存しないこと）を宣言するために返すことがあります—そのような依存関係は呼び出し側がスキップし、実行時に解決されます。例外を送出する場合は、その値が本当に無効であることを意味します。 #### 次と一致させてください `run()` 読み込む 宣言された識別子は、まさに `run()` が要求するものと完全に同じでなければなりません—version field から生成された id も含みます。ブロックが `f"my_family/{self.version}"` の `load_core_model(...)` 呼び出し箇所で組み立てているなら、宣言も同じ id を組み立てる必要があります（そして version field に selector が入っている場合は、そのままの selector にフォールバックしてください）。 #### 例 自分のモデルと、任意のアクティブラーニング対象プロジェクトを宣言する model block: ```python 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 field からモデル id を生成するブロック: ```python 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): # selector はそのまま返され、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}")] ``` 実行せずにモデルエンティティだけを参照する sink: ```python 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, ) ] ``` #### 宣言すべきでない場合 単に *持つ* model-id や project 種別の値を、一般的なペイロードとして保持するだけの field（例: 値を query parameter として転送する webhook sink）はリソース依存ではありません—そのような block は意図的に既定値のままにします。 core リポジトリの貢献者は、unit test (`tests/workflows/unit_tests/core_steps/test_dependent_resources.py`) が境界を守っていることを認識してください。manifest が `roboflow_model_id` または `roboflow_project` 型の field を宣言している core block は、必ずオーバーライドするか `discover_dependent_resources()` そうでなければ、明示的に carry-only として allowlist に載っていなければなりません。 model の重みを読み込む block は **model manager の外部で** （何を `run()` 実際に行うかではなく `model_manager` init parameters に見えるかを確認してください）は、現時点ではこのメソッドを実装しないでください—その依存関係は未宣言のままです（`None`). カスタム Python block は常に `None`: そのコードは静的解析には不透明なので、 *不明* が唯一正直な答えです。