File size: 2,572 Bytes
c7893fa | 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 | """Keep new training data away from the evaluation sets.
Two rules:
1. Task families used only for zero-shot evaluation (data/heldout.jsonl: trec question types,
formality, spam) must never be trained on. `blocked_task(name)` checks dataset/task names.
2. No training state may overlap an eval state (heldout, test, bench): exact match after
normalization, or any shared 12-word shingle. `is_contaminated(state)` checks this.
from jevlike.data.decontam import Decontaminator
dc = Decontaminator() # loads the eval sets once (~1 s)
if dc.blocked_task(task_name) or dc.is_contaminated(state): skip
CLI: python -m jevlike.data.decontam IN.jsonl OUT.jsonl (filters a Record JSONL)
"""
from __future__ import annotations
import re
import sys
from pathlib import Path
from jevlike.types import read_jsonl, write_jsonl
ROOT = Path(__file__).resolve().parents[2]
EVAL_FILES = ["data/heldout.jsonl", "data/test.jsonl", "data/bench/decisions.jsonl",
"data/eval_newsfin/zeroshot.jsonl", "data/eval_newsfin/zeroshot_long.jsonl"]
BLOCKED_PATTERNS = [r"\btrec\b", r"question[\s_-]*(type|classification)", r"formal", r"spam", r"enron"]
SHINGLE = 12
def _norm(s: str) -> str:
return re.sub(r"\W+", " ", s.lower()).strip()
def _shingles(s: str) -> set[str]:
w = _norm(s).split()
if len(w) < SHINGLE:
return {" ".join(w)}
return {" ".join(w[i:i + SHINGLE]) for i in range(len(w) - SHINGLE + 1)}
class Decontaminator:
def __init__(self, eval_files: list[str] = EVAL_FILES):
self.exact: set[str] = set()
self.shingles: set[str] = set()
for f in eval_files:
p = ROOT / f
if not p.exists():
continue
for r in read_jsonl(str(p)):
self.exact.add(_norm(r.state))
self.shingles |= _shingles(r.state)
@staticmethod
def blocked_task(name: str) -> bool:
"""True for dataset/task names in the held-out task families (match on name or description)."""
return any(re.search(p, name.lower()) for p in BLOCKED_PATTERNS)
def is_contaminated(self, state: str) -> bool:
return _norm(state) in self.exact or not self.shingles.isdisjoint(_shingles(state))
if __name__ == "__main__":
src, dst = sys.argv[1], sys.argv[2]
dc = Decontaminator()
rs = read_jsonl(src)
keep = [r for r in rs if not (dc.blocked_task(r.source) or dc.is_contaminated(r.state))]
write_jsonl(dst, keep)
print(f"kept {len(keep)}/{len(rs)} ({len(rs) - len(keep)} dropped)")
|