Codeseys/composer-replication-framework

1"""dedup.py — MinHash near-duplicate detection (finding D-12).2 3Cross-generation dedup is a flywheel-collapse mitigation: a self-training loop4that re-ingests its own outputs accumulates near-identical rows, and per-batch5`document_deduplicator` (the only dedup the old designs had) never sees across6runs. This module computes MinHash signatures over word 5-shingles so a run can7(a) dedup within itself and (b) accept the PRIOR run's signature file and dedup8against it (lineage threaded by `RunManifest.parent_run_id`).9 10Pragmatic v1: builtin-hash permutation MinHash with N=64 seeds, no banding/LSH11(O(n^2) pair scan — fine for Stage-0 corpus sizes, thousands of rows).12`datasketch` (MinHashLSH) is the documented upgrade path when row counts make13the pair scan bite.14 15NOTE on hash stability: Python's builtin `hash()` over str is salted per16process (PYTHONHASHSEED), which would make signatures non-portable across17runs — exactly what cross-generation dedup needs. We therefore use md5-based18hashing (stable everywhere) despite the small speed cost.19"""20from __future__ import annotations21 22import hashlib23import json24import re25from typing import IO, Callable, Iterable, Sequence26 27N_PERMUTATIONS = 6428_SHINGLE_W = 529_WORD_RE = re.compile(r"\w+")30_MAX64 = (1 << 64) - 131 32 33def _shingles(text: str, w: int = _SHINGLE_W) -> set[str]:34    words = _WORD_RE.findall(text.lower())35    if len(words) <= w:36        return {" ".join(words)} if words else set()37    return {" ".join(words[i:i + w]) for i in range(len(words) - w + 1)}38 39 40def _stable_hash(s: str, seed: int) -> int:41    h = hashlib.md5(f"{seed}:{s}".encode()).digest()42    return int.from_bytes(h[:8], "big")43 44 45def minhash_signature(text: str, n_perm: int = N_PERMUTATIONS) -> tuple[int, ...]:46    """MinHash signature: per-seed minimum over the shingle set."""47    sh = _shingles(text)48    if not sh:49        return tuple([_MAX64] * n_perm)50    return tuple(min(_stable_hash(s, seed) for s in sh) for seed in range(n_perm))51 52 53def jaccard_estimate(sig_a: Sequence[int], sig_b: Sequence[int]) -> float:54    """Estimated Jaccard similarity = fraction of agreeing signature slots."""55    if len(sig_a) != len(sig_b) or not sig_a:56        raise ValueError("signatures must be equal-length and non-empty")57    return sum(1 for a, b in zip(sig_a, sig_b) if a == b) / len(sig_a)58 59 60def find_near_duplicates(61    rows: Sequence[dict],62    key_fn: Callable[[dict], str],63    threshold: float = 0.85,64    *,65    prior_signatures: Sequence[Sequence[int]] | None = None,66) -> list[tuple[int, int]]:67    """All (i, j) index pairs whose estimated Jaccard >= threshold.68 69    `prior_signatures` (from a previous run) participate as virtual rows with70    negative indices -(k+1), so a pair (i, -1) means "row i duplicates prior71    signature 0" — the cross-generation case.72    """73    sigs = [minhash_signature(key_fn(r)) for r in rows]74    pairs: list[tuple[int, int]] = []75    for i in range(len(sigs)):76        for j in range(i + 1, len(sigs)):77            if jaccard_estimate(sigs[i], sigs[j]) >= threshold:78                pairs.append((i, j))79        for k, prior in enumerate(prior_signatures or []):80            if jaccard_estimate(sigs[i], prior) >= threshold:81                pairs.append((i, -(k + 1)))82    return pairs83 84 85def dedup(86    rows: Sequence[dict],87    key_fn: Callable[[dict], str],88    threshold: float = 0.85,89    *,90    prior_signatures: Sequence[Sequence[int]] | None = None,91) -> tuple[list[dict], dict]:92    """Keep-first dedup. Returns (kept_rows, stats).93 94    A row duplicating a PRIOR-run signature is dropped outright (the prior run95    already owns it); within-run duplicates keep the earliest occurrence.96    """97    pairs = find_near_duplicates(rows, key_fn, threshold,98                                 prior_signatures=prior_signatures)99    # Partition into disjoint drop-reason sets (Wave-21 review P2: a row that100    # is both a within-run AND cross-generation duplicate must count once;101    # cross-generation wins the attribution since the prior run owns the row).102    drop_cross: set[int] = {i for i, j in pairs if j < 0}103    drop_within: set[int] = {j for _, j in pairs if j >= 0} - drop_cross104    drop = drop_cross | drop_within105    kept = [r for i, r in enumerate(rows) if i not in drop]106    return kept, {107        "rows_in": len(rows),108        "rows_kept": len(kept),109        "dropped_within_run": len(drop_within),110        "dropped_cross_generation": len(drop_cross),111        "threshold": threshold,112    }113 114 115def signatures_to_jsonl(rows: Sequence[dict], key_fn: Callable[[dict], str],116                        fh: IO[str]) -> int:117    """Persist this run's signatures so the NEXT generation can dedup against118    them (pass the loaded list as `prior_signatures`)."""119    n = 0120    for r in rows:121        fh.write(json.dumps(list(minhash_signature(key_fn(r)))) + "\n")122        n += 1123    return n124 125 126def load_signatures(fh: IO[str]) -> list[tuple[int, ...]]:127    return [tuple(json.loads(line)) for line in fh if line.strip()]128 129 130__all__ = [131    "N_PERMUTATIONS",132    "dedup",133    "find_near_duplicates",134    "jaccard_estimate",135    "load_signatures",136    "minhash_signature",137    "signatures_to_jsonl",138]139