Buckets:
| """Build training data by SELECTION, not imitation: keep the king's best thoughts. | |
| Every path tried so far imitated something and failed -- merging (-0.0006), | |
| teacher distillation (-0.032 measured), stronger-teacher (refuted: with the | |
| answer-pairing broken the king beats GLM on 64% of pairs). headroom.py then | |
| showed where the score actually is: within-turn sd across the king's OWN samples | |
| is 0.0196 against a mean of 0.0340, and a perfect best-of-8 selector gains | |
| +0.0237 -- 91% of the crown bar, still climbing at K=8. | |
| This turns that into a corpus. For each turn, sample K rollouts from the king, | |
| score every thought against the teacher answers already published in the dumps, | |
| and keep the best. The answers cost nothing: 2,052 turns across the dumps carry | |
| teacher_refs, so the expensive half of Phase B is already on disk. | |
| Two properties that Phase B lacked: | |
| * ON-POLICY. The model already emits these thoughts, just not often enough, | |
| so training is a reweighting of its own distribution rather than a shift | |
| toward someone else's. That is precisely the mechanism that broke Phase B. | |
| * SCORED BY THE REAL METRIC. Selection uses L1lift itself, not a proxy for it. | |
| The reported `selection gain` is the in-sample ceiling this corpus encodes. What | |
| survives training is the realization fraction -- the one number this whole | |
| experiment exists to measure, and the one Phase B proved can be negative. | |
| python harness/select_data.py --repo <king> --revision <sha> \ | |
| --n-turns 700 --samples 8 --out data/best.jsonl | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import asyncio | |
| import gzip | |
| import json | |
| import pathlib | |
| import random | |
| import statistics as st | |
| import time | |
| import score | |
| import screen | |
| ROOT = pathlib.Path(__file__).resolve().parent.parent | |
| EVALS = ROOT / "evals" | |
| def load_refs(exclude_ref: str) -> tuple[dict, float]: | |
| """Teacher answers per turn, pooled over every dump, minus the screen slice.""" | |
| refs, excluded, clip = {}, set(), score.DEFAULT_L1_CLIP | |
| for f in sorted(EVALS.glob("chal-*.json.gz")): | |
| d = json.load(gzip.open(f)) | |
| if f.stem.startswith(exclude_ref): | |
| excluded |= set(d["turn_ids"]) | |
| clip = d["verdict"].get("gates", {}).get("l1_clip", clip) | |
| for tid, rows in (d.get("teacher_refs") or {}).items(): | |
| rows = [r for r in rows if r.get("y") and r.get("z")] | |
| if rows and tid not in refs: | |
| refs[tid] = rows | |
| for tid in excluded: | |
| refs.pop(tid, None) # the screen slice must never be trained on | |
| return refs, clip | |
| async def one_turn(model, prefix, refs, n, temperature, max_thought, max_action, | |
| clip, top): | |
| """Sample n rollouts, score each thought by real L1lift, keep the best `top`.""" | |
| rolls = await asyncio.gather(*[ | |
| model.sample(prefix, temperature, max_thought + max_action) | |
| for _ in range(n)], return_exceptions=True) | |
| rolls = [r for r in rolls if not isinstance(r, Exception) and r[1]] | |
| ys = [r["y"] for r in refs] | |
| if len(rolls) < 2 or not ys: | |
| return None | |
| jobs = [model.score_action(prefix, z, y) for y in ys for z, _ in rolls] | |
| jobs += [model.score_action(prefix, "", y) for y in ys] | |
| res = await asyncio.gather(*jobs, return_exceptions=True) | |
| if any(isinstance(r, Exception) for r in res): | |
| return None | |
| cl = lambda x: min(clip, max(-clip, x)) | |
| nz = len(rolls) | |
| empties = [res[len(ys) * nz + i]["lp_per_byte"] for i in range(len(ys))] | |
| # score each thought by its mean L1lift over that turn's teacher answers | |
| lifts = [st.mean(cl(res[i * nz + k]["lp_per_byte"] - empties[i]) | |
| for i in range(len(ys))) for k in range(nz)] | |
| order = sorted(range(nz), key=lambda k: lifts[k], reverse=True) | |
| return {"rolls": rolls, "lifts": lifts, "keep": order[:top]} | |
| async def run(args) -> dict: | |
| import httpx | |
| from vllm_client import Served, VllmModel | |
| # Open the sink BEFORE the expensive work and write incrementally. Buffering | |
| # everything for a single write at the end cost 90 min of completed | |
| # selection to a missing directory on 2026-08-10; now a bad path fails in | |
| # milliseconds and a late crash loses one line, not the whole run. | |
| outp = pathlib.Path(args.out) | |
| outp.parent.mkdir(parents=True, exist_ok=True) | |
| sink = outp.open("w") | |
| refs, clip = load_refs(args.exclude_ref) | |
| corpus = screen.load_corpus() | |
| pool = sorted(t for t in refs if t in corpus) | |
| # Drop turns the TRAINER would throw away anyway. Phase B generated over the | |
| # full length distribution and then lost 57% of it to the 8192-token cap -- | |
| # paying GPU for examples that never reached a gradient. Filtering here is | |
| # free money twice over: every kept example is trainable, and the kept set | |
| # averages 2.7x shorter prefixes, so prefill on every scoring call shrinks. | |
| # 3.0 chars/token is deliberately conservative (real ratio is nearer 4). | |
| if args.max_prefix_chars: | |
| before = len(pool) | |
| pool = [t for t in pool | |
| if corpus[t].get("n_prefix_chars", 0) <= args.max_prefix_chars] | |
| print(f"length filter: {len(pool)}/{before} turns with prefix " | |
| f"<= {args.max_prefix_chars} chars", flush=True) | |
| random.Random(args.seed).shuffle(pool) | |
| pool = pool[: args.n_turns] if args.n_turns else pool | |
| print(f"{len(pool)} turns with teacher answers (screen slice excluded), " | |
| f"K={args.samples}, keeping top {args.top}", flush=True) | |
| served = Served("model", args.repo, args.revision or None, args.port) | |
| sem = asyncio.Semaphore(args.concurrency) | |
| out, all_l, sel_l, done, t0 = [], [], [], 0, time.time() | |
| async with httpx.AsyncClient( | |
| limits=httpx.Limits(max_connections=args.concurrency + 8)) as http: | |
| model = VllmModel(served, http, sem) | |
| async def one(tid): | |
| nonlocal done | |
| r = await one_turn(model, corpus[tid]["prefix"], refs[tid], | |
| args.samples, args.temperature, args.max_thought, | |
| args.max_action, clip, args.top) | |
| done += 1 | |
| if done % 50 == 0: | |
| el = time.time() - t0 | |
| print(f" {done}/{len(pool)} turns {len(out)} kept " | |
| f"eta {(len(pool)-done)*el/max(done,1)/60:.0f} min", flush=True) | |
| if not r: | |
| return | |
| all_l.extend(r["lifts"]) | |
| for k in r["keep"]: | |
| z, y = r["rolls"][k] | |
| sel_l.append(r["lifts"][k]) | |
| rec = {"turn_id": tid, "source": "selected", | |
| "prefix": corpus[tid]["prefix"], | |
| "target": "</think>\nTHOUGHT: " + z + "\n\n" + y, | |
| "l1lift": r["lifts"][k]} | |
| out.append(rec) | |
| sink.write(json.dumps(rec) + "\n") | |
| sink.flush() # durable per turn, not per run | |
| await asyncio.gather(*[one(t) for t in pool]) | |
| sink.close() | |
| return { | |
| "examples": len(out), "turns_used": len({r["turn_id"] for r in out}), | |
| "turns_offered": len(pool), "samples": args.samples, "top": args.top, | |
| "mean_l1lift_all": st.mean(all_l) if all_l else None, | |
| "mean_l1lift_selected": st.mean(sel_l) if sel_l else None, | |
| "selection_gain": (st.mean(sel_l) - st.mean(all_l)) if all_l else None, | |
| "elapsed_s": time.time() - t0, "out": args.out, | |
| } | |
| def main() -> None: | |
| ap = argparse.ArgumentParser(description=__doc__.split("\n")[0]) | |
| ap.add_argument("--repo", required=True) | |
| ap.add_argument("--revision", default="") | |
| ap.add_argument("--port", type=int, default=8002) | |
| ap.add_argument("--n-turns", type=int, default=700) | |
| ap.add_argument("--samples", type=int, default=8) | |
| ap.add_argument("--top", type=int, default=1) | |
| ap.add_argument("--temperature", type=float, default=0.8) | |
| ap.add_argument("--max-thought", type=int, default=1024) | |
| ap.add_argument("--max-action", type=int, default=768) | |
| ap.add_argument("--concurrency", type=int, default=16) | |
| ap.add_argument("--exclude-ref", default="chal-00388", | |
| help="duel whose turns are the screen slice — never trained on") | |
| ap.add_argument("--max-prefix-chars", type=int, default=23700, | |
| help="skip turns the trainer would drop anyway " | |
| "(8192-token budget at a conservative 3 chars/token)") | |
| ap.add_argument("--seed", type=int, default=7) | |
| ap.add_argument("--out", required=True) | |
| args = ap.parse_args() | |
| r = asyncio.run(run(args)) | |
| print("\n" + "=" * 62) | |
| print(f" wrote {r['examples']} examples from {r['turns_used']}/" | |
| f"{r['turns_offered']} turns in {r['elapsed_s']:.0f}s") | |
| print(f" mean L1lift, all samples {r['mean_l1lift_all']:+.5f}") | |
| print(f" mean L1lift, SELECTED {r['mean_l1lift_selected']:+.5f}") | |
| print(f" selection gain (in-sample) {r['selection_gain']:+.5f}") | |
| print("=" * 62) | |
| print(" This is the CEILING the corpus encodes. What survives training is") | |
| print(" the realization fraction — Phase B showed it can be negative.") | |
| pathlib.Path(args.out + ".stats.json").write_text(json.dumps(r, indent=2)) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 9.3 kB
- Xet hash:
- 39475e42c2619db8605a466a2d4f11bd2946dc00a4627f596f623f300aaccd2e
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.