File size: 4,711 Bytes
12eb3a5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 | #!/usr/bin/env python3
"""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() |