> 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 実行エンジンによって導入されるオーバーヘッドを定量化することです。

## セルフホスト結果

すべての時間は **ミリ秒**単位です。ベンチマークは、TensorRT 最適化モデルを使用し、サーバーグレードの NVIDIA GPU 上で実行されました。モデルは `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()` を直接呼び出します。
* ワークフロー推論：同じモデルを最小構成の単一ステップワークフローでラップし、 [実行エンジン](/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("画像の準備ができました。")

    # ワークフローエンジン用の共有モデルマネージャーを初期化（モデル間で再利用）
    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{len(models)} 個のモデルに対して、各手法 {args.iterations} 回ずつ実行します...\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"  Direct   [{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"  Direct   → avg={avg:.2f}  min={mn:.2f}  max={mx:.2f}  stddev={sd:.2f} ms")

        except Exception as e:
            print(f"  Direct FAILED: {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"  Workflow [{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"  Workflow → avg={avg:.2f}  min={mn:.2f}  max={mx:.2f}  stddev={sd:.2f} ms")

        except Exception as e:
            print(f"  Workflow FAILED: {e}")

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

        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>

## クラウドホスト結果

以下の結果は、ホストされた [サーバーレス API](https://docs.roboflow.com/deployment/roboflow-cloud/serverless-api)上でベンチマークされました。注意点:

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

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