> 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 内に定義するだけで通常は十分です - 参照してください [カスタムブロック](/workflows/ja/burokku/blocks/custom-blocks.md).
{% endhint %}
Workflows のブロック開発には、Workflow エコシステムの理解が必要です。詳細に入る前に、必要な知識を要約しておきましょう:
の理解、特に: [ワークフロー実行](/workflows/ja/gaido/developer-guide/workflow-execution.md)、特に:
* Workflow definition における Workflow ブロックとステップの関係は何か
* Workflow ブロックとその manifest がどのように [Workflows Compiler](/workflows/ja/gaido/developer-guide/compiler.md)
* 何であるか `次元レベル` Workflow を流れるバッチ指向データの
* どのように [Execution Engine](/workflows/ja/gaido/developer-guide/execution-engine.md) ステップと、その入力および出力に関して相互作用するか
* の性質と役割は何か [ワークフロー `kinds`](/workflows/ja/gaido/developer-guide/kinds.md)
* どのように [`pydantic`](https://docs.pydantic.dev/latest/) が動作するかを理解する
## 環境のセットアップ
すぐにわかるように、Workflow ブロックの作成は、特定のインターフェースを実装する Python クラスを定義するだけです。この設計により、他の Python コードと同様に、Python インタープリタでブロックを実行できます。ただし、Workflow 実行中に他のブロックから通常提供される必要な入力をすべて組み立てる際に、困難に直面することがあります。そのため、スムーズに作業を進めるには開発環境を適切にセットアップすることが重要です。標準的な開発プロセスの一部として、以下の手順に従うことを推奨します（最初の手順は、以降の作業では省略できます）:
1. **をセットアップし `conda` 環境** をインストールする `inference`で説明されているように、 [`inference` コントリビューターガイド](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 server を実行することを推奨します（ `inference` を変更するたびにサーバーを再ビルドするよりもずっと高速です）:
```bash
inference_repo$ docker run -p 9001:9001 \
-v ./inference:/app/inference \
roboflow/roboflow-inference-server-cpu:latest
```
をローカルサーバーに接続して Roboflow UI に:
プレビューを素早く実行するには:
私のブロックには追加の依存関係が必要です - 事前構築済みの `inference` server は使えません
ブロックが追加の依存関係を必要とすることは自然です。依存関係を追加するには、 [requirements files](https://github.com/roboflow/inference/tree/main/requirements)のどれかに含めればよいです。それらは関連する Docker イメージ（通常は [CPU ビルド](https://github.com/roboflow/inference/blob/main/docker/dockerfiles/Dockerfile.onnx.cpu) であるオブジェクトです。対象は `inference` server）にインストールされています。
その後、次を実行します:
```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 definition を作成してサーバーにリクエストを送る必要があります。
UI なしで Workflow を実行する
次のコードスニペットは、Workflow を実行するために `inference` サーバーへリクエストを送る方法を示しています。 `inference_sdk` は `inference` パッケージに、サーバー向けの軽量クライアントライブラリとして含まれています。
```python
from inference_sdk import InferenceHTTPClient
YOUR_WORKFLOW_DEFINITION = ...
client = InferenceHTTPClient(
api_url=object_detection_service_url,
api_key="XXX", # 任意、Workflow が Roboflow Platform を使用する場合のみ必須
)
result = client.run_workflow(
specification=YOUR_WORKFLOW_DEFINITION,
images={
"image": your_image_np, # これは例の入力です。適宜調整してください
},
parameters={
"my_parameter": 37, # これは例の入力です。適宜調整してください
},
)
```
### 通常のコントリビューター向けの推奨方法
で統合テストを作成することは、 `tests/workflows/integration_tests/execution` ディレクトリで行う開発反復プロセスの自然な一部です。この方法では、開発とテストを同時に進めることができ、コードを洗練する際に有益なフィードバックが得られます。ある程度の経験は必要ですが、長期的なコード保守性を大きく高めます。
手順は簡単です:
1. **新しいテストモジュールを作成する:** たとえば、次のように名付けます `test_workflows_with_my_custom_block.py`.
2. **サンプル Workflow を開発する**: 1つ以上のサンプル Workflow を作成します。ブロックがエコシステム内の他のブロックと協調できるのが理想です。
3. **サンプルデータでテストを実行する:** テスト内でこれらの Workflow をサンプルデータを使って実行します（通常使用している例データを見つけるには、 [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:
# given
workflow_init_parameters = {
"workflows_core.model_manager": model_manager,
"workflows_core.api_key": roboflow_api_key,
"workflows_core.step_execution_mode": StepExecutionMode.LOCAL,
}
execution_engine = ExecutionEngine.init(
workflow_definition=,
init_parameters=workflow_init_parameters,
max_concurrent_steps=WORKFLOWS_MAX_CONCURRENT_STEPS,
)
# when
result = execution_engine.run(
runtime_parameters={
"image": dogs_image,
}
)
# then
assert isinstance(result, list), "リストが返されることを期待"
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` は、テストする Workflow 内のいずれかのブロックが Roboflow API キーを必要とする場合に使える任意の fixture です。その場合は、テストを実行する前に、 `ROBOFLOW_API_KEY` 環境変数を有効なキーでエクスポートしてください。
* 行 `7-11` は、Workflow definition に基づいて Execution Engine が実行時に作成するブロックの初期化パラメータのセットアップを示しています。
* 行 `19-23` は、入力パラメータを注入して Workflow を実行する方法を示しています。runtime\_parameters のキーが Workflow definition で宣言された入力と一致していることを確認してください。
* 行 `26`以降に、テスト内の例示的なアサーションがあります。
## プロトタイプ
Workflow ブロックを作成するには、Workflows ライブラリのコアからいくつかの import が必要です。ブロック作成時に役立つ import の一覧を以下に示します:
```python
from inference.core.workflows.execution_engine.entities.base import (
Batch, # データのバッチは Batch[X] コンテナで渡される
OutputDefinition, # manifest で出力を宣言するために使うクラス
WorkflowImageData, # 画像の内部表現
# - 入力の種類が image のときは常に使う
)
from inference.core.workflows.prototypes.block import (
BlockResult, # `run(...)` メソッドの結果の型エイリアス
WorkflowBlock, # あなたのブロックの基底クラス
WorkflowBlockManifest, # ブロック manifest の基底クラス
)
from inference.core.workflows.execution_engine.entities.types import *
# core ライブラリの `kinds` を含むモジュール
```
最も重要なのは次のものです:
* `WorkflowBlock` - あなたのブロックの基底クラス
* `WorkflowBlockManifest` - ブロック manifest の基底クラス
{% hint style="warning" %}
**内部データ表現の理解**
次の import を推奨していることに気づいたかもしれません。 `Batch` と `WorkflowImageData` これらのクラスは、私たちのシステムでビルディングブロックを構築する際に使われる基本コンポーネントです。これらのクラスが全体アーキテクチャの中でどのように位置づくかをより深く理解するには、 [Data Representations](/workflows/ja/gaido/developer-guide/data-representations.md) ページを参照して、より詳細な情報をご覧ください。
{% endhint %}
## block manifest
manifest は Workflow ブロックの重要なコンポーネントであり、ブロックを使うために Workflow definition に配置できるステップ宣言のプロトタイプを定義します。特に、以下を行います:
* **を使用します `pydantic` Workflow definitions の構文解析を支えるために:** は [`pydantic BaseModel`](https://docs.pydantic.dev/latest/api/base_model/) の機能を継承しており、Workflow definitions の解析と検証を行います。このスキーマは、 `pydantic の` OpenAPI 標準との統合により、Workflows UI と互換性のある形式に自動エクスポートすることもできます。
* **データバインディングを定義する:** manifest のどのフィールドが、実行中にワークフローを流れるデータのセレクタであり、その種類が何かを指定します。
* **ブロック出力を記述する:** ブロックが生成する出力を概説します。
* **次元性を指定する:** 入力と出力の次元性に関連するプロパティを詳述します。
* **バッチ入力と空値を示す:** ステップがバッチ入力と空値を受け入れるかどうかを Execution Engine に知らせます。
* **互換性を確保する:** 安定性を保つため、異なる Execution Engine のバージョンとの互換性を規定します。詳細は [バージョニング](/workflows/ja/gaido/developer-guide/versioning.md).
### manifest のスキャフォールディング
manifest の動作を理解するために、1つずつ定義してみましょう。ここで作成する例のブロックは、画像の類似度を計算するものです。まず 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` - blocks の動的プールに基づく Workflow definitions の構文を解析するために必要 - これが [`pydantic` type discriminator](https://docs.pydantic.dev/latest/concepts/unions/#discriminated-unions) であり、Compiler が Workflow definition 内の特定のステップを解析する際に、どの block manifest を検証すべきかを理解できるようにします
* `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.` で始まる文字列で、データへの参照を示します。
# 実行時に利用可能です。この場合、通常はデータの種類を指定して、
# コンパイラに、期待するデータの形を知らせます。
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 のプロトタイプなので、ステップで使う画像について伝える唯一の方法はセレクタを提供することです - core ライブラリには専用の型があり、 `Selector`を使えます。コードベースをさらに深く見ると、これは type alias のコンストラクタ関数であることがわかります - `pydantic` に次のような文字列を期待するよう指示します `$inputs.{name}` と `$steps.{name}.*` というパターンそれぞれに一致し、さらに Workflows エコシステムの各コンポーネントに対して、 `種類` セレクタの背後にあるデータの [image](/workflows/ja/gaido/developer-guide/kinds/image.md). **重要な注意:** 私たちは *種類* をリストとして表し、特定の種類のリストは *種類の和集合* 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 は次のことを行います:
* この type を宣言する 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.` で始まる文字列で、データへの参照を示します。
# 実行時に利用可能です。この場合、通常はデータの種類を指定して、
# コンパイラに、期待するデータの形を知らせます。
image_1: Selector(kind=[IMAGE_KIND]) = Field(
description="類似度を計算する最初の画像",
)
image_2: Selector(kind=[IMAGE_KIND]) = Field(
description="類似度を計算する2枚目の画像",
)
similarity_threshold: Union[
float,
Selector(kind=[FLOAT_ZERO_TO_ONE_KIND]),
] = Field(
default=0.4,
description="画像が類似しているとみなすしきい値",
)
```
* 行 `9` imports [`float_zero_to_one`](/workflows/ja/gaido/developer-guide/kinds/float-zero-to-one.md) `種類` パラメータを定義するために使用される定義です。
* 行で `27` というパラメータの定義を始めます `similarity_threshold`. manifest は float 値か、 `種類` [`float_zero_to_one`](/workflows/ja/gaido/developer-guide/kinds/float-zero-to-one.md)でインポートされた `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
}
```
### ブロック出力の宣言
これでブロックの入力は正常に定義できましたが、ブロックを正常に実行するために必要な要素がまだいくつか足りません。ブロック出力を定義しましょう。
ブロック出力の宣言
必要最小限の情報は出力の説明です。さらに、ブロックの安定性を高めるために、Execution Engine の互換性情報を提供することを推奨します。
```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` ステップ出力を記述するために使われるクラスを import する
* 行 `11` imports [`boolean`](/workflows/ja/gaido/developer-guide/kinds/boolean.md) `種類` 出力定義で使用するため
* 行 `32-39` block からの出力を指定するクラスメソッドを定義する - リスト内の各要素は、バッチ要素ごとの1つの返却プロパティとその `種類`。私たちのブロックは boolean フラグを返します `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 が `*` と kind が `*` （別名 `WILDCARD_KIND`)
* を含むリストを返す必要があります。さらに、block manifest はインスタンスメソッド `get_actual_outputs(...)` を実装し、埋められた 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="類似度を計算する最初の画像",
)
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="類似度を計算する最初の画像",
)
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` blockクラスとそのすべてのメソッドシグネチャを適切に定義するために必要な追加シンボルを提供するように、import構造に変更を追加しました
* 行 `53-55` クラスメソッドを定義します `get_manifest(...)` 先ほど作成したmanifestクラスをそのまま返すために
* 行 `57-63` define `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).
### blockロジックの実装を提供する
それでは、 `run(...)` メソッドの実装例をblockに追加しましょう。そうすることで、意味のある結果を生成できるようにします。
{% hint style="info" %}
このセクションの内容は、blockの堅牢な実装を提供するというより、Workflowエコシステムとblock作成者としてどのようにやり取りするかの例を示すことを目的としています。
{% 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をimportします
* 行 `55-57` blockコンストラクタを定義します。これにより、blockの状態は一度初期化され、連続する `run(...)` メソッド呼び出しの間も保持されます。たとえば、Execution Engineが動画の連続フレーム上で実行される場合です
* 行 `69-80` block機能の実装を提供します。詳細はWorkflowエコシステムに関してそれほど重要ではありませんが、いくつか注目すべき点があります。
* 行 `69` と `70` を使用します `WorkflowImageData` 抽象化であり、 `numpy_image` プロパティをどのように使って取得できるかを示しています `np.ndarray` Workflowにおける画像の内部表現から取得します。残りの `WorkflowImageData` プロパティも試してみることをお勧めします。
* Workflow blockの実行結果。行 `78-80` の場合は単なる辞書です **で、キーはmanifestで宣言された出力名です**、行 `43`で宣言されています。必ず宣言されたすべての出力を提供してください。そうしないとExecution Engineがエラーを発生させます。
## blockを `pluginに公開する`
これでblockは使用可能な状態になりましたが、Execution Engineはその存在を認識していません。これは、登録済みのどのpluginも、あなたが今作成したblockを公開していないためです。blockのバンドリングの詳細は [別ページ](/workflows/ja/gaido/developer-guide/block-bundling.md)で説明されていますが、残る作業はpluginsの `load_blocks(...)` 関数から返されるリストにblockクラスを追加することです:
```python
# あなたのpluginの__init__.py（または、`inference`に直接貢献する場合はroboflow_core plugin）
from my_plugin.images_similarity.v1 import ImagesSimilarityBlock
# これはサンプルのimportです！調整が必要です
def load_blocks():
return [ImagesSimilarityBlock]
```
## 高度なトピック
### 入力バッチを処理するblocks
場合によっては、すべての入力データをバッチとして一度に処理すると、blockの性能が向上することがあります。これはGPU上で動作するモデルで起こりえます。このような動作モードはWorkflows blocksでサポートされています。ここでは、あなたのblockでの使い方の例を示します。
バッチを受け付けるblocksの実装
```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` imports `Batch` Workflowsライブラリのコアから。これはバッチ要素を保持するための、listによく似た（ただし読み取り専用の）コンテナを表すクラスです
* 行 `40-42` blockのデフォルト動作を変更し、バッチを処理できるようにするクラスメソッドを定義します。各パラメータを、 `run(...)` method **がバッチ指向として認識する**.
* 上記の変更により、 `run(...)` メソッドのシグネチャが変更され、今では `image_1` と `image_2` は `WorkflowImageData`のインスタンスではなく、この型の要素のバッチになります。 **重要な注意：** 複数のバッチ指向パラメータがある場合、それらのバッチは対応する位置で互いに関連する要素を持つことを期待します。つまり、たとえば画像を比較する私たちのblockでは、 `image_1[1]` に `image_2[1]` が実際に意味のある論理操作を実行するようになります。
* 行 `74-77`, `85-86` すべてのバッチ要素に対して処理を実行するために導入する必要があった変更を示し、必要に応じてバッチ要素を反復処理する方法を示します
* 行で出力がどのように構築されるかに注意することが重要です `85` - バッチの各要素には、 `run(...)` メソッドから返されるリスト内にそれぞれのエントリが与えられます。順序はバッチ要素の順序と一致していなければなりません。各出力辞書は、block出力で宣言されたすべてのキーを提供する必要があります。
バッチとスカラーの両方を受け付ける入力
これは **比較的まれですが**、blockが1つの入力パラメータ内でバッチ指向データとスカラーの両方を受け付ける必要がある場合があります。Execution Engineはそれを `get_parameters_accepting_batches_and_scalars(...)` block manifestのメソッドで認識します。以下の例を見てください:
```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` 混在（スカラーとバッチ指向の両方）の入力データを受け付けると想定されるmanifestパラメータを指定します。この段階では、前の例と比べて定義上の違いはないことに注意してください。
* 行 `24-26` 指定する `get_parameters_accepting_batches_and_scalars(...)` メソッドにより、blockが `run(...)` メソッドは、指定されたパラメータについてスカラー入力とバッチ指向入力の両方を扱えることをExecution Engineに伝えます。
* 行 `45-47` 混在した性質のパラメータを `run(...)` メソッドシグネチャに示します。
* 行 `49` 期待される出力サイズを追跡する必要があることがわかります **blockロジック内で**。そのため、混在入力を持つblocksを実装するのはかなり難しいのです。通常、blockの `run(...)` メソッドはスカラー上で動作します。ほとんどの場合（例外は下で説明します）、そのメソッドは単一の出力辞書を構築します。同様に、バッチ指向入力を受け付ける場合、それらの入力が期待される出力サイズを定義します。ただしこの場合は、バッチを手動で検出し、そのサイズを取得する必要があります。
* 行 `50-54` バッチ指向データが検出されたときに異なるロジックを適用する、混在パラメータの通常の扱い方を示します
* 先に述べたように、出力の構築も混在入力の性質に合わせて調整する必要があります。それは行で示されています `65-70`
### フロー制御blockの実装
Flow-control blocksは、単にデータを処理する他のblocksとはかなり異なります。ここではフロー制御blockの作り方を示しますが、その前に少し理論を説明します。
* フロー制御blockとは、manifest内でstepセレクタとの互換性を宣言するblockのことです（stepへのセレクタは `$steps.{step_name}` で定義されます。- step出力セレクタに似ていますが、出力名の指定はありません）
* flow-control blocksは出力を登録できず、返すことを意図しています `FlowControl` オブジェクト
* `FlowControl` オブジェクトは、与えられたバッチ要素（SIMDフロー制御）またはワークフロー全体の実行（非SIMDフロー制御）に対して、次に処理すべき次のstep（step manifestで提供されたセレクタから）を指定します
フロー制御の実装
例として、ランダムcontinue blockの実装を示し、コメントアウトしています
```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と評価された場合に実行されるstepへの参照",
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` blockがフローを制御することをExecution Engineに通知するために使用されるstep selectorの型注釈をimportします
* 行 `14` imports `FlowControl` フロー制御blockからの唯一の有効な応答であるクラス
* 行 `28` stepセレクタのリストを定義します **これにより実質的にblockがフロー制御blockになります**
* 行 `55` と `56` 出力の構築方法を示します - `FlowControl` オブジェクトはコンテキストを受け取り、それは `None`, `string` または `文字列のリストです` - `None` バッチ要素のフロー終了を表します。文字列は入力で渡された次のstepのセレクタであることが期待されます。
フロー制御の実装 - バッチ版
例として、ランダムcontinue blockの実装を示し、コメントアウトしています
```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と評価された場合に実行されるstepへの参照",
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` blockがフローを制御することをExecution Engineに通知するために使用されるstep selectorの型注釈をimportします
* 行 `15` imports `FlowControl` フロー制御blockからの唯一の有効な応答であるクラス
* 行 `29-32` stepセレクタのリストを定義します **これにより実質的にblockがフロー制御blockになります**
* 行 `38-40` の定義を含みます `get_parameters_accepting_batches(...)` メソッドは、blockが `run(...)` メソッドがバッチ指向の `image` パラメータ。
* 行 `59` の各要素ごとにフロー制御ガイドを返す必要があることを示しています `image` バッチ。
* そのために、行 `60` ではバッチの内容を反復処理します。
* 行 `61-63` 出力の構築方法を示します - `FlowControl` オブジェクトはコンテキストを受け取り、それは `None`, `string` または `文字列のリストです` - `None` バッチ要素のフロー終了を表します。文字列は入力で渡された次のstepのセレクタであることが期待されます。
### ネストされたセレクタ
いくつかのblockでは、block manifestフィールドでセレクタのリストまたはセレクタの辞書を提供する必要があります。Version `v1` のExecution Engineは1段階のネストのみをサポートしています。そのため、セレクタのリストのリストや、セレクタのリストを含む辞書は正しく認識されません。
ネストされたセレクタの実用的なユースケースを以下に示します。
#### 可変数のモデルからの予測の融合
複数の分類器の予測に対して多数決を取るblockを作りたいとしましょう。その場合、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="step outputsへの参照",
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` セレクタのリストを受け付けられるmanifestフィールドの定義方法を示します
* 行 `50` blockの `run(...)` メソッドへの入力として何を想定すべきかを示します - 特定のkindの表現であるオブジェクトのリストです。blockがバッチを受け付ける場合、 `predictions` フィールドの入力型は `List[Batch[sv.Detections]`
このようなblockは次のstep宣言と互換性があります:
```json
{
"type": "my_plugin/fusion_of_predictions@v1",
"name": "my_step",
"predictions": [
"$steps.model_1.predictions",
"$steps.model_2.predictions"
]
}
```
#### 動的パラメータを許可するデータ変換を伴うblock
場合によっては、Workflow定義の作成者によって名前と値が定義される、「名前付き」セレクタのグループを受け付ける必要があることがあります。このような場合、block manifestはセレクタの辞書を受け付けるべきであり、キーがそれらのセレクタの名前として機能します。
ネストされたセレクタ - 名前付きセレクタ
```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="step outputsへの参照",
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` blockの `run(...)` メソッドへの入力として何を想定すべきかを示します - セレクタで参照されるオブジェクトの辞書です。blockがバッチを受け付ける場合、 `データ` フィールドの入力型は `Dict[str, Union[Batch[Any], Any]]`。非バッチの場合、セレクタで参照される非バッチ指向データは自動的にブロードキャストされますが、バッチを受け付けるblockでは `Batch` コンテナはバッチ指向入力のみをラップし、他の入力は単一値として渡されます。
このようなblockは次のstep宣言と互換性があります:
```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(...)` method
block入力の次元は、 `run(...)` メソッドのシグネチャ形成において重要な役割を果たします。そのためシステムは、入力間の次元レベル差に厳密な上限を課しています（許容される最大差は `1`）。この制限は、blockを記述する際の一貫性と予測可能性を確保するために重要です。
次元の差が制御されていなければ、 `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`、blockは空の画像を以降の処理から除外しますが、その代わりに `None` 出力を持つ辞書の代わりにこれを置きます。これは条件付き実行で使われるのと同じExecution Engineの挙動を利用します。つまり、データポイントは下流の処理から除外されます（ただし、空入力を要求するstepが下流にある場合を除きます）。
* 行内で `64-65` 単一入力の結果は `image` と `predictions` 収集されます。これは、登録されたすべての出力をキーとして含む辞書のリストであることを意味します。Execution engineは、stepが各入力要素ごとに要素のバッチを返すことを理解し、下流stepの実行中に追跡するためのインデックスのネスト構造を作成します。
**出力次元の減少**
この例では、blockはクロップの予測を可視化し、すべてのクロップ予測を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`
* を参照してください [changelog](/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` manifest クラスは入力次元オフセットを宣言しており、 `image` パラメータが最上位であり、 `image_predictions` 予測のネストされたバッチであることを示します
* 異なる入力次元数が宣言される場合は、次元参照プロパティを指定する必要があります（行を参照 `38-40`) - この次元レベルが出力次元数の計算に使用されます - この特定のケースでは、次を指定します `image`. この選択は結果の期待フォーマットに影響します - 選択したシナリオでは、登録済みのすべての出力キーを含む単一の辞書を返す必要があります。もし選択が `image_predictions`の場合、辞書のリスト（サイズは〜の長さに等しい）を返します `image_predictions` バッチ）。別の言い方をすれば、 `get_dimensionality_reference_property(...)` どの次元レベルを出力に対応付けるべきか。
* 行 `63-64` 行で指定された次元オフセットの影響が示されています `31-36`. 明らかに、 `image_predictions` は〜に関するネストされたバッチです `image`. もちろん、特定の〜に関連するネストされた予測のみが `images` バッチとしてまとめられ、実行時にメソッドへ渡されます。
* 前述のとおり、行 `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`、blockは空の画像を以降の処理から除外しますが、その代わりに `None` 出力を持つ辞書の代わりにこれを置きます。これは条件付き実行で使われるのと同じExecution Engineの挙動を利用します。つまり、データポイントは下流の処理から除外されます（ただし、空入力を要求するstepが下流にある場合を除きます）。
* 行で示されている出力の構築 `71-73` ネストが 2 層あることを示しています。まず、ブロックはバッチに対して動作するため、入力バッチ要素ごとに 1 つ、合計で出力のリストを返すことが期待されます。さらに、この各入力バッチ要素の出力要素はネストされたバッチになります。したがって、各入力画像と予測に対して、ブロックは出力のリストを生成します - そのリストの要素は、それぞれ宣言された各出力の値を提供する辞書です。
**出力次元の減少**
この例では、blockはクロップの予測を可視化し、すべてのクロップ予測を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 つのオプションの classmethod を公開します。これらをオーバーライドすると、エアギャップ対応の workflow builder がどのブロックをオフラインで使用できるか判断しやすくなります。
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 モデルブロックでは、このブロックが扱えるタスクタイプを返します。エアギャップ対応 builder はこれを使って、キャッシュされたユーザーモデルを互換ブロックに対応付けます。
```python
class BlockManifest(WorkflowBlockManifest):
# ...
@classmethod
def get_compatible_task_types(cls) -> Optional[List[str]]:
return ["object-detection"]
```
デフォルトでは次を返します `None` - Roboflow モデルでパラメータ化されていないブロック（基盤モデル、ロジックブロック、シンクなど）に適しています。
### 実行時制約
一部のブロックは、デプロイ先のランタイム（ホスト型サーバーレス、専用デプロイ、自前ホスト、推論パイプライン）、ステップの実行モード（ローカル vs. リモート）、入力モード（画像 vs. 動画）に応じて挙動が変わったり、まったく失敗したりします。次をオーバーライドしてください `get_restrictions()` で `WorkflowBlockManifest` これらの注意点をブロック内で一度だけ宣言しておけば、execution engine、スキーマエンドポイント、自動生成される block gallery のすべてで一貫して表示できます。
デフォルトでは次を返します `[]`、そのため既存のブロックは **変更不要です**.
#### 重大度: `soft` 対 `ハード`
各制約には `重大度`:
* **`Severity.SOFT`** - ブロックは最後まで実行され、正しい出力形状を返しますが、値が劣化していたり意味をなさなかったりします（例: tracker ID がリクエスト間でリセットされる、クールダウンが制限として機能しない、ファイルが一時ディスクに書き込まれる）。workflow は継続しますが、結果はユーザーの期待どおりではありません。
* **`Severity.HARD`** - このランタイムではブロックが実行されない / 例外を投げる / 利用可能な出力を生成できません。engine はコンパイルを拒否するか、fail-fast で失敗すべきです。
#### 制約の適用範囲を定める
A `RuntimeRestriction` は 3 つの軸の任意の組み合わせに沿って範囲指定できます。ある軸が次のまま残されている場合 `None`、その制約はその軸のすべての値に適用されます。
* `applies_to_runtimes`: `Runtime.HOSTED_SERVERLESS`, `Runtime.DEDICATED_DEPLOYMENT`, `Runtime.SELF_HOSTED_CPU`, `Runtime.SELF_HOSTED_GPU`, `Runtime.INFERENCE_PIPELINE`.
* `applies_to_step_execution_modes`: `StepExecutionMode.LOCAL`, `StepExecutionMode.REMOTE`.
* `applies_to_input_modes`: `RuntimeInputMode.IMAGE`, `RuntimeInputMode.VIDEO`.
この `注記` フィールドは、失敗モードまたは劣化した挙動を 1 行で人間が読める形で説明するものです - 抽象的な前提条件ではなく、何が起こるかを記述してください（例: 「track\_ids がリクエスト間でリセットされる」「一時的な /tmp に書き込む」）。
#### 共有プリセット
ほとんどの注意点は少数の共通パターンに収まるため、コードベース全体で文言を統一する目的で、再利用可能なプリセットが dataclass と一緒にエクスポートされています。まずはこれらを使ってください:
* `STATEFUL_VIDEO_HTTP_SOFT_RESTRICTION` - ビデオ追跡 / カウント / 集約ブロック向け。動画ごとの状態がプロセスメモリにあり、ステートレスな HTTP リクエスト間でリセットされます。
* `COOLDOWN_HTTP_SOFT_RESTRICTION` - クールダウン / レート制限タイマーがプロセスメモリに保持されるため、複数レプリカの HTTP ランタイムではスロットリングしないブロック向け。
* `STILL_IMAGE_INPUT_SOFT_RESTRICTION` - 時間的文脈（動画や繰り返しフレーム）に依存し、静止画ではほとんど、あるいはまったく効果がないブロック向け。
#### 例
動画ごとの状態を保持し、静止画では意味をなさないラインカウンターブロックは、両方の制約を宣言します:
```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="ブロックには CUDA GPU が必要です; CPU のみのワーカーでは例外を発生します。",
applies_to_runtimes=[
Runtime.HOSTED_SERVERLESS,
Runtime.SELF_HOSTED_CPU,
],
),
]
```
このように宣言された制約は 3 か所に表示されます:
1. この `describe_interface` HTTP ペイロード（〜経由で `RuntimeRestriction.to_dict()`）として、workflow クライアントと builder が workflow 実行前にユーザーへ警告できます。
2. ブロックの自動生成された block gallery ページの「Runtime compatibility」セクションに、次の直後で表示されます **プロパティ**.
3. Execution engine は、次の条件で fail-fast することを選べます `Severity.HARD` 現在のランタイムに対する制約