From 3e5a37a21f71243469c3d554d480bbb02a95cd0d Mon Sep 17 00:00:00 2001 From: Frank Date: Thu, 29 Oct 2026 15:26:08 -0700 Subject: [PATCH] refactor(quant): one corpus plumbing module, and the builders out of main() MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The three corpus tools now share scripts/qwen38_common.py: tokenizer load with the venv guidance, one n_tokens (so a KLD count cannot drift from the corpus it scores), JSONL read/write, an atomic writer, and refuse(). make-calib-corpus.py's 170-line main() with three nested closures became classify / split_pools / fragment_text / build_frags / pack_class / emit_calib / emit_kld / manifest; its magic numbers are named constants. rechunk-kld-corpus.py's main became validate / cut / rechunk around a Window record, so the round-trip proof sits in one function that can be tested without a tokenizer. calib-push.py grew parse_args / load_prompts / push_round, and a malformed corpus now refuses with a line number instead of raising KeyError. Parity, measured on dankserver against the pre-refactor scripts at the same seed: calib.jsonl and kld.jsonl carry identical records in identical order, the manifest is identical, and the re-chunked corpus is byte-identical (md5 e824e67d...). The one byte-level change is JSON escaping — write_jsonl emits literal UTF-8, which is what the re-chunker already wrote; the parsed records do not move. --- scripts/calib-push.py | 53 ++-- scripts/make-calib-corpus.py | 303 +++++++++++++--------- scripts/qwen38_common.py | 94 +++++++ scripts/rechunk-kld-corpus.py | 179 +++++++------ serverAppConfigs-docs/LLM/frank-archon.md | 5 +- 5 files changed, 405 insertions(+), 229 deletions(-) create mode 100755 scripts/qwen38_common.py diff --git a/scripts/calib-push.py b/scripts/calib-push.py index 9a4114d..beffdfb 100755 --- a/scripts/calib-push.py +++ b/scripts/calib-push.py @@ -22,6 +22,9 @@ Mechanics notes: prompts do not bias the Hessians (the corpus is drawn round-robin over source classes, so loss is class-neutral as long as failures are rare; a systematic 4xx is loud in the log and must be fixed, not ignored). +- a round that seals no tap bytes is a FAILURE, not a retry: the engine + answering 200 while the tap directory is wrong, full or read-only is + how a corpus gets pushed four times for nothing. Usage: calib-push.py [--port 7770] [--parallel 64] [--container arbench] [--corpus /home/dank/corpora/out/calib.jsonl] [--max-rounds 4] @@ -35,6 +38,12 @@ import sys import time import urllib.request +_HERE = os.path.dirname(os.path.abspath(__file__)) if "__file__" in globals() else os.getcwd() +if _HERE not in sys.path: + sys.path.insert(0, _HERE) + +import qwen38_common as qc # noqa: E402 - sibling module, same directory + DONE_MARK = "calibration wrote" @@ -69,6 +78,15 @@ def push_one(url, prompt, model): return False +def push_round(url, prompts, model, parallel): + """Push the whole corpus once; returns how many requests were accepted.""" + ok = 0 + with concurrent.futures.ThreadPoolExecutor(parallel) as ex: + for good in ex.map(lambda p: push_one(url, p, model), prompts): + ok += 1 if good else 0 + return ok + + def tap_bytes(path): """Bytes the tap has sealed so far; 0 when the directory is absent. @@ -89,7 +107,7 @@ def tap_bytes(path): return total -def main(): +def parse_args(): ap = argparse.ArgumentParser() ap.add_argument("--port", type=int, default=7770) ap.add_argument("--parallel", type=int, default=64) @@ -98,31 +116,36 @@ def main(): ap.add_argument("--tap-dir", default="/home/dank/campaign/qwen38-w5nl64/calib") ap.add_argument("--corpus", default="/home/dank/corpora/out/calib.jsonl") ap.add_argument("--max-rounds", type=int, default=4) - a = ap.parse_args() - url = f"http://127.0.0.1:{a.port}/v1/completions" + return ap.parse_args() + +def load_prompts(a) -> list[str]: + if not os.path.isfile(a.corpus): + qc.refuse(f"no corpus at {a.corpus}") prompts = [] - with open(a.corpus) as f: - for line in f: - line = line.strip() - if line: - prompts.append(json.loads(line)["prompt"]) + for lineno, rec in qc.iter_jsonl(a.corpus): + if not isinstance(rec.get("prompt"), str): + qc.refuse(f"{a.corpus}:{lineno} has no string 'prompt' field") + prompts.append(rec["prompt"]) + if not prompts: + qc.refuse(f"the corpus at {a.corpus} holds no prompts") + return prompts + + +def main(): + a = parse_args() + url = f"http://127.0.0.1:{a.port}/v1/completions" + prompts = load_prompts(a) chars = sum(len(p) for p in prompts) print(f"corpus: {len(prompts)} prompts, ~{chars // 4} tokens (chars/4)", flush=True) - if not prompts: - print("REFUSED: the corpus holds no prompts", flush=True) - sys.exit(2) for rnd in range(1, a.max_rounds + 1): if tap_done(a.container): break before = tap_bytes(a.tap_dir) t0 = time.time() - ok = 0 - with concurrent.futures.ThreadPoolExecutor(a.parallel) as ex: - for good in ex.map(lambda p: push_one(url, p, a.model), prompts): - ok += 1 if good else 0 + ok = push_round(url, prompts, a.model, a.parallel) dt = max(time.time() - t0, 1e-9) print(f"round {rnd}: {ok}/{len(prompts)} ok in {dt:.0f}s " f"({chars / 4 / dt:.0f} tokens/s est)", flush=True) diff --git a/scripts/make-calib-corpus.py b/scripts/make-calib-corpus.py index f5bacfd..8cd2cae 100755 --- a/scripts/make-calib-corpus.py +++ b/scripts/make-calib-corpus.py @@ -34,14 +34,25 @@ import json import os import random import re +import sys from jinja2 import Environment, FileSystemLoader -from tokenizers import Tokenizer + +_HERE = os.path.dirname(os.path.abspath(__file__)) if "__file__" in globals() else os.getcwd() +if _HERE not in sys.path: + sys.path.insert(0, _HERE) + +import qwen38_common as qc # noqa: E402 - sibling module, same directory CODE_EXT = (".sh", ".py", ".jinja", ".recipe", ".env", ".example", ".yaml", ".yml", ".bats", ".service", ".container", ".toml") JSONL_EXTS = (".jsonl", ".json") DOC_DIR = "serverAppConfigs-docs" +FRAGMENT_TOKENS = 2400 # what one fragment may contribute +CLASS_MIX = {"code": .45, "prose": .30, "tools": .25} +HF_CAP = {"code": 40000, "prose": 60000, "tools": 0} +POOL_CYCLES = 40 # how many times a class pool may repeat before giving up +KLD_GUARD = 5000 # rendered conversations tried before declaring the pool short FLEET_CMDS = [ "systemctl status frank-archon --no-pager", @@ -142,7 +153,7 @@ def load_parquet_texts(path: str, kind: str, rng: random.Random, return out[:cap] -def main() -> None: +def parse_args() -> argparse.Namespace: ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("--repo", required=True) ap.add_argument("--tokenizer-json", required=True) @@ -156,22 +167,26 @@ def main() -> None: ap.add_argument("--hf-code-parquet", default="") ap.add_argument("--hf-prose-parquet", default="") ap.add_argument("--seed", type=int, default=20261006) - a = ap.parse_args() + return ap.parse_args() - rng = random.Random(a.seed) - tokenizer = Tokenizer.from_file(a.tokenizer_json) - env = Environment(loader=FileSystemLoader(os.path.dirname(a.template)), + +def load_template(template: str) -> tuple[Environment, str]: + """The chat template as the engine will render it, plus its short name.""" + env = Environment(loader=FileSystemLoader(os.path.dirname(template)), trim_blocks=True, lstrip_blocks=True) env.filters["tojson"] = lambda v: json.dumps(v, ensure_ascii=False) env.filters["raise_exception"] = lambda m: (_ for _ in ()).throw( ValueError(m)) - tname = os.path.basename(a.template) + return env, os.path.basename(template) + +def classify(repo: str) -> dict[str, list[str]]: + """Walk the checkout into the three source classes the mix draws from.""" code, prose, jsonl = [], [], [] - for root, dirs, files in os.walk(a.repo): + for root, dirs, files in os.walk(repo): dirs[:] = [d for d in dirs if not d.startswith(".")] for f in sorted(files): - p = os.path.relpath(os.path.join(root, f), a.repo) + p = os.path.relpath(os.path.join(root, f), repo) if DOC_DIR in p and f.endswith(".md"): prose.append(p) elif p.startswith(DOC_DIR): @@ -180,153 +195,185 @@ def main() -> None: code.append(p) elif f.endswith(JSONL_EXTS) and f != "package-lock.json": jsonl.append(p) - # file-level split: bucket 0 -> KLD pool, 1..4 -> calibration pool - pools: dict[str, tuple[list, list]] = {} - for cls, lst in (("code", code), ("prose", prose), ("tools", jsonl)): - pools[cls] = ([p for p in lst if fnv(p) % 5 == 0], - [p for p in lst if fnv(p) % 5 != 0]) + return {"code": code, "prose": prose, "tools": jsonl} - def text_of(p: str) -> str: - path = os.path.join(a.repo, p) - raw = open(path, encoding="utf-8", errors="replace").read() - if p.endswith(JSONL_EXTS): - # A .json in this repo is not necessarily a content-keyed JSONL: the - # substring test passes on any line that merely mentions "content", - # and a list-shaped document raises TypeError — both crashed the whole - # build. Skip what does not fit instead. - lines = [] - for line in raw.splitlines(): - if not line.strip() or "content" not in line: - continue - try: - rec = json.loads(line) - except json.JSONDecodeError: - continue - if isinstance(rec, dict) and isinstance(rec.get("content"), str): - lines.append(rec["content"]) - raw = "\n".join(lines) - toks = tokenizer.encode(raw, add_special_tokens=False).ids - return tokenizer.decode(toks[:2400], skip_special_tokens=True) +def split_pools(classes: dict[str, list[str]]) -> dict[str, tuple[list, list]]: + """File-level split: bucket 0 -> KLD pool, everything else -> calibration.""" + return {cls: ([p for p in lst if fnv(p) % 5 == 0], + [p for p in lst if fnv(p) % 5 != 0]) + for cls, lst in classes.items()} + + +def jsonl_content(raw: str) -> str: + """The content fields of a content-keyed JSONL, or "" for anything else. + + A .json in this repo is not necessarily a content-keyed JSONL: the substring + test passes on any line that merely mentions "content", and a list-shaped + document raises TypeError — both used to crash the whole build. Skip what + does not fit instead. + """ + lines = [] + for line in raw.splitlines(): + if not line.strip() or "content" not in line: + continue + try: + rec = json.loads(line) + except json.JSONDecodeError: + continue + if isinstance(rec, dict) and isinstance(rec.get("content"), str): + lines.append(rec["content"]) + return "\n".join(lines) + + +def fragment_text(repo: str, rel: str, tok) -> str: + """One file as a fragment: its first FRAGMENT_TOKENS tokens, decoded.""" + raw = open(os.path.join(repo, rel), encoding="utf-8", + errors="replace").read() + if rel.endswith(JSONL_EXTS): + raw = jsonl_content(raw) + ids = tok.encode(raw, add_special_tokens=False).ids + return tok.decode(ids[:FRAGMENT_TOKENS], skip_special_tokens=True) + + +def build_frags(a: argparse.Namespace, tok, + rng: random.Random) -> dict[str, dict[str, list[str]]]: + """Both fragment pools per class, seeded so the same seed rebuilds the corpus.""" + pools = split_pools(classify(a.repo)) cmds = harvest_cmds(a.repo, rng) - # fragment pools per class, corpus-split first frags: dict[str, dict[str, list[str]]] = {} - for cls in pools: - kld_files, cal_files = pools[cls] - fr = {} - src = {"kld": kld_files, "cal": cal_files} + for cls, (kld_files, cal_files) in pools.items(): if cls == "tools": - fr["kld"] = [tool_call(rng, cmds) for _ in range(600)] - fr["cal"] = [tool_call(rng, cmds) for _ in range(40000)] + frags[cls] = {"kld": [tool_call(rng, cmds) for _ in range(600)], + "cal": [tool_call(rng, cmds) for _ in range(40000)]} else: - fr["kld"] = [text_of(p) for p in src["kld"]] - fr["cal"] = [text_of(p) for p in src["cal"]] - frags[cls] = fr + frags[cls] = {"kld": [fragment_text(a.repo, p, tok) for p in kld_files], + "cal": [fragment_text(a.repo, p, tok) for p in cal_files]} # per-class fragment caps: prose rows are short (~66 tok) so need many; # codeparrot files are long (~2.6k tok) so a small cap suffices - hfcap = {"code": 40000, "prose": 60000, "tools": 0} for cls, spec in (("code", a.hf_code_parquet), ("prose", a.hf_prose_parquet)): for path in [p for p in spec.split(",") if p]: frags[cls]["cal"].extend( - load_parquet_texts(path, cls, rng, cap=hfcap[cls])) + load_parquet_texts(path, cls, rng, cap=HF_CAP[cls])) for cls in frags: rng.shuffle(frags[cls]["cal"]) + return frags - def toks(s: str) -> int: - return len(tokenizer.encode(s, add_special_tokens=False).ids) - def emit_calib() -> tuple[int, list[dict]]: - targets = {"code": .45, "prose": .30, "tools": .25} - rows: list[dict] = [] - for cls, frac in targets.items(): - want = int(a.calib_tokens * frac) - pool = frags[cls]["cal"] +def pack_class(cls: str, pool: list[str], want: int, tok, + chunk_tokens: int) -> list[dict]: + """Greedily pack `pool` into chunk_tokens-sized calibration prompts.""" + if not pool: + return [] + rows: list[dict] = [] + # Count every fragment once: the loop cycles the pool up to POOL_CYCLES + # times, and re-tokenizing a 2.4k-token fragment on each pass dominated the + # build's wall time. + counts = [qc.n_tokens(tok, f) for f in pool] + got = 0 + i = 0 + buf: list[str] = [] + nbuf = 0 + while got < want: + j = i % len(pool) + frag = pool[j] + i += 1 + n = counts[j] + if nbuf + n > chunk_tokens and buf: + rows.append({"prompt": "\n".join(buf), "cls": cls}) + got += nbuf + buf, nbuf = [], 0 + if got >= want: + break + buf.append(frag) + nbuf += n + if i > len(pool) * POOL_CYCLES: + break # pool exhausted with repeats; stop + if buf and got < want: + rows.append({"prompt": "\n".join(buf), "cls": cls}) + got += nbuf + print(f"calib {cls}: {got} tokens, {i} fragments") + return rows + + +def emit_calib(a: argparse.Namespace, frags, tok, + rng: random.Random) -> tuple[int, list[dict]]: + """calib.jsonl rows: the class mix, shuffled, prompts only.""" + rows: list[dict] = [] + for cls, frac in CLASS_MIX.items(): + rows.extend(pack_class(cls, frags[cls]["cal"], int(a.calib_tokens * frac), + tok, a.chunk_tokens)) + rng.shuffle(rows) + return len(rows), [{"prompt": r["prompt"]} for r in rows] + + +def emit_kld(a: argparse.Namespace, frags, tok, rng: random.Random, + env: Environment, tname: str) -> tuple[int, list[dict], int]: + """kld.jsonl rows: multi-turn conversations, each with a scored tail.""" + template = env.get_template(tname) + rows: list[dict] = [] + total = scored_total = 0 + classes = ["code", "prose", "tools"] + cursor = {c: 0 for c in classes} + guard = 0 + while total < a.kld_tokens and guard < KLD_GUARD: + guard += 1 + msgs, srcs = [], [] + for t in range(rng.choice([2, 3, 4])): + cls = classes[(len(rows) + t) % 3] + pool = frags[cls]["kld"] if not pool: continue - got, i, buf, nbuf = 0, 0, [], 0 - # Count every fragment once: the loop cycles the pool up to 40 times, - # and re-tokenizing a 2.4k-token fragment on each pass dominated the - # build's wall time. - counts = [toks(f) for f in pool] - while got < want: - j = i % len(pool) - frag = pool[j] - i += 1 - n = counts[j] - if nbuf + n > a.chunk_tokens and buf: - rows.append({"prompt": "\n".join(buf), "cls": cls}) - got += nbuf - buf, nbuf = [], 0 - if got >= want: - break - buf.append(frag) - nbuf += n - if i > len(pool) * 40: - break # pool exhausted with repeats; stop - if buf and got < want: - rows.append({"prompt": "\n".join(buf), "cls": cls}) - got += nbuf - print(f"calib {cls}: {got} tokens, {i} fragments") - rng.shuffle(rows) - return len(rows), [{k: v for k, v in r.items() if k == "prompt"} - for r in rows] + frag = pool[cursor[cls] % len(pool)] + cursor[cls] += 1 + msgs.append({"role": "user", + "content": "Summarize and explain this " + f"engineering material:\n{frag}"}) + msgs.append({"role": "assistant", + "content": frag[:4000]}) + srcs.append(cls) + rendered = template.render(messages=msgs, add_generation_prompt=False) + n = qc.n_tokens(tok, rendered) + if n < a.kld_min_prompt: + continue + rows.append({"prompt": rendered, + "score_from": n - a.score_tail, + "source": "+".join(sorted(set(srcs)))}) + total += n + scored_total += a.score_tail + return len(rows), rows, scored_total - def emit_kld() -> tuple[int, list[dict], int]: - rows, total, scored_total = [], 0, 0 - classes = ["code", "prose", "tools"] - i = {c: 0 for c in classes} - guard = 0 - while total < a.kld_tokens and guard < 5000: - guard += 1 - msgs, srcs = [], [] - for t in range(rng.choice([2, 3, 4])): - cls = classes[(len(rows) + t) % 3] - pool = frags[cls]["kld"] - if not pool: - continue - frag = pool[i[cls] % len(pool)] - i[cls] += 1 - msgs.append({"role": "user", - "content": "Summarize and explain this " - f"engineering material:\n{frag}"}) - msgs.append({"role": "assistant", - "content": frag[:4000]}) - srcs.append(cls) - rendered = env.get_template(tname).render( - messages=msgs, add_generation_prompt=False) - n = toks(rendered) - if n < a.kld_min_prompt: - continue - rows.append({"prompt": rendered, - "score_from": n - a.score_tail, - "source": "+".join(sorted(set(srcs)))}) - total += n - scored_total += a.score_tail - return len(rows), rows, scored_total - os.makedirs(a.out_dir, exist_ok=True) - nc, rows = emit_calib() - with open(os.path.join(a.out_dir, "calib.jsonl"), "w") as f: - for r in rows: - f.write(json.dumps(r) + "\n") - nk, krows, scored = emit_kld() - with open(os.path.join(a.out_dir, "kld.jsonl"), "w") as f: - for r in krows: - f.write(json.dumps(r) + "\n") - manifest = { +def manifest(a: argparse.Namespace, tname: str, nc: int, nk: int, + scored: int) -> dict: + return { "seed": a.seed, "template": tname, "tokenizer": os.path.basename(a.tokenizer_json), "calib_prompts": nc, "kld_prompts": nk, "kld_scored_positions": scored, "kld_score_tail": a.score_tail, "disjoint_file_pools": "local KLD material only; HF code feeds calib", - "calib_mix": {"code": .45, "prose": .30, "tools": .25}, + "calib_mix": dict(CLASS_MIX), "hf_code": a.hf_code_parquet, "hf_prose": a.hf_prose_parquet, } - with open(os.path.join(a.out_dir, "manifest.json"), "w") as f: - json.dump(manifest, f, indent=1) - print(json.dumps(manifest, indent=1)) + + +def main() -> None: + a = parse_args() + rng = random.Random(a.seed) + tok = qc.load_tokenizer(a.tokenizer_json) + env, tname = load_template(a.template) + + frags = build_frags(a, tok, rng) + os.makedirs(a.out_dir, exist_ok=True) + nc, calib_rows = emit_calib(a, frags, tok, rng) + qc.write_jsonl(os.path.join(a.out_dir, "calib.jsonl"), calib_rows) + nk, kld_rows, scored = emit_kld(a, frags, tok, rng, env, tname) + qc.write_jsonl(os.path.join(a.out_dir, "kld.jsonl"), kld_rows) + man = manifest(a, tname, nc, nk, scored) + qc.write_json(os.path.join(a.out_dir, "manifest.json"), man) + print(json.dumps(man, indent=1)) if __name__ == "__main__": diff --git a/scripts/qwen38_common.py b/scripts/qwen38_common.py new file mode 100755 index 0000000..28eb6cf --- /dev/null +++ b/scripts/qwen38_common.py @@ -0,0 +1,94 @@ +#!/usr/bin/env python3 +"""Shared plumbing for the qwen38 campaign corpus tools. + +The three tools move the same two shapes — a JSONL of prompt records and a +one-artifact JSON manifest — and they all tokenize with the SAME tokenizer. That +last part is not a detail: a KLD reference is only meaningful against the exact +token sequence the corpus recorded, so `n_tokens` exists once, here, and every +tool counts the same way. A window that decode/encode does not round-trip is a +refusal, never a warning. + +Loaded as a sibling module. Each tool puts its own directory on `sys.path` +first, so `python3 scripts/.py` and the driver's container mount +(`runpy.run_path('/work/.py')`, `-v scripts:/work:ro`) both resolve it — +run the tools by path, not piped through stdin. +""" +import contextlib +import json +import os +import sys +import tempfile + + +def refuse(msg: str) -> None: + """Exit 2 with a REFUSED line — the code argparse uses for unusable input.""" + print(f"REFUSED: {msg}", file=sys.stderr) + sys.exit(2) + + +def load_tokenizer(path: str): + try: + from tokenizers import Tokenizer + except ImportError as exc: # the corpus tools carry their own venv + refuse(f"tokenizers is not importable by {sys.executable} — run the " + f"corpus venv's python (on dankserver: " + f"/home/dank/venvs/radcorpus/bin/python) or set " + f"FRANK_QWEN38_TOKENIZER: {exc}") + if not os.path.isfile(path): + refuse(f"no tokenizer at {path}") + return Tokenizer.from_file(path) + + +def n_tokens(tok, text: str) -> int: + """The corpus's own count: no special tokens added, prompt side only.""" + return len(tok.encode(text, add_special_tokens=False).ids) + + +def iter_jsonl(path: str): + """Yield `(lineno, record)` for every non-blank line; refuse on a bad line. + + Field validation stays with the caller: the three corpora carry different + fields (`prompt` alone, `prompt`+`cls`, `prompt`+`score_from`+`source`), and + a record missing the one the tool needs is that tool's refusal to make. + """ + with open(path, encoding="utf-8", errors="replace") as fh: + for lineno, line in enumerate(fh, 1): + if not line.strip(): + continue + try: + rec = json.loads(line) + except json.JSONDecodeError as exc: + refuse(f"{path}:{lineno} is not JSON: {exc}") + yield lineno, rec + + +def write_jsonl(path: str, rows) -> None: + with open(path, "w", encoding="utf-8") as fh: + for row in rows: + fh.write(json.dumps(row, ensure_ascii=False) + "\n") + + +@contextlib.contextmanager +def atomic_writer(path: str): + """Write `path` through a temporary sibling and rename on success. + + An interrupted run then leaves no half corpus behind, and the temporary name + keeps the one-artifact-one-name law the driver applies to a score report. + """ + d = os.path.dirname(os.path.abspath(path)) or "." + fd, tmp = tempfile.mkstemp(dir=d, prefix="." + os.path.basename(path) + "-") + try: + with os.fdopen(fd, "w", encoding="utf-8") as fh: + yield fh + os.replace(tmp, path) + except BaseException: + if os.path.exists(tmp): + os.unlink(tmp) + raise + os.chmod(path, 0o644) + + +def write_json(path: str, obj) -> None: + with atomic_writer(path) as fh: + json.dump(obj, fh, indent=1) + fh.write("\n") diff --git a/scripts/rechunk-kld-corpus.py b/scripts/rechunk-kld-corpus.py index e8b69d1..d02969c 100755 --- a/scripts/rechunk-kld-corpus.py +++ b/scripts/rechunk-kld-corpus.py @@ -38,12 +38,19 @@ one-artifact-one-name law run-qwen38-bf16.sh applies to a score report. The file is written through a temporary name and renamed, so an interrupted run leaves no half corpus behind. A JSON summary of what was written goes to stdout. """ - import argparse +import dataclasses import json import os import sys -import tempfile + +_HERE = os.path.dirname(os.path.abspath(__file__)) if "__file__" in globals() else os.getcwd() +if _HERE not in sys.path: + sys.path.insert(0, _HERE) + +import qwen38_common as qc + +MISSING = object() # "the parent record carried no source field" def parse_args() -> argparse.Namespace: @@ -69,100 +76,96 @@ def parse_args() -> argparse.Namespace: return ap.parse_args() -def load_tokenizer(path: str): - try: - from tokenizers import Tokenizer - except ImportError as exc: # the corpus scripts carry their own venv - raise SystemExit( - f"REFUSED: tokenizers is not importable by {sys.executable} — run the " - f"corpus venv's python (on dankserver: /home/dank/venvs/radcorpus/bin/python) " - f"or set FRANK_QWEN38_TOKENIZER: {exc}") from None - if not os.path.isfile(path): - raise SystemExit(f"REFUSED: no tokenizer at {path}") - return Tokenizer.from_file(path) +@dataclasses.dataclass(frozen=True) +class Window: + """One scoring document cut out of a parent record.""" + prompt: str + score_from: int + n_tokens: int + clamped: bool + source: object = MISSING + + def record(self) -> dict: + rec = {"prompt": self.prompt, "score_from": self.score_from} + if self.source is not MISSING: + rec["source"] = self.source + return rec -def refuse(msg: str) -> None: - print(f"REFUSED: {msg}", file=sys.stderr) - sys.exit(2) # 2 is the code argparse uses for unusable arguments - - -def main() -> None: - a = parse_args() +def validate(a: argparse.Namespace) -> None: + """Everything that makes this run unusable, named before a token is cut.""" if a.max_doc_tokens <= 0: - refuse(f"--max-doc-tokens must be positive (got {a.max_doc_tokens})") + qc.refuse(f"--max-doc-tokens must be positive (got {a.max_doc_tokens})") if a.score_tail <= 0: - refuse(f"--score-tail must be positive (got {a.score_tail})") + qc.refuse(f"--score-tail must be positive (got {a.score_tail})") if a.max_doc_tokens <= a.score_tail: - refuse(f"--max-doc-tokens {a.max_doc_tokens} cannot hold a {a.score_tail}-position " - "score tail: every window would be all tail and no context") + qc.refuse(f"--max-doc-tokens {a.max_doc_tokens} cannot hold a " + f"{a.score_tail}-position score tail: every window would be all " + "tail and no context") if not os.path.isfile(a.corpus): - refuse(f"no corpus at {a.corpus}") + qc.refuse(f"no corpus at {a.corpus}") if os.path.realpath(a.out) == os.path.realpath(a.corpus): - refuse(f"--out {a.out} is the corpus itself; move it to trash first — the trash law " - "means the old segmentation stays recoverable") + qc.refuse(f"--out {a.out} is the corpus itself; move it to trash first — the " + "trash law means the old segmentation stays recoverable") if os.path.exists(a.out) and os.path.getsize(a.out) > 0 and not a.force: - refuse(f"{a.out} exists and is not empty; pass --force after moving it to trash") + qc.refuse(f"{a.out} exists and is not empty; pass --force after moving it to trash") - tok = load_tokenizer(a.tokenizer_json) +def cut(rec: dict, tok, a: argparse.Namespace, + where: str) -> tuple[list[Window], int]: + """Cut one record's token ids into windows, proving each one round-trips. + + The round-trip encode is the whole reason this tool may claim the reference + scores the corpus's own token sequence: decoding a window to text and + encoding it back must return the identical ids, or the score is of a + different document than the one on disk. + """ + ids = tok.encode(rec["prompt"], add_special_tokens=False).ids + windows: list[Window] = [] + for start in range(0, max(len(ids), 1), a.max_doc_tokens): + window = ids[start:start + a.max_doc_tokens] + if not window: + continue + text = tok.decode(window) + if tok.encode(text, add_special_tokens=False).ids != window: + qc.refuse(f"{where} window at token {start} does not round-trip through " + "text — the reference would score a different token sequence " + "than the corpus holds") + windows.append(Window(prompt=text, + score_from=max(len(window) - a.score_tail, 0), + n_tokens=len(window), + clamped=len(window) < a.score_tail, + source=rec["source"] if "source" in rec else MISSING)) + return windows, len(ids) + + +def rechunk(a: argparse.Namespace, tok) -> dict: + """Write the re-chunked corpus and return the summary it printed.""" docs_in = docs_out = tokens_in = tokens_out = 0 - longest_child = 0 - clamped = scored = 0 + longest_child = clamped = scored = 0 sources: dict[str, int] = {} - fd, tmp = tempfile.mkstemp(dir=os.path.dirname(os.path.abspath(a.out)) or ".", - prefix=".rechunk-") - try: - with os.fdopen(fd, "w", encoding="utf-8") as out: - with open(a.corpus, encoding="utf-8") as src: - for lineno, line in enumerate(src, 1): - if not line.strip(): - continue - try: - rec = json.loads(line) - except json.JSONDecodeError as exc: - refuse(f"{a.corpus}:{lineno} is not JSON: {exc}") - prompt = rec.get("prompt") - if not isinstance(prompt, str): - refuse(f"{a.corpus}:{lineno} has no string 'prompt' field") - ids = tok.encode(prompt, add_special_tokens=False).ids - docs_in += 1 - tokens_in += len(ids) - for start in range(0, max(len(ids), 1), a.max_doc_tokens): - window = ids[start:start + a.max_doc_tokens] - if not window: - continue - score_from = max(len(window) - a.score_tail, 0) - if len(window) < a.score_tail: - clamped += 1 - scored += len(window) - score_from - text = tok.decode(window) - if tok.encode(text, add_special_tokens=False).ids != window: - refuse(f"{a.corpus}:{lineno} window at token {start} does not " - "round-trip through text — the reference would score a " - "different token sequence than the corpus holds") - child = {"prompt": text, - "score_from": score_from} - if "source" in rec: - child["source"] = rec["source"] - out.write(json.dumps(child, ensure_ascii=False) + "\n") - docs_out += 1 - tokens_out += len(window) - longest_child = max(longest_child, len(window)) - if "source" in rec: - for cls in str(rec["source"]).split("+"): - sources[cls] = sources.get(cls, 0) + 1 - if tokens_out != tokens_in: - raise SystemExit(f"REFUSED: the windows hold {tokens_out} tokens and the corpus " - f"has {tokens_in} — re-chunking lost or invented tokens") - os.replace(tmp, a.out) - except BaseException: - if os.path.exists(tmp): - os.unlink(tmp) - raise - os.chmod(a.out, 0o644) - - print(json.dumps({ + with qc.atomic_writer(a.out) as out: + for lineno, rec in qc.iter_jsonl(a.corpus): + where = f"{a.corpus}:{lineno}" + if not isinstance(rec.get("prompt"), str): + qc.refuse(f"{where} has no string 'prompt' field") + windows, n_parent = cut(rec, tok, a, where) + docs_in += 1 + tokens_in += n_parent + for w in windows: + out.write(json.dumps(w.record(), ensure_ascii=False) + "\n") + docs_out += 1 + tokens_out += w.n_tokens + longest_child = max(longest_child, w.n_tokens) + clamped += 1 if w.clamped else 0 + scored += w.n_tokens - w.score_from + if "source" in rec: + for cls in str(rec["source"]).split("+"): + sources[cls] = sources.get(cls, 0) + 1 + if tokens_out != tokens_in: + raise SystemExit(f"REFUSED: the windows hold {tokens_out} tokens and the corpus " + f"has {tokens_in} — re-chunking lost or invented tokens") + return { "corpus": a.corpus, "out": a.out, "tokenizer": os.path.basename(a.tokenizer_json), @@ -176,7 +179,13 @@ def main() -> None: "scored_positions": scored, "clamped_score_from": clamped, "source_documents": sources, - }, indent=1)) + } + + +def main() -> None: + a = parse_args() + validate(a) + print(json.dumps(rechunk(a, qc.load_tokenizer(a.tokenizer_json)), indent=1)) if __name__ == "__main__": diff --git a/serverAppConfigs-docs/LLM/frank-archon.md b/serverAppConfigs-docs/LLM/frank-archon.md index 263ed8b..89ec957 100644 --- a/serverAppConfigs-docs/LLM/frank-archon.md +++ b/serverAppConfigs-docs/LLM/frank-archon.md @@ -641,7 +641,10 @@ vllm-radiance era (this table opens on 09-26; the era page claims 10-10). - Convert campaign (dormant, kept for a future quant): `scripts/convert-swift15-w4dn8.sh` `scripts/recipes/qwen4exp-swift15-w4dn8-i8-hc8m.recipe`, `scripts/run-swift15-bf16.sh` `scripts/kld-gate-swift15.sh`, `scripts/kld-then-calib.sh`, `scripts/calib-push.py` - `scripts/make-calib-corpus.py`, `scripts/derive-protected-experts.py` + `scripts/make-calib-corpus.py`, `scripts/rechunk-kld-corpus.py`, + `scripts/qwen38_common.py` (the tokenizer, JSONL and refusal plumbing those three + corpus tools share — one `n_tokens`, so every tool counts the same way), + `scripts/derive-protected-experts.py` `scripts/apply-protected-cast.sh`, `scripts/verify-swift15-ple.py`. - Probes: `scripts/probe-frank-archon-kv.py` (reads slots/window/dtype from the deployed env so it cannot drift; leg D = decode), `scripts/frank-archon-expert-activity.py`