Workflows 벤치마크
워크플로에 래핑된 동일한 모델과 비교한 직접 모델 추론의 지연 시간입니다.
마지막 업데이트
도움이 되었나요?
워크플로에 래핑된 동일한 모델과 비교한 직접 모델 추론의 지연 시간입니다.
마지막 업데이트
도움이 되었나요?
도움이 되었나요?
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",
]
# 테스트 이미지를 한 번만 다운로드
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" 직접 [{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 "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()