Implementing semantic deduplication with vector similarity

Why Exact Matching Fails in Practice

Exact or hash-based deduplication works only when two records are byte-for-byte identical. In practice, duplicates arrive from multiple sources with inconsistent formatting, different levels of detail, varying punctuation, and partial updates that change one field while keeping everything else the same.

Consider a product catalogue where these three entries all represent the same physical item:

"Samsung Galaxy S24 Ultra 256GB Phantom Black"
"Galaxy S24 Ultra - 256 GB, Black (Samsung)"
"SAMSUNG S24 ULTRA 256G BLK"

No hash-based or fuzzy-string approach handles all three reliably. Embedding-based deduplication does, because all three sentences encode to vectors that cluster tightly in semantic space.

Semantic deduplication applies to a broad class of problems: support ticket merging, news article clustering, academic citation deduplication, customer address normalisation, e-commerce catalogue cleaning, and knowledge base quality control. The core algorithm is the same in each case: embed all records, find pairs (or clusters) with cosine similarity above a threshold, and decide how to merge or flag them.

Choosing the Right Embedding Model and Threshold

The embedding model and similarity threshold are the two most consequential decisions. They must be calibrated together on your specific data.

Embedding model selection:

For short texts like product titles or support ticket subjects, 768-dimensional models are often sufficient. Longer documents benefit from larger models or chunked-embedding strategies.

Threshold calibration:

A cosine similarity of 1.0 means identical vectors. Real duplicates typically cluster between 0.90 and 0.99. Non-duplicates in the same domain usually fall between 0.70 and 0.89. The exact boundary depends on your domain's vocabulary diversity.

To calibrate, manually label 200 pairs across the similarity range and plot precision-recall curves:

import numpy as np
from sklearn.metrics import precision_recall_curve
from typing import List, Tuple

def calibrate_threshold(
    labeled_pairs: List[Tuple[float, int]]  # (similarity, is_duplicate: 0 or 1)
) -> float:
    """Find the threshold that maximises F1 score on labeled pairs."""
    similarities = np.array([p[0] for p in labeled_pairs])
    labels = np.array([p[1] for p in labeled_pairs])

    precisions, recalls, thresholds = precision_recall_curve(labels, similarities)
    f1_scores = 2 * (precisions * recalls) / (precisions + recalls + 1e-8)
    best_idx = np.argmax(f1_scores[:-1])  # last element has no threshold

    best_threshold = thresholds[best_idx]
    print(f"Best threshold: {best_threshold:.4f}")
    print(f"Precision at threshold: {precisions[best_idx]:.3f}")
    print(f"Recall at threshold: {recalls[best_idx]:.3f}")
    print(f"F1 at threshold: {f1_scores[best_idx]:.3f}")
    return best_threshold

For a production system, different thresholds for different record types are normal. Product titles in a homogeneous catalogue might use 0.95; support ticket bodies in a diverse helpdesk might use 0.88.

Database Schema for Scale

The core challenge in deduplication at scale is that naive O(n²) pairwise comparison is prohibitive. For 100,000 records, that's 5 billion comparisons. The solution is approximate nearest neighbour (ANN) search: for each record, find only the records most likely to be duplicates.

-- Enable pgvector
CREATE EXTENSION IF NOT EXISTS vector;

-- Records to deduplicate (e.g., product listings)
CREATE TABLE records (
  id          BIGSERIAL PRIMARY KEY,
  external_id TEXT UNIQUE,
  content     TEXT NOT NULL,
  source      TEXT NOT NULL,
  embedding   VECTOR(1536),
  canonical_id BIGINT REFERENCES records(id),  -- NULL = is canonical
  status      TEXT DEFAULT 'unprocessed' 
              CHECK (status IN ('unprocessed','canonical','duplicate','review')),
  processed_at TIMESTAMPTZ,
  created_at  TIMESTAMPTZ DEFAULT NOW()
);

-- HNSW index for fast ANN search
CREATE INDEX records_embedding_hnsw
ON records
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

-- Pair candidates found by ANN search, awaiting resolution
CREATE TABLE duplicate_candidates (
  id              BIGSERIAL PRIMARY KEY,
  record_a_id     BIGINT NOT NULL REFERENCES records(id),
  record_b_id     BIGINT NOT NULL REFERENCES records(id),
  similarity      FLOAT NOT NULL,
  resolution      TEXT CHECK (resolution IN ('duplicate','not_duplicate','merged')),
  resolved_at     TIMESTAMPTZ,
  resolved_by     TEXT,  -- 'auto' or user ID
  created_at      TIMESTAMPTZ DEFAULT NOW(),
  CONSTRAINT ordered_pair CHECK (record_a_id < record_b_id),
  UNIQUE (record_a_id, record_b_id)
);

CREATE INDEX dup_candidates_similarity ON duplicate_candidates(similarity DESC);
CREATE INDEX dup_candidates_unresolved ON duplicate_candidates(resolution)
  WHERE resolution IS NULL;

The canonical_id pattern lets you maintain a clean view of the dataset. Canonical records (the "source of truth" for a cluster) have canonical_id = NULL. Duplicates point to their canonical record. Queries against the clean dataset filter WHERE canonical_id IS NULL OR id = canonical_id.

The Embedding and Candidate Generation Pipeline

import openai
import psycopg2
import psycopg2.extras
from typing import List, Dict

client = openai.OpenAI()

def preprocess_for_embedding(record: Dict) -> str:
    """
    Normalise before embedding: lowercasing, unit standardisation,
    removing noise. Domain-specific normalisation here is worth the effort.
    """
    text = record["content"].lower().strip()
    # Standardise common abbreviations in your domain
    replacements = {
        " gb": " gigabytes",
        " tb": " terabytes",
        " blk": " black",
        " wht": " white",
    }
    for abbr, full in replacements.items():
        text = text.replace(abbr, full)
    return text

def embed_records_batch(records: List[Dict], model: str = "text-embedding-3-small") -> List[List[float]]:
    texts = [preprocess_for_embedding(r) for r in records]
    response = client.embeddings.create(input=texts, model=model)
    return [item.embedding for item in response.data]

def ingest_records(conn, records: List[Dict]) -> None:
    """Embed and store records without triggering deduplication yet."""
    embeddings = embed_records_batch(records)

    rows = [
        (r["external_id"], r["content"], r["source"], emb)
        for r, emb in zip(records, embeddings)
    ]

    with conn.cursor() as cur:
        psycopg2.extras.execute_values(
            cur,
            """
            INSERT INTO records (external_id, content, source, embedding)
            VALUES %s
            ON CONFLICT (external_id) DO UPDATE
              SET content = EXCLUDED.content,
                  embedding = EXCLUDED.embedding,
                  status = 'unprocessed'
            """,
            rows,
            template="(%s, %s, %s, %s::vector)"
        )
    conn.commit()

Finding Duplicate Candidates with ANN Search

The candidate generation step uses pgvector's nearest-neighbour search to find candidate pairs. For each unprocessed record, find its N nearest neighbours above the threshold and insert them into duplicate_candidates:

def find_candidates_for_record(
    conn,
    record_id: int,
    threshold: float = 0.92,
    top_k: int = 10
) -> int:
    """
    Find near-duplicate candidates for a single record.
    Returns the number of new candidates inserted.
    """
    with conn.cursor() as cur:
        cur.execute(
            """
            WITH neighbours AS (
                SELECT
                    r.id AS neighbour_id,
                    1 - (r.embedding <=> ref.embedding) AS similarity
                FROM records ref
                JOIN records r ON r.id != ref.id
                WHERE ref.id = %s
                  AND r.status != 'duplicate'
                  AND 1 - (r.embedding <=> ref.embedding) >= %s
                ORDER BY ref.embedding <=> r.embedding
                LIMIT %s
            )
            INSERT INTO duplicate_candidates (record_a_id, record_b_id, similarity)
            SELECT
                LEAST(%s, neighbour_id),
                GREATEST(%s, neighbour_id),
                similarity
            FROM neighbours
            ON CONFLICT (record_a_id, record_b_id) DO NOTHING
            """,
            (record_id, threshold, top_k, record_id, record_id)
        )
        inserted = cur.rowcount

    conn.commit()
    return inserted

def run_candidate_generation(conn, threshold: float = 0.92, batch_size: int = 500) -> None:
    """Process all unprocessed records in batches."""
    with conn.cursor() as cur:
        cur.execute(
            "SELECT id FROM records WHERE status = 'unprocessed' ORDER BY id"
        )
        record_ids = [row[0] for row in cur.fetchall()]

    total_candidates = 0
    for i, record_id in enumerate(record_ids):
        total_candidates += find_candidates_for_record(conn, record_id, threshold)

        if (i + 1) % 100 == 0:
            print(f"Processed {i + 1}/{len(record_ids)} records, {total_candidates} candidates found")

        # Mark as processed so it won't be re-queried
        with conn.cursor() as cur:
            cur.execute(
                "UPDATE records SET status = 'canonical', processed_at = NOW() WHERE id = %s",
                (record_id,)
            )
        conn.commit()

Auto-Resolution and Human Review Triage

High-confidence pairs (similarity above a hard threshold like 0.97) can be resolved automatically. Lower-confidence pairs go to a review queue. This stratified approach maximises automation while protecting against false positives:

def auto_resolve_candidates(
    conn,
    auto_merge_threshold: float = 0.97,
    auto_reject_threshold: float = 0.88
) -> Dict[str, int]:
    """
    Auto-resolve high and low confidence pairs.
    Returns counts by resolution type.
    """
    counts = {"auto_duplicate": 0, "auto_not_duplicate": 0, "sent_to_review": 0}

    with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
        cur.execute(
            """
            SELECT id, record_a_id, record_b_id, similarity
            FROM duplicate_candidates
            WHERE resolution IS NULL
            ORDER BY similarity DESC
            """
        )
        candidates = cur.fetchall()

    for candidate in candidates:
        sim = candidate["similarity"]

        if sim >= auto_merge_threshold:
            resolution = "duplicate"
            resolved_by = "auto"
            # Mark the lower-id record as the duplicate of the higher-id canonical
            mark_as_duplicate(conn, candidate["record_b_id"], candidate["record_a_id"])
            counts["auto_duplicate"] += 1
        elif sim < auto_reject_threshold:
            resolution = "not_duplicate"
            resolved_by = "auto"
            counts["auto_not_duplicate"] += 1
        else:
            # Middle band goes to human review
            counts["sent_to_review"] += 1
            continue

        with conn.cursor() as cur:
            cur.execute(
                """
                UPDATE duplicate_candidates
                SET resolution = %s, resolved_by = %s, resolved_at = NOW()
                WHERE id = %s
                """,
                (resolution, resolved_by, candidate["id"])
            )
        conn.commit()

    return counts

def mark_as_duplicate(conn, duplicate_id: int, canonical_id: int) -> None:
    """Mark a record as a duplicate of another."""
    with conn.cursor() as cur:
        cur.execute(
            """
            UPDATE records
            SET canonical_id = %s, status = 'duplicate'
            WHERE id = %s
            """,
            (canonical_id, duplicate_id)
        )
    conn.commit()

Querying the Deduplicated Dataset

After deduplication runs, queries against the clean dataset should filter on status:

-- Clean catalogue view: one row per canonical item
CREATE VIEW clean_records AS
SELECT r.*
FROM records r
WHERE r.status = 'canonical'
   OR (r.status = 'unprocessed' AND r.canonical_id IS NULL);

-- Count deduplication effectiveness
SELECT
    status,
    COUNT(*) as record_count,
    ROUND(100.0 * COUNT(*) / SUM(COUNT(*)) OVER (), 1) as pct
FROM records
GROUP BY status
ORDER BY record_count DESC;

For applications that need to resolve duplicates at query time (e.g., showing all variants of an item), use the canonical_id join:

-- All variants of a canonical item
SELECT r.*
FROM records r
WHERE r.canonical_id = :canonical_id
   OR r.id = :canonical_id
ORDER BY r.created_at;

Incremental Deduplication for Ongoing Ingestion

A production system ingests new records continuously. You don't want to re-process the entire corpus for each new batch. The incremental pattern only compares new records against existing canonical records:

def deduplicate_new_records(
    conn,
    new_record_ids: List[int],
    threshold: float = 0.92
) -> None:
    """
    Compare only new records against existing canonical records.
    This avoids O(n²) growth as the corpus scales.
    """
    for record_id in new_record_ids:
        with conn.cursor() as cur:
            cur.execute(
                """
                INSERT INTO duplicate_candidates (record_a_id, record_b_id, similarity)
                SELECT
                    LEAST(%s, r.id),
                    GREATEST(%s, r.id),
                    1 - (r.embedding <=> new_rec.embedding)
                FROM records new_rec
                CROSS JOIN records r
                WHERE new_rec.id = %s
                  AND r.id != %s
                  AND r.status = 'canonical'
                  AND 1 - (r.embedding <=> new_rec.embedding) >= %s
                ORDER BY r.embedding <=> new_rec.embedding
                LIMIT 10
                ON CONFLICT (record_a_id, record_b_id) DO NOTHING
                """,
                (record_id, record_id, record_id, record_id, threshold)
            )
        conn.commit()

The operational cost of semantic deduplication is modest compared to its impact on data quality. A catalogue with 5% duplicate rate that receives 1 million searches per day is showing users irrelevant results in thousands of cases. Embedding-based deduplication with a well-calibrated threshold and a human-in-the-loop review layer for the ambiguous middle band eliminates the majority of these failures without requiring manual review of every record pair.