File size: 3,701 Bytes
76b78ee | 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 | #!/usr/bin/env python3
"""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()
|