AI Content Deduplicator with Similarity Score

AI Content Deduplicator with Similarity Score

Problem Statement

Content pipelines — RSS aggregators, news scrapers, CMS imports, and LLM-generated article feeds — frequently produce near-duplicate documents. Two articles may cover the same wire story with slightly different headlines, or an LLM may regenerate a summary that is 95% identical to one already published. Exact string matching misses these, and naive fuzzy matching (e.g., Levenshtein, Jaccard on words) fails badly on paraphrases and reordering.

This script solves the problem by embedding each document with OpenAI's embedding model, then computing cosine similarity between embeddings to detect semantic duplicates. It uses a union-find (disjoint-set) structure to cluster transitively similar items, so A≈B and B≈C correctly collapses A, B, and C into one cluster even if A and C fall just under the threshold. The output is a cleaned JSONL file containing one representative per cluster.

How It Works

The script streams input records from a JSONL file, extracts a text field from each, batches them through openai.embeddings.create, and stores L2-normalized vectors in a NumPy array. Because vectors are normalized, cosine similarity reduces to a single dot product, so the full N×N similarity matrix is computed as E @ E.T — one BLAS call, no Python loops.

A threshold (default 0.92) converts the similarity matrix into a boolean adjacency mask. The upper triangle is scanned, and any pair above threshold is unioned in a disjoint-set structure. Finally, clusters are collapsed to their first-seen member, and the representative records are written to the output JSONL. The script handles batching, retries with exponential backoff, malformed JSON lines, and missing text fields gracefully.

Full Script

#!/usr/bin/env python3
"""
AI Content Deduplicator with Similarity Score.

Reads a JSONL file of documents, embeds each document's text field via the
OpenAI embeddings API, clusters near-duplicates using cosine similarity,
and writes one representative per cluster to an output JSONL file.

Usage:
    python3 dedup_ai.py --input articles.jsonl --threshold 0.92 --output unique.jsonl
"""

from __future__ import annotations

import argparse
import json
import logging
import os
import sys
import time
from pathlib import Path
from typing import Any, Iterable

import numpy as np
from openai import OpenAI, APIError, RateLimitError, APIConnectionError

# ---------------------------------------------------------------------------
# Configuration (overridable via environment variables)
# ---------------------------------------------------------------------------
OPENAI_API_KEY: str | None = os.environ.get("OPENAI_API_KEY")
EMBED_MODEL: str = os.environ.get("EMBED_MODEL", "text-embedding-3-small")
BATCH_SIZE: int = int(os.environ.get("EMBED_BATCH_SIZE", "64"))
MAX_RETRIES: int = int(os.environ.get("EMBED_MAX_RETRIES", "5"))
TEXT_FIELD: str = os.environ.get("TEXT_FIELD", "text")
LOG_LEVEL: str = os.environ.get("LOG_LEVEL", "INFO")

logging.basicConfig(
    level=LOG_LEVEL,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%H:%M:%S",
)
log = logging.getLogger("dedup")


# ---------------------------------------------------------------------------
# Disjoint-set (union-find) with path compression + union by rank
# ---------------------------------------------------------------------------
class UnionFind:
    """Tracks equivalence classes of document indices."""

    def __init__(self, n: int) -> None:
        self.parent: list[int] = list(range(n))
        self.rank: list[int] = [0] * n

    def find(self, x: int) -> int:
        # Iterative path compression to avoid recursion limits on long chains
        root = x
        while self.parent[root] != root:
            root = self.parent[root]
        while self.parent[x] != root:
            self.parent[x], x = root, self.parent[x]
        return root

    def union(self, a: int, b: int) -> None:
        ra, rb = self.find(a), self.find(b)
        if ra == rb:
            return
        if self.rank[ra] < self.rank[rb]:
            ra, rb = rb, ra
        self.parent[rb] = ra
        if self.rank[ra] == self.rank[rb]:
            self.rank[ra] += 1


# ---------------------------------------------------------------------------
# I/O helpers
# ---------------------------------------------------------------------------
def load_jsonl(path: Path) -> list[dict[str, Any]]:
    """Load records from a JSONL file, skipping malformed lines."""
    records: list[dict[str, Any]] = []
    with path.open("r", encoding="utf-8") as fh:
        for lineno, line in enumerate(fh, start=1):
            line = line.strip()
            if not line:
                continue
            try:
                obj = json.loads(line)
            except json.JSONDecodeError as exc:
                log.warning("Skipping malformed JSON on line %d: %s", lineno, exc)
                continue
            if not isinstance(obj, dict):
                log.warning("Skipping non-object record on line %d", lineno)
                continue
            records.append(obj)
    return records


def extract_texts(records: list[dict[str, Any]], field: str) -> list[str]:
    """Pull the text field from each record, dropping records with no text."""
    texts: list[str] = []
    for i, rec in enumerate(records):
        value = rec.get(field)
        if not isinstance(value, str) or not value.strip():
            log.warning("Record %d missing usable '%s' field; dropping", i, field)
            continue
        texts.append(value)
    return texts


# ---------------------------------------------------------------------------
# Embedding with retry/backoff
# ---------------------------------------------------------------------------
def embed_texts(client: OpenAI, texts: list[str], model: str) -> np.ndarray:
    """Embed a list of texts in batches; returns an (N, D) float32 array."""
    all_vectors: list[list[float]] = []
    for start in range(0, len(texts), BATCH_SIZE):
        batch = texts[start : start + BATCH_SIZE]
        attempt = 0
        while True:
            try:
                resp = client.embeddings.create(model=model, input=batch)
                # API preserves input order; sort defensively by index
                ordered = sorted(resp.data, key=lambda d: d.index)
                all_vectors.extend(item.embedding for item in ordered)
                log.info("Embedded %d/%d", min(start + BATCH_SIZE, len(texts)), len(texts))
                break
            except (RateLimitError, APIConnectionError) as exc:
                attempt += 1
                if attempt > MAX_RETRIES:
                    raise RuntimeError(f"Embedding failed after {MAX_RETRIES} retries") from exc
                backoff = 2 ** attempt
                log.warning("Transient error (%s); retrying in %ds", type(exc).__name__, backoff)
                time.sleep(backoff)
            except APIError as exc:
                # Non-retryable API error (bad request, auth, etc.)
                raise RuntimeError(f"OpenAI API error: {exc}") from exc

    arr = np.asarray(all_vectors, dtype=np.float32)
    # L2-normalize so cosine similarity == dot product
    norms = np.linalg.norm(arr, axis=1, keepdims=True)
    norms[norms == 0] = 1.0
    return arr / norms


# ---------------------------------------------------------------------------
# Clustering
# ---------------------------------------------------------------------------
def cluster_duplicates(embeddings: np.ndarray, threshold: float) -> list[int]:
    """
    Return a list of cluster IDs, one per input row.
    Two rows share a cluster ID iff they are transitively similar.
    """
    n = embeddings.shape[0]
    if n == 0:
        return []
    # Full pairwise cosine similarity (vectors are normalized -> dot product)
    sim = embeddings @ embeddings.T
    uf = UnionFind(n)
    # Only inspect the strict upper triangle
    iu, ju = np.triu_indices(n, k=1)
    mask = sim[iu, ju] >= threshold
    for i, j in zip(iu[mask], ju[mask]):
        uf.union(int(i), int(j))
    return [uf.find(i) for i in range(n)]


# ---------------------------------------------------------------------------
# Main pipeline
# ---------------------------------------------------------------------------
def run(input_path: Path, output_path: Path, threshold: float) -> tuple[int, int]:
    if not OPENAI_API_KEY:
        raise EnvironmentError("OPENAI_API_KEY is not set")
    if not input_path.exists():
        raise FileNotFoundError(f"Input file not found: {input_path}")

    records = load_jsonl(input_path)
    if not records:
        log.warning("No valid records found in %s", input_path)
        output_path.write_text("", encoding="utf-8")
        return 0, 0

    texts = extract_texts(records, TEXT_FIELD)
    if not texts:
        raise ValueError(f"No records contained a usable '{TEXT_FIELD}' field")

    # Keep a parallel list of records that actually had text
    kept_records = [r for r in records if isinstance(r.get(TEXT_FIELD), str) and r[TEXT_FIELD].strip()]

    client = OpenAI(api_key=OPENAI_API_KEY)
    embeddings = embed_texts(client, texts, EMBED_MODEL)
    log.info("Computed embeddings for %d documents (dim=%d)", *embeddings.shape)

    cluster_ids = cluster_duplicates(embeddings, threshold)
    # First occurrence of each cluster ID wins
    seen: dict[int, int] = {}
    unique_records: list[dict[str, Any]] = []
    for idx, cid in enumerate(cluster_ids):
        if cid in seen:
            continue
        seen[cid] = idx
        unique_records.append(kept_records[idx])

    with output_path.open("w", encoding="utf-8") as fh:
        for rec in unique_records:
            fh.write(json.dumps(rec, ensure_ascii=False) + "\n")

    log.info(
        "Wrote %d unique records to %s (removed %d duplicates)",
        len(unique_records), output_path, len(kept_records) - len(unique_records),
    )
    return len(kept_records), len(unique_records)


def parse_args(argv: Iterable[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="AI content deduplicator using embeddings.")
    parser.add_argument("--input", required=True, type=Path, help="Input JSONL file")
    parser.add_argument("--output", required=True, type=Path, help="Output JSONL file")
    parser.add_argument("--threshold", type=float, default=0.92,
                        help="Cosine similarity threshold for duplicates (0-1)")
    return parser.parse_args(argv)


def main() -> int:
    args = parse_args()
    if not 0.0 < args.threshold <= 1.0:
        log.error("Threshold must be in (0, 1]; got %s", args.threshold)
        return 2
    try:
        run(args.input, args.output, args.threshold)
    except (EnvironmentError, FileNotFoundError, ValueError, RuntimeError) as exc:
        log.error("%s", exc)
        return 1
    except KeyboardInterrupt:
        log.warning("Interrupted by user")
        return 130
    return 0


if __name__ == "__main__":
    sys.exit(main())

Configuration

For reference: 0.85 is permissive (groups topically related docs), 0.92 is a good default for paraphrases, and 0.97+ only catches near-verbatim copies.

Usage Examples

Basic run — deduplicate a feed of scraped articles:

$ export OPENAI_API_KEY=sk-...
$ python3 dedup_ai.py --input articles.jsonl --output unique.jsonl --threshold 0.92
14:02:11 [INFO] Embedded 64/312
14:02:14 [INFO] Embedded 128/312
...
14:02:29 [INFO] Computed embeddings for 312 documents (dim=1536)
14:02:29 [INFO] Wrote 187 unique records to unique.jsonl (removed 125 duplicates)

Aggressive dedup — collapse topically similar summaries:

$ python3 dedup_ai.py --input summaries.jsonl --output unique.jsonl --threshold 0.85
14:05:02 [INFO] Wrote 42 unique records to unique.jsonl (removed 158 duplicates)

Custom field name and larger model:

$ TEXT_FIELD=body EMBED_MODEL=text-embedding-3-large EMBED_BATCH_SIZE=32 \
    python3 dedup_ai.py --input posts.jsonl --output posts_unique.jsonl --threshold 0.94

Error Handling

• Missing API key — EnvironmentError is raised before any network call; the process exits with code 1 and a clear message.
• Missing input file — FileNotFoundError is caught in main and reported without a traceback.
• Malformed JSONL lines — logged as warnings and skipped; the rest of the file is processed normally.
• Records without the text field — dropped with a warning, so a partially malformed file still produces useful output.
• Rate limits / connection errors — RateLimitError and APIConnectionError trigger exponential backoff (2^attempt seconds) up to EMBED_MAX_RETRIES.
• Non-retryable API errors — auth failures, invalid model names, oversized inputs raise APIError, which is re-raised as a RuntimeError and reported cleanly.
• Bad threshold — values outside (0, 1] exit with code 2 before touching the API.
• Ctrl-C — KeyboardInterrupt is trapped and returns exit code 130 so shell scripts can distinguish user cancellation.

Extending Ideas

• Cache embeddings on disk — key by sha256(text) and store in a small SQLite or .npz file so re-runs on overlapping corpora don't re-bill the API.
• Swap in a local model — replace the OpenAI call with sentence-transformers (all-MiniLM-L6-v2) to run fully offline; the rest of the pipeline is unchanged.
• Add a similarity report — emit a sidecar JSONL mapping each dropped record to its representative and the similarity score, useful for auditing what got merged.
• Scale to large corpora — replace the O(N²) similarity matrix with FAISS or hnswlib approximate nearest-neighbor search to handle hundreds of thousands of documents.