Spaces:
Running on Zero
Running on Zero
File size: 5,198 Bytes
9936912 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 | #!/usr/bin/env python3
"""Validate ControlAI DAPT splits, provenance, deduplication, and token counts."""
from __future__ import annotations
import argparse
import hashlib
import json
import re
import sys
from pathlib import Path
from transformers import AutoTokenizer
def normalized_hash(text: str) -> str:
text = re.sub(r"\s+", " ", text).casefold().strip()
return hashlib.sha256(text.encode("utf-8")).hexdigest()
def load(path: Path) -> tuple[list[dict], list[str]]:
rows = []
errors = []
if not path.exists():
return rows, [f"missing file: {path}"]
with path.open(encoding="utf-8") as stream:
for line_number, line in enumerate(stream, start=1):
try:
row = json.loads(line)
except json.JSONDecodeError as exc:
errors.append(f"{path}:{line_number}: invalid JSON ({exc.msg})")
continue
row["_location"] = f"{path}:{line_number}"
rows.append(row)
return rows, errors
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("dataset", type=Path, nargs="?", default=Path("data/training/dapt_v1"))
parser.add_argument("--tokenizer", default="Qwen/Qwen3-4B-Instruct-2507")
parser.add_argument("--allow-download", action="store_true")
args = parser.parse_args()
train, errors = load(args.dataset / "train.jsonl")
valid, valid_errors = load(args.dataset / "valid.jsonl")
errors.extend(valid_errors)
summary_path = args.dataset / "summary.json"
summary = json.loads(summary_path.read_text()) if summary_path.exists() else {}
seen_hashes = {}
documents = {"train": set(), "valid": set()}
source_counts = {"train": {}, "valid": {}}
for split, rows in (("train", train), ("valid", valid)):
for row in rows:
location = row["_location"]
for field in (
"text",
"token_count",
"document_id",
"source_id",
"source_pool",
"text_sha256",
"simhash64",
):
if row.get(field) in (None, ""):
errors.append(f"{location}: missing {field}")
text = row.get("text")
if not isinstance(text, str) or not text.strip():
continue
digest = normalized_hash(text)
if digest != row.get("text_sha256"):
errors.append(f"{location}: text_sha256 mismatch")
if digest in seen_hashes:
errors.append(
f"{location}: exact duplicate also present at {seen_hashes[digest]}"
)
else:
seen_hashes[digest] = location
if row.get("source_pool") == "mathworks_r2026a":
errors.append(f"{location}: proprietary MathWorks text is disallowed")
if row.get("document_id"):
documents[split].add(f"{row.get('source_pool')}::{row['document_id']}")
pool = row.get("source_pool", "unknown")
source_counts[split][pool] = source_counts[split].get(pool, 0) + 1
leaked_documents = documents["train"] & documents["valid"]
if leaked_documents:
errors.append(
f"document split leakage: {len(leaked_documents)} document ids occur in both splits"
)
expected = {
"train_rows": len(train),
"valid_rows": len(valid),
"train_tokens": sum(row.get("token_count", 0) for row in train),
"valid_tokens": sum(row.get("token_count", 0) for row in valid),
}
for key, value in expected.items():
if summary.get(key) != value:
errors.append(f"summary {key}={summary.get(key)!r}, actual={value}")
tokenizer = AutoTokenizer.from_pretrained(
args.tokenizer, local_files_only=not args.allow_download
)
# Full token recount is deliberate: a stale count changes training estimates.
for row in train + valid:
actual = len(tokenizer.encode(row["text"], add_special_tokens=False))
if actual != row.get("token_count"):
errors.append(
f"{row['_location']}: token_count={row.get('token_count')}, actual={actual}"
)
print(f"train rows: {len(train):,}")
print(f"validation rows: {len(valid):,}")
print(f"unique normalized texts: {len(seen_hashes):,}")
print(f"train documents: {len(documents['train']):,}")
print(f"validation documents: {len(documents['valid']):,}")
print(f"train tokens: {expected['train_tokens']:,}")
print(f"validation tokens: {expected['valid_tokens']:,}")
print("source rows:")
for split in ("train", "valid"):
print(f" {split}: {dict(sorted(source_counts[split].items()))}")
if errors:
print("validation failed:", file=sys.stderr)
for error in errors[:100]:
print(f"- {error}", file=sys.stderr)
if len(errors) > 100:
print(f"- ... {len(errors) - 100} additional errors", file=sys.stderr)
return 1
print("validation passed")
return 0
if __name__ == "__main__":
raise SystemExit(main())
|