Spaces:
Running
Running
| from typing import List, Dict, Any, Set, Optional | |
| from datasketch import MinHash, MinHashLSH | |
| class DuplicateDetector: | |
| def __init__(self, threshold: float = 0.92, num_perm: int = 128, ngram_size: int = 5): | |
| self.threshold = threshold | |
| self.num_perm = num_perm | |
| self.ngram_size = ngram_size | |
| self.lsh = MinHashLSH(threshold=threshold, num_perm=num_perm) | |
| self.seen_ids: Set[str] = set() | |
| def _text_to_shingles(self, text: str) -> Set[str]: | |
| tokens = text.lower().split() | |
| shingles: Set[str] = set() | |
| n = self.ngram_size | |
| if len(tokens) < n: | |
| shingles.add(" ".join(tokens)) | |
| return shingles | |
| for i in range(len(tokens) - n + 1): | |
| shingles.add(" ".join(tokens[i : i + n])) | |
| return shingles | |
| def _compute_minhash(self, text: str) -> MinHash: | |
| shingles = self._text_to_shingles(text) | |
| m = MinHash(num_perm=self.num_perm) | |
| for s in shingles: | |
| m.update(s.encode("utf-8")) | |
| return m | |
| def is_duplicate(self, doc_id: str, text: str) -> bool: | |
| if doc_id in self.seen_ids: | |
| return True | |
| m = self._compute_minhash(text) | |
| candidates = self.lsh.query(m) | |
| if candidates: | |
| return True | |
| self.lsh.insert(doc_id, m) | |
| self.seen_ids.add(doc_id) | |
| return False | |
| def filter_duplicates(self, documents: List[Dict[str, Any]]) -> List[Dict[str, Any]]: | |
| unique: List[Dict[str, Any]] = [] | |
| for doc in documents: | |
| doc_id = doc.get("chunk_id") or doc.get("node_id") or "" | |
| text = doc.get("text", "") | |
| if not text: | |
| continue | |
| if not self.is_duplicate(doc_id, text): | |
| unique.append(doc) | |
| return unique | |
| def reset(self): | |
| self.lsh = MinHashLSH(threshold=self.threshold, num_perm=self.num_perm) | |
| self.seen_ids.clear() | |