Workflows benchmarks
Latency of direct model inference compared with the same model wrapped in a Workflow.
Last updated
Was this helpful?
Latency of direct model inference compared with the same model wrapped in a Workflow.
Last updated
Was this helpful?
Was this helpful?
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:
"""Build a minimal workflow definition for a given model (detection or classification)."""
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="Benchmark direct inference vs. Workflow inference latency.")
parser.add_argument("--iterations", type=int, default=10, help="Number of timed iterations per method (default: 10)")
args = parser.parse_args()
# Models, same list as 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",
]
# Download the test image once
print("Downloading test image...")
url = "https://media.roboflow.com/inference/people-walking.jpg"
image_np = sv.load_image_from_url(url)
print("Image ready.")
# Initialize shared model manager for Workflow engine (reused across models)
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"\nRunning {args.iterations} iterations per method for {len(models)} models...\n")
for model_id in models:
print(f"--- Model: {model_id} ---")
# ── 1. Direct inference ──────────────────────────────────────────────
direct_latencies = None
try:
model = get_model(model_id=model_id)
# warmup
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 inference ────────────────────────────────────────────
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,
)
# warmup
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}")
# ── Write row ────────────────────────────────────────────────────────
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" Saved.\n")
print(f"All done! Results: {results_file}")
if __name__ == "__main__":
main()