blakeslee-research/nucleic-purpose-classifier
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
f4e52143987832af7c950090376bf4a6b42028e6
nucleic-purpose-classifier/validate-data.py
T

763 lines
26 KiB
Python
Raw Normal View History

Merge nucleic/brisk-umber-viper-rg2p into dev
2026-07-29 23:52:13 -07:00
#!/usr/bin/env python3
"""Validate purpose-classifier JSONL data.
The checks in this script are derived from datagen-prompt.md. Definite format and
process violations are errors. Approximate targets (the requirements written as
"~" or "≈") are warnings; pass --strict to make warnings fail the command too.
Usage:
python3 ml/purpose-classifier/validate-data.py
python3 ml/purpose-classifier/validate-data.py path/to/batch.jsonl
python3 ml/purpose-classifier/validate-data.py --strict
"""
from __future__ import annotations
import argparse
import json
import math
import re
import sys
import unicodedata
from collections import Counter
from dataclasses import dataclass
from datetime import date
from pathlib import Path
from typing import Any, Iterable
SCRIPT_DIR = Path(__file__).resolve().parent
DEFAULT_DATA_DIR = SCRIPT_DIR / "data"
DEFAULT_FIXTURES = (
SCRIPT_DIR.parent.parent
/ "Tests"
/ "NucleicCoreTests"
/ "Fixtures"
/ "purpose-prompts.json"
)
FIELDS = frozenset(
{"prompt", "purpose", "secondary", "mixed", "difficulty", "slice", "lang"}
)
PURPOSES = (
"planning",
"backendImpl",
"frontendImpl",
"quickFix",
"refactor",
"debugging",
"review",
"writing",
)
PURPOSE_SET = frozenset(PURPOSES)
SLICES = ("core", "boundary", "mixed", "pasted-context", "vague-eval")
SLICE_SET = frozenset(SLICES)
SLICE_TARGETS = {
"core": 0.55,
"boundary": 0.20,
"mixed": 0.10,
"pasted-context": 0.10,
"vague-eval": 0.05,
}
EXPECTED_NON_ENGLISH = frozenset({"es", "de", "fr", "pt", "zh", "ja"})
# This is deliberately a conservative list. It is only used for the prompt's
# mechanical "same opening verb no more than three times per 50 examples" check;
# unknown first words are not guessed to be verbs.
OPENING_VERBS = frozenset(
{
"add",
"analyze",
"architect",
"audit",
"break",
"build",
"bump",
"change",
"check",
"clean",
"compare",
"consolidate",
"correct",
"create",
"debug",
"decouple",
"design",
"diagnose",
"document",
"draft",
"evaluate",
"explain",
"extract",
"figure",
"find",
"fix",
"flip",
"help",
"implement",
"investigate",
"look",
"make",
"map",
"migrate",
"modularize",
"move",
"outline",
"plan",
"polish",
"proofread",
"propose",
"refactor",
"remove",
"rename",
"replace",
"restructure",
"review",
"rewrite",
"set",
"simplify",
"sketch",
"split",
"summarize",
"track",
"translate",
"tweak",
"update",
"walk",
"wire",
"write",
}
)
TOKEN_RE = re.compile(r"\w+|[^\w\s]", re.UNICODE)
WORD_RE = re.compile(r"[^\W_]+(?:['’-][^\W_]+)*", re.UNICODE)
ENUMERATED_TASK_RE = re.compile(r"^\s*(?:[-*]\s*)?task\s+\d+\s*:", re.IGNORECASE)
LABEL_LEAK_RE = re.compile(
r"\b(?:this|it)\s+is\s+(?:an?\s+)?"
r"(?:planning|backendimpl|frontendimpl|quickfix|refactor|debugging|review|writing)"
r"\s+(?:task|prompt)\b",
re.IGNORECASE,
)
META_RE = re.compile(r"\b(?:classify this prompt|as an ai)\b", re.IGNORECASE)
MID_CONVERSATION_RE = re.compile(
r"^\s*(?:"
r"yes[,\s]+do\s+option\s+\d+|"
r"that\s+didn['’]?t\s+work|"
r"try\s+again\b|"
r"same\s+error\s+as\s+before|"
r"looks\s+good[,\s]+ship\s+it|"
r"no[,\s]+the\s+other\b"
r")",
re.IGNORECASE,
)
@dataclass(frozen=True)
class Issue:
severity: str
path: Path
message: str
line: int | None = None
def render(self) -> str:
location = str(self.path)
if self.line is not None:
location += f":{self.line}"
return f"{location}: {self.severity}: {self.message}"
@dataclass(frozen=True)
class Record:
path: Path
line: int
value: dict[str, Any]
class Validator:
def __init__(
self,
*,
batch_size: int = 200,
expected_total: int = 8_000,
fixture_path: Path | None = DEFAULT_FIXTURES,
process_checks: bool = True,
) -> None:
self.batch_size = batch_size
self.expected_total = expected_total
self.fixture_path = fixture_path
self.process_checks = process_checks
self.issues: list[Issue] = []
def error(self, path: Path, message: str, line: int | None = None) -> None:
self.issues.append(Issue("error", path, message, line))
def warning(self, path: Path, message: str, line: int | None = None) -> None:
self.issues.append(Issue("warning", path, message, line))
def validate(self, paths: list[Path], roots: list[Path]) -> list[Record]:
records_by_file: dict[Path, list[Record]] = {}
for path in paths:
records_by_file[path] = self._read_jsonl(path)
records = [record for path in paths for record in records_by_file[path]]
self._check_duplicates(records)
self._check_fixture_contamination(records)
if self.process_checks:
for path, file_records in records_by_file.items():
self._check_file_batches(path, file_records)
self._check_global_distribution(records, roots)
self._check_manifests(roots)
return records
def _read_jsonl(self, path: Path) -> list[Record]:
records: list[Record] = []
try:
lines = path.read_text(encoding="utf-8").splitlines()
except (OSError, UnicodeError) as exc:
self.error(path, f"cannot read UTF-8 JSONL: {exc}")
return records
if not lines:
self.error(path, "file is empty")
return records
for line_number, line in enumerate(lines, 1):
if not line.strip():
self.error(path, "blank lines are not allowed in strict JSONL", line_number)
continue
try:
value = json.loads(line, parse_constant=self._reject_json_constant)
except (json.JSONDecodeError, ValueError) as exc:
self.error(path, f"invalid JSON: {exc}", line_number)
continue
if not isinstance(value, dict):
self.error(path, "each JSONL line must be an object", line_number)
continue
if self._check_record(path, line_number, value):
records.append(Record(path, line_number, value))
return records
@staticmethod
def _reject_json_constant(value: str) -> None:
raise ValueError(f"{value} is not valid strict JSON")
def _check_record(self, path: Path, line: int, value: dict[str, Any]) -> bool:
valid = True
keys = set(value)
missing = sorted(FIELDS - keys)
extra = sorted(keys - FIELDS)
if missing:
self.error(path, f"missing fields: {', '.join(missing)}", line)
valid = False
if extra:
self.error(path, f"unexpected fields: {', '.join(extra)}", line)
valid = False
if missing:
return False
prompt = value["prompt"]
if not isinstance(prompt, str):
self.error(path, "prompt must be a string", line)
valid = False
elif not prompt.strip():
self.error(path, "prompt must not be empty or whitespace-only", line)
valid = False
purpose = value["purpose"]
if not isinstance(purpose, str) or purpose not in PURPOSE_SET:
self.error(path, f"purpose must be one of: {', '.join(PURPOSES)}", line)
valid = False
secondary = value["secondary"]
if secondary is not None and (
not isinstance(secondary, str) or secondary not in PURPOSE_SET
):
self.error(path, "secondary must be null or a valid purpose label", line)
valid = False
mixed = value["mixed"]
if type(mixed) is not bool:
self.error(path, "mixed must be a boolean", line)
valid = False
difficulty = value["difficulty"]
if (
isinstance(difficulty, bool)
or not isinstance(difficulty, (int, float))
or not math.isfinite(difficulty)
or not 0.0 <= difficulty <= 1.0
):
self.error(path, "difficulty must be a finite number from 0.0 to 1.0", line)
valid = False
slice_name = value["slice"]
if not isinstance(slice_name, str) or slice_name not in SLICE_SET:
self.error(path, f"slice must be one of: {', '.join(SLICES)}", line)
valid = False
lang = value["lang"]
if not isinstance(lang, str) or not is_bcp47(lang):
self.error(path, "lang must be a syntactically valid BCP 47 tag", line)
valid = False
if type(mixed) is bool:
if mixed and secondary is None:
self.error(path, "mixed=true requires a secondary purpose", line)
valid = False
if not mixed and secondary is not None:
self.error(path, "mixed=false requires secondary=null", line)
valid = False
if (slice_name == "mixed") != mixed:
self.error(path, "the mixed slice and mixed field must agree", line)
valid = False
if purpose in PURPOSE_SET and secondary == purpose:
self.error(path, "secondary must differ from the primary purpose", line)
valid = False
if isinstance(prompt, str) and prompt.strip():
self._check_prompt_antipatterns(path, line, prompt)
tokens = token_count(prompt)
if slice_name == "pasted-context" and not 100 <= tokens <= 400:
self.warning(
path,
"pasted-context prompt should be approximately 100-400 tokens "
f"(estimated {tokens})",
line,
)
elif tokens > 400:
self.warning(
path,
f"prompt exceeds the approximately 400-token maximum (estimated {tokens})",
line,
)
return valid
def _check_prompt_antipatterns(self, path: Path, line: int, prompt: str) -> None:
if ENUMERATED_TASK_RE.search(prompt):
self.error(path, "enumerated 'Task N:' prefix is forbidden", line)
if LABEL_LEAK_RE.search(prompt):
self.error(path, "prompt leaks its dataset label as a task/prompt hint", line)
if META_RE.search(prompt):
self.error(path, "assistant-directed classification/meta prompt is forbidden", line)
if MID_CONVERSATION_RE.search(prompt):
self.error(path, "prompt reads as a mid-conversation reply", line)
def _check_duplicates(self, records: list[Record]) -> None:
seen: dict[str, Record] = {}
for record in records:
key = normalize_prompt(record.value["prompt"])
previous = seen.get(key)
if previous is not None:
self.error(
record.path,
f"duplicate prompt; first seen at {previous.path}:{previous.line}",
record.line,
)
else:
seen[key] = record
def _check_fixture_contamination(self, records: list[Record]) -> None:
if self.fixture_path is None:
return
try:
fixture_data = json.loads(self.fixture_path.read_text(encoding="utf-8"))
fixture_prompts = {
normalize_prompt(item["prompt"])
for item in fixture_data
if isinstance(item, dict) and isinstance(item.get("prompt"), str)
}
except (OSError, UnicodeError, json.JSONDecodeError, TypeError, KeyError) as exc:
self.warning(self.fixture_path, f"could not load eval-only fixtures: {exc}")
return
for record in records:
if normalize_prompt(record.value["prompt"]) in fixture_prompts:
self.error(
record.path,
"prompt duplicates an eval-only purpose-prompts.json fixture",
record.line,
)
def _check_file_batches(self, path: Path, records: list[Record]) -> None:
if self.batch_size <= 0:
return
if len(records) % self.batch_size:
self.error(
path,
f"contains {len(records)} valid records; generation batches must contain "
f"{self.batch_size} records",
)
for start in range(0, len(records), self.batch_size):
batch = records[start : start + self.batch_size]
if batch:
self._check_batch(path, start // self.batch_size + 1, batch)
def _check_batch(self, path: Path, number: int, records: list[Record]) -> None:
name = f"batch {number}"
total = len(records)
minimum_length_count = math.ceil(total * 0.15)
short_count = sum(word_count(r.value["prompt"]) < 8 for r in records)
long_count = sum(token_count(r.value["prompt"]) > 60 for r in records)
if short_count < minimum_length_count:
self.error(
path,
f"{name} has {short_count}/{total} prompts under 8 words; "
f"at least {minimum_length_count} required",
)
if long_count < minimum_length_count:
self.error(
path,
f"{name} has {long_count}/{total} prompts over 60 estimated tokens; "
f"at least {minimum_length_count} required",
)
slices = Counter(r.value["slice"] for r in records)
for slice_name, target in SLICE_TARGETS.items():
actual = slices[slice_name] / total
if abs(actual - target) > 0.05:
self.warning(
path,
f"{name} {slice_name} share is {actual:.1%}; target is approximately "
f"{target:.0%} (±5 percentage points)",
)
mixed_share = sum(r.value["mixed"] for r in records) / total
if mixed_share > 0.12:
self.error(
path,
f"{name} mixed-intent share is {mixed_share:.1%}; maximum is approximately 12%",
)
english = sum(primary_language(r.value["lang"]) == "en" for r in records)
english_share = english / total
if not 0.90 <= english_share <= 0.98:
self.warning(
path,
f"{name} English share is {english_share:.1%}; target is approximately 95%",
)
foreign_languages = {
primary_language(r.value["lang"])
for r in records
if primary_language(r.value["lang"]) != "en"
}
unexpected = sorted(foreign_languages - EXPECTED_NON_ENGLISH)
if unexpected:
self.warning(
path,
f"{name} uses non-English languages outside the requested set: "
f"{', '.join(unexpected)}",
)
for window_start in range(0, total, 50):
window = records[window_start : window_start + 50]
verbs: dict[str, list[Record]] = {}
for record in window:
verb = opening_verb(record.value["prompt"])
if verb is not None:
verbs.setdefault(verb, []).append(record)
for verb, matches in sorted(verbs.items()):
if len(matches) > 3:
locations = ", ".join(str(r.line) for r in matches)
self.error(
path,
f"{name}, records {window_start + 1}-{window_start + len(window)} "
f"open with '{verb}' {len(matches)} times (lines {locations}); maximum is 3",
)
def _check_global_distribution(self, records: list[Record], roots: list[Path]) -> None:
label = roots[0] if roots else DEFAULT_DATA_DIR
total = len(records)
if self.expected_total > 0 and total != self.expected_total:
self.warning(
label,
f"dataset contains {total} valid records; generation target is "
f"{self.expected_total}",
)
if not records:
return
non_vague = [r for r in records if r.value["slice"] != "vague-eval"]
if len(non_vague) >= len(PURPOSES):
counts = Counter(r.value["purpose"] for r in non_vague)
expected = len(non_vague) / len(PURPOSES)
low = expected * 0.85
high = expected * 1.15
outside = [
f"{purpose}={counts[purpose]}"
for purpose in PURPOSES
if not low <= counts[purpose] <= high
]
if outside:
self.error(
label,
"non-vague primary labels are not within ±15% of uniform "
f"(expected about {expected:.1f} each): {', '.join(outside)}",
)
mixed_share = sum(r.value["mixed"] for r in records) / total
if mixed_share > 0.12:
self.error(
label,
f"dataset mixed-intent share is {mixed_share:.1%}; maximum is approximately 12%",
)
slices = Counter(r.value["slice"] for r in records)
for slice_name, target in SLICE_TARGETS.items():
actual = slices[slice_name] / total
if abs(actual - target) > 0.05:
self.warning(
label,
f"dataset {slice_name} share is {actual:.1%}; target is approximately "
f"{target:.0%} (±5 percentage points)",
)
def _check_manifests(self, roots: list[Path]) -> None:
directories = sorted({root if root.is_dir() else root.parent for root in roots})
for directory in directories:
candidates = sorted(directory.glob("*manifest*.json"))
if not candidates:
self.warning(
directory,
"generation manifest not found; record the model, date, and batch topics",
)
continue
for candidate in candidates:
self._check_manifest(candidate)
def _check_manifest(self, path: Path) -> None:
try:
value = json.loads(path.read_text(encoding="utf-8"))
except (OSError, UnicodeError, json.JSONDecodeError) as exc:
self.error(path, f"invalid generation manifest: {exc}")
return
if not isinstance(value, dict):
self.error(path, "generation manifest must be a JSON object")
return
values_by_key: dict[str, list[Any]] = {}
for key, item in walk_mapping_items(value):
normalized_key = re.sub(r"[-_]", "", key).casefold()
values_by_key.setdefault(normalized_key, []).append(item)
models = values_by_key.get("model", []) + values_by_key.get("generatingmodel", [])
if not any(isinstance(item, str) and item.strip() for item in models):
self.error(path, "generation manifest must record a non-empty model")
dates = values_by_key.get("date", []) + values_by_key.get("generationdate", [])
valid_dates = [item for item in dates if isinstance(item, str) and is_iso_date(item)]
if not valid_dates:
self.error(path, "generation manifest must record an ISO date (YYYY-MM-DD)")
topics = values_by_key.get("topics", []) + values_by_key.get("batchtopics", [])
if not any(
(isinstance(item, str) and item.strip())
or (isinstance(item, list) and len(item) > 0)
for item in topics
):
self.error(path, "generation manifest must record non-empty batch topics")
def is_bcp47(value: str) -> bool:
"""A dependency-free structural BCP 47 check.
Full registry validation would make this script network- or package-dependent.
This rejects the common malformed forms while accepting normal language,
script, region, variant, extension, and private-use tags.
"""
if not value or "_" in value or value.startswith("-") or value.endswith("-"):
return False
parts = value.split("-")
if any(not part.isascii() or not part.isalnum() or not 1 <= len(part) <= 8 for part in parts):
return False
if parts[0].casefold() == "x":
return len(parts) > 1
return parts[0].isalpha() and 2 <= len(parts[0]) <= 8
def normalize_prompt(prompt: str) -> str:
normalized = unicodedata.normalize("NFKC", prompt).casefold()
return " ".join(normalized.split())
def word_count(prompt: str) -> int:
return len(WORD_RE.findall(prompt))
def token_count(prompt: str) -> int:
return len(TOKEN_RE.findall(prompt))
def primary_language(tag: str) -> str:
return tag.split("-", 1)[0].casefold()
def opening_verb(prompt: str) -> str | None:
words = [word.casefold() for word in WORD_RE.findall(prompt[:160])]
if not words:
return None
if words[0] in OPENING_VERBS:
return words[0]
# Recognize common request wrappers without mistaking a later noun ("a small
# change") for the sentence's opening verb.
start = 0
if words[0] in {"please", "kindly"}:
start = 1
elif len(words) >= 2 and words[0] in {"can", "could", "would", "will"}:
start = 2 if words[1] == "you" else 1
elif len(words) >= 2 and words[0] in {"i", "we"} and words[1] in {
"need",
"want",
"would",
}:
start = 2
elif len(words) >= 2 and words[0] == "help" and words[1] in {"me", "us"}:
start = 2
else:
return None
for word in words[start : start + 3]:
if word in OPENING_VERBS:
return word
return None
def walk_mapping_items(value: Any) -> Iterable[tuple[str, Any]]:
if isinstance(value, dict):
for key, item in value.items():
if isinstance(key, str):
yield key, item
yield from walk_mapping_items(item)
elif isinstance(value, list):
for item in value:
yield from walk_mapping_items(item)
def is_iso_date(value: str) -> bool:
try:
date.fromisoformat(value)
except ValueError:
return False
return bool(re.fullmatch(r"\d{4}-\d{2}-\d{2}", value))
def discover_paths(targets: list[Path]) -> tuple[list[Path], list[Path], list[str]]:
files: set[Path] = set()
roots: list[Path] = []
errors: list[str] = []
for target in targets:
target = target.resolve()
if not target.exists():
errors.append(f"{target}: path does not exist")
continue
roots.append(target)
if target.is_dir():
files.update(path.resolve() for path in target.rglob("*.jsonl") if path.is_file())
elif target.is_file() and target.suffix.casefold() == ".jsonl":
files.add(target)
else:
errors.append(f"{target}: expected a .jsonl file or directory")
if not files and not errors:
errors.append("no .jsonl files found")
return sorted(files), roots, errors
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description="Validate purpose-classifier JSONL data against datagen-prompt.md."
)
parser.add_argument(
"paths",
nargs="*",
type=Path,
help=f"JSONL files or directories (default: {DEFAULT_DATA_DIR})",
)
parser.add_argument(
"--strict",
action="store_true",
help="return failure for approximate-target warnings as well as errors",
)
parser.add_argument(
"--batch-size",
type=int,
default=200,
help="required generation batch size; use 0 to disable (default: 200)",
)
parser.add_argument(
"--expected-total",
type=int,
default=8_000,
help="expected total record count; use 0 to disable (default: 8000)",
)
parser.add_argument(
"--fixtures",
type=Path,
default=DEFAULT_FIXTURES,
help="eval-only fixture JSON checked for contamination",
)
parser.add_argument(
"--no-fixture-check",
action="store_true",
help="do not check prompts against the eval-only fixture set",
)
parser.add_argument(
"--no-process-checks",
action="store_true",
help="only check JSONL records, schema, duplicates, and fixture contamination",
)
parser.add_argument(
"--max-issues",
type=int,
default=200,
help="maximum diagnostics to print; 0 prints all (default: 200)",
)
return parser
def main(argv: list[str] | None = None) -> int:
args = build_parser().parse_args(argv)
if args.batch_size < 0 or args.expected_total < 0 or args.max_issues < 0:
print("error: numeric options must be non-negative", file=sys.stderr)
return 2
targets = args.paths or [DEFAULT_DATA_DIR]
paths, roots, discovery_errors = discover_paths(targets)
if discovery_errors:
for message in discovery_errors:
print(f"error: {message}", file=sys.stderr)
return 2
validator = Validator(
batch_size=args.batch_size,
expected_total=args.expected_total,
fixture_path=None if args.no_fixture_check else args.fixtures.resolve(),
process_checks=not args.no_process_checks,
)
records = validator.validate(paths, roots)
errors = sum(issue.severity == "error" for issue in validator.issues)
warnings = sum(issue.severity == "warning" for issue in validator.issues)
visible = validator.issues if args.max_issues == 0 else validator.issues[: args.max_issues]
for issue in visible:
print(issue.render(), file=sys.stderr)
hidden = len(validator.issues) - len(visible)
if hidden:
print(f"... {hidden} additional issues omitted", file=sys.stderr)
print(
f"Validated {len(records)} records in {len(paths)} JSONL files: "
f"{errors} error(s), {warnings} warning(s)."
)
return 1 if errors or (args.strict and warnings) else 0
if __name__ == "__main__":
raise SystemExit(main())
Reference in New Issue Copy Permalink