> 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 ブロックを作成する 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 定義における Workflow ブロックとステップの関係 * Workflow ブロックとその manifest がどのように [Workflows Compiler](/workflows/ja/gaido/developer-guide/compiler.md) * とは何か `次元レベル` Workflow を流れるバッチ指向データの * どのように [Execution Engine](/workflows/ja/gaido/developer-guide/execution-engine.md) が、入力と出力に関してステップとやり取りするか * の性質と役割は何か [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/workflows/roboflow_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 を実行する 以下のコードスニペットは、Workflow を実行するために `inference` サーバーへリクエストを送る方法を示しています。 `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 があり、通常はモデルブロックで必要になります。この fixture は、 `ModelManager` という抽象化を `inference`から提供し、モデルの読み込みとアンロードに使われます。 * 行 `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 ライブラリのコアからある程度の import が必要です。ブロック作成時に役立つ import の一覧は次のとおりです: ```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, # ブロック manifest のベースクラス ) from inference.core.workflows.execution_engine.entities.types import * # コアライブラリの `kinds` を含むモジュール ``` 最も重要なのは次のとおりです: * `WorkflowBlock` - ブロックのベースクラス * `WorkflowBlockManifest` - ブロック manifest のベースクラス {% hint style="warning" %} **内部データ表現の理解** すでにお気づきかもしれませんが、 `Batch` と `WorkflowImageData` クラスを import することを推奨しています。これらは、私たちのシステムでビルディングブロックを構築する際の基本要素です。これらのクラスが全体アーキテクチャの中でどう位置付けられるかをより深く理解するには、 [Data Representations](/workflows/ja/gaido/developer-guide/data-representations.md) ページを参照して、より詳細な情報を確認してください。 {% endhint %} ## ブロック manifest manifest は Workflow ブロックの重要なコンポーネントで、Workflow definition に配置してブロックを使うためのステップ宣言のプロトタイプを定義します。特に、次の役割を果たします: * **使用 `pydantic` Workflow 定義の構文解析を支える:** これには [`pydantic BaseModel`](https://docs.pydantic.dev/latest/api/base_model/) の機能を継承し、Workflow 定義の解析と検証を行います。このスキーマは、さらに `pydantic の` OpenAPI 標準との統合により、Workflows UI と互換性のある形式へ自動エクスポートすることもできます。 * **データバインディングを定義する:** 実行中にワークフローを流れるデータのセレクタとなる manifest のどのフィールドかを指定し、その kind を示します。 * **ブロック出力を記述する:** ブロックが生成する出力を概説します。 * **次元性を指定する:** 入力と出力の次元性に関連するプロパティを詳述します。 * **バッチ入力と空値を示す:** そのステップがバッチ入力と空値を受け付けるかどうかを Execution Engine に知らせます。 * **互換性を確保する:** 安定性を保つために、異なる Execution Engine バージョンとの互換性を規定します。詳細は [versioning](/workflows/ja/gaido/developer-guide/versioning.md). ### manifest のスキャフォールディング manifest の動作を理解するために、ステップごとに定義してみましょう。ここで作成する例のブロックは、画像類似度の計算を行います。まず import とクラスのスキャフォールディングから始めます: ```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` - 動的なブロックプールに基づく Workflow 定義の構文解析に必要です。これは [`pydantic` type discriminator](https://docs.pydantic.dev/latest/concepts/unions/#discriminated-unions) であり、Workflow 定義内の特定ステップを解析するときに、どのブロック manifest を検証すべきかを Compiler が理解できるようにします * `name` - このプロパティは、ステップに一意の名前を付け、他のステップがセレクタを通じてそれを選択できるようにするために使われます ### 入力の追加 私たちのステップには、比較対象の 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` 以外のすべてのプロパティは、 # ハードコードされたパラメータかデータセレクタのいずれかとして扱われます。データセレクタは、 # から始まる文字列で、データ参照を示します # `$steps.` または `$inputs.` です。これは runtime で利用可能なデータを参照します。 # この場合、通常はデータの kind を指定し、 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`。コードベースをさらに深く見ると、これは type alias のコンストラクタ関数であることがわかるでしょう - つまり、 `pydantic` には次の文字列に一致するものを期待するよう指示します `$inputs.{name}` と `$steps.{name}.*` というパターンです。さらに、追加の schema フィールドメタデータを提供し、Workflows エコシステムのコンポーネントに対して、その `kind` がセレクタの背後にあるデータであることを伝えます [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 definition 内の次のステップ宣言を扱えます: ```json { "type": "my_plugin/images_similarity@v1", "name": "my_step", "image_1": "$inputs.my_image", "image_2": "$steps.image_transformation.image" } ``` この定義により Compiler と Execution Engine は次のようになります: * 次の型を宣言する Workflow ブロックからステップを初期化する `my_plugin/images_similarity@v1` * ステップの run メソッドに 2 つのパラメータを供給する: * `input_1` という型の `WorkflowImageData` で埋められ、Workflow 実行入力として送信された画像が入ります。名前は `my_image`. * `imput_2` という型の `WorkflowImageData` で、別のステップ `image_transformation` ### manifest へのパラメータ追加 では、ステップ実行に影響するパラメータを追加しましょう。
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` 以外のすべてのプロパティは、 # ハードコードされたパラメータかデータセレクタのいずれかとして扱われます。データセレクタは、 # から始まる文字列で、データ参照を示します # `$steps.` または `$inputs.` です。これは runtime で利用可能なデータを参照します。 # この場合、通常はデータの kind を指定し、 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` import [`float_zero_to_one`](/workflows/ja/gaido/developer-guide/kinds/float-zero-to-one.md) `kind` パラメータを定義するために使われる定義。 * の行で `27` というパラメータの定義を始めます `similarity_threshold`。manifest は float 値か、 `kind` [`float_zero_to_one`](/workflows/ja/gaido/developer-guide/kinds/float-zero-to-one.md)で import した Workflow 入力へのセレクタのいずれかを受け入れます `9`.
のような manifest 定義で、Workflow definition 内の次のステップ宣言を扱えます: ```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="類似度を計算する最初の画像", ) 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` import [`boolean`](/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(...)` は name が1つの要素を持つリストを返す必要があります `*` および kind `*` （別名 `WILDCARD_KIND`) * さらに、ブロックのマニフェストはインスタンスメソッド `get_actual_outputs(...)` を実装する必要があり、これは埋められたマニフェストデータに基づいて生成可能な実際の出力のリストを提供します ```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="類似度を計算する最初の画像", ) 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 ] ```
## ブロッククラスの定義 この段階で、シンプルなブロックのマニフェストは準備できました。続けて例を見ていきましょう。より詳しくは、今は気を散らすだけなので [応用トピック](#advanced-topics) セクションをご覧ください。 ### 基本実装 マニフェストの準備ができたので、ブロックの基本実装を用意できます。
ブロックの足場づくり ```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="類似度を計算する最初の画像", ) 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 がデータを渡して所望の結果を得るために呼び出します。なお、入力を定義するマニフェストフィールドは [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="類似度を計算する最初の画像", ) 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` ブロックのコンストラクタを定義します。これにより、ブロックの状態は1回だけ初期化され、連続した呼び出しをまたいで保持されます `run(...)` メソッド - たとえば Execution Engine が動画の連続フレームで実行される場合 * 行 `69-80` ブロック機能の実装を提供します。詳細は Workflows エコシステムにおいて本質的には重要ではありませんが、注目すべき点がいくつかあります: * 行 `69` と `70` を活用します `WorkflowImageData` という抽象化を使って、次の方法を示します `numpy_image` プロパティを使って取得できます `np.ndarray` Workflows 内の画像の内部表現から取得します。残りの `WorkflowImageData` プロパティも確認して、さらに詳しく理解することをお勧めします。 * Workflow ブロック実行の結果は、次の行で定義されています `78-80` 今回の場合は単なる辞書です **そのキーは、マニフェストで宣言された出力名です**、次の行で `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 # これは例としてのインポートです！調整が必要です 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="類似度を計算する最初の画像", ) 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` import `Batch` Workflows ライブラリのコアからのもので、このクラスはバッチ要素を保持するための、リスト（ただし読み取り専用）に非常によく似たコンテナを表します * 行 `40-42` ブロックのデフォルト動作を変更し、バッチを処理できるようにするクラスメソッドを定義します。ここでは、各パラメータが `run(...)` メソッド **バッチ指向として認識する**. * 上記の変更により、次のシグネチャが変更されました `run(...)` メソッドは変更され、現在では `image_1` と `image_2` は `WorkflowImageData`のインスタンスではなく、むしろこの型の要素のバッチです。 **重要な注意:** 複数のバッチ指向パラメータがある場合、それらのバッチでは対応する位置に互いに関連する要素が入っていることを想定しています。つまり、私たちのブロックが比較する `image_1[1]` と `image_2[1]` が実際に論理的に意味のある操作になるようにするためです。 * 行 `74-77`, `85-86` すべてのバッチ要素にわたって処理を実行するために導入が必要だった変更を示します。必要に応じてバッチ要素を反復する方法も示しています * 出力が次の行でどのように構成されるかに注意することが重要です `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 = ... # スカラーパラメータに対して何かを行う 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(...)` Execution Engine に、そのブロックが `run(...)` 指定されたパラメータについてスカラー入力とバッチ指向入力の両方を処理できることを伝えるメソッドです。 * 行 `45-47` 混在した性質を持つパラメータを `run(...)` メソッドシグネチャに示します。 * 行 `49` これにより、期待される出力サイズを把握しておく必要があることがわかります **ブロックロジック内で**。そのため、混在入力を持つブロックの実装はかなり難しいです。通常、ブロックの `run(...)` メソッドはスカラーで動作します。ほとんどの場合（例外は以下で説明します）- メソッドは単一の出力辞書を構築します。同様に、バッチ指向入力が受け入れられる場合、それらの入力が期待される出力サイズを定義します。しかしこの場合は、バッチを手動で検出し、そのサイズを取得する必要があります。 * 行 `50-54` バッチ指向データが検出されたときに異なるロジックを適用することで、混在パラメータを通常どのように扱うかを示します * 前述のとおり、出力の構築も混在入力の性質に合わせて調整する必要があります。これは次の行で示されています `65-70`
### フロー制御ブロックの実装 フロー制御ブロックは、単にデータを処理する他のブロックとはかなり異なります。ここではフロー制御ブロックの作成方法を示しますが、その前に少し理論を説明します: * フロー制御ブロックとは、マニフェスト内のステップセレクタとの互換性を宣言するブロックです（ステップへのセレクタは次のように定義されます `$steps.{step_name}` — ステップ出力セレクタに似ていますが、出力名の指定はありません） * フロー制御ブロックは出力を登録できません。返すことが想定されているのは `FlowControl` オブジェクト * `FlowControl` オブジェクトは、次に進むべき次のステップ（ステップマニフェストで提供されたセレクタから）を指定します。これは、与えられたバッチ要素（SIMD フロー制御）またはワークフロー全体の実行（非 SIMD フロー制御）に対して適用されます
フロー制御の実装 例では random continue ブロックの実装を示し、コメントアウトしています ```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` import `FlowControl` フロー制御ブロックから返せる唯一の有効な応答となるクラス * 行 `28` ステップセレクタのリストを定義します **これにより、ブロックは実質的にフロー制御ブロックになります** * 行 `55` と `56` 出力の構築方法を示します — `FlowControl` オブジェクトはコンテキストとして次のものを受け入れます `None`, `文字列` または `文字列のリスト` - `None` バッチ要素のフロー終了を表します。文字列は入力で渡された次のステップのセレクタであることが期待されます。
フロー制御の実装 - バッチ版 例では random continue ブロックの実装を示し、コメントアウトしています ```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` import `FlowControl` フロー制御ブロックから返せる唯一の有効な応答となるクラス * 行 `29-32` ステップセレクタのリストを定義します **これにより、ブロックは実質的にフロー制御ブロックになります** * 行 `38-40` の定義を含みます `get_parameters_accepting_batches(...)` Execution Engine に、そのブロックが `run(...)` メソッドがバッチ指向の `image` パラメータを期待していることを示します。 * 行 `59` 各要素ごとにフロー制御ガイドを返す必要があることを示しています `image` バッチ。 * その目的を達成するために、次の行で `60` バッチの内容を反復処理します。 * 行 `61-63` 出力の構築方法を示します — `FlowControl` オブジェクトはコンテキストとして次のものを受け入れます `None`, `文字列` または `文字列のリスト` - `None` バッチ要素のフロー終了を表します。文字列は入力で渡された次のステップのセレクタであることが期待されます。
### ネストされたセレクタ 一部のブロックでは、ブロックマニフェストのフィールドにセレクタのリストまたはセレクタの辞書を指定する必要があります。バージョン `v1` の Execution Engine は 1 階層のネストしかサポートしていないため、セレクタのリストのリストや、セレクタのリストを含む辞書は正しく認識されません。 ネストされたセレクタの使用例となる実践的なユースケースを以下に示します。 #### 可変数のモデルからの予測の融合 複数の分類器の予測に対して多数決を取るブロックを作りたいとします。その場合、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]} ```
ネストされたセレクタ - モデルのアンサンブル ```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="ステップ出力へのセレクタ", 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]`
このようなブロックは、次のステップ宣言と互換性があります: ```json { "type": "my_plugin/fusion_of_predictions@v1", "name": "my_step", "predictions": [ "$steps.model_1.predictions", "$steps.model_2.predictions" ] } ``` #### 動的パラメータを許可するデータ変換付きブロック 場合によっては、ブロックが「名前付き」セレクタのグループを受け取る必要があります。その名前と値は、Workflow 定義の作成者によって定義されます。そのような場合、ブロックのマニフェストはセレクタの辞書を受け入れなければならず、キーがそれらのセレクタの名前として機能します。
ネストされたセレクタ - 名前付きセレクタ ```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="ステップ出力へのセレクタ", 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` コンテナはバッチ指向の入力のみをラップし、その他の入力は単一値として渡されます。
このようなブロックは、次のステップ宣言と互換性があります: ```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` もし `model_1` が物体検出モデルであれば * の下で `data["b"]` 内部で `run(...)`、次の名前の入力パラメータの値が見つかります `my_parameter` ### 入力と出力の次元数対 `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` マニフェストクラスは出力次元数オフセットを宣言します - 値 `1` は、次元数レベルに `1` を加えるものと理解されるべきです * 行 `63`では、ブロックはその後の処理から空の画像を除外しますが、その代わりに `None` 出力の辞書ではなく、出力を持つ辞書の代わりに None を置くことで、これを行います。これは条件分岐実行で使われるのと同じ Execution Engine の挙動を利用します - データポイントは下流処理から除外されます（途中に空入力を要求するステップが存在しない限り）。 * 行で `64-65` 単一入力の結果 `image` と `predictions` が収集されます - これは、登録済みのすべての出力をキーに持つ辞書のリストであることが意図されています。Execution Engine は、ステップが各入力要素に対して要素のバッチを返すことを理解し、下流ステップの実行中に追跡するためのインデックスのネスト構造を作成します。 **出力の次元数減少** この例では、ブロックはクロップ予測を可視化し、すべてのクロップ予測を単一の出力画像に配置したタイルを作成します。 ```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` マニフェストクラスは出力次元数オフセットを宣言します - 値 `-1` 次元数レベルを `1` * 行で `34-36` マニフェストクラスは `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 つの入力画像をクロップした場合、その 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` マニフェストクラスは入力次元数オフセットを宣言しており、それは `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` マニフェストは、ブロックが入力のバッチを受け取ることを宣言しています * 行で `33-35` マニフェストクラスは出力次元数オフセットを宣言します - 値 `1` は、次元数レベルに `1` を加えるものと理解されるべきです * 行で `55-66`、入力パラメータのシグネチャは、 `run(...)` メソッドが同じ次元数の入力に対して実行され、それらの入力がバッチとして提供されることを反映しています * 行 `70`では、ブロックはその後の処理から空の画像を除外しますが、その代わりに `None` 出力の辞書ではなく、出力を持つ辞書の代わりに None を置くことで、これを行います。これは条件分岐実行で使われるのと同じ Execution Engine の挙動を利用します - データポイントは下流処理から除外されます（途中に空入力を要求するステップが存在しない限り）。 * 出力の構築は、行で示されています `71-73` は 2 つのネストレベルを示しています。まず、ブロックはバッチ上で動作するため、各入力バッチ要素ごとに 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` マニフェストは、ブロックが入力としてバッチを受け取ることを想定している * 行で `33-35` マニフェストクラスは出力次元数オフセットを宣言します - 値 `-1` 次元数レベルを `1` * 行で `52-53` 出力次元数の減少とバッチ処理がメソッドシグネチャに与える影響を確認できます。最初の「層」の `Batch[]` は、この事実の副次的な結果です。つまり、マニフェストがブロックが入力のバッチを受け取ることを宣言しているためです。2 つ目の「層」は出力次元数の減少によって生じます。Execution Engine は、減少させるべき次元を、入力に用意された追加の `Batch[]` コンテナに包み込みます。これにより、プログラマは特定のトップレベルバッチ要素に属するネストされたバッチ要素をすべて集められます。 * 行 `66-67` 出力がどのように構築されるかを示します - 各トップレベルバッチ要素ごとに、ブロックはすべてのクロップと予測を集約し、単一のタイルを作成します。ブロックは入力のバッチを受け取るため、この手順の結果、トップレベルバッチ要素ごとに 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` マニフェストは、ブロックが入力としてバッチを受け取ることを想定している * 行で `35-40` マニフェストクラスは入力次元数オフセットを宣言しており、それは `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` batch
### 空の入力を受け入れるブロック 前述のとおり、ワークフローの実行中に一部のバッチ要素が「空」になることがあります。これはいくつかの要因で発生します: * **フロー制御の仕組み:** 実行の特定の分岐が一部のバッチ要素をマスクし、その後続のステップで処理されないようにする場合があります。 * **データ処理ブロックでは:** 場合によっては、ブロックが特定のデータポイントに対して意味のある出力を生成できないことがあります。たとえば、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 の役割は、それらのエンティティをブロックに渡し、使用可能にすることです。この仕組みは、 [Workflow 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) ブロックを定義しながら初期化引数を要求する方法を見てみましょう。
コンストラクタ引数を要求するブロック ```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 つのオプションのクラスメソッドを公開します。これらをオーバーライドしてください。 既存のブロックには **変更は不要です**. #### `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` その注意事項をブロック内に一度だけ宣言して、実行エンジン、スキーマエンドポイント、そして自動生成されるブロックギャラリーが一貫して表示できるようにします。 デフォルトは `[]`ので、既存のブロックには **変更は不要です**. #### 重大度: `ソフト` vs. `ハード` 各制約には `重大度`: * **`Severity.SOFT`** - ブロックは最後まで実行され、正しい出力形状を返しますが、値が劣化しているか意味を持ちません（例: トラッカー ID がリクエスト間でリセットされる、クールダウンがスロットリングしない、ファイルが一時ディスクに書き込まれる）。ワークフローは引き続き実行されますが、結果はユーザーの期待どおりではありません。 * **`Severity.HARD`** - ブロックは実行されない / 例外を投げる / この実行環境では有用な出力を生成できません。エンジンはコンパイルを拒否するか、即時失敗すべきです。 #### 制約のスコープ設定 1 つの `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`. The `note` フィールドは、失敗モードまたは劣化した動作についての 1 行の、人間が読める説明です。抽象的な前提条件ではなく、何が起こるのかを記述してください（例: "track\_ids がリクエスト間でリセットされる", "一時的な /tmp に書き込む"）。 #### 共有プリセット ほとんどの注意事項は少数の一般的なパターンに収まるため、コードベース全体で表記を統一する目的で、再利用可能なプリセットがデータクラスと一緒にエクスポートされています。まずはこちらを使ってください: * `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` ホスト型サーバーレス実行環境には存在しない GPU ハードウェアを必要とするブロック）は、インラインで宣言されます: ```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="Block requires a CUDA GPU; raises on CPU-only workers.", applies_to_runtimes=[ Runtime.HOSTED_SERVERLESS, Runtime.SELF_HOSTED_CPU, ], ), ] ``` この方法で宣言された制約は、次の 3 か所に表示されます: 1. The `describe_interface` HTTP ペイロード（ `RuntimeRestriction.to_dict()`経由）で、ワークフローが実行される前にワークフロークライアントやビルダーがユーザーに警告できるようにします。 2. ブロックの自動生成ブロックギャラリーページの、「Runtime compatibility」セクション内。 **Properties**. 3. 実行エンジン。現在の実行環境に対する `Severity.HARD` 制約で fail-fast するかどうかを選べます。 ### 依存リソースの宣言 多くのブロックは実行時に外部リソースを必要とします。Roboflow で学習したモデルや基盤モデルの重み、Roboflow プロジェクト（例: アクティブラーニングの対象、データセットアップロード先）、またはサードパーティのホスト済みモデル（OpenAI、Anthropic、OpenRouter、...）です。ワークフローが実際に実行されるまで、システムはどのリソースが必要か知りません。 オーバーライドする `discover_dependent_resources()` をブロックマニフェストで行うと、このギャップが埋まります。このメソッドは **解析済みマニフェストインスタンス**に対して呼び出されるため、特定のステップの具体的なフィールド値から宣言を計算できます。これにより呼び出し側は、何も実行せずにコンパイル時点でワークフロー全体のリソースを静的に列挙できます。これによって、予測可能な実行時間のためにモデル重みを事前読み込みする、あるいは API キーが参照されたすべてのモデルとプロジェクトにアクセスできることを事前に検証するといった用途が可能になります。 ```python def discover_dependent_resources(self) -> Optional[List[DependentResource]]: ... ``` デフォルトは `None`、つまりブロックは **その依存関係を** 宣言しません *unknown*として扱う必要があります。これは `[]`とは意図的に異なり、後者はそのブロックが外部リソースを一切必要としないことを明示的に宣言します。既存のブロックに変更は不要です。リソースを使うブロックはオーバーライドしてください。 #### リソースのエンベロープ ブロックがリソース形状を独自に作り出すことはありません。エンベロープは 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` （モデルエンティティにプラットフォーム上で到達可能であることだけが必要です。例: メタデータを添付するモニタリングシンク）。実行の場合、 `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.` 参照を置き換えられます。 `$steps..` 参照は静的にはまったく解決できません。すべてのメタデータエンティティは `requires_runtime_resolution()` を公開しており、呼び出し側は具体的な識別子と、まだ解決が必要な参照を区別できます。 最終的な識別子がフィールド値の *関数* である場合（ `clip/`のようなファミリープレフィックス、カタログ検索）、置換された入力値だけでは実行される ID にはなりません。そのような宣言には **`model_id_resolver`** を付けます。これは、必要なものをすべてクロージャに束縛した callable で、置換された値を最終 ID に変換します。この resolver はプロセス内での補助にすぎません。シリアライズ、JSON スキーマ、等価性の対象外です。プロセス内の呼び出し側（例: エンジンの初回実行前ロード）は、入力値を置換したあとにこれを呼び出します。resolver は `None` を返して、その値が静的には解決不可能であることを宣言できます（最終 ID はその 1 つの値だけではなく他の要素にも依存する）— 呼び出し側はそのような依存関係をスキップし、実行時に解決します。例外を送出する場合は、その値が本当に無効であることを意味します。 #### に倣ってください `run()` が読み込みます 宣言された識別子は、 `run()` が要求するものと完全に一致していなければなりません — バージョンフィールドから合成された ID を含みます。ブロックが `f"my_family/{self.version}"` を `load_core_model(...)` 呼び出し箇所で構築しているなら、宣言も同じ ID を構築しなければなりません（そして version フィールドにセレクターが入っている場合は、そのままのセレクターにフォールバックしてください）。 #### 例 モデルと、任意のアクティブラーニング対象プロジェクトを宣言するモデルブロック: ```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 フィールドからモデル 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): # セレクターはそのまま返される; 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}")] ``` 実行せずにモデルエンティティのみを参照するシンク: ```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, ) ] ``` #### 宣言しない場合 単に *保持する* だけの、モデル ID やプロジェクト種別の値を一般的なペイロードとして扱うフィールド（例: 値をクエリパラメータとして転送する webhook シンク）はリソース依存関係ではありません。そうしたブロックは意図的にデフォルトのままにしておきます。 コアリポジトリへの貢献者は、その単体テスト（`tests/workflows/unit_tests/core_steps/test_dependent_resources.py`）が境界を守っていることを認識してください。マニフェストで `roboflow_model_id` または `roboflow_project` kind `discover_dependent_resources()` のフィールドを宣言するすべてのコアブロックは、必ず モデル重みを読み込むブロック **モデルマネージャーの外部で** （実際に何をするかを確認してください、 `run()` 存在するかどうかではなく `model_manager` ）は、今のところこのメソッドを実装すべきではありません。依存関係は未宣言のままです（`None`). カスタム Python ブロックは常に `None`を返します: そのコードは静的解析からは不透明なので、 *unknown* が唯一の正直な答えです。