#!/usr/bin/env python3 """Re-chunk a KLD corpus so every document fits the container that scores it. A KLD reference is recorded by booting one container and prefilling every corpus document through it. radiance refuses a document the KV pool cannot hold on its own (core/mem/kv.cpp `can_ever_fit`: a paged group holds `n_blocks * block_size` tokens), and the number that matters is printed by the boot ledger, not by a flag: W kv group 'kv_attn' holds 7436 tokens but the context limit is 32768, so a prompt longer than 7436 tokens is refused (400) ... or lower --max-model-len to 7436 The BF16 qwen4exp container at the campaign's shape gets 270.00 MiB of cache -- the expert plane owns both cards -- so its real window is 7,436 tokens however `--max-model-len` is spelled. The corpus the campaign inherited from the swift15 era is 197 documents of 8,193-11,970 tokens, because make-calib-corpus.py's `--kld-min-prompt 8192` sets a floor and no ceiling: every one of them is unscoreable there, and three drained windows have proved it. This script moves the document boundaries and nothing else. Each record's token ids -- counted with the same tokenizer the generator used -- are cut into windows of at most `--max-doc-tokens`, so a record's windows concatenate back to that record's own token sequence, unchanged: no token dropped, rewritten or reworded. Each window gets its own `score_from = len - --score-tail`, the same relation the generator wrote for the parent, so the reference scores 256 positions per window instead of per document: 100,864 measured positions over the same text against the 50,432 the inherited corpus scored, at a shorter context. The calibration corpus is untouched, so the quantization itself does not move. Usage: rechunk-kld-corpus.py --corpus kld.jsonl --out kld.jsonl.new \ [--tokenizer-json PATH] [--max-doc-tokens 7168] [--score-tail 256] [--force] --out must not exist and be non-empty without --force: the same 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 json import os import sys import tempfile def parse_args() -> argparse.Namespace: ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) ap.add_argument("--corpus", required=True, help="the kld.jsonl to re-chunk (records: prompt, score_from, source)") ap.add_argument("--out", required=True, help="file to write; refuse if it exists and is non-empty without --force") ap.add_argument("--tokenizer-json", default=os.environ.get("FRANK_QWEN38_TOKENIZER", "/home/dank/models/Qwen3.8-Flash-Next/tokenizer.json"), help="the tokenizer the corpus was authored with (the same one the " "manifest in the corpus output dir names)") ap.add_argument("--max-doc-tokens", type=int, default=7168, help="the widest document the scoring container can hold; 7,168 is " "7x1024, the largest whole kilopage under the 7,436 the bf16 " "container's ledger reports") ap.add_argument("--score-tail", type=int, default=256, help="positions scored at the end of every document, as the generator " "recorded it") ap.add_argument("--force", action="store_true", help="overwrite a non-empty --out (after moving the old file to trash)") 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) 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() if a.max_doc_tokens <= 0: 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})") 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") if not os.path.isfile(a.corpus): 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") 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") tok = load_tokenizer(a.tokenizer_json) docs_in = docs_out = tokens_in = tokens_out = 0 longest_child = 0 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 child = {"prompt": tok.decode(window), "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({ "corpus": a.corpus, "out": a.out, "tokenizer": os.path.basename(a.tokenizer_json), "documents_in": docs_in, "documents_out": docs_out, "tokens_in": tokens_in, "tokens_out": tokens_out, "longest_document": longest_child, "max_doc_tokens": a.max_doc_tokens, "score_tail": a.score_tail, "scored_positions": scored, "clamped_score_from": clamped, "source_documents": sources, }, indent=1)) if __name__ == "__main__": main()