| |
| """Validate handcrafted preference-pair readiness for TinyLiquid DPO. |
| |
| This script never creates training content. It only counts already-authored |
| JSONL preference rows and reports whether the researched DPO gate is met. |
| """ |
| import argparse |
| import collections |
| import json |
| import re |
| from pathlib import Path |
|
|
| VERDICT_RE = re.compile(r"Verdict:\s*([^.\n]+)\.", re.IGNORECASE) |
|
|
| DEFAULT_CLASSES = [ |
| "true", "false", "refutes", "contradiction", "not enough information", |
| "unsubstantiated", "overclaim", "misleading", "not a contradiction", |
| "mixed", "low confidence", "abstain", "cannot provide", |
| "partially true", "conflict", "unsupported", "inaccurate", |
| "unverifiable", "cannot confirm", "not a discrepancy", |
| "no meaningful pattern", |
| ] |
|
|
|
|
| def verdict(text): |
| m = VERDICT_RE.search(text or "") |
| return m.group(1).strip().lower() if m else "<missing>" |
|
|
|
|
| def main(): |
| ap = argparse.ArgumentParser() |
| ap.add_argument("data", help="preference JSONL with prompt/chosen/rejected") |
| ap.add_argument("--min-total", type=int, default=1500) |
| ap.add_argument("--target-total", type=int, default=3000) |
| ap.add_argument("--min-per-class", type=int, default=60) |
| ap.add_argument("--max-median-ratio", type=float, default=2.0) |
| args = ap.parse_args() |
|
|
| path = Path(args.data) |
| rows = [] |
| seen_prompts = set() |
| duplicates = 0 |
| missing = [] |
| counts = collections.Counter() |
|
|
| for line_no, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1): |
| if not line.strip(): |
| continue |
| ex = json.loads(line) |
| rows.append(ex) |
| prompt = ex.get("prompt", "") |
| if prompt in seen_prompts: |
| duplicates += 1 |
| seen_prompts.add(prompt) |
| for key in ("persona", "prompt", "chosen", "rejected"): |
| if key not in ex: |
| missing.append((line_no, key)) |
| counts[verdict(ex.get("chosen", ""))] += 1 |
|
|
| print(f"file: {path}") |
| print(f"pairs: {len(rows)} unique_prompts: {len(seen_prompts)} duplicates: {duplicates}") |
| print("chosen verdict counts:") |
| for k, v in counts.most_common(): |
| print(f" {k:24s} {v}") |
|
|
| required = DEFAULT_CLASSES |
| deficits = {c: max(0, args.min_per_class - counts.get(c, 0)) for c in required} |
| deficits = {c: d for c, d in deficits.items() if d} |
| present = sorted(v for c, v in counts.items() if c in required and v > 0) |
| median = present[len(present) // 2] if present else 0 |
| max_allowed = int(args.max_median_ratio * median) if median else 0 |
| oversized = {c: v for c, v in counts.items() if median and v > max_allowed} |
|
|
| ok = True |
| if len(rows) < args.min_total: |
| ok = False |
| print(f"FAIL total: need {args.min_total}, have {len(rows)}, target {args.target_total}") |
| if deficits: |
| ok = False |
| print("FAIL per-class floor:") |
| for c, d in sorted(deficits.items()): |
| print(f" {c:24s} need +{d}") |
| if oversized: |
| ok = False |
| print(f"FAIL imbalance: median={median}, max_allowed={max_allowed}") |
| for c, v in sorted(oversized.items(), key=lambda kv: (-kv[1], kv[0])): |
| print(f" {c:24s} {v}") |
| if duplicates: |
| ok = False |
| print("FAIL duplicates: prompt-level duplicates must be reviewed") |
| if missing: |
| ok = False |
| print("FAIL schema:") |
| for line_no, key in missing[:20]: |
| print(f" line {line_no}: missing {key}") |
| if len(missing) > 20: |
| print(f" ... {len(missing) - 20} more") |
|
|
| print("PASS preference DPO gate" if ok else "BLOCK DPO") |
| raise SystemExit(0 if ok else 1) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
