rag.indexing.deduplication

Chunk deduplication for cleaner vector indices.

class ChunkDeduplicator(mode='exact', similarity_threshold=0.95)[source]

Bases: object

Detect and filter duplicate/near-duplicate chunks.

Supports multiple deduplication modes: - exact: Hash-based exact content matching - similarity: Embedding cosine similarity threshold

Parameters:

mode (str) – Deduplication mode (“exact” or “similarity”).
similarity_threshold (float) – Threshold for similarity-based dedup (0.0-1.0).

Example

>>> dedup = ChunkDeduplicator(mode="exact")
>>> chunks, embeddings = dedup.filter_duplicates(chunks, embeddings)

compute_hash(content)[source]

Compute content hash for exact deduplication.

Parameters:: content (str) – Chunk text content.
Returns:: SHA256 hash (first 16 chars).
Return type:: str

is_duplicate(content, embedding=None)[source]

Check if chunk is a duplicate of previously seen content.

Parameters:

content (str) – Chunk text content.
embedding (list[float] | None) – Optional embedding vector for similarity-based dedup.

Returns:

True if chunk is a duplicate.

Return type:

bool

filter_duplicates(chunks, embeddings)[source]

Filter duplicate chunks from a batch.

Parameters:

chunks (list[Any]) – List of chunk objects (must have .content attribute or be string).
embeddings (list[list[float]]) – List of embedding vectors.

Returns:

Tuple of (filtered_chunks, filtered_embeddings).

Return type:

tuple[list[Any], list[list[float]]]

clear()[source]

Clear seen hashes and embeddings.

Return type:: None

get_stats()[source]

Get deduplicator statistics.

Returns:: Dict with mode, threshold, and counts.
Return type:: dict[str, Any]

Classes

class ChunkDeduplicator(mode='exact', similarity_threshold=0.95)[source]

Bases: object

Detect and filter duplicate/near-duplicate chunks.

Supports multiple deduplication modes: - exact: Hash-based exact content matching - similarity: Embedding cosine similarity threshold

Parameters:

mode (str) – Deduplication mode (“exact” or “similarity”).
similarity_threshold (float) – Threshold for similarity-based dedup (0.0-1.0).

Example

>>> dedup = ChunkDeduplicator(mode="exact")
>>> chunks, embeddings = dedup.filter_duplicates(chunks, embeddings)

compute_hash(content)[source]

Compute content hash for exact deduplication.

Parameters:: content (str) – Chunk text content.
Returns:: SHA256 hash (first 16 chars).
Return type:: str

is_duplicate(content, embedding=None)[source]

Check if chunk is a duplicate of previously seen content.

Parameters:

content (str) – Chunk text content.
embedding (list[float] | None) – Optional embedding vector for similarity-based dedup.

Returns:

True if chunk is a duplicate.

Return type:

bool

filter_duplicates(chunks, embeddings)[source]

Filter duplicate chunks from a batch.

Parameters:

chunks (list[Any]) – List of chunk objects (must have .content attribute or be string).
embeddings (list[list[float]]) – List of embedding vectors.

Returns:

Tuple of (filtered_chunks, filtered_embeddings).

Return type:

tuple[list[Any], list[list[float]]]

clear()[source]

Clear seen hashes and embeddings.

Return type:: None

get_stats()[source]

Get deduplicator statistics.

Returns:: Dict with mode, threshold, and counts.
Return type:: dict[str, Any]