Repository Public
1"""Memorisation and contamination audit.23* **Nearest-neighbour overlap** — for each reconstruction on held-out items, the maximum 5-gram Jaccard (MinHash) and4 the longest common word n-gram against the TRAINING targets. A reconstruction that reproduces a training original5 (not its own test original) signals memorisation / regurgitation.6* **Verbatim-copy rate** — fraction of outputs with normalised edit similarity ≥ 0.95 to any training target.7* **Canaries (optional, lawful: synthetic text only)** — synthetic passages with unique markers inserted into the8 training set a fixed number of times; at test time we check whether the model emits the marker given an unrelated9 rewrite. Implemented as helpers; the canary set is generated locally and never contains real data.10"""11from __future__ import annotations1213import random14import re1516from datasketch import MinHash, MinHashLSH1718from htr.dedup.minhash import minhash19from htr.utils.hashing import normalize_for_hash2021_TOK = re.compile(r"\w+")222324def _ngrams(text: str, n: int) -> set[tuple[str, ...]]:25 t = _TOK.findall(normalize_for_hash(text))26 return {tuple(t[i : i + n]) for i in range(max(0, len(t) - n + 1))}272829class TrainingIndex:30 def __init__(self, train_targets: dict[str, str], threshold: float = 0.3, num_perm: int = 128):31 self.lsh = MinHashLSH(threshold=threshold, num_perm=num_perm)32 self.num_perm = num_perm33 self.targets = train_targets34 self.mh: dict[str, MinHash] = {}35 for pid, t in train_targets.items():36 m = minhash(t, num_perm, 5)37 self.mh[pid] = m38 self.lsh.insert(pid, m)3940 def nearest(self, text: str) -> tuple[str | None, float]:41 m = minhash(text, self.num_perm, 5)42 best, best_j = None, 0.043 for cand in self.lsh.query(m):44 j = float(m.jaccard(self.mh[cand]))45 if j > best_j:46 best, best_j = cand, j47 return best, best_j484950def longest_common_ngram(a: str, b: str, max_n: int = 40) -> int:51 ta, tb = _TOK.findall(normalize_for_hash(a)), _TOK.findall(normalize_for_hash(b))52 sb = set()53 best = 054 for n in range(3, max_n + 1):55 grams_a = {tuple(ta[i : i + n]) for i in range(max(0, len(ta) - n + 1))}56 grams_b = {tuple(tb[i : i + n]) for i in range(max(0, len(tb) - n + 1))}57 if grams_a & grams_b:58 best = n59 else:60 break61 return best626364def audit_outputs(outputs: dict[str, str], eval_pairs: list[dict], train_targets: dict[str, str]) -> dict:65 idx = TrainingIndex(train_targets)66 rows = []67 for p in eval_pairs:68 out = outputs.get(p["pair_id"])69 if out is None:70 continue71 nn, jac = idx.nearest(out)72 lcn = longest_common_ngram(out, train_targets[nn]) if nn else 073 rows.append({"pair_id": p["pair_id"], "nn_train_passage": nn, "jaccard_to_train": jac, "longest_common_ngram": lcn, "same_passage_leak": nn == p["passage_id"]})74 if not rows:75 return {"n": 0}76 jac = [r["jaccard_to_train"] for r in rows]77 return {78 "n": len(rows),79 "mean_max_jaccard_to_train": sum(jac) / len(jac),80 "frac_jaccard_ge_0_5": sum(j >= 0.5 for j in jac) / len(jac),81 "frac_jaccard_ge_0_8": sum(j >= 0.8 for j in jac) / len(jac),82 "max_longest_common_ngram": max(r["longest_common_ngram"] for r in rows),83 "frac_lcn_ge_12": sum(r["longest_common_ngram"] >= 12 for r in rows) / len(rows),84 "n_same_passage_leak": sum(r["same_passage_leak"] for r in rows), # must be 0 with correct splits85 "worst": sorted(rows, key=lambda r: -r["jaccard_to_train"])[:5],86 }878889def make_canaries(n: int = 5, seed: int = 0) -> list[dict]:90 """Synthetic canary pairs: an invented study with a unique marker token in the target only."""91 rng = random.Random(seed)92 out = []93 for i in range(n):94 marker = f"ZQX{rng.randrange(10**6):06d}"95 tgt = f"The specimen series labelled {marker} was examined in triplicate, and the recorded mass of 12.5 g did not vary between trials. No statistically significant difference emerged between the two preparations (p = 0.41)."96 src = "The specimen series was examined in triplicate; its recorded mass of 12.5 g remained constant across trials, and the two preparations did not differ significantly (p = 0.41)."97 out.append({"pair_id": f"canary_{i}", "passage_id": f"canary_{i}", "doc_id": f"canary_{i}", "variant": "A", "sample_index": 0, "source_text": src, "target_text": tgt, "marker": marker, "metadata": {"discipline": "other", "era": "2010-2017", "section_role": "methods", "source": "local"}})98 return out99100101def canary_leak_rate(outputs: dict[str, str], canaries: list[dict]) -> float:102 hits = sum(1 for c in canaries if c["marker"] in outputs.get(c["pair_id"], ""))103 return hits / max(1, len(canaries))104