| """Evaluate frozen decisions against a running, untrained-model adapter. |
| |
| No retries, generation, calibration, truncation, or training. Latency includes |
| adapter HTTP, tokenization, engine scheduling and forward work; it is NOT CUDA |
| kernel time. Model load and warmup are excluded. Failed runs have no headline. |
| """ |
|
|
| import argparse |
| import asyncio |
| import hashlib |
| import json |
| import math |
| import os |
| import platform |
| import random |
| import sys |
| import time |
| from collections import defaultdict |
| from datetime import UTC, datetime |
| from pathlib import Path |
|
|
| import httpx |
|
|
| from jev_adapter.protocol import SystemOneRequest |
|
|
| from .metrics import latency_summary, prediction_rows, summarize |
|
|
|
|
| def digest(data): |
| return hashlib.sha256(data).hexdigest() |
|
|
|
|
| def write_json(path, value): |
| path.write_text( |
| json.dumps(value, ensure_ascii=False, indent=2, allow_nan=False) + "\n" |
| ) |
|
|
|
|
| def load_data(paths, model, allow_test=False, limit=None): |
| items, seen, sources = [], set(), [] |
| for path in paths: |
| raw = path.read_bytes() |
| loaded = [json.loads(line) for line in raw.splitlines() if line.strip()] |
| source = { |
| "path": str(path.resolve()), |
| "sha256": digest(raw), |
| "records": len(loaded), |
| } |
| manifest_path = path.with_suffix(".manifest.json") |
| if manifest_path.exists(): |
| prepared = json.loads(manifest_path.read_text()) |
| if prepared.get("data_sha256") != digest(raw): |
| raise ValueError(f"prepared dataset checksum mismatch: {path}") |
| if prepared.get("selected", {}).get("records") != len(loaded): |
| raise ValueError(f"prepared dataset count mismatch: {path}") |
| source["prepared_manifest"] = prepared |
| sources.append(source) |
| for item in loaded: |
| key = item["suite"], item["split"], item["id"] |
| if key in seen: |
| raise ValueError(f"duplicate input record: {key}") |
| seen.add(key) |
| if item["split"] == "test" and not allow_test: |
| raise ValueError( |
| "test data requires --allow-test; use development first" |
| ) |
| request = SystemOneRequest.model_validate( |
| {**item["record"], "model": model} |
| ) |
| if set(request.questions) != set(item["expected"]): |
| raise ValueError(f"question/label mismatch: {key}") |
| for qid, expected in item["expected"].items(): |
| labels, target, label = ( |
| expected["labels"], |
| expected["target"], |
| expected["label"], |
| ) |
| if ( |
| len(labels) < 2 |
| or len(labels) != len(set(labels)) |
| or type(label) is not int |
| or not 0 <= label < len(labels) |
| or target != [int(i == label) for i in range(len(labels))] |
| ): |
| raise ValueError( |
| "this runner requires valid one-hot objective labels" |
| ) |
| question = request.questions[qid] |
| actual = ( |
| list(question.criteria) |
| if question.type == "choice" |
| else ["false", "true"] |
| if question.type == "noul" |
| else [str(i) for i in range(len(question.criteria))] |
| ) |
| if set(labels) != set(actual) or expected["type"] != question.type: |
| raise ValueError( |
| f"expected labels/type differ from request: {key}/{qid}" |
| ) |
| reference = expected.get("reference_probs") |
| if reference is not None and ( |
| not isinstance(reference, list) |
| or len(reference) != len(labels) |
| or any( |
| type(p) not in (int, float) |
| or not math.isfinite(p) |
| or not 0 <= p <= 1 |
| for p in reference |
| ) |
| or abs(math.fsum(reference) - 1) > 1e-8 |
| ): |
| raise ValueError("invalid reference probability distribution") |
| items.append(item) |
| if limit is not None: |
| items = items[:limit] |
| if not items: |
| raise ValueError("no input records") |
| return items, sources |
|
|
|
|
| async def engine_snapshot(client, base_url, cache_mode): |
| headers = {} |
| if key := os.environ.get("SGLANG_API_KEY"): |
| headers["Authorization"] = f"Bearer {key}" |
| results = {} |
| allowed = { |
| "model_path", |
| "model_type", |
| "architectures", |
| "served_model_name", |
| "revision", |
| "tokenizer_revision", |
| "dtype", |
| "quantization", |
| "tp_size", |
| "tp", |
| "version", |
| "context_length", |
| "max_total_tokens", |
| "max_running_requests", |
| "chunked_prefill_size", |
| "mem_fraction_static", |
| "disable_radix_cache", |
| "mm_preprocess_cache_size_mb", |
| "enable_prefix_mm_cache", |
| "enable_mm_global_cache", |
| "speculative_algorithm", |
| "is_generation", |
| "has_image_understanding", |
| "attention_backend", |
| "moe_runner_backend", |
| } |
| for name in ("model_info", "server_info"): |
| response = await client.get(base_url.rstrip("/") + "/" + name, headers=headers) |
| if response.status_code == 404: |
| response = await client.get( |
| base_url.rstrip("/") + "/get_" + name, headers=headers |
| ) |
| response.raise_for_status() |
| value = response.json() |
| if not isinstance(value, dict): |
| raise TypeError(f"invalid engine {name}") |
| results[name] = {k: v for k, v in value.items() if k in allowed} |
| config = results["server_info"] |
| if config.get("speculative_algorithm") is not None: |
| raise ValueError("disable speculative decoding for this benchmark") |
| if cache_mode == "full-prefill": |
| if config.get("disable_radix_cache") is not True: |
| raise ValueError("full-prefill requires engine --disable-radix-cache") |
| if config.get("mm_preprocess_cache_size_mb") != 0: |
| raise ValueError("full-prefill requires --mm-preprocess-cache-size-mb 0") |
| if config.get("enable_prefix_mm_cache") or config.get("enable_mm_global_cache"): |
| raise ValueError("full-prefill requires multimodal feature caches disabled") |
| return results |
|
|
|
|
| def checked_usage(response): |
| usage, meta = response.get("usage", {}), response.get("metadata", {}) |
| if type(usage.get("output_tokens")) is not int or usage["output_tokens"] != 0: |
| raise ValueError("engine must return exactly zero output tokens") |
| if type(usage.get("input_tokens")) is not int or usage["input_tokens"] < 1: |
| raise ValueError("missing or invalid input-token accounting") |
| if type(meta.get("evaluations")) is not int or meta["evaluations"] < 1: |
| raise ValueError("missing forward evaluation count") |
| elapsed = meta.get("adapter_elapsed_ms") |
| if type(elapsed) not in (int, float) or not math.isfinite(elapsed) or elapsed < 0: |
| raise ValueError("missing or invalid adapter elapsed time") |
| if not isinstance(response.get("model"), str) or not response["model"]: |
| raise ValueError("missing served model identity") |
| return usage, meta |
|
|
|
|
| def validate_launch(snapshot, launch): |
| model, engine = launch["model"], launch["engine"] |
| server, info = snapshot["server_info"], snapshot["model_info"] |
| for field, expected in { |
| "model_path": model["repo_id"], |
| "revision": model["revision"], |
| "dtype": model["dtype"], |
| "quantization": model["quantization"], |
| }.items(): |
| if server.get(field) != expected: |
| raise ValueError(f"launch manifest {field} differs from running engine") |
| if info.get("model_path") != model["repo_id"]: |
| raise ValueError("launch manifest model differs from running engine model_info") |
| if not engine.get("revision") or not launch.get("profile"): |
| raise ValueError("launch manifest lacks pinned engine/profile identity") |
|
|
|
|
| async def evaluate(args, client=None): |
| items, inputs = load_data(args.data, args.model, args.allow_test, args.limit) |
| args.output.mkdir(parents=True, exist_ok=False) |
| manifest = { |
| "created_at": datetime.now(UTC).isoformat(), |
| "status": "started", |
| "python": sys.version, |
| "platform": platform.platform(), |
| "requested_model": args.model, |
| "base_url": args.base_url, |
| "engine_url": args.engine_url, |
| "data": inputs, |
| "limit": args.limit, |
| "concurrency": args.concurrency, |
| "warmup_requests": args.warmup, |
| "repeats": args.repeats, |
| "seed": args.seed, |
| "cache_mode": args.cache_mode, |
| "options": {"temperature": 1.0, "permutations": 1, "enable_thinking": False}, |
| "latency_scope": "client HTTP wall time, excluding queue before dispatch; NOT GPU kernel time", |
| "quality_repeats": "first measured repetition only", |
| "assistant_prefix": args.assistant_prefix, |
| "code_sha256": { |
| str(p.relative_to(Path(__file__).parents[1])): digest(p.read_bytes()) |
| for p in Path(__file__).parents[1].rglob("*.py") |
| }, |
| } |
| write_json(args.output / "manifest.json", manifest) |
| owned = client is None |
| if owned: |
| client = httpx.AsyncClient( |
| timeout=args.timeout, limits=httpx.Limits(max_connections=args.concurrency) |
| ) |
| headers = {} |
| if key := os.environ.get("JEV_API_KEY"): |
| headers["Authorization"] = f"Bearer {key}" |
| results, failures = [], [] |
| try: |
| manifest["engine"] = await engine_snapshot( |
| client, args.engine_url, args.cache_mode |
| ) |
| expected_served = manifest["engine"]["model_info"].get("served_model_name") |
| if not expected_served: |
| raise ValueError("engine did not identify its served model") |
| if args.engine_manifest: |
| manifest["launch_manifest"] = json.loads(args.engine_manifest.read_text()) |
| validate_launch(manifest["engine"], manifest["launch_manifest"]) |
| write_json(args.output / "manifest.json", manifest) |
|
|
| async def call(item): |
| request = { |
| **item["record"], |
| "model": args.model, |
| "options": { |
| "temperature": 1.0, |
| "permutations": 1, |
| "return_logprobs": True, |
| }, |
| } |
| if args.assistant_prefix is not None: |
| request["options"]["assistant_prefix"] = args.assistant_prefix |
| started = time.perf_counter() |
| response = await client.post( |
| args.base_url.rstrip("/") + "/v1/systemone", |
| headers=headers, |
| json=request, |
| ) |
| response.raise_for_status() |
| body = response.json() |
| elapsed = (time.perf_counter() - started) * 1000 |
| usage, meta = checked_usage(body) |
| if body["model"] != expected_served: |
| raise ValueError("adapter response model differs from engine snapshot") |
| if meta["evaluations"] != len(item["record"]["questions"]): |
| raise ValueError("expected one forward evaluation per question") |
| rows = prediction_rows(item, body) |
| return { |
| "request_sha256": digest(json.dumps(request, sort_keys=True).encode()), |
| "id": item["id"], |
| "suite": item["suite"], |
| "split": item["split"], |
| "latency_ms": elapsed, |
| "adapter_ms": meta["adapter_elapsed_ms"], |
| "input_tokens": usage["input_tokens"], |
| "evaluations": meta["evaluations"], |
| "served_model": body["model"], |
| "response": body, |
| "rows": rows, |
| } |
|
|
| for i in range(args.warmup): |
| |
| item = items[(i * len(items) // max(args.warmup, 1)) % len(items)] |
| print( |
| f"Warmup {i + 1}/{args.warmup}: {item['suite']} {item['id']}", |
| flush=True, |
| ) |
| await call(item) |
| jobs = [ |
| (i, repeat) for repeat in range(args.repeats) for i in range(len(items)) |
| ] |
| random.Random(args.seed).shuffle(jobs) |
| semaphore = asyncio.Semaphore(args.concurrency) |
| started = time.perf_counter() |
| with (args.output / "predictions.jsonl").open("w") as handle: |
|
|
| async def job(index, repeat): |
| async with semaphore: |
| try: |
| result = await call(items[index]) |
| result.update(index=index, repeat=repeat) |
| results.append(result) |
| handle.write( |
| json.dumps(result, ensure_ascii=False, allow_nan=False) |
| + "\n" |
| ) |
| handle.flush() |
| except (httpx.HTTPError, ValueError, KeyError, TypeError) as error: |
| failures.append( |
| { |
| "index": index, |
| "repeat": repeat, |
| "id": items[index]["id"], |
| "suite": items[index]["suite"], |
| "error_type": type(error).__name__, |
| "error": str(error), |
| } |
| ) |
| done = len(results) + len(failures) |
| if done % 50 == 0 or done == len(jobs): |
| print( |
| f"{done}/{len(jobs)} requests; errors={len(failures)}", |
| flush=True, |
| ) |
|
|
| await asyncio.gather(*(job(index, repeat) for index, repeat in jobs)) |
| elapsed = time.perf_counter() - started |
| served = {r["served_model"] for r in results} |
| if len(served) > 1: |
| failures.append({"error": "served model changed during run"}) |
| report = { |
| "status": "failed" if failures else "complete", |
| "requested_records": len(items), |
| "requested_questions": sum(len(i["expected"]) for i in items), |
| "attempted_requests": len(jobs), |
| "successful_requests": len(results), |
| "errors": len(failures), |
| "served_models": sorted(served), |
| "elapsed_measurement_s": elapsed, |
| "requests_per_second": len(results) / elapsed, |
| "questions_per_second": sum(r["evaluations"] for r in results) / elapsed, |
| "latency_ms": latency_summary([r["latency_ms"] for r in results]), |
| "adapter_ms": latency_summary([r["adapter_ms"] for r in results]), |
| "input_tokens_per_request": latency_summary( |
| [r["input_tokens"] for r in results] |
| ), |
| "suites": {}, |
| "latency_scope": manifest["latency_scope"], |
| "partial_dataset": args.limit is not None |
| or any( |
| not source.get("prepared_manifest", {}) |
| .get("selection", {}) |
| .get("is_full_partition", False) |
| for source in inputs |
| ), |
| } |
| if not failures: |
| grouped, times = defaultdict(list), defaultdict(list) |
| for result in sorted(results, key=lambda r: (r["index"], r["repeat"])): |
| key = result["suite"] + "/" + result["split"] |
| times[key].append(result["latency_ms"]) |
| if result["repeat"] == 0: |
| grouped[key].extend(result["rows"]) |
| report["suites"] = { |
| key: {**summarize(rows), "latency_ms": latency_summary(times[key])} |
| for key, rows in grouped.items() |
| } |
| write_json(args.output / "report.json", report) |
| if failures: |
| write_json(args.output / "failures.json", failures) |
| manifest.update(status=report["status"], served_models=sorted(served)) |
| write_json(args.output / "manifest.json", manifest) |
| return report |
| except BaseException as error: |
| manifest.update( |
| status="failed", failure_type=type(error).__name__, failure=str(error) |
| ) |
| write_json(args.output / "manifest.json", manifest) |
| raise |
| finally: |
| if owned: |
| await client.aclose() |
|
|
|
|
| def main(): |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument("--data", type=Path, action="append", required=True) |
| parser.add_argument("--output", type=Path, required=True) |
| parser.add_argument("--base-url", default="http://127.0.0.1:30120") |
| parser.add_argument("--engine-url", default="http://127.0.0.1:30000") |
| parser.add_argument("--engine-manifest", type=Path) |
| parser.add_argument("--model", default="decision-model") |
| parser.add_argument("--concurrency", type=int, default=1) |
| parser.add_argument("--warmup", type=int, default=20) |
| parser.add_argument("--repeats", type=int, default=1) |
| parser.add_argument("--seed", type=int, default=42) |
| parser.add_argument("--limit", type=int) |
| parser.add_argument("--timeout", type=float, default=120) |
| parser.add_argument("--assistant-prefix") |
| parser.add_argument("--allow-test", action="store_true") |
| parser.add_argument( |
| "--cache-mode", |
| choices=["full-prefill", "server-default"], |
| default="full-prefill", |
| ) |
| args = parser.parse_args() |
| if ( |
| args.concurrency < 1 |
| or args.repeats < 1 |
| or args.warmup < 0 |
| or (args.limit is not None and args.limit < 1) |
| or not math.isfinite(args.timeout) |
| or args.timeout <= 0 |
| ): |
| parser.error("counts and timeout must be positive; warmup may be zero") |
| report = asyncio.run(evaluate(args)) |
| print(json.dumps(report, ensure_ascii=False, indent=2)) |
| raise SystemExit(0 if report["status"] == "complete" else 1) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
