実行エンジン

Workflows実行エンジンの内部: 入力検証、実行順序、ステップの入力と出力、フロー制御。

このコンパイルプロセス Workflow定義を実行するために必要なすべての詳細を保持する Workflow Execution graph を作成します。このセクションでは、実行プロセスの詳細を説明します。

高レベルでは、このプロセスは次のことを行います:

実行時入力を検証: Workflow定義のすべての必須プレースホルダーにデータが入力されていることを確認し、データ型が正しいことを保証します。
実行順序を決定: ステップが実行される順序を定義します。
ステップ入力を準備し、出力をキャッシュ: 各ステップの入力を整理し、出力を将来の使用のために保存します。
最終的な Workflow 出力を構築: Workflow の全体結果を組み立てます。

実行時入力の検証

Workflow 定義では、Workflow 実行に必要な入力が指定されています。で述べたように以前の、入力はステップで処理されるバッチ指向データにも、ステップ実行を設定するパラメータにもなり得ます。この違いは Workflow の動作にとって非常に重要であり、このページ全体で詳しく見ていきます。

入力検証から始めると、Execution Engine には入力を解析して利用できるように準備する専用コンポーネントがあります。これは Workflow 定義からバッチ指向の入力を認識し、内部表現に変換します（例: WorkflowImage になります Batch[WorkflowImageData]）。これにより、ブロック開発者はデータを簡単に扱えます。非バッチ指向のパラメータは、それらを必要とするステップを作成するために使われたブロックマニフェストに対して型の整合性がチェックされます。これにより、型エラーを実行プロセスの早い段階で検出できます。

すべてのバッチ指向入力は、サイズが 1 か n のいずれかでなければなりません。バッチに要素が 1 つしかない場合、その要素は自動的にバッチ全体へブロードキャストされます。

実行順序の決定

Workflow Execution graph は有向非巡回グラフ（DAG）であり、これによりトポロジカル順序を決定できます。トポロジカル順序とは、Workflow の各ステップが実行される順序で、各ステップが実行される前に依存関係が満たされることを保証します。言い換えると、あるステップが別のステップの出力に依存している場合、Workflow Engine は依存先のステップを先に実行するようにします。

さらに、トポロジカル構造により、競合状態を起こさずに並列実行できるステップを特定できます。並列実行は Workflows Execution Engine の既定のモードです。つまり、モデルアンサンブルで使われるような複数の独立したステップを同時に実行でき、順次処理と比べて実行速度が大幅に向上します。

Execution Engine の並列実行モードのため（また、各ステップに渡す際の不要なデータコピーを避けるため）、ブロック開発者の皆さんには、ブロックの run(...) メソッドに渡されたデータを変更しないよう強く推奨します。変更が必要な場合は、必ず入力オブジェクトをコピーしてから変更してください！

ステップ入力と出力の扱い

ステップ入力と出力の扱いは、Execution Engine にとって複雑な作業です。これには次が含まれます:

SIMD（Single Instruction, Multiple Data）ブロックと非SIMDブロックを、その入力との関係で区別すること。
条件付き実行と期待される入力の次元数を考慮しながら、ステップ入力を準備すること。
実行の流れを制御するステップからの出力を管理すること。
データ処理ステップからの出力を登録し、ブロックで宣言された出力の次元数と一致するようにすること。

これらのトピックをそれぞれ詳しく見ていきましょう。

SIMD 対非SIMDステップ

定義が示すとおり、SIMD（Single Instruction, Multiple Data）ステップはバッチ指向データを処理し、同じ操作が各データポイントに適用されます。設定には非バッチ指向パラメータを使用することもあります。このようなステップの出力は要素のバッチであることが期待され、入力バッチ要素の順序は保持されます。これは通常の処理ステップにもフロー制御ステップにも当てはまります（参照ブロック開発ガイド）では、フロー制御の判断が各バッチ要素に個別に影響します。

要するに、ステップに入力されるデータの種類によって、それが SIMD か非SIMD かが決まります。ステップが何らかのバッチ指向入力を要求する場合、そのステップは SIMD ステップとして扱われます。

これに対して、非SIMDステップは入力データに対して単一の結果を返すことが想定されています。非SIMDのフロー制御ステップの場合、バッチ内の各要素ごとではなく、下流のすべてのステップ全体に影響します。

歴史的に、Execution Engine は、非SIMDステップの出力がSIMDステップの入力に渡されるすべてのシナリオをうまく扱えず、そのような出力をSIMDステップに渡す際に自動的にバッチへキャストする機能がなかったため、コンパイルエラーが発生していました。Execution Engine 以降、 v1.6.0SIMD および非SIMDブロックの扱いは改善されました。 Auto Batch Casting:

SIMD 入力が検出されたにもかかわらずスカラーデータを受け取った場合、Execution Engine はそれを自動的にバッチへキャストします。
バッチの次元数は、コンパイル時に次を用いて決定されます。来歴利用可能な場合は他のバッチ指向入力からの情報を使用します。不足している次元は、次に似た方法で生成されます torch.unsqueeze(...).
出力はキャスティングコンテキストに照らして評価され、ブロックが出力次元を維持または減少させる場合はスカラーのままとなり、 新しいバッチを作成し 次元数の増加が想定される場合です。

ステップ入力の準備

要求された各入力要素は、バッチ指向である場合とそうでない場合があります。非バッチ入力は比較的簡単で、特別な処理は不要です。バッチ指向のものは、かなり手間がかかります。Execution Engine は、たとえば各バッチ指向データポイントのインデックスを管理します。

入力画像のバッチがある場合、各要素にはそれぞれ固有のインデックスが割り当てられます。たとえば入力画像が 4 枚あるなら、バッチのインデックスは [(0, ), (1, ), (2, ), (3, )].
ネストされていないバッチであるステップ出力にも同様にインデックスが付けられます。たとえば、上記の各画像に対するモデルの予測にもインデックスが付けられます [(0, ), (1, ), (2, ), (3, )].
次元数を増やすブロックがある場合 次元レベル — たとえば、物体検出モデルの予測に基づく Dynamic Crop で、1 枚目の画像に 2 つ、2 枚目に 1 つ、4 枚目に 3 つのクロップがある場合、そのようなステップの出力は次のようにインデックス付けされます: [(0, 0), (0, 1), (1, 0), (3, 0), (3, 1), (3, 2)].

要素のインデックス付けは、ステップ実行の入力を収集する際に重要です。これにより、すべてのバッチ指向入力を整列でき、Execution Engine は常に予測を (3, ) 画像と (3, ) クロップのバッチを [(3, 0), (3, 1), (3, 2)] いずれかのステップがそれを要求したときに。

各要求入力要素は、バッチ指向にも非バッチにもなり得ます。非バッチ入力は単純で、特別な扱いは必要ありません。しかし、バッチ指向入力にははるかに多くの複雑さが伴います。Execution Engine は、たとえば各バッチ指向データポイントのインデックスを保持します。

入力画像のバッチがある場合、各要素にはそれぞれ固有のインデックスが割り当てられます。4 枚の画像のバッチがあるとすると、インデックスは [(0,), (1,), (2,), (3,)].
次元数を増やさないステップ出力も同様にインデックスが付けられます。たとえば、上記 4 枚の画像それぞれに対するモデル予測にはインデックス [(0,), (1,), (2,), (3,)].
ブロックが次元を増やす dimensionality_level （たとえば、物体検出モデルの予測に基づく動的クロップ）場合、その出力は別の方法でインデックス付けされます。1 枚目の画像に 2 つ、2 枚目に 1 つ、4 枚目に 3 つのクロップがあるとすると、この出力のインデックスは [(0, 0), (0, 1), (1, 0), (3, 0), (3, 1), (3, 2)].

インデックス付けは、ステップ実行中の入力を揃えるうえで重要です。Execution Engine は、すべてのバッチ指向入力が同期するようにします。たとえば、予測 (3,) 画像と (3,) と対応するクロップのバッチを [(3, 0), (3, 1), (3, 2)] ステップが要求したときに一致させます。

コンパイル中にデータの来歴を整理しておくと、実行が簡単になります。Execution Engine は、動的に作成されたネストされたバッチが同じソースに由来するかどうかを検証する必要はありません。その役割は、ステップ入力を準備する際にインデックスを揃えることです。

追加の考慮事項

入力次元オフセット

Workflow ブロックは、入力の次元数がどのように扱われるかを定義します。Execution Engine が入力の次元数の違いを検出すると、より大きい次元をバッチとして包みます。たとえば、あるブロックが入力画像と動的に切り出された画像の両方を処理する場合、後者はバッチに包まれ、各トップレベルの画像が対応するクロップのバッチとともに処理されるようになります。

深くネストされたバッチは、ステップ実行前に平坦化されます

ブロックがすべての入力を同じ次元レベルで定義している場合、入力バッチのネストがどれだけ深くても、ステップ入力は単一のバッチに平坦化され、出力内のインデックスは Execution Engine によって自動的に維持されます。

条件付き実行

フロー制御ブロックは、特定の条件に基づいてどのステップを実行すべきかを管理します。コンパイル中、これらの条件の影響を受けるステップにはフラグが立てられます。入力を構築する際、フロー制御除外用のマスク（SIMD 向けおよび非SIMD向けの両方）が適用されます。このマスクに基づき、特定の入力要素は次の値に置き換えられます None、空の値を表します。

既定では、ブロックは空の値を受け付けないため、いずれかの None インデックスの (i,) バッチ内にある場合、そのインデックスは処理から除外されます。これが Execution Engine におけるフロー制御の仕組みです。ただし、一部のブロックは空の入力を扱うように設計されています。その場合でもフロー制御マスクは適用されますが、空の入力は入力バッチから除外されません。

バッチ処理モード

ブロックは、バッチ入力を一度にまとめて処理することも、既定では Execution Engine に各入力をループ処理させ、ブロックの run(...) メソッド。

フロー制御ステップの出力の管理

フロー制御ステップの出力は独特です。というのも、これらのステップは、どのデータポイントを後続のステップに渡すべきかを決定するからです。これはおおむね次の擬似コードの結果に似ています:

条件(A) なら:
    step_1(A)
    step_2(A)
else:
    step_3(A)

Workflows Execution Engine は、フロー制御ステップの出力を解析し、実行ブランチを作成します。各ブランチには対応するマスクがあります:

次の場合は SIMD ブランチの場合、マスクには処理対象としてアクティブのまま残るインデックスの集合が含まれます。
次の場合は 非SIMD ブランチの場合、マスクは単純な True / False 値であり、ブランチ全体がアクティブかどうかを決定します。

フロー制御ステップが実行されると、このマスクが登録され、判定の影響を受けるすべてのステップに適用されます。これにより、エンジンは下流ブランチで処理する特定のデータポイントを除外できます。あるデータポイントがブランチの最初のステップから除外された場合（マスキングの結果として）、そのデータポイントはバッチ内の空の入力が既定で除外されるため、ブランチ全体から自動的に取り除かれます。

ステップ出力のキャッシュ

注意深く管理する必要があるのはフロー制御ステップの結果だけではありません。データ処理ステップについても、その結果が他のステップに正しく渡されるように注意が必要です。ここで重要なのは、出力を適切にインデックス付けすることです。

すべての入力が同じ 次元レベル で、出力も同じ次元数を維持する単純なケースでは、Execution Engine の主な役割は入力インデックスの順序を維持することです。ただし、入力の次元数が異なる場合は、ステップの作成に使われた Workflow ブロックが、インデックスの扱いを決定します。

処理中に次元数が変わる場合、Execution Engine は高レベルのインデックスを使うか、出力内の要素リストの長さに基づいてネストされた次元を動的に作成します。これにより、ステップ間でのデータの適切な整合と追跡が保証されます。

Workflow 出力の構築

出力の構築方法の詳細については、 Workflow 定義ページおよび出力の構築セクションを参照してください。

Python で Execution Engine を実行する

Compiler と Execution Engine は次のパッケージに同梱されています。 inference そのため、HTTP API を経由せずに Python アプリケーション内で直接 Workflow を実行できます。これは次のようなチームに適しています:

アプリケーションを Python で保っている
アプリケーションプロセス内でリソースを大量に消費する計算を受け入れられる
HTTP 経由の遅延や失敗モードを避けたい
Workflow 実行を完全に制御したい。

これは最も要求の厳しい統合モードです。Workflow 内のブロックが必要とするすべての初期化パラメータを指定します。それらのパラメータとその解決方法は、ブロックからの Workflow ステップの初期化.

from inference.core.env import WORKFLOWS_MAX_CONCURRENT_STEPS, MAX_ACTIVE_MODELS
from inference.core.managers.base import ModelManager
from inference.core.managers.decorators.fixed_size_cache import WithFixedSizeCache
from inference.core.registries.roboflow import RoboflowModelRegistry
from inference.core.workflows.core_steps.common.entities import StepExecutionMode
from inference.core.workflows.execution_engine.core import ExecutionEngine
from inference.models.utils import ROBOFLOW_MODEL_TYPES

# Roboflow モデルを公開するコアブロックに必要なモデルレジストリ
model_registry = RoboflowModelRegistry(ROBOFLOW_MODEL_TYPES)
model_manager = ModelManager(model_registry=model_registry)
model_manager = WithFixedSizeCache(model_manager, max_size=MAX_ACTIVE_MODELS)

OBJECT_DETECTION_WORKFLOW = {
    "version": "1.0",
    "inputs": [
        {"type": "WorkflowImage", "name": "image"},
        {"type": "WorkflowParameter", "name": "model_id"},
        {"type": "WorkflowParameter", "name": "confidence", "default_value": 0.3},
    ],
    "steps": [
        {
            "type": "RoboflowObjectDetectionModel",
            "name": "detection",
            "image": "$inputs.image",
            "model_id": "$inputs.model_id",
            "confidence": "$inputs.confidence",
        }
    ],
    "outputs": [
        {"type": "JsonField", "name": "result", "selector": "$steps.detection.*"}
    ],
}

# ブロックの初期化パラメータ - 必要なものは Workflow 内のブロックによって異なります
workflow_init_parameters = {
    "workflows_core.model_manager": model_manager,
    "workflows_core.api_key": "<YOUR-API-KEY>",
    "workflows_core.step_execution_mode": StepExecutionMode.LOCAL,
}

# init(...) を呼び出すとコンパイルが実行されます
execution_engine = ExecutionEngine.init(
    workflow_definition=OBJECT_DETECTION_WORKFLOW,
    init_parameters=workflow_init_parameters,
    max_concurrent_steps=WORKFLOWS_MAX_CONCURRENT_STEPS,
)

result = execution_engine.run(
    runtime_parameters={
        "image": [<your-image>],
        "model_id": "rfdetr-small",
    }
)

HTTP、WebRTC ビデオストリーミング、CLI、バッチ処理を含む、Workflow を実行する他の方法については、 Workflowをデプロイする.

前へWorkflowsコンパイラ次へ内部ワークフロー

最終更新 5 日前

役に立ちましたか？