| """CPU-only aggregate depth sensitivity model. Never a measured static-depth fit. |
| |
| Reads bench2 records, emits only aggregate counters/timings and file hashes. |
| No network, torch, tokens, prompts, telemetry or captured tensors are exported. |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| import hashlib |
| import itertools |
| import json |
| import math |
| import statistics |
| from collections import defaultdict |
| from dataclasses import dataclass |
| from pathlib import Path |
| from typing import cast |
|
|
| VERIFY = (28.79, 31.86, 34.44, 37.76, 41.56, 43.68) |
| |
| DRAFT = (2.0275, 2.5122, 2.8961, 3.7219) |
|
|
|
|
| def obj(value: object) -> dict[str, object]: |
| if not isinstance(value, dict) or not all(isinstance(k, str) for k in value): |
| raise ValueError('expected JSON object') |
| return cast(dict[str, object], value) |
|
|
|
|
| def number(value: object) -> float: |
| if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value): |
| raise ValueError('expected finite number') |
| return float(value) |
|
|
|
|
| def integer(value: object) -> int: |
| v = number(value) |
| if v != int(v): |
| raise ValueError('expected integer') |
| return int(v) |
|
|
|
|
| def vector(value: object, size: int) -> list[float]: |
| if not isinstance(value, list) or len(value) != size: |
| raise ValueError(f'expected {size} values') |
| return [number(v) for v in value] |
|
|
|
|
| @dataclass(frozen=True) |
| class Run: |
| rounds: float |
| counts: tuple[float, ...] |
| step_ms: float |
|
|
|
|
| def read_suite(paths: list[Path], suite: str) -> list[dict[str, object]]: |
| cells: dict[tuple[int, int], list[Run]] = defaultdict(list) |
| sources: dict[tuple[int, int], set[tuple[str, str]]] = defaultdict(set) |
| for path in paths: |
| records = [obj(json.loads(s)) for s in path.read_text().splitlines() if s.strip()] |
| digest = hashlib.sha256(path.read_bytes()).hexdigest() |
| for r in records: |
| if r.get('type') != 'run': |
| continue |
| if r.get('error') is not None: |
| raise ValueError(f'{path}: failed run') |
| settings = obj(r['settings']) |
| if number(settings['temperature']) != 1 or number(settings['top_p']) != 1: |
| raise ValueError(f'{path}: requires T=1, top_p=1') |
| context, batch = integer(r['target_ctx']), integer(r['concurrency']) |
| if r.get('mixed') is True: |
| continue |
| if not 1 <= batch <= 4 or r.get('mixed') is not False: |
| raise ValueError(f'{path}: requires unmixed B1..B4') |
| spec = obj(r['spec']) |
| rounds = number(spec['rounds']) |
| if rounds <= 0: |
| raise ValueError(f'{path}: no request-rounds') |
| counts = obj(spec['accepted_per_position']) |
| |
| cs = tuple(number(counts[str(i)]) for i in range(5)) |
| if any(c < 0 or c > rounds for c in cs) or any(b > a for a, b in zip(cs, cs[1:])): |
| raise ValueError(f'{path}: invalid prefix counters') |
| if not math.isclose(sum(cs), number(spec['accepted_tokens']), abs_tol=1e-6): |
| raise ValueError(f'{path}: position counters do not sum to accepted tokens') |
| reqs = [q for q in records if q.get('type') == 'request' and q.get('run_id') == r['run_id']] |
| if len(reqs) != batch or any(q.get('error') is not None for q in reqs): |
| raise ValueError(f'{path}: missing/failed requests') |
| step = statistics.median(number(obj(q['step_gap_ms'])['p50']) for q in reqs) |
| if step <= 0: |
| raise ValueError(f'{path}: nonpositive step time') |
| key = context, batch |
| cells[key].append(Run(rounds, cs, step)) |
| sources[key].add((path.name, digest)) |
| out: list[dict[str, object]] = [] |
| for (context, batch), runs in sorted(cells.items()): |
| total = sum(r.rounds for r in runs) |
| surv = [sum(r.counts[i] for r in runs) / total for i in range(5)] |
| out.append(dict(suite=suite, context=context, batch=batch, runs=len(runs), request_rounds=total, |
| prefix_survival_MEASURED=surv, |
| step_p50_median_MEASURED=statistics.median(r.step_ms for r in runs), |
| step_p50_min_MEASURED=min(r.step_ms for r in runs), |
| step_p50_max_MEASURED=max(r.step_ms for r in runs), |
| per_run_survival_MEASURED=[[c/r.rounds for c in r.counts] for r in runs], |
| sources_MEASURED=[dict(file=p, sha256=h) for p, h in sorted(sources[context,batch])])) |
| return out |
|
|
|
|
| def expected(survival: list[float], depth: int) -> float: |
| if not 0 <= depth <= 5: |
| raise ValueError('supported depth is 0..5; no extrapolation above trained maximum') |
| if len(survival) != 5 or any(not math.isfinite(x) or not 0 <= x <= 1 for x in survival): |
| raise ValueError('invalid prefix survival') |
| if any(b > a for a,b in zip(survival,survival[1:])): |
| raise ValueError('prefix survival must not increase') |
| return 1 + sum(survival[:depth]) |
|
|
|
|
| def rate(survival: list[float], depth: int, batch: int, anchor_ms: float, |
| slope_scale: float, overlap: float, overhead: float) -> float: |
| """ESTIMATED nominal-cap anchor. See ESTIMATES.md for non-identifiability limits.""" |
| if not 1 <= batch <= 4 or anchor_ms <= 0 or not 0 <= overlap <= 1 or overhead < 0: |
| raise ValueError('invalid timing assumptions') |
| if not 0 <= depth <= 5: |
| raise ValueError('supported depth is 0..5') |
| cap = 5 if batch <= 2 else 4 |
| draft = DRAFT[batch-1] |
| |
| central_exposed = .25*draft + .3 |
| verify = anchor_ms - central_exposed + slope_scale*batch*(VERIFY[depth]-VERIFY[cap]) |
| cost = verify + (1-overlap)*draft + overhead |
| if cost <= 0: |
| raise ValueError('nonpositive model cost') |
| return 1000*batch*expected(survival,depth)/cost |
|
|
|
|
| def candidates(cell: dict[str, object]) -> list[dict[str, object]]: |
| batch = integer(cell['batch']) |
| s = vector(cell['prefix_survival_MEASURED'],5) |
| if not isinstance(cell['per_run_survival_MEASURED'],list): |
| raise ValueError('expected per-run arrays') |
| sruns = [vector(v,5) for v in cell['per_run_survival_MEASURED']] |
| anchor = number(cell['step_p50_median_MEASURED']) |
| rows: list[dict[str, object]] = [] |
| for k in range(1,9): |
| if k > 5: |
| rows.append(dict(k=k, status='UNSUPPORTED trained block / HELLO maximum five')) |
| continue |
| central = rate(s,k,batch,anchor,.5,.75,.3) |
| sensitivities = [rate(v,k,batch,t,m,o,h) for v,t,m,o,h in itertools.product( |
| sruns, (number(cell['step_p50_min_MEASURED']),number(cell['step_p50_max_MEASURED'])), |
| (0.,1.), (0.,1.), (0.,.6))] |
| cap = 5 if batch <= 2 else 4 |
| gains = [100*(rate(v,k,batch,t,m,o,h)/rate(v,cap,batch,t,m,o,h)-1) |
| for v,t,m,o,h in itertools.product(sruns, |
| (number(cell['step_p50_min_MEASURED']),number(cell['step_p50_max_MEASURED'])), |
| (0.,1.), (0.,1.), (0.,.6))] |
| rows.append(dict(k=k, status='ESTIMATED censored-survival nominal-cap screen', |
| tokens_per_request_round_ESTIMATED=expected(s,k), |
| aggregate_tps_ESTIMATED=central, |
| sensitivity_min_ESTIMATED=min(sensitivities), |
| sensitivity_max_ESTIMATED=max(sensitivities), |
| vs_nominal_cap_gain_percent_ESTIMATED=100*(central/rate(s,cap,batch,anchor,.5,.75,.3)-1), |
| gain_sensitivity_min_ESTIMATED=min(gains), |
| gain_sensitivity_max_ESTIMATED=max(gains))) |
| return rows |
|
|
|
|
| def markdown(cells: list[dict[str, object]]) -> str: |
| lines = ['# SP2 aggregate depth screen (generated)', '', |
| 'MEASURED input summaries; ALL candidate rates are ESTIMATED. See ESTIMATES.md.', '', |
| 'Rates are ideal steady full-batch decode aggregate tokens/s; divide by B for per stream.', |
| 'Sensitivity ranges are assumption envelopes, not confidence intervals or achieved speedups.', '', |
| '| Suite / context / B | Runs | MEASURED step p50 median, ms | MEASURED prefix survival, positions 1–5 |', |
| '|---|---:|---:|---|'] |
| for c in cells: |
| s = vector(c['prefix_survival_MEASURED'],5) |
| lines.append(f"| {c['suite']} / {c['context']} / {c['batch']} | {c['runs']} | " |
| f"{number(c['step_p50_median_MEASURED']):.2f} | "+', '.join(f'{v:.3f}' for v in s)+' |') |
| lines += ['', '| Suite / context / B | k1 | k2 | k3 | k4 | k5 | k6–8 |', '|---|---:|---:|---:|---:|---:|---|'] |
| for c in cells: |
| cols = [] |
| for r in candidates(c)[:5]: |
| cols.append(f"{number(r['aggregate_tps_ESTIMATED']):.1f} " |
| f"[{number(r['sensitivity_min_ESTIMATED']):.1f}, {number(r['sensitivity_max_ESTIMATED']):.1f}]") |
| lines.append(f"| {c['suite']} / {c['context']} / {c['batch']} | "+' | '.join(cols)+' | unsupported |') |
| return '\n'.join(lines)+'\n' |
|
|
|
|
| def main() -> None: |
| ap = argparse.ArgumentParser(description=__doc__) |
| ap.add_argument('--results-root',type=Path,required=True) |
| ap.add_argument('--coordinator-results-root',type=Path,required=True) |
| ap.add_argument('--out-dir',type=Path,required=True) |
| a = ap.parse_args() |
| cells: list[dict[str,object]] = [] |
| for suite in ('W4-E1p','W5-E1p','W5-prod'): |
| paths = sorted((a.results_root/suite).glob(f'{suite}-*.jsonl')) |
| if not paths: |
| raise ValueError(f'missing {suite} suite') |
| cells.extend(read_suite(paths,suite)) |
| k1 = a.coordinator_results_root/'K1' |
| paths = [k1/f'R1-{cell}.jsonl' for cell in ('short','c32','c150','cap4')] |
| cells.extend(read_suite(paths,'K1-R1')) |
| a.out_dir.mkdir(parents=True,exist_ok=True) |
| payload = dict(status='ESTIMATED screening only; not a qualified recalibration', |
| measured_worker_source='offload/results/n2-pair-w5/summary.json: k2-tp2 draft_p50_us B1..4 k5_n5', |
| measured_worker_ms=list(DRAFT), |
| estimated_assumptions=dict(nominal_anchor_caps=[5,5,4,4],central_overlap_fraction=.75, |
| overlap_sensitivity=[0,1],central_fixed_overhead_ms=.3,overhead_sensitivity_ms=[0,.6], |
| central_row_slope_scale=.5,row_slope_scale_sensitivity=[0,1]), |
| cells=[dict(c,candidates_ESTIMATED=candidates(c)) for c in cells]) |
| (a.out_dir/'screen.json').write_text(json.dumps(payload,indent=2,allow_nan=False)+'\n') |
| (a.out_dir/'TABLES.md').write_text(markdown(cells)) |
|
|
|
|
| if __name__ == '__main__': |
| main() |
|
|
