Deduplicator
On this page
Module: chunks.deduplicator
Import: from zerda.chunks import Deduplicator
Table of Contents
- Purpose
- Constructor
- Public Methods
- Deduplication Algorithm
- Hash-Based vs Similarity-Based
- Complexity
- Examples
Purpose
Deduplicator removes duplicate and near-duplicate chunks from a list, using fast SHA-256 hash comparison first and optional cosine similarity comparison second.
It runs as the fourth stage in the ChunkManager pipeline, after metadata enrichment.
Constructor
Deduplicator(
config: Optional[ChunkConfig] = None,
*,
similarity_threshold: float = 0.95,
use_hash: bool = True,
use_similarity: bool = True,
)
Public Methods
deduplicate(chunks: List[DocumentChunk]) → List[DocumentChunk]
Returns a new list with duplicate chunks removed. Does not mutate the input list.
Returns: List[DocumentChunk] — unique chunks in original order. First occurrence wins.
Empty input: returns [].
Single chunk: returns [chunk] (no dedup needed).
Deduplication Algorithm
Stage 1 — Hash dedup (if use_hash=True):
seen_hashes = set()
for each chunk:
if chunk.content_hash in seen_hashes:
discard
else:
seen_hashes.add(chunk.content_hash)
keep chunk
Stage 2 — Similarity dedup (if use_similarity=True):
embed all surviving chunk texts via EmbeddingProvider
for each new chunk i:
compute cosine similarity against all previously kept chunks
if max(similarity) >= similarity_threshold:
discard (near-duplicate)
else:
keep
content_hash is computed in DocumentChunk.__post_init__ as sha256(text.encode()).hexdigest().
Hash-Based vs Similarity-Based
Use use_similarity=False for large corpora (>10k chunks) where similarity is too slow.
Complexity
Where N = total chunks, M = hash-unique chunks, D = embedding dimension.
Examples
Standalone Use
from zerda.chunks import Deduplicator, SemanticChunker
chunker = SemanticChunker()
chunks = chunker.chunk(text_with_duplicates)
dedup = Deduplicator(similarity_threshold=0.95)
unique = dedup.deduplicate(chunks)
print(f"Before: {len(chunks)}, After: {len(unique)}")
Hash Only (Fast Mode)
dedup = Deduplicator(use_similarity=False)
unique = dedup.deduplicate(chunks)
Custom Threshold
# More aggressive: remove chunks that are 85%+ similar
dedup = Deduplicator(similarity_threshold=0.85)
Disable in ChunkManager
config = ChunkConfig(deduplication_enabled=False)
manager = ChunkManager(config=config)