| |
| """Render the raw bench JSON into the markdown tables used by README.md and docs/.""" |
|
|
| from __future__ import annotations |
|
|
| import json |
| import statistics |
| from pathlib import Path |
|
|
| HERE = Path(__file__).resolve().parent |
| RES = HERE / "results" |
|
|
|
|
| def med(rows, key): |
| vals = [r[key] for r in rows if r.get(key) is not None] |
| return statistics.median(vals) if vals else None |
|
|
|
|
| def fmt(v, nd=2): |
| return "—" if v is None else f"{v:.{nd}f}" |
|
|
|
|
| def budget_curve(): |
| f = RES / "strict_budgets.json" |
| if not f.exists(): |
| return None |
| rows = json.loads(f.read_text()) |
| budgets = sorted({r["budget_mb"] for r in rows}) |
| lines = ["| Budget | Arena (GiB) | 槽位 | 命中率 | SSD MB/token | Decode tok/s | 峰值匿名 RSS | VmSwap |", |
| "| ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |"] |
| for b in budgets: |
| sel = [r for r in rows if r["budget_mb"] == b] |
| ok = [r for r in sel if r["strict_ok"]] |
| lines.append( |
| f"| {b // 1024} GiB | {fmt(med(sel, 'arena_gib'))} | " |
| f"{int(med(sel, 'slots') or 0)} | {fmt(med(sel, 'hit_pct'))}% | " |
| f"{fmt(med(sel, 'mb_per_token'), 1)} | {fmt(med(sel, 'decode_tps'))} | " |
| f"{fmt(med(sel, 'peak_anon_mb'), 0)} | {fmt(med(sel, 'peak_swap_mb'), 3)} |") |
| return "\n".join(lines) |
|
|
|
|
| def ram_sweep(): |
| f = RES / "ram_sweep_n9.json" |
| src = f if f.exists() else (RES / "ram_sweep.json") |
| if not src.exists(): |
| return None, None |
| data = json.loads(src.read_text()) |
| if "per_budget" in data: |
| per = data["per_budget"] |
| tests = data.get("welch_vs_8192", {}) |
| n = max(v["n"] for v in per.values()) |
| lines = [f"| Budget | Arena (GiB) | 槽位 | 命中率 | SSD MB/token | 平均 tok/s | 單輪標準差 | 峰值 RSS | VmSwap |", |
| "| ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |"] |
| for b, v in sorted(per.items(), key=lambda kv: int(kv[0])): |
| lines.append( |
| f"| {int(b) // 1024} GiB | {fmt(v['arena_gib'])} | {int(v['slots'])} | " |
| f"{fmt(v['hit_pct'])}% | {fmt(v['mb_per_token'], 1)} | " |
| f"{fmt(v['mean_tps'])} ± {fmt(v['stdev_tps'])} | " |
| f"{fmt(v['peak_anon_mb'], 0)} | {fmt(v['peak_swap_mb'], 3)} |") |
| tlines = ["| 比較 | 只改變了 | 平均 tok/s | Welch t | p | 判定 |", "| --- | --- | ---: | ---: | ---: | --- |"] |
| base = sorted(per, key=lambda k: int(k))[1] if len(per) > 1 else None |
| for name, t in tests.items(): |
| other, _, b = name.partition("_vs_") |
| o, bb = per[other], per[b] |
| verdict = "顯著" if t["p"] < 0.05 else "無法分辨" |
| tlines.append( |
| f"| {int(other) // 1024} GiB 比 {int(b) // 1024} GiB | RAM 預算 | " |
| f"{o['mean_tps']:.2f} → {bb['mean_tps']:.2f} " |
| f"({(o['mean_tps'] / bb['mean_tps'] - 1) * 100:+.0f}%) | " |
| f"{t['t']:+.2f} | {t['p']:.3f} | {verdict} |") |
| return "\n".join(lines), "\n".join(tlines) |
| rows = data["rows"] if isinstance(data, dict) else data |
| budgets = sorted({r["budget_mb"] for r in rows}) |
| lines = ["| Budget | Arena (GiB) | 槽位 | 命中率 | SSD MB/token | Decode tok/s | 峰值 RSS | VmSwap |", |
| "| ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |"] |
| for b in budgets: |
| sel = [r for r in rows if r["budget_mb"] == b] |
| lines.append( |
| f"| {b // 1024} GiB | {fmt(med(sel, 'arena_gib'))} | {int(med(sel, 'slots') or 0)} | " |
| f"{fmt(med(sel, 'hit_pct'))}% | {fmt(med(sel, 'mb_per_token'), 1)} | " |
| f"{fmt(med(sel, 'decode_tps'))} | {fmt(med(sel, 'peak_anon_mib'), 0)} | " |
| f"{fmt(med(sel, 'peak_swap_mb'), 3)} |") |
| return "\n".join(lines), None |
|
|
|
|
| def headline(): |
| for name in ("strict_headline.json",): |
| f = RES / name |
| if not f.exists(): |
| return None |
| return json.loads(f.read_text()) |
|
|
|
|
| def negatives(): |
| out = {} |
| f = RES / "strict_threads.json" |
| if f.exists(): |
| out["threads"] = json.loads(f.read_text())["summary"] |
| f = RES / "strict_prefetch.json" |
| if f.exists(): |
| out["prefetch"] = json.loads(f.read_text())["summary"] |
| return out |
|
|
|
|
| def main() -> None: |
| parts = {} |
| parts["budget_curve"] = budget_curve() |
| sweep, tests = ram_sweep() |
| parts["ram_sweep"] = sweep |
| parts["ram_sweep_tests"] = tests |
| parts["negatives"] = negatives() |
| print(json.dumps(parts, indent=2, ensure_ascii=False)) |
| (RES / "summary.json").write_text(json.dumps(parts, indent=2, ensure_ascii=False)) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
