#!/usr/bin/env python3 """ calibrate-jev.py — accuracy and calibration for the REAL, hosted TypeSafe Jev API (not the open-source jeff imitation, not a DIY LLM hack). Same method as calibrate-jeff.py: identical 200 test cases, identical disjoint 200-case holdout, identical temperature-scaling procedure. This is the missing piece both earlier posts explicitly said we did not have. Design (fixed before this ran): Data TEST = data/decision-cases.jsonl (the same 200 used for every other backend in this post — apples to apples) HOLDOUT = data/decision-holdout.jsonl (the same disjoint 200 used for the OpenAI calibration run) Metric accuracy, ECE (10 equal-width bins), confidence-bucket accuracy, then the same after temperature scaling fit on HOLDOUT (T minimises holdout NLL, single parameter, grid search) Latency p50/p95 from the sequential run only. No concurrency/throughput sweep here — this is TypeSafe's shared production service, not ours to load-test; jeff's own local sweep (calibrate-jeff.py) is the throughput data point in this series. Usage python scripts/calibrate-jev.py \ --out public/data/decision-layer-jev.csv """ from __future__ import annotations import argparse import csv import json import math import os import sys import time from collections import defaultdict from concurrent.futures import ThreadPoolExecutor import httpx CLASSES = ["world", "sports", "business", "scitech"] DESCRIPTIONS = { "world": "World news: international affairs, politics, conflicts, diplomacy, governments", "sports": "Sports: games, matches, athletes, teams, leagues, tournaments", "business": "Business: companies, markets, economy, finance, earnings, deals", "scitech": "Science and technology: research, space, computing, internet, software, gadgets", } TASK = "Classify this news article into one topic." def call(client: httpx.Client, base_url: str, api_key: str, model: str, state: str) -> dict: t0 = time.perf_counter() r = client.post( base_url, headers={"Authorization": f"Bearer {api_key}"}, json={ "model": model, "state": state, "questions": {"topic": {"type": "choice", "instructions": TASK, "criteria": DESCRIPTIONS}}, }, timeout=60.0, ) dt = (time.perf_counter() - t0) * 1000 r.raise_for_status() body = r.json() ans = body["answers"]["topic"] usage = body.get("usage", {}) return { "choice": ans["choice"], "confidence": ans["confidence"], "probabilities": ans["probabilities"], # {class: p}, straight from the server "latency_ms": dt, "input_tokens": usage.get("input_tokens", 0), "output_tokens": usage.get("output_tokens", 0), } def softmax_T(probs: dict, T: float) -> dict: # Re-derive logits from the returned probabilities, rescale by T, re-normalise. # jeff doesn't expose raw logits, so this is the only way to apply temperature # scaling post-hoc; it is exact when the reported probabilities are themselves # already a softmax (true for this API). logp = {c: math.log(max(probs.get(c, 1e-9), 1e-9)) for c in CLASSES} z = {c: logp[c] / T for c in CLASSES} m = max(z.values()) e = {c: math.exp(v - m) for c, v in z.items()} s = sum(e.values()) return {c: v / s for c, v in e.items()} def fit_temperature(rows: list[dict]) -> float: def nll(T: float) -> float: total = 0.0 for r in rows: p = softmax_T(r["probabilities"], T) total -= math.log(max(p[r["label"]], 1e-12)) return total / len(rows) grid = [0.25 * 1.06 ** i for i in range(90)] return min(grid, key=nll) def evaluate(rows: list[dict], T: float) -> dict: preds = [] for r in rows: p = softmax_T(r["probabilities"], T) if T != 1.0 else r["probabilities"] best = max(CLASSES, key=lambda c: p[c]) preds.append((p[best], best == r["label"], best, r["label"])) n = len(preds) bins = defaultdict(list) for conf, ok, *_ in preds: bins[min(int(conf * 10), 9)].append((conf, ok)) ece = sum(len(b) / n * abs(sum(o for _, o in b) / len(b) - sum(c for c, _ in b) / len(b)) for b in bins.values()) hi = [ok for conf, ok, *_ in preds if conf >= 0.9] lo = [ok for conf, ok, *_ in preds if conf < 0.8] recall = {c: sum(ok for _, ok, _, lab in preds if lab == c) / max(1, sum(1 for *_, lab in preds if lab == c)) for c in CLASSES} return { "n": n, "accuracy": sum(ok for _, ok, *_ in preds) / n, "ece": ece, "mean_conf": sum(c for c, *_ in preds) / n, "hi_n": len(hi), "hi_acc": (sum(hi) / len(hi)) if hi else float("nan"), "lo_n": len(lo), "lo_acc": (sum(lo) / len(lo)) if lo else float("nan"), **{f"recall_{c}": recall[c] for c in CLASSES}, } def throughput_test(client: httpx.Client, base_url, api_key, model, cases, concurrency) -> dict: """Fire `concurrency` requests at once, repeated to cover ~40 calls, measure aggregate requests/sec. Sequential concurrency=1 is comparable to the p50 latency numbers used elsewhere in this post.""" n_calls = max(concurrency * 5, 20) pool_cases = (cases * ((n_calls // len(cases)) + 1))[:n_calls] t0 = time.perf_counter() with ThreadPoolExecutor(concurrency) as pool: list(pool.map(lambda c: call(client, base_url, api_key, model, c["state"]), pool_cases)) dt = time.perf_counter() - t0 return {"concurrency": concurrency, "n_calls": n_calls, "wall_s": round(dt, 2), "req_per_sec": round(n_calls / dt, 2)} def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--test", default="data/decision-cases.jsonl") ap.add_argument("--holdout", default="data/decision-holdout.jsonl") ap.add_argument("--base-url", default="https://api.typesafe.ai/v1/systemone") ap.add_argument("--api-key", default=os.environ.get("TYPESAFE_API_KEY", "")) ap.add_argument("--model", default="jev-latest") ap.add_argument("--out", default="public/data/decision-layer-jev.csv") ap.add_argument("--raw-out", default="public/data/decision-layer-jev-raw.csv") ap.add_argument("--concurrency-levels", default="1,2,4,8,16") args = ap.parse_args() test = [json.loads(line) for line in open(args.test)] holdout = [json.loads(line) for line in open(args.holdout)] assert not ({c["id"] for c in test} & {c["id"] for c in holdout}), "holdout overlaps test" print(f"test={len(test)} holdout={len(holdout)} (disjoint) — server: {args.base_url}", file=sys.stderr) client = httpx.Client() # Warm the server (first call after load carries a one-off JIT/kernel-compile # cost on MPS; excluded from every timing below). call(client, args.base_url, args.api_key, args.model, test[0]["state"]) def run_split(cases, label): t0 = time.time() with ThreadPoolExecutor(4) as pool: out = list(pool.map(lambda c: {**call(client, args.base_url, args.api_key, args.model, c["state"]), "id": c["id"], "label": c["label"]}, cases)) print(f"{label}: {len(out)} calls in {time.time()-t0:.0f}s", file=sys.stderr) return out t_rows = run_split(test, "test") h_rows = run_split(holdout, "holdout") T = fit_temperature(h_rows) before, after = evaluate(t_rows, 1.0), evaluate(t_rows, T) acc_only = sum(r["choice"] == r["label"] for r in t_rows) / len(t_rows) lat = sorted(r["latency_ms"] for r in t_rows) p50, p95 = lat[len(lat) // 2], lat[int(len(lat) * 0.95)] mean_in = sum(r["input_tokens"] for r in t_rows) / len(t_rows) print(f"\naccuracy={acc_only:.3f} p50={p50:.1f}ms p95={p95:.1f}ms " f"ECE {before['ece']:.3f} -> {after['ece']:.3f} (T={T:.2f}) " f"hi_conf_acc={before['hi_acc']:.3f}(n={before['hi_n']}) " f"lo_conf_acc={before['lo_acc']:.3f}(n={before['lo_n']}) " f"mean_input_tokens={mean_in:.1f}", file=sys.stderr) with open(args.out, "w", newline="") as fh: w = csv.writer(fh) w.writerow(["backend", "n", "accuracy", "p50_ms", "p95_ms", "mean_input_tokens", "mean_output_tokens", "stage", "temperature", "ece", "mean_conf", "hi_n", "hi_acc", "lo_n", "lo_acc", "recall_world", "recall_sports", "recall_business", "recall_scitech"]) for stage, ev, T_val in (("raw", before, 1.0), ("temperature_scaled", after, T)): w.writerow(["jev", ev["n"], round(acc_only, 4), round(p50, 1), round(p95, 1), round(mean_in, 1), round(sum(r["output_tokens"] for r in t_rows) / len(t_rows), 1), stage, round(T_val, 3), round(ev["ece"], 4), round(ev["mean_conf"], 4), ev["hi_n"], round(ev["hi_acc"], 4), ev["lo_n"], round(ev["lo_acc"], 4), round(ev["recall_world"], 4), round(ev["recall_sports"], 4), round(ev["recall_business"], 4), round(ev["recall_scitech"], 4)]) print(f"wrote {args.out}", file=sys.stderr) with open(args.raw_out, "w", newline="") as fh: w = csv.writer(fh) w.writerow(["split", "id", "label", "predicted", "correct", "confidence", "latency_ms", "input_tokens", "output_tokens"] + [f"p_{c}" for c in CLASSES]) for split, rows in (("test", t_rows), ("holdout", h_rows)): for r in rows: w.writerow([split, r["id"], r["label"], r["choice"], r["choice"] == r["label"], round(r["confidence"], 6), round(r["latency_ms"], 1), r["input_tokens"], r["output_tokens"]] + [round(r["probabilities"].get(c, 0.0), 6) for c in CLASSES]) print(f"wrote {args.raw_out}", file=sys.stderr) print("\nSkipping the concurrency/throughput sweep against the real hosted " "API: that would load-test TypeSafe's shared infrastructure with our " "own synthetic traffic, which is a different thing to publish than " "jeff's own local throughput. p50/p95 from the sequential run above " "is in the summary CSV.", file=sys.stderr) return 0 if __name__ == "__main__": raise SystemExit(main())