Skip to content

Commit 027c662

Browse files
authored
feat(classifier): serve int8 encoder, size ORT threads, add triage benchmark (#62)
- trainer: quantize the classification encoder to int8 (dynamic, per-channel) after export and score the eval set with the int8 graph, so the report reflects the shipped model. Scoring and embedding encoders stay fp32. encoder.json records the precision. - serving: size ONNX Runtime's intra-op threads to the pod's CPU request (ORT_INTRA_OP_THREADS) with spin-waiting off. The default spun one thread per visible core, so a 2-vCPU pod kept all 32 cores of a c9g.8xlarge busy. Add GET /stats (process CPU seconds, requests served) for benchmarking. - demo: add `make bench-triage`, an in-cluster Job that measures throughput, latency and CPU per request at several concurrency levels, prices it at the node's on-demand rate, and compares it with the frontier model's cost from `TRIAGE=llm make triage-acc`, which now prints the per-call cost.
1 parent ebf390e commit 027c662

11 files changed

Lines changed: 506 additions & 17 deletions

File tree

‎docs/deep-dive/serving.md‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,8 +17,12 @@ How a model is served depends on its `project.task`:
1717
A lean CPU pod (onnxruntime + tokenizers, **no torch**) downloads the project's
1818
`encoder.onnx`, `tokenizer.json`, and `head.json` from S3, embeds the query
1919
with ONNX Runtime (CLS pooling + L2 normalize), applies the logistic head, and
20-
returns a label. The encoder is exported to ONNX by the training job, so the
21-
embeddings match training exactly and serving carries no heavyweight ML deps.
20+
returns a label. The encoder is exported to ONNX (int8 for classification) by the training job, so serving carries no heavyweight
21+
ML deps. ONNX Runtime's thread pool is sized to the pod's CPU request
22+
(`ORT_INTRA_OP_THREADS`) with spin-waiting off; its default of one spinning
23+
thread per visible core lets a 2-vCPU pod keep every core of the node busy.
24+
`GET /stats` returns the process CPU seconds and requests served, so a
25+
benchmark can price the pod on the CPU it actually used.
2226
- **Scoring** (`task: scoring`) → **the same ONNX encoder pod with a regression
2327
head**. The pod embeds the query identically, then computes `weights · vector +
2428
intercept` and clamps the result to `[0,1]`, returning a single number instead

‎docs/deep-dive/training.md‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,11 @@ head over those vectors: a logistic head (`embedding_dim × num_classes` weights
2929
for classification, or a ridge regression head (`embedding_dim` weights + an
3030
intercept) for scoring. There is no backpropagation through the encoder, no
3131
epochs, and no GPU. The same job then exports the frozen encoder to ONNX so
32-
serving needs neither torch nor sentence-transformers. The expensive, GPU-hungry
32+
serving needs neither torch nor sentence-transformers. For `classification` the
33+
exported encoder is quantized to int8 (dynamic, per-channel weights), which runs
34+
on the CPU's native int8 matrix instructions (Arm i8mm, x86 VNNI), and the eval
35+
metrics are computed with that int8 encoder. Scoring and embedding encoders stay
36+
fp32, because their output is a number or a vector rather than a label. The expensive, GPU-hungry
3337
work (learning language) was already done once when the encoder was pretrained;
3438
every encoder-head task rides on top of it cheaply.
3539

‎examples/k8s-autoscaling/demo/Makefile‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@
33

44
SCRIPTS := ./scripts
55

6-
.PHONY: demo reset clean infra eval eval-incluster scoreboard scoreboard-save triage-acc recall index-tuned check-infra swap forward forward-check
6+
.PHONY: demo reset clean infra eval eval-incluster scoreboard scoreboard-save triage-acc bench-triage recall index-tuned check-infra swap forward forward-check
77

88
# One command to start a demo session: reset the scenarios to their broken
99
# state, then port-forward the orchestrator, open the browser, and launch the
@@ -46,6 +46,13 @@ eval-incluster:
4646
triage-acc:
4747
python3 eval/triage_acc.py
4848

49+
# Price-performance of the triage classifier on CPU: throughput, latency, the CPU
50+
# it used per message, and that CPU priced at the node's on-demand rate. Runs as a
51+
# Job next to the model (no port-forward); about a minute. Compares against the
52+
# frontier model when a `TRIAGE=llm make triage-acc` result is in eval/results/.
53+
bench-triage:
54+
bash scripts/run-bench-incluster.sh triage
55+
4956
recall:
5057
python3 eval/recall.py $(if $(MODE),--mode $(MODE),)
5158

‎examples/k8s-autoscaling/demo/eval/.gitignore‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,3 +3,4 @@
33
results/scorecard-*.json
44
results/triage-*.json
55
results/recall-*.json
6+
results/bench-*.json
Lines changed: 145 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,145 @@
1+
#!/usr/bin/env python3
2+
"""Price-performance benchmark for one encoder model served on CPU.
3+
4+
Sends the held-out messages to the model's chat endpoint from N concurrent
5+
clients (closed loop: each client sends the next message when the last one
6+
returns), at several concurrency levels. Before and after each level it reads
7+
the model pod's own /stats (process CPU seconds and requests served), so the
8+
cost is priced on the CPU the model actually used, not on what it requested:
9+
10+
CPU ms per request = CPU seconds used / requests served
11+
$ per 1M requests = CPU seconds per request x node $/hour / node vCPUs / 3600 x 1e6
12+
13+
Runs in-cluster (scripts/run-bench-incluster.sh), next to the model, so a
14+
port-forward is not part of what is measured.
15+
"""
16+
import argparse
17+
import json
18+
import statistics
19+
import threading
20+
import time
21+
22+
import httpx
23+
24+
25+
def pct(xs, p):
26+
xs = sorted(xs)
27+
return xs[min(len(xs) - 1, max(0, round(p * (len(xs) - 1))))] if xs else 0.0
28+
29+
30+
def stats(client, base):
31+
return client.get(f"{base}/stats", timeout=10).json()
32+
33+
34+
def body(q):
35+
return {"model": "model", "messages": [{"role": "user", "content": q}],
36+
"max_tokens": 32, "temperature": 0.1}
37+
38+
39+
def run_level(base, queries, concurrency, duration):
40+
lat, errors, lock, nxt = [], [0], threading.Lock(), [0]
41+
stop_at = time.perf_counter() + duration
42+
43+
def worker():
44+
with httpx.Client(timeout=30) as c:
45+
while time.perf_counter() < stop_at:
46+
with lock:
47+
q = queries[nxt[0] % len(queries)]
48+
nxt[0] += 1
49+
t0 = time.perf_counter()
50+
try:
51+
ok = c.post(f"{base}/v1/chat/completions", json=body(q)).status_code == 200
52+
except Exception:
53+
ok = False
54+
ms = (time.perf_counter() - t0) * 1000
55+
with lock:
56+
if ok:
57+
lat.append(ms)
58+
else:
59+
errors[0] += 1
60+
61+
with httpx.Client() as c:
62+
s0 = stats(c, base)
63+
t0 = time.perf_counter()
64+
threads = [threading.Thread(target=worker) for _ in range(concurrency)]
65+
for t in threads:
66+
t.start()
67+
for t in threads:
68+
t.join()
69+
wall = time.perf_counter() - t0
70+
s1 = stats(c, base)
71+
served = max(1, s1["requests"] - s0["requests"])
72+
cpu_s = s1["cpu_seconds"] - s0["cpu_seconds"]
73+
return {"concurrency": concurrency, "requests": len(lat), "errors": errors[0],
74+
"rps": len(lat) / wall, "p50_ms": pct(lat, 0.5), "p95_ms": pct(lat, 0.95),
75+
"cores_busy": cpu_s / wall, "cpu_ms_per_req": cpu_s * 1000 / served,
76+
"mean_ms": statistics.fmean(lat) if lat else 0.0}
77+
78+
79+
def main():
80+
ap = argparse.ArgumentParser()
81+
ap.add_argument("--url", required=True, help="model base URL (no path)")
82+
ap.add_argument("--name", default="model")
83+
ap.add_argument("--queries", required=True, help="JSON list of messages")
84+
ap.add_argument("--levels", default="1,2,4,8")
85+
ap.add_argument("--duration", type=float, default=10)
86+
ap.add_argument("--instance-type", default="")
87+
ap.add_argument("--vcpus", type=float, default=0)
88+
ap.add_argument("--usd-per-hour", type=float, default=0)
89+
ap.add_argument("--cpu-request", type=float, default=0, help="the pod's CPU request")
90+
args = ap.parse_args()
91+
92+
queries = json.load(open(args.queries))
93+
with httpx.Client() as c:
94+
info = stats(c, args.url)
95+
for q in queries[:20]: # warm the inference path
96+
c.post(f"{args.url}/v1/chat/completions", json=body(q), timeout=30)
97+
98+
usd_per_vcpu_s = (args.usd_per_hour / args.vcpus / 3600) if args.usd_per_hour and args.vcpus else 0
99+
print(f"Model: {args.name} precision {info.get('precision') or '?'}, "
100+
f"{info.get('intra_op_threads') or 'default'} inference threads, CPU request {args.cpu_request:g}")
101+
if usd_per_vcpu_s:
102+
print(f"Node: {args.instance_type} ({args.vcpus:g} vCPUs), on-demand ${args.usd_per_hour:.4f}/hour "
103+
f"= ${args.usd_per_hour / args.vcpus:.4f} per vCPU-hour")
104+
print(f"{len(queries)} held-out messages, {args.duration:g} s per level\n")
105+
print(f" {'concurrent':>10} {'req/s':>7} {'p50 ms':>7} {'p95 ms':>7} {'cores busy':>10} "
106+
f"{'CPU ms/req':>10} {'$ per 1M':>9}")
107+
rows = []
108+
for lvl in [int(x) for x in args.levels.split(",")]:
109+
r = run_level(args.url, queries, lvl, args.duration)
110+
r["usd_per_1m"] = r["cpu_ms_per_req"] / 1000 * usd_per_vcpu_s * 1e6 if usd_per_vcpu_s else None
111+
rows.append(r)
112+
usd = f"${r['usd_per_1m']:.2f}" if r["usd_per_1m"] is not None else "n/a"
113+
print(f" {lvl:>10} {r['rps']:>7.1f} {r['p50_ms']:>7.1f} {r['p95_ms']:>7.1f} "
114+
f"{r['cores_busy']:>10.2f} {r['cpu_ms_per_req']:>10.1f} {usd:>9}", flush=True)
115+
116+
cpu_ms = statistics.median(r["cpu_ms_per_req"] for r in rows)
117+
one = rows[0]
118+
summary = {"name": args.name, "precision": info.get("precision"),
119+
"threads": info.get("intra_op_threads"), "instance_type": args.instance_type,
120+
"usd_per_hour": args.usd_per_hour, "vcpus": args.vcpus,
121+
"cpu_request": args.cpu_request, "p50_ms_single": one["p50_ms"],
122+
"p95_ms_single": one["p95_ms"], "cpu_ms_per_req": cpu_ms,
123+
"usd_per_1m": cpu_ms / 1000 * usd_per_vcpu_s * 1e6 if usd_per_vcpu_s else None,
124+
"max_rps": max(r["rps"] for r in rows), "levels": rows}
125+
if args.cpu_request:
126+
# What one pod sized at its request can sustain, priced on that slice.
127+
summary["per_pod_per_hour"] = args.cpu_request * 3600 / (cpu_ms / 1000)
128+
summary["pod_usd_per_hour"] = args.cpu_request * args.usd_per_hour / args.vcpus if args.vcpus else None
129+
print()
130+
line = (f"=== Bench [{args.name} {summary['precision']}]: p50 {one['p50_ms']:.0f} ms one at a time, "
131+
f"{cpu_ms:.0f} CPU ms per request")
132+
if summary["usd_per_1m"] is not None:
133+
line += f", ${summary['usd_per_1m']:.2f} per 1M requests"
134+
print(line + " ===")
135+
if summary.get("per_pod_per_hour"):
136+
print(f" one pod at its {args.cpu_request:g}-vCPU request: about {summary['per_pod_per_hour']:,.0f} "
137+
f"requests per hour" + (f" for ${summary['pod_usd_per_hour']:.3f}/hour"
138+
if summary.get("pod_usd_per_hour") else ""))
139+
print("===BENCH_JSON_BEGIN===")
140+
print(json.dumps(summary))
141+
print("===BENCH_JSON_END===")
142+
143+
144+
if __name__ == "__main__":
145+
main()

‎examples/k8s-autoscaling/demo/eval/triage_acc.py‎

Lines changed: 12 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -11,8 +11,10 @@
1111
TRIAGE=classifier python3 eval/triage_acc.py
1212
TRIAGE=llm python3 eval/triage_acc.py
1313
14-
Prints per-label accuracy, the confusion pairs, and latency p50, and writes
15-
eval/results/triage-<step>-<stamp>.json.
14+
Prints per-label accuracy, the confusion pairs, latency p50, and (for
15+
TRIAGE=llm) the frontier model's cost per call, and writes
16+
eval/results/triage-<step>-<stamp>.json. `make bench-triage` reads the cost
17+
from the latest llm run to compare the two models on price.
1618
"""
1719
import argparse
1820
import json
@@ -28,7 +30,7 @@
2830
HERE = os.path.dirname(os.path.abspath(__file__))
2931
sys.path.insert(0, os.path.dirname(HERE)) # the demo dir, for `agent`
3032

31-
from agent import classify, config # noqa: E402
33+
from agent import classify, config, metrics # noqa: E402
3234

3335
RESULTS_DIR = os.path.join(HERE, "results")
3436

@@ -63,15 +65,18 @@ def main():
6365
per_label = defaultdict(lambda: [0, 0]) # label -> [correct, total]
6466
confusions = Counter()
6567
latencies = []
68+
usd = []
6669
rows = []
6770
for it in items:
71+
meter = metrics.open_meter() # records the Bedrock cost when TRIAGE=llm
6872
t0 = time.perf_counter()
6973
try:
7074
pred = classify.classify(it["query"])
7175
except Exception as e:
7276
pred = {"category": f"error: {e}", "confidence": "unknown"}
7377
ms = round((time.perf_counter() - t0) * 1000)
7478
latencies.append(ms)
79+
usd.append(meter.summary()["usd"])
7580
ok = pred["category"] in it["accept"]
7681
per_label[it["label"]][1] += 1
7782
if ok:
@@ -89,6 +94,9 @@ def main():
8994
correct = sum(v[0] for v in per_label.values())
9095
print(f"\n=== Triage accuracy [{step}]: {correct}/{total} = {correct / total:.1%} "
9196
f"latency p50 {statistics.median(latencies):.0f} ms ===")
97+
usd_mean = statistics.fmean(usd) if usd else 0.0
98+
if usd_mean:
99+
print(f" frontier cost: ${usd_mean:.6f} per call = ${usd_mean * 1e6:,.0f} per 1M calls")
92100
for label in sorted(per_label):
93101
c, n = per_label[label]
94102
print(f" {label:<18} {c}/{n}")
@@ -103,6 +111,7 @@ def main():
103111
with open(out, "w", encoding="utf-8") as f:
104112
json.dump({"step": step, "accuracy": correct / total, "correct": correct,
105113
"total": total, "latency_p50_ms": statistics.median(latencies),
114+
"usd_per_call_mean": round(usd_mean, 8),
106115
"per_label": {k: {"correct": v[0], "total": v[1]} for k, v in per_label.items()},
107116
"rows": rows}, f, indent=2)
108117
print(f"Wrote {out}")

0 commit comments

Comments
 (0)