#!/usr/bin/env python3 """ calibrate-jeff.py — accuracy, calibration and throughput for jeff (the open-source, self-hostable, jev-API-compatible server, GLiFormer-400M), run LOCALLY against our own server instance. Nobody's numbers but ours. Companion to calibrate-decision-layer.py (which did the same exercise for constrained-token OpenAI models). jeff's API already returns a probability per class directly — no logprob reconstruction, no letter-token indirection, no risk of the exact-match bug that corrupted the earlier OpenAI run. That is a real structural advantage of the jev-shaped API, and it shows up here as "nothing to get wrong," not as a claim about accuracy. 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) Throughput sequential (repeats, one request in flight) AND concurrent (N requests in flight via a thread pool) at several concurrency levels, on THIS machine (Apple M4 Pro, MPS). This is not a cloud GPU. Labelled as such throughout; the EU hosting-cost table scales it against jeff's own published L4 numbers rather than presenting it as a cloud figure. Usage # 1. Start the server first (see the post / jeff's README): # JEFF_API_KEYS=devkey JEFF_MODEL= JEFF_DEVICE=mps jeff # 2. Then: python scripts/calibrate-jeff.py \ --base-url http://localhost:8000/v1/systemone --api-key devkey \ --out public/data/decision-layer-jeff.csv """ from __future__ import annotations import argparse import csv import json import math 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="http://localhost:8000/v1/systemone") ap.add_argument("--api-key", default="devkey") ap.add_argument("--model", default="jev-latest") ap.add_argument("--out", default="public/data/decision-layer-jeff.csv") ap.add_argument("--raw-out", default="public/data/decision-layer-jeff-raw.csv") ap.add_argument("--throughput-out", default="public/data/decision-layer-jeff-throughput.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(["jeff-local", 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("\nthroughput (this machine — Apple M4 Pro, MPS, NOT a cloud GPU):", file=sys.stderr) tp_rows = [] for c in [int(x) for x in args.concurrency_levels.split(",")]: tp = throughput_test(client, args.base_url, args.api_key, args.model, test, c) tp_rows.append(tp) print(f" concurrency={tp['concurrency']:3} {tp['req_per_sec']:7.1f} req/s " f"({tp['n_calls']} calls in {tp['wall_s']}s)", file=sys.stderr) with open(args.throughput_out, "w", newline="") as fh: w = csv.DictWriter(fh, fieldnames=list(tp_rows[0])) w.writeheader() w.writerows(tp_rows) print(f"wrote {args.throughput_out}", file=sys.stderr) return 0 if __name__ == "__main__": raise SystemExit(main())