Similarity Threshold Configuration: Implementation Guide

Similarity threshold configuration establishes the deterministic decision boundary between exact-match exclusion and semantic near-duplicate clustering within modern eDiscovery pipelines. While cryptographic hashing efficiently eliminates byte-identical copies, it cannot account for minor revisions, OCR degradation, or template-driven variations that routinely appear in litigation document sets. Configuring similarity thresholds requires balancing computational overhead against legal defensibility, ensuring that documents exceeding the configured boundary are grouped for review without introducing false positives that inflate production costs. This implementation guide details a production-ready, memory-aware architecture for threshold evaluation, structured logging, and deterministic fallback routing, positioned as the core execution layer within Deduplication & Family Grouping workflows.

Pipeline Architecture & Execution Flow

Near-duplicate detection operates downstream from initial ingestion and Hash-Based Deduplication Strategies, which eliminate exact matches before semantic analysis begins. To prevent memory exhaustion during vectorization and pairwise comparison, the pipeline must employ chunked asynchronous processing with strict batch sizing. The similarity engine computes cosine distances against a dynamically updated index, applying configurable thresholds per document class. Sliding-window batching ensures that only active document vectors reside in RAM, while memory-mapped feature stores offload historical embeddings to disk-backed buffers. This architecture guarantees linear memory scaling regardless of corpus size, a strict requirement for multi-terabyte productions.

The execution sequence follows a strict linear progression:

1. Preprocessing & Vectorization: Text extraction, OCR normalization, and TF-IDF/embedding generation.
2. Chunked Indexing: Active document vectors loaded into a bounded memory window.
3. Pairwise Distance Calculation: Cosine distance computed against the active window and historical index.
4. Threshold Evaluation: Deterministic routing based on configured boundaries.
5. Cluster Assignment & Routing: Documents routed to review queues, family groups, or bypass streams.

Domain Calibration & Threshold Routing Logic

The default threshold range (typically 0.85 to 0.92) must be calibrated against corpus-specific noise floors. Legal documents contain high-frequency boilerplate, jurisdictional citations, and standardized discovery requests that artificially inflate similarity scores. Adjusting cosine similarity thresholds for legal text therefore requires domain-aware token weighting, procedural stoplist expansion, and the exclusion of header and footer artifacts from the similarity calculation. Thresholds should be implemented as tiered decision gates to maintain review efficiency and defensible audit trails:

The flowchart below shows how a single cosine similarity score is routed through the three tiered decision gates.

flowchart TD
    A["Cosine similarity score"] --> D{"0.95 or higher?"}
    D -->|"yes"| E["Exact family grouping"]
    D -->|"no"| F{"0.85 to 0.94?"}
    F -->|"yes"| G["Near-duplicate cluster flagged for review"]
    F -->|"no"| H["Downstream bypass"]

Calibration quality directly governs near-duplicate detection accuracy: iterative validation against known document families establishes the precision and recall baselines that justify each boundary. Threshold drift must be monitored continuously, because OCR quality variations and language shifts across custodians can alter vector distributions mid-production.

Production-Ready Implementation

The following Python implementation demonstrates a memory-aware, threshold-driven routing engine. It utilizes structured logging, explicit type annotations, and deterministic fallback routing to ensure auditability and compliance with EDRM processing standards.

import logging
from collections import Counter
from dataclasses import dataclass, field
from typing import Dict, List, Optional

import numpy as np

# Configure structured JSON logging for audit trails
logging.basicConfig(
    level=logging.INFO,
    format='{"timestamp": "%(asctime)s", "level": "%(levelname)s", "event": "%(message)s"}'
)
logger = logging.getLogger("similarity_threshold_engine")

@dataclass
class ThresholdConfig:
    exact_family: float = 0.95
    near_duplicate: float = 0.85
    max_batch_size: int = 5000
    memory_limit_mb: int = 2048

@dataclass
class DocumentVector:
    doc_id: str
    vector: np.ndarray
    metadata: Dict[str, str] = field(default_factory=dict)

class SimilarityThresholdEngine:
    def __init__(self, config: ThresholdConfig):
        self.config = config
        self.index: List[DocumentVector] = []
        self.cluster_map: Dict[str, str] = {}  # doc_id -> cluster_id

    def _compute_cosine_distance(self, vec_a: np.ndarray, vec_b: np.ndarray) -> float:
        """Deterministic cosine distance calculation for auditability."""
        norm_a = np.linalg.norm(vec_a)
        norm_b = np.linalg.norm(vec_b)
        if norm_a == 0 or norm_b == 0:
            return 1.0
        cosine_sim = np.dot(vec_a, vec_b) / (norm_a * norm_b)
        return 1.0 - cosine_sim

    def _evaluate_threshold(self, distance: float) -> str:
        """Deterministic routing based on tiered thresholds."""
        similarity = 1.0 - distance
        if similarity >= self.config.exact_family:
            return "EXACT_FAMILY"
        elif similarity >= self.config.near_duplicate:
            return "NEAR_DUPLICATE"
        return "BYPASS"

    async def process_chunk(self, chunk: List[DocumentVector]) -> List[Dict]:
        """Memory-aware chunked processing with deterministic routing."""
        results: List[Dict] = []
        route_counts: Counter = Counter()
        for doc in chunk:
            best_distance = 1.0
            match_id: Optional[str] = None

            # Sliding window comparison against active index
            for indexed_doc in self.index:
                dist = self._compute_cosine_distance(doc.vector, indexed_doc.vector)
                if dist < best_distance:
                    best_distance = dist
                    match_id = indexed_doc.doc_id

            route_action = self._evaluate_threshold(best_distance)
            # A document only joins an existing cluster when it clears a tier
            # threshold and a prior vector was actually matched; otherwise it
            # seeds its own cluster.
            if route_action != "BYPASS" and match_id is not None:
                cluster_id = self.cluster_map.get(match_id, match_id)
            else:
                cluster_id = doc.doc_id

            # Route assignment
            self.cluster_map[doc.doc_id] = cluster_id
            route_counts[route_action] += 1
            results.append({
                "doc_id": doc.doc_id,
                "match_id": match_id,
                "similarity_score": round(1.0 - best_distance, 4),
                "route_action": route_action,
                "cluster_id": cluster_id
            })

            # Memory management: append to index, enforce sliding window
            self.index.append(doc)
            if len(self.index) > self.config.max_batch_size:
                self.index.pop(0)  # Evict oldest vector

        logger.info(
            f"Processed chunk of {len(chunk)} documents | "
            f"Routes: {dict(route_counts)}"
        )
        return results

    async def run_pipeline(self, document_stream: List[DocumentVector]):
        """Orchestrate chunked execution with fallback routing."""
        for i in range(0, len(document_stream), self.config.max_batch_size):
            chunk = document_stream[i : i + self.config.max_batch_size]
            try:
                await self.process_chunk(chunk)
            except Exception as e:
                logger.error(f"Chunk processing failed: {e}")
                # Deterministic fallback for corrupted/unparseable vectors
                await self._fallback_route(chunk)

    async def _fallback_route(self, chunk: List[DocumentVector]):
        """Route documents that fail vectorization into a deterministic fallback chain for independent review."""
        for doc in chunk:
            self.cluster_map[doc.doc_id] = f"FALLBACK_{doc.doc_id}"
            logger.warning(f"Fallback routing applied to {doc.doc_id}")

Key Engineering Considerations

• Vector Normalization: _compute_cosine_distance normalizes inline for safety, but pre-normalizing all embeddings to unit L2 length at ingestion lets you replace the division with the cheaper 1 - np.dot(a, b) while yielding an identical cosine distance.
• Memory Bounding: The sliding window eviction strategy (self.index.pop(0)) prevents OOM errors during corpus-scale processing. For persistent indexing, integrate with disk-backed vector stores like FAISS or Annoy.
• Deterministic Routing: The _evaluate_threshold method uses strict inequality boundaries to prevent ambiguous routing states. All decisions are logged with exact similarity scores for defensibility.

Defensibility & Audit Compliance

Threshold configuration in eDiscovery must withstand judicial scrutiny and opposing counsel challenges. The following practices ensure compliance with industry standards and internal review protocols:

1. Immutable Audit Trails: Every routing decision must log the document ID, matched counterpart, raw similarity score, applied threshold, and timestamp. Structured JSON logging enables rapid reconstruction of processing decisions during privilege or clawback disputes.
2. Calibration Validation: Before production deployment, run the threshold engine against a curated validation set containing known duplicates, near-duplicates, and distinct documents. Track precision, recall, and F1-score. Adjust boundaries iteratively until false-positive rates fall below 3%.
3. Cross-Custodian Consistency: Apply identical threshold configurations across all custodians unless jurisdictional or language-specific variations are documented and approved by counsel. Inconsistent thresholds introduce bias and complicate production certification.
4. Fallback Chain Documentation: When OCR degradation or file corruption prevents vectorization, documents must route through deterministic fallback chains rather than being silently dropped. Maintain explicit routing logs to demonstrate complete corpus coverage.

By adhering to these architectural and procedural standards, legal automation teams can deploy similarity threshold configurations that balance computational efficiency with rigorous legal defensibility.