| """Build the tajwid distillation corpus: teacher answers to Ahmad's REAL prompts. |
| |
| WHY THIS AND NOT A PUBLIC DATASET: |
| The student is not meant to be generally frontier — it is meant to be frontier ON |
| AHMAD'S WORKLOAD. That distribution is already measured and on disk: 167,258 unique |
| messages over 3.7 months, tool mix dominated by terminal / Read / Bash / PowerShell / |
| Grep. Public instruction sets do not have that shape. train.jsonl is the shape. |
| |
| STRATA (never blended — measured 2026-07-26): |
| Arabic is 0.0227% of real usage and pays a ~2.3-2.8x tokenizer tax. Proportional |
| sampling gives Arabic almost no signal, so it is generated as its OWN stratum with its |
| own file, and mixed at training time at a chosen ratio rather than at harvest time. |
| |
| RESUME SAFETY: |
| Every record is keyed by a hash of (stratum, prompt). A re-run skips what exists, so a |
| killed run costs nothing. Learned the hard way: a "stopped" background run once kept |
| writing the SAME output file as its replacement for 20 minutes. |
| |
| *** BEFORE POINTING THIS AT A PAID/MANAGED API — READ *** |
| Generating training data from a hosted model's outputs is governed by that provider's |
| terms, and some providers restrict using outputs to train other models. Ahmad SELLS |
| software, so this is commercial exposure, not a theoretical concern. Two clean routes: |
| 1. SELF-HOST an open-weight teacher whose licence permits distillation (rent the GPU, |
| run the weights yourself) — no ambiguity, and you also get logprobs. |
| 2. Use openly-licensed existing traces. |
| This script therefore REQUIRES --i-checked-the-licence when TEACHER_BASE_URL is not |
| localhost. It is a speed bump against an expensive mistake, not legal advice. |
| |
| Usage: |
| python make_traces.py --stratum workload --limit 200 |
| python make_traces.py --stratum arabic --limit 100 |
| python make_traces.py --stratum workload --limit 5000 --i-checked-the-licence |
| """ |
| import argparse |
| import hashlib |
| import json |
| import os |
| import re |
| import sys |
| import time |
|
|
| sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) |
| import teacher_client as tc |
|
|
| CORPUS = r"E:\ABU\NEOCORTEX\corpus" |
| OUTDIR = r"E:\ABU\NEOCORTEX\cloud\traces" |
| ARABIC = re.compile(r"[\u0600-\u06FF]") |
|
|
| SYSTEM = ("You are a precise engineering assistant for a Windows/PowerShell power user " |
| "who runs local AI infrastructure. Answer concretely and completely. Prefer " |
| "exact commands, file paths and numbers over generalities.") |
|
|
|
|
| def key(stratum, prompt): |
| return hashlib.sha256(f"{stratum}\x00{prompt}".encode("utf-8")).hexdigest()[:20] |
|
|
|
|
| def load_prompts(stratum, min_len=40, max_len=6000): |
| """User-role messages from the real record, filtered and de-duplicated.""" |
| src = os.path.join(CORPUS, "train.jsonl") |
| seen, out = set(), [] |
| with open(src, encoding="utf-8") as fh: |
| for line in fh: |
| try: |
| d = json.loads(line) |
| except Exception: |
| continue |
| if d.get("role") != "user": |
| continue |
| t = (d.get("text") or "").strip() |
| if not (min_len <= len(t) <= max_len): |
| continue |
| is_ar = len(ARABIC.findall(t)) > max(8, 0.15 * len(t)) |
| if (stratum == "arabic") != is_ar: |
| continue |
| h = key(stratum, t) |
| if h in seen: |
| continue |
| seen.add(h) |
| out.append(t) |
| return out |
|
|
|
|
| def main(): |
| ap = argparse.ArgumentParser() |
| ap.add_argument("--stratum", choices=("workload", "arabic"), default="workload") |
| ap.add_argument("--limit", type=int, default=100) |
| ap.add_argument("--max-tokens", type=int, default=tc.DEFAULT_MAX_TOKENS) |
| ap.add_argument("--sleep", type=float, default=0.0, help="seconds between calls") |
| ap.add_argument("--i-checked-the-licence", action="store_true") |
| a = ap.parse_args() |
|
|
| remote = not re.search(r"(127\.0\.0\.1|localhost)", tc.BASE) |
| if remote and not a.i_checked_the_licence: |
| sys.exit("REFUSING: TEACHER_BASE_URL is remote and --i-checked-the-licence was not\n" |
| "passed. Confirm the provider's terms permit using outputs as training\n" |
| "data, or self-host an open-weight teacher instead. See the header.") |
|
|
| os.makedirs(OUTDIR, exist_ok=True) |
| out_path = os.path.join(OUTDIR, f"traces_{a.stratum}.jsonl") |
|
|
| done = set() |
| if os.path.exists(out_path): |
| with open(out_path, encoding="utf-8") as fh: |
| for line in fh: |
| try: |
| done.add(json.loads(line)["key"]) |
| except Exception: |
| pass |
|
|
| prompts = load_prompts(a.stratum) |
| todo = [p for p in prompts if key(a.stratum, p) not in done][: a.limit] |
| print(tc.describe()) |
| print(f"stratum={a.stratum} available={len(prompts):,} already done={len(done):,} " |
| f"this run={len(todo):,} -> {out_path}") |
| if not todo: |
| print("nothing to do") |
| return |
|
|
| t0, ok, tok = time.time(), 0, 0 |
| with open(out_path, "a", encoding="utf-8") as fh: |
| for i, p in enumerate(todo, 1): |
| r = tc.complete([{"role": "system", "content": SYSTEM}, |
| {"role": "user", "content": p}], |
| max_tokens=a.max_tokens) |
| |
| |
| if not r["text"].strip(): |
| print(f" [{i}/{len(todo)}] EMPTY completion (finish={r['finish_reason']}) — skipped") |
| continue |
| rec = {"key": key(a.stratum, p), "stratum": a.stratum, |
| "system": SYSTEM, "prompt": p, "completion": r["text"], |
| "finish_reason": r["finish_reason"], "usage": r["usage"], |
| "teacher": f"{tc.MODEL}@{tc.BASE}"} |
| fh.write(json.dumps(rec, ensure_ascii=False) + "\n") |
| fh.flush() |
| ok += 1 |
| tok += (r["usage"] or {}).get("completion_tokens", 0) or 0 |
| if i % 10 == 0 or i == len(todo): |
| el = time.time() - t0 |
| print(f" [{i}/{len(todo)}] kept={ok} {tok:,} out-tok " |
| f"{el/60:.1f} min {tok/max(el,1e-9):.1f} tok/s", flush=True) |
| if a.sleep: |
| time.sleep(a.sleep) |
|
|
| el = time.time() - t0 |
| print(f"\nDONE kept {ok}/{len(todo)} {tok:,} completion tokens in {el/60:.1f} min") |
| print(f"wrote {out_path}") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
