Spaces:
Running
Running
File size: 2,835 Bytes
ce2d64b 9916edb ce2d64b 9916edb ce2d64b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 | from __future__ import annotations
import math
from statistics import mean, median
from typing import Any
from .simulator import run_simulation
SUPPORTED_METRICS = {
"goodput_rps": ("summary", "goodput_rps"),
"request_throughput_rps": ("summary", "request_throughput_rps"),
"slo_attainment": ("summary", "slo_attainment"),
"p95_ttft_ms": ("latency", "ttft_ms", "p95"),
"p99_ttft_ms": ("latency", "ttft_ms", "p99"),
"p95_e2e_ms": ("latency", "e2e_ms", "p95"),
"p99_e2e_ms": ("latency", "e2e_ms", "p99"),
}
def _predicted(result: dict[str, Any], metric: str) -> float:
path = SUPPORTED_METRICS[metric]
value: Any = result
for key in path:
value = value[key]
return float(value)
def validate_cases(cases: list[dict[str, Any]]) -> dict[str, Any]:
"""Compare simulator predictions with externally measured serving cases.
Each case contains a normal SimulationConfig dictionary and a `measured`
mapping. No measured data ships as benchmark truth with InferScale; this
function is the explicit integration point for future empirical validation.
"""
rows: list[dict[str, Any]] = []
absolute_percentage_errors: list[float] = []
for index, case in enumerate(cases):
if "config" not in case or "measured" not in case:
raise ValueError(f"Validation case {index} requires config and measured fields")
result = run_simulation(case["config"])
name = str(case.get("name", f"case-{index + 1}"))
for metric, measured_raw in case["measured"].items():
if metric not in SUPPORTED_METRICS:
raise ValueError(f"Unsupported validation metric: {metric}")
measured = float(measured_raw)
predicted = _predicted(result, metric)
error = predicted - measured
ape = abs(error) / abs(measured) * 100.0 if abs(measured) > 1e-12 else math.nan
if math.isfinite(ape):
absolute_percentage_errors.append(ape)
rows.append(
{
"case": name,
"metric": metric,
"measured": measured,
"predicted": predicted,
"error": error,
"absolute_error": abs(error),
"absolute_percentage_error": ape,
}
)
return {
"case_count": len(cases),
"observation_count": len(rows),
"mape_pct": mean(absolute_percentage_errors) if absolute_percentage_errors else 0.0,
"median_ape_pct": median(absolute_percentage_errors) if absolute_percentage_errors else 0.0,
"max_ape_pct": max(absolute_percentage_errors, default=0.0),
"rows": rows,
"provenance": "external-measurements-vs-analytical-reference",
}
|