File size: 4,024 Bytes
164e036 | 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 | #!/usr/bin/env python3
"""Validate a claim-drafter JSONL dataset before spending training credits.
python3 scripts/validate_dataset.py data/sft
Checks structural integrity, claim-format correctness, dependency validity,
input/target leakage, and length distribution.
Usage:
python3 validate_dataset.py <dir_with_train.jsonl_and_val.jsonl>
"""
import json
import os
import re
import statistics
import sys
def ngrams(text, n=8):
w = re.findall(r"[a-z]+", text.lower())
return set(tuple(w[i : i + n]) for i in range(len(w) - n + 1))
def check(path):
rows = []
for i, line in enumerate(open(path), 1):
try:
rows.append(json.loads(line))
except json.JSONDecodeError as e:
print(f" FAIL malformed JSON at line {i}: {e}")
return None
print(f"\n=== {os.path.basename(path)}: {len(rows)} examples ===")
bad_roles = bad_first = fwd_ref = dep_no_antecedent = 0
ulens, alens, ncl, leaks = [], [], [], []
for r in rows:
m = r.get("messages", [])
if [x.get("role") for x in m] != ["system", "user", "assistant"]:
bad_roles += 1
continue
user, asst = m[1]["content"], m[2]["content"]
claims = asst.split("\n\n")
ulens.append(len(user))
alens.append(len(asst))
ncl.append(len(claims))
if not claims[0].lstrip().startswith("1."):
bad_first += 1
for i, c in enumerate(claims, 1):
if not c.lstrip().startswith(f"{i}."):
bad_first += 1
break
# A dependent claim must reference a strictly earlier claim.
for i, c in enumerate(claims, 1):
for mm in re.finditer(r"\bclaims?\s+(\d+)", c):
if int(mm.group(1)) >= i:
fwd_ref += 1
break
else:
continue
break
# Leakage: share of claim-1 8-grams already present in the prompt.
cg = ngrams(claims[0])
if len(cg) >= 20:
leaks.append(len(ngrams(user) & cg) / len(cg))
print(f" role sequence errors: {bad_roles}")
print(f" claim numbering errors: {bad_first}")
print(f" invalid forward references: {fwd_ref} ({fwd_ref/max(len(rows),1):.1%})")
if leaks:
print(f" claim-1 8-gram leakage: median {statistics.median(leaks):.1%} "
f"mean {statistics.mean(leaks):.1%} >=25%: {sum(1 for x in leaks if x>=.25)/len(leaks):.1%}")
print(f" claims per example: min {min(ncl)} median {int(statistics.median(ncl))} max {max(ncl)}")
print(f" prompt chars: median {int(statistics.median(ulens))} max {max(ulens)}")
print(f" target chars: median {int(statistics.median(alens))} max {max(alens)}")
approx = [(u + a) / 4 for u, a in zip(ulens, alens)]
print(f" approx tokens/example: median {int(statistics.median(approx))} "
f"p99 {int(sorted(approx)[int(len(approx)*.99)])} max {int(max(approx))}")
return {"n": len(rows), "errors": bad_roles + bad_first}
def main():
if len(sys.argv) < 2:
sys.exit(__doc__)
d = sys.argv[1]
total_err = 0
for name in ("train.jsonl", "val.jsonl"):
p = os.path.join(d, name)
if not os.path.exists(p):
print(f"missing {p}")
continue
res = check(p)
if res:
total_err += res["errors"]
# No patent may appear in both splits.
man = os.path.join(d, "manifest.jsonl")
if os.path.exists(man):
splits = {}
for line in open(man):
r = json.loads(line)
splits.setdefault(r["patent_number"], set()).add(r["split"])
overlap = [p for p, s in splits.items() if len(s) > 1]
print(f"\ntrain/val patent overlap: {len(overlap)}")
total_err += len(overlap)
print(f"\n{'PASS' if total_err == 0 else 'ISSUES: ' + str(total_err)}")
if __name__ == "__main__":
main()
|