Anomaly and Duplicate Detection

Objectives

• Detect abnormal texts within a corpus
• Identify semantic duplicates
• Build a deduplication pipeline

Anomaly Detection

The two applications in this lesson share the same intuition, applied to opposite ends of the similarity scale. A text is “abnormal” if it is semantically distant from everything else in the corpus; a text is a duplicate if it is too close. Embeddings let you measure this distance in both cases. Most of the work involves calibrating the thresholds.

The first approach builds the corpus centroid—the average of all vectors, which represents “the usual topic of this corpus”—then measures how far each text deviates from it. The threshold is intentionally not an absolute value but a percentile: you cannot know in advance how far away an anomaly lies, but you do know you want to examine the most atypical 5%.

from openai import OpenAI
import numpy as np

client = OpenAI()

def detect_anomalies(
    texts: list[str],
    percentile_threshold: float = 95,
    model: str = "text-embedding-3-large"
) -> list[dict]:
    """Detects abnormal texts by distance to the centroid."""
    response = client.embeddings.create(
        input=texts, model=model
    )
    embeddings = np.array(
        [d.embedding for d in sorted(response.data, key=lambda x: x.index)]
    )

    # Corpus centroid
    centroid = embeddings.mean(axis=0)
    centroid /= np.linalg.norm(centroid)

    # Distance of each text to the centroid
    similarities = embeddings @ centroid
    threshold = np.percentile(similarities, 100 - percentile_threshold)

    anomalies = []
    for i, sim in enumerate(similarities):
        if sim < threshold:
            anomalies.append({
                "index": i,
                "text": texts[i],
                "similarity": float(sim),
                "is_anomaly": True
            })

    return anomalies

# Example
corpus = [
    "Our leave policy provides 25 days per year",
    "Leave must be requested 2 weeks in advance",
    "Leave balances are visible on the HR portal",
    "Exceptional leave is granted with supporting documentation",
    "SELECT * FROM users WHERE admin=1; DROP TABLE users;--",  # Anomaly!
    "The remote work policy allows 3 days per week",
]

anomalies = detect_anomalies(corpus)
for a in anomalies:
    print(f"  ANOMALY [{a['similarity']:.3f}] : {a['text'][:60]}...")

The example corpus tells the story: five HR policy sentences and one SQL injection attempt. No keyword filter would have caught the injection without a pre-written rule; distance to the centroid flags it without any prior anticipation. This is the main argument for using the method to monitor a stream of incoming documents.

The centroid does have a weakness: it assumes the corpus revolves around a single theme. If your documents cover three very different domains, the average point falls into a semantic void, and all texts appear equally distant. The k-nearest neighbors approach sidesteps this by judging each text based on its immediate neighborhood rather than the whole set: a document belonging to a minority group remains normal as long as it has neighbors.

def detect_anomalies_knn(
    texts: list[str],
    k: int = 3,
    percentile_threshold: float = 90,
    model: str = "text-embedding-3-large"
) -> list[dict]:
    """Detects anomalies by distance to the k nearest neighbors."""
    response = client.embeddings.create(
        input=texts, model=model
    )
    embeddings = np.array(
        [d.embedding for d in sorted(response.data, key=lambda x: x.index)]
    )

    # Similarity matrix
    sim_matrix = embeddings @ embeddings.T

    # For each text, average of the k nearest neighbors
    isolation_scores = []
    for i in range(len(texts)):
        sims = sim_matrix[i].copy()
        sims[i] = -1  # Exclude self
        top_k_sims = np.sort(sims)[-k:]
        isolation_scores.append(top_k_sims.mean())

    threshold = np.percentile(isolation_scores, 100 - percentile_threshold)

    anomalies = []
    for i, score in enumerate(isolation_scores):
        if score < threshold:
            anomalies.append({
                "index": i,
                "text": texts[i],
                "isolation_score": float(score)
            })

    return anomalies

Semantic Duplicate Detection

At the other end of the scale lie texts that are too close. Two entries can be duplicates even with different words—“How do I reset my password?” and “Forgot password, what do I do?” share only three common words—and that is exactly what hashing or string comparison will never catch. Embeddings detect them.

The function calculates the full similarity matrix and returns all pairs above the threshold. Watch the cost: this comparison is quadratic, which remains instant for a few thousand texts but becomes impractical for a million—you must then use an approximate index like those covered with vector databases.

def find_duplicates(
    texts: list[str],
    threshold: float = 0.90,
    model: str = "text-embedding-3-large"
) -> list[dict]:
    """Finds pairs of semantically identical texts."""
    response = client.embeddings.create(
        input=texts, model=model
    )
    embeddings = np.array(
        [d.embedding for d in sorted(response.data, key=lambda x: x.index)]
    )

    # Similarity matrix
    sim_matrix = embeddings @ embeddings.T

    duplicates = []
    for i in range(len(texts)):
        for j in range(i + 1, len(texts)):
            if sim_matrix[i][j] >= threshold:
                duplicates.append({
                    "text_a": texts[i],
                    "text_b": texts[j],
                    "index_a": i,
                    "index_b": j,
                    "similarity": float(sim_matrix[i][j])
                })

    return sorted(duplicates, key=lambda x: x["similarity"], reverse=True)

# Example
texts = [
    "How do I reset my password?",
    "I cannot log in to my account",
    "Forgot password, what do I do?",           # Duplicate of [0]
    "What is the refund policy?",
    "Unable to access my personal space",       # Duplicate of [1]
    "How do I get a refund?",                   # Duplicate of [3]
]

dupes = find_duplicates(texts, threshold=0.85)
print(f"{len(dupes)} duplicate pairs found:")
for d in dupes:
    print(f"  [{d['similarity']:.3f}]")
    print(f"    A: {d['text_a']}")
    print(f"    B: {d['text_b']}")

The call here uses a threshold of 0.85 rather than the default 0.90 because reformulations of the same support question often fall well below 0.90. This is the setting you will spend the most time tuning: set it too high, and you let duplicates slip through; set it too low, and you merge “how do I change my password” with “how do I change my email address,” two distinct questions that look very similar.

Deduplication Pipeline

Detecting pairs is not enough to clean a corpus. If A resembles B and B resembles C, all three belong to the same group, even if A and C were never directly compared above the threshold. Transitive grouping is the real work, and the Union-Find structure handles it in a single pass: each detected pair merges two sets, yielding fully formed duplicate groups at the end.

You still need to choose which member survives. The rule used here—the shortest text—works well for support questions, where the most concise formulation is often the most canonical. For a document corpus, you will likely want the opposite, or the most recent: this is a business decision, not a technical detail. The returned dictionary keeps the list of deleted texts so a human can verify the edge cases before approving.

class Deduplicator:
    def __init__(self, threshold: float = 0.90):
        self.client = OpenAI()
        self.threshold = threshold

    def deduplicate(self, texts: list[str]) -> dict:
        """Removes semantic duplicates."""
        response = self.client.embeddings.create(
            input=texts, model="text-embedding-3-large"
        )
        embeddings = np.array(
            [d.embedding for d in sorted(response.data, key=lambda x: x.index)]
        )

        # Union-Find to group duplicates
        parent = list(range(len(texts)))

        def find(x):
            while parent[x] != x:
                parent[x] = parent[parent[x]]
                x = parent[x]
            return x

        def union(a, b):
            pa, pb = find(a), find(b)
            if pa != pb:
                parent[pb] = pa

        # Find similar pairs
        sim_matrix = embeddings @ embeddings.T
        for i in range(len(texts)):
            for j in range(i + 1, len(texts)):
                if sim_matrix[i][j] >= self.threshold:
                    union(i, j)

        # Keep one representative per group
        groups = {}
        for i in range(len(texts)):
            root = find(i)
            if root not in groups:
                groups[root] = []
            groups[root].append(i)

        # Select unique texts (shortest in each group)
        uniques = []
        deleted = []
        for root, members in groups.items():
            representative = min(members, key=lambda i: len(texts[i]))
            uniques.append(texts[representative])
            for m in members:
                if m != representative:
                    deleted.append(texts[m])

        return {
            "uniques": uniques,
            "deleted": deleted,
            "n_original": len(texts),
            "n_deduplicated": len(uniques),
            "n_deleted": len(deleted)
        }

# Usage
dedup = Deduplicator(threshold=0.85)
result = dedup.deduplicate(texts)
print(f"Before: {result['n_original']} texts")
print(f"After: {result['n_deduplicated']} texts")
print(f"Deleted: {result['n_deleted']}")

Applied upstream of a RAG index, this deduplication has a direct effect on quality: it prevents the five results returned for a question from being five versions of the same passage, which would deprive the generator of any additional information.

Summary

• Distance to the centroid or k-NN detects abnormal texts
• Cosine similarity above a threshold identifies duplicates
• Union-Find efficiently groups transitive duplicates
• Calibrate thresholds on your data (typically 0.85–0.95 for duplicates)