> 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/benchmarks.md).

# Workflows ベンチマーク

このページでは比較します **直接** [モデル推論](https://docs.roboflow.com/deployment/self-hosted/self-hosted#run-a-model) レイテンシーと **ワークフローでラップされた推論** 人気の検出モデルにおける、直接推論のレイテンシーとワークフローでラップされた推論のレイテンシーを比較します。目的は、Workflows 実行エンジンによって追加されるオーバーヘッドを定量化することです。

## 自己ホスト環境の結果

すべての時間は **ミリ秒**。ベンチマークは、サーバーグレードの NVIDIA GPU 上で TensorRT 最適化モデルを使用して実行され、以下の `inference` Python パッケージです。各モデルは計測前にウォームアップされ、その後 10 回の反復で測定されました。

| モデル           | 直接平均 (ms) | ワークフロー平均 (ms) | ワークフローのオーバーヘッド (ms) |
| ------------- | --------- | ------------- | ------------------- |
| rfdetr-nano   | 2.65      | 4.40          | 1.75                |
| rfdetr-small  | 3.17      | 4.62          | 1.45                |
| rfdetr-medium | 3.79      | 5.38          | 1.59                |
| rfdetr-large  | 4.87      | 7.46          | 2.59                |
| rfdetr-xlarge | 8.44      | 10.65         | 2.21                |
| yolo26n-640   | 2.42      | 4.07          | 1.65                |
| yolo26s-640   | 3.25      | 5.48          | 2.23                |
| yolo26m-640   | 4.56      | 6.29          | 1.73                |
| yolo26l-640   | 5.76      | 8.01          | 2.25                |
| yolo26x-640   | 7.75      | 9.36          | 1.61                |

### 要点

* **ワークフローのオーバーヘッドは最小限です** — 通常、モデルサイズに関係なく、直接推論に 1.5〜2.6 ms 上乗せされる程度です。
* より大きなモデル（例： `rfdetr-xlarge`）では、ワークフローのオーバーヘッドはモデル推論時間に比べて小さく（約 26%）です。
* より小さく高速なモデル（例： `yolo26n-640`）では、オーバーヘッドは相対的に大きくなりますが、絶対値では依然として 2 ms 未満です。
* ワークフローのオーバーヘッドは CPU による処理（グラフのスケジューリング、入力準備、出力ルーティング）であり、モデル推論自体は通常 GPU 上で実行されます（利用可能な場合）。そのため、GPU の速度に関係なくオーバーヘッドは比較的一定に保たれます。

### 方法

* GPU アクセラレーションによる推論（結果はハードウェアによって異なります）。
* ウォームアップ: 計測開始前に、各モデルとワークフローエンジンを 1 回の推論呼び出しでウォームアップします。
* 反復回数: 各メソッド、各モデルにつき 10 回の計測反復。
* 直接推論: 使用するのは `get_model()` で、 `model.infer()` を直接呼び出します。
* ワークフロー推論: 同じモデルを最小限の 1 ステップワークフローにラップし、以下の [実行エンジン](/workflows/ja/gaido/developer-guide/execution-engine.md).

<details>

<summary>クリックしてベンチマークスクリプトを展開</summary>

```python
import os
import statistics
import time
import argparse
import csv
import supervision as sv

from inference import get_model
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

def build_workflow(model_id: str) -> dict:
    """指定されたモデル（検出または分類）用の最小限のワークフロー定義を構築します。"""
    if "classifiers" in model_id or "classification" in model_id:
        step_type = "RoboflowClassificationModel"
    else:
        step_type = "RoboflowObjectDetectionModel"

    return {
        "version": "1.0",
        "inputs": [
            {"type": "WorkflowImage", "name": "image"},
        ],
        "steps": [
            {
                "type": step_type,
                "name": "model_step",
                "image": "$inputs.image",
                "model_id": model_id,
            }
        ],
        "outputs": [
            {
                "type": "JsonField",
                "name": "predictions",
                "selector": "$steps.model_step.predictions",
            },
        ],
    }

def main():
    parser = argparse.ArgumentParser(description="直接推論とワークフロー推論のレイテンシーをベンチマークします。")
    parser.add_argument("--iterations", type=int, default=10, help="各メソッドごとの計測反復回数（既定: 10）")
    args = parser.parse_args()

    # モデル、benchmark.py と同じリスト
    models = [
        "classifiers/3",
        "yolo26n-640", "yolo26s-640", "yolo26m-640", "yolo26l-640", "yolo26x-640",
        "rfdetr-nano", "rfdetr-small", "rfdetr-medium", "rfdetr-large", "rfdetr-xlarge", "rfdetr-2xlarge",
    ]

    # テスト画像を 1 回だけダウンロード
    print("テスト画像をダウンロード中...")
    url = "https://media.roboflow.com/inference/people-walking.jpg"
    image_np = sv.load_image_from_url(url)
    print("画像の準備ができました。")

    # Workflow エンジン用の共有モデルマネージャーを初期化（モデル間で再利用）
    model_registry = RoboflowModelRegistry(ROBOFLOW_MODEL_TYPES)
    model_manager = ModelManager(model_registry=model_registry)
    model_manager = WithFixedSizeCache(model_manager, max_size=MAX_ACTIVE_MODELS)

    workflow_init_parameters = {
        "workflows_core.model_manager": model_manager,
        "workflows_core.step_execution_mode": StepExecutionMode.LOCAL,
    }

    results_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), "benchmark_inference_vs_workflows.csv")
    fieldnames = [
        "model_id",
        "avg_latency_direct_ms", "min_direct_ms", "max_direct_ms", "stddev_direct_ms",
        "avg_latency_workflow_ms", "min_workflow_ms", "max_workflow_ms", "stddev_workflow_ms",
    ]

    with open(results_file, "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()

    print(f"\n各メソッドにつき {args.iterations} 回の反復を {len(models)} 個のモデルで実行します...\n")

    for model_id in models:
        print(f"--- モデル: {model_id} ---")

        # ── 1. 直接推論 ──────────────────────────────────────────────
        direct_latencies = None
        try:
            model = get_model(model_id=model_id)

            # ウォームアップ
            model.infer(image_np)

            direct_latencies = []
            for i in range(args.iterations):
                t0 = time.perf_counter()
                model.infer(image_np)
                t1 = time.perf_counter()
                ms = (t1 - t0) * 1000.0
                direct_latencies.append(ms)
                print(f"  直接   [{i+1:>2}/{args.iterations}]: {ms:.2f} ms")

            avg = statistics.mean(direct_latencies)
            mn  = min(direct_latencies)
            mx  = max(direct_latencies)
            sd  = statistics.stdev(direct_latencies) if len(direct_latencies) > 1 else 0.0
            print(f"  直接   → 平均={avg:.2f}  最小={mn:.2f}  最大={mx:.2f}  標準偏差={sd:.2f} ms")

        except Exception as e:
            print(f"  直接失敗: {e}")

        # ── 2. ワークフロー推論 ────────────────────────────────────────────
        workflow_latencies = None
        try:
            workflow_def = build_workflow(model_id)

            engine = ExecutionEngine.init(
                workflow_definition=workflow_def,
                init_parameters=workflow_init_parameters,
                max_concurrent_steps=WORKFLOWS_MAX_CONCURRENT_STEPS,
            )

            # ウォームアップ
            engine.run(runtime_parameters={"image": [image_np]})

            workflow_latencies = []
            for i in range(args.iterations):
                t0 = time.perf_counter()
                engine.run(runtime_parameters={"image": [image_np]})
                t1 = time.perf_counter()
                ms = (t1 - t0) * 1000.0
                workflow_latencies.append(ms)
                print(f"  ワークフロー [{i+1:>2}/{args.iterations}]: {ms:.2f} ms")

            avg = statistics.mean(workflow_latencies)
            mn  = min(workflow_latencies)
            mx  = max(workflow_latencies)
            sd  = statistics.stdev(workflow_latencies) if len(workflow_latencies) > 1 else 0.0
            print(f"  ワークフロー → 平均={avg:.2f}  最小={mn:.2f}  最大={mx:.2f}  標準偏差={sd:.2f} ms")

        except Exception as e:
            print(f"  ワークフロー失敗: {e}")

        # ── 行を書き込み ────────────────────────────────────────────────────────
        def fmt(vals, fn):
            return round(fn(vals), 2) if vals else "失敗"

        row = {
            "model_id": model_id,
            "avg_latency_direct_ms":  fmt(direct_latencies,   statistics.mean),
            "min_direct_ms":          fmt(direct_latencies,   min),
            "max_direct_ms":          fmt(direct_latencies,   max),
            "stddev_direct_ms":       fmt(direct_latencies,   lambda v: statistics.stdev(v) if len(v) > 1 else 0.0),
            "avg_latency_workflow_ms": fmt(workflow_latencies, statistics.mean),
            "min_workflow_ms":         fmt(workflow_latencies, min),
            "max_workflow_ms":         fmt(workflow_latencies, max),
            "stddev_workflow_ms":      fmt(workflow_latencies, lambda v: statistics.stdev(v) if len(v) > 1 else 0.0),
        }

        with open(results_file, "a", newline="") as f:
            writer = csv.DictWriter(f, fieldnames=fieldnames)
            writer.writerow(row)

        print(f"  保存しました。\n")

    print(f"完了しました！結果: {results_file}")

if __name__ == "__main__":
    main()

```

</details>

## クラウドホストの結果

以下の結果は、ホストされた [Serverless Cloud API](https://docs.roboflow.com/deployment/roboflow-cloud/serverless-api)。いくつかの注意点:

* 自己ホスト結果と比べると、Serverless サーバーはワークフローのスキーマとモデルの重みも取得する必要があります（いっぽう、Serverless のモデル推論は重みのみを取得します）
* Workflows には固定の 10〜50 ms のレイテンシーオーバーヘッドがあります

![Serverless ベンチマーク用ワークフロー](/files/7fa017bd750bd03e1ad5ad9a82ecfef40679306c)