refactor(quant): one corpus plumbing module, and the builders out of main()

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.
This commit is contained in:
Frank committed 2026-10-29 15:26:08 -07:00
1 parent 1e4fa14772
commit 3e5a37a21f
5 files changed
+405 -229

No files matched your search

+175 -128
View File
@@ -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__":