Buckets:
| """Build the LoRA SFT dataset: teacher trajectories in the exact scored format. | |
| What the score actually rewards | |
| ------------------------------- | |
| L1lift = lpA(y_C|z_A) − lpA(y_C|∅) (challenger-side, ρ=+0.900 with margin) | |
| Λ2 = lpC(y_C|z_A) − lpC(y_C|∅) (teacher-side) | |
| Both are measured on the CANONICAL injected rendering built by chat.py: | |
| gen_prompt(...) + "</think>\\nTHOUGHT: " + z + "\\n\\n" + y | |
| so training examples are emitted in byte-identical form, with the loss masked | |
| to the `</think>...y` span. Training text == scored text, by construction. | |
| Two sources | |
| ----------- | |
| --from-teacher (default, durable) sample fresh (z_C, y_C) from a served | |
| teacher over corpus turns that appear in NO published dump. 16,645 of the | |
| 18,384 corpus turns (90.5%) qualify, so train and eval stay disjoint. | |
| --from-dumps (free, fragile) reuse the 30,360 teacher rollouts already | |
| published inside the eval dumps. Costs nothing and is ideal for shaking | |
| down the pipeline, BUT those 1,739 turns are exactly the ones duels have | |
| drawn from — training on them is the RT-6 memorisation channel the | |
| validator patched on 2026-08-06. It flatters the screen and the effect | |
| dies at the next corpus refresh (epoch 4 added 8,513 turns on 2026-08-08). | |
| The current screen slice is always excluded, so the screen stays honest. | |
| Usage | |
| ----- | |
| # free shakedown | |
| python harness/distill_data.py --from-dumps --out data/sft_dumps.jsonl | |
| # real run (teacher served on 2 GPUs, port 8000) | |
| vllm serve zai-org/GLM-4.5-Air-FP8 --served-model-name zai-org/GLM-4.5-Air-FP8 \ | |
| --tensor-parallel-size 2 --max-model-len 32768 \ | |
| --gpu-memory-utilization 0.80 --port 8000 | |
| python harness/distill_data.py --from-teacher --n-turns 4000 --out data/sft.jsonl | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import asyncio | |
| import gzip | |
| import json | |
| import pathlib | |
| import random | |
| import time | |
| import screen # load_corpus / EVALS, imports no GPU deps at module level | |
| ROOT = pathlib.Path(__file__).resolve().parent.parent | |
| TEACHER = "zai-org/GLM-4.5-Air-FP8" | |
| def dump_turn_ids() -> set[str]: | |
| """Every turn id that appears in any published duel — the eval side.""" | |
| out: set[str] = set() | |
| for path in sorted(screen.EVALS.glob("*.json.gz")): | |
| with gzip.open(path) as fh: | |
| out.update(json.load(fh).get("turn_ids", [])) | |
| return out | |
| def screen_turn_ids(ref: str) -> set[str]: | |
| with gzip.open(screen.EVALS / f"{ref}.json.gz") as fh: | |
| return set(json.load(fh)["turn_ids"]) | |
| def record(prefix: list[dict], z: str, y: str, tid: str, src: str) -> dict: | |
| """One SFT example. `target` is the span the loss is computed on.""" | |
| return {"turn_id": tid, "source": src, "prefix": prefix, | |
| "target": "</think>\nTHOUGHT: " + z + "\n\n" + y, | |
| "z": z, "y": y} | |
| def from_dumps(exclude: set[str], limit: int) -> list[dict]: | |
| corpus = screen.load_corpus() | |
| out, seen = [], set() | |
| for path in sorted(screen.EVALS.glob("*.json.gz")): | |
| with gzip.open(path) as fh: | |
| d = json.load(fh) | |
| for tid, refs in (d.get("teacher_refs") or {}).items(): | |
| if tid in exclude or tid not in corpus: | |
| continue | |
| for i, r in enumerate(refs): | |
| key = (tid, r["y"][:200], i) | |
| if key in seen or not r.get("y"): | |
| continue | |
| seen.add(key) | |
| out.append(record(corpus[tid]["prefix"], r["z"], r["y"], | |
| tid, "dump")) | |
| random.shuffle(out) | |
| return out[:limit] if limit else out | |
| async def from_teacher(args, exclude: set[str]) -> list[dict]: | |
| import httpx | |
| from vllm_client import Served, VllmModel | |
| corpus = screen.load_corpus() | |
| pool = sorted(set(corpus) - exclude) | |
| random.Random(args.seed).shuffle(pool) | |
| pool = pool[: args.n_turns] | |
| if args.num_shards > 1: | |
| # Deterministic split so N workers, each pinned to its own teacher | |
| # replica, cover the pool exactly once with no coordination. | |
| pool = pool[args.shard::args.num_shards] | |
| print(f"sampling {args.samples} teacher rollouts on {len(pool)} held-out " | |
| f"turns ({len(exclude)} excluded)") | |
| served = Served("teacher", args.teacher, None, args.port) | |
| sem = asyncio.Semaphore(args.concurrency) | |
| out, 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 | |
| prefix = corpus[tid]["prefix"] | |
| rolls = await asyncio.gather(*[ | |
| model.sample(prefix, args.temperature, | |
| args.max_thought + args.max_action) | |
| for _ in range(args.samples) | |
| ], return_exceptions=True) | |
| for r in rolls: | |
| if isinstance(r, Exception): | |
| continue | |
| z, y = r | |
| if y: | |
| out.append(record(prefix, z, y, tid, "teacher")) | |
| done += 1 | |
| if done % 100 == 0: | |
| rate = done / max(time.time() - t0, 1e-9) | |
| print(f" {done}/{len(pool)} turns {len(out)} examples " | |
| f"eta {(len(pool)-done)/rate/60:.0f} min", flush=True) | |
| await asyncio.gather(*[one(t) for t in pool]) | |
| return out | |
| def main() -> None: | |
| ap = argparse.ArgumentParser() | |
| src = ap.add_mutually_exclusive_group(required=True) | |
| src.add_argument("--from-dumps", action="store_true") | |
| src.add_argument("--from-teacher", action="store_true") | |
| ap.add_argument("--out", required=True) | |
| ap.add_argument("--ref", default="chal-00388", | |
| help="screen slice to hold out no matter what") | |
| ap.add_argument("--n-turns", type=int, default=4000) | |
| ap.add_argument("--samples", type=int, default=4) | |
| ap.add_argument("--limit", type=int, default=0) | |
| ap.add_argument("--teacher", default=TEACHER) | |
| ap.add_argument("--port", type=int, default=8000) | |
| 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=24) | |
| ap.add_argument("--seed", type=int, default=0) | |
| ap.add_argument("--shard", type=int, default=0) | |
| ap.add_argument("--num-shards", type=int, default=1) | |
| args = ap.parse_args() | |
| screen_slice = screen_turn_ids(args.ref) | |
| if args.from_dumps: | |
| # Only the live screen slice is withheld; everything else in the dumps | |
| # is fair game for a shakedown run. Read the module docstring first. | |
| rows = from_dumps(screen_slice, args.limit) | |
| print(f"NOTE: --from-dumps trains on turns duels actually draw from. " | |
| f"Shakedown only — see the RT-6 warning in this file.") | |
| else: | |
| rows = asyncio.run(from_teacher(args, dump_turn_ids() | screen_slice)) | |
| out = pathlib.Path(args.out) | |
| out.parent.mkdir(parents=True, exist_ok=True) | |
| with out.open("w") as fh: | |
| for r in rows: | |
| fh.write(json.dumps(r) + "\n") | |
| turns = len({r["turn_id"] for r in rows}) | |
| print(f"\nwrote {len(rows)} examples over {turns} turns -> {out}") | |
| print(f"held out screen slice {args.ref} ({len(screen_slice)} turns): " | |
| f"{len(screen_slice & {r['turn_id'] for r in rows})} leaked (must be 0)") | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 7.58 kB
- Xet hash:
- f20ed3b37ccb2238b227b744c884f87d477b2bfaa0ae1e750c1bbc070bb3fc12
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.