Building a Document Similarity Application

This closing lesson builds a complete, standalone application that finds near-duplicate and closely related documents across a collection — a genuinely different project from Unit 10's notes-RAG assistant and from Lesson 5's ranked search engine earlier in this unit. Where a search engine answers "what matches this specific query," a document similarity application answers a different question entirely: "which documents in this collection are similar to each other." This shape of problem shows up constantly in real systems — flagging duplicate support tickets before an agent works the same issue twice, detecting near-duplicate job postings or product listings, grouping customer feedback into clusters of the same underlying complaint, or catching plagiarized or copy-pasted submissions.

Defining the Problem Precisely

Given a collection of documents, the application needs to:

1. Embed every document.
2. Compute similarity between every pair of documents (not one query against many, as in search).
3. Group documents whose pairwise similarity exceeds a threshold into clusters of "the same or near-identical content."
4. Report each cluster, so a human can review and act on it.

Step 3 is the part that differs meaningfully from anything built earlier in this unit. Similarity above a threshold is not automatically transitive in a simple pairwise sense — document A might be similar enough to B, and B similar enough to C, without A and C directly exceeding the threshold. Grouping decisions need a proper mechanism for combining these pairwise relationships into clusters, which is where a union-find (disjoint-set) structure comes in.

Computing All Pairwise Similarities

def cosine_similarity(a: list[float], b: list[float]) -> float:
    dot = sum(x * y for x, y in zip(a, b))
    mag_a = sum(x * x for x in a) ** 0.5
    mag_b = sum(x * x for x in b) ** 0.5
    if mag_a == 0 or mag_b == 0:
        return 0.0
    return dot / (mag_a * mag_b)


def find_similar_pairs(doc_ids: list[str], embeddings: list[list[float]],
                        threshold: float = 0.9) -> list[tuple[str, str, float]]:
    """Return all document pairs whose cosine similarity exceeds `threshold`."""
    pairs = []
    for i in range(len(doc_ids)):
        for j in range(i + 1, len(doc_ids)):
            score = cosine_similarity(embeddings[i], embeddings[j])
            if score >= threshold:
                pairs.append((doc_ids[i], doc_ids[j], score))
    return pairs

Why compare every pair instead of using a search-style top-k lookup? Search retrieves the best matches for one query vector. This application needs to know, for every document, which other documents are close to it — a fundamentally different access pattern. The nested loop here compares each pair exactly once (range(i + 1, len(doc_ids)) skips redundant and self-comparisons), which is the correct approach for small to medium collections. This is the part of the application least tolerant of scale: comparing every pair is O(n²) — for 1,000 documents that is roughly 500,000 comparisons, which is fine; for 1,000,000 documents it is roughly 500 billion, which is not. Real large-scale duplicate-detection systems use approximate nearest-neighbor indexes (the same kind of vector database index from Lesson 6) to find only the likely close pairs for each document instead of comparing everything to everything — worth knowing as the scaling path beyond what this lesson implements directly.

Grouping Similar Documents With Union-Find

A union-find structure keeps track of which items belong to the same group, efficiently merging two groups together whenever a new connection (a similar pair) is found.

class UnionFind:
    """A disjoint-set structure: tracks groups of connected items and
    merges two groups in near-constant time when a connection is found.
    """

    def __init__(self, items: list[str]):
        self._parent = {item: item for item in items}

    def find(self, item: str) -> str:
        # Path compression: point every visited node directly at the root,
        # so future lookups for these items are faster.
        if self._parent[item] != item:
            self._parent[item] = self.find(self._parent[item])
        return self._parent[item]

    def union(self, item_a: str, item_b: str) -> None:
        root_a, root_b = self.find(item_a), self.find(item_b)
        if root_a != root_b:
            self._parent[root_a] = root_b

    def groups(self) -> dict[str, list[str]]:
        clusters: dict[str, list[str]] = {}
        for item in self._parent:
            root = self.find(item)
            clusters.setdefault(root, []).append(item)
        return clusters

Why is this the right structure for this problem, instead of simpler ad hoc grouping? The transitivity issue raised above — A similar to B, B similar to C, but A not directly similar enough to C — is exactly what union-find resolves correctly and efficiently: calling union("A", "B") and then union("B", "C") places all three in the same group automatically, because find() follows the chain of merges back to a shared root, regardless of the order pairs are processed in. Writing this grouping logic manually with sets and loops is easy to get subtly wrong (for example, forgetting to merge two existing groups together when a new pair connects them); union-find is a well-established, easy-to-verify way to get it right the first time.

find() uses path compression — every node visited while chasing down to the root gets rewired to point directly at that root — which keeps repeated lookups fast even after many merges. union() simply points one group's root at the other's root, merging the two groups in one step.

Assembling the Application

from dataclasses import dataclass


@dataclass
class DuplicateCluster:
    doc_ids: list[str]
    max_similarity: float


class DocumentSimilarityApp:
    """Finds clusters of near-duplicate documents in a collection."""

    def __init__(self, embed_fn, threshold: float = 0.9):
        self._embed_fn = embed_fn
        self._threshold = threshold

    def find_clusters(self, documents: dict[str, str]) -> list[DuplicateCluster]:
        """`documents` maps doc_id -> text. Returns clusters containing
        more than one document (singletons are not near-duplicates of
        anything and are omitted from the report).
        """
        doc_ids = list(documents.keys())
        texts = [documents[doc_id] for doc_id in doc_ids]
        embeddings = self._embed_fn(texts)

        pairs = find_similar_pairs(doc_ids, embeddings, threshold=self._threshold)

        uf = UnionFind(doc_ids)
        pair_scores: dict[tuple[str, str], float] = {}
        for doc_a, doc_b, score in pairs:
            uf.union(doc_a, doc_b)
            pair_scores[(doc_a, doc_b)] = score

        clusters = []
        for members in uf.groups().values():
            if len(members) < 2:
                continue
            relevant_scores = [
                score for (a, b), score in pair_scores.items()
                if a in members and b in members
            ]
            clusters.append(DuplicateCluster(
                doc_ids=sorted(members),
                max_similarity=max(relevant_scores) if relevant_scores else 0.0,
            ))
        return clusters

find_clusters ties the whole pipeline together: embed every document once, find all pairs above the threshold, feed those pairs into UnionFind to resolve them into clusters, and finally discard any "cluster" of size one — a document with no near-duplicates is not useful in a duplicate-detection report. Following the same dependency-injection pattern used throughout this unit, embed_fn is passed in rather than hard-coded, keeping the class fully testable without real API calls.

Testing the Application End to End

def fake_embed_fn(texts: list[str]) -> list[list[float]]:
    """Deterministic fake embedder: documents sharing more words end up
    with more similar vectors, giving predictable test behavior.
    """
    vocabulary = ["refund", "broken", "item", "shipping", "delayed", "package"]
    vectors = []
    for text in texts:
        lowered = text.lower()
        vectors.append([1.0 if word in lowered else 0.0 for word in vocabulary])
    return vectors


def test_document_similarity_app_finds_duplicate_cluster():
    documents = {
        "ticket-1": "My item arrived broken, I want a refund.",
        "ticket-2": "The item I received was broken, please refund me.",
        "ticket-3": "My shipping is delayed, when will my package arrive?",
        "ticket-4": "Completely unrelated question about account settings.",
    }
    app = DocumentSimilarityApp(embed_fn=fake_embed_fn, threshold=0.99)
    clusters = app.find_clusters(documents)

    assert len(clusters) == 1
    assert clusters[0].doc_ids == ["ticket-1", "ticket-2"]
    print(f"PASS: found duplicate cluster {clusters[0].doc_ids} "
          f"(similarity={clusters[0].max_similarity:.2f})")


test_document_similarity_app_finds_duplicate_cluster()

fake_embed_fn gives ticket-1 and ticket-2 identical vectors (both mention "broken" and "item" and imply "refund" via shared vocabulary), while ticket-3 shares no vocabulary with them and ticket-4 shares none with anything — so the test can assert a specific, predictable clustering outcome without depending on a real model's exact numeric output. This is the same reasoning behind every fake embedder used throughout this unit: the test verifies the application's logic (pairing, thresholding, grouping) is correct, which is a separate concern from verifying that a real embedding model produces good vectors for real text — the latter is what Lesson 9's evaluation methodology is for.

Wiring in a Real Embedding Model and Reporting Results

from openai import OpenAI

client = OpenAI()

def real_embed_fn(texts: list[str]) -> list[list[float]]:
    response = client.embeddings.create(model="text-embedding-4", input=texts)
    return [item.embedding for item in response.data]

app = DocumentSimilarityApp(embed_fn=real_embed_fn, threshold=0.92)

support_tickets = {
    "ticket-101": "The package I ordered never showed up and tracking hasn't updated in a week.",
    "ticket-102": "My order tracking says delivered but I never received the package.",
    "ticket-103": "I'd like to cancel my subscription before the next billing cycle.",
    "ticket-104": "How do I cancel my subscription before I get charged again?",
}

clusters = app.find_clusters(support_tickets)
for cluster in clusters:
    print(f"Possible duplicates (similarity={cluster.max_similarity:.3f}): {cluster.doc_ids}")

In a real support system, this report becomes the input to a workflow decision: automatically merging tickets above a very high similarity threshold, or surfacing a "possible duplicate" suggestion to an agent for confirmation rather than acting automatically — the appropriate threshold and level of automation should, as with everything else in this unit, be set by evaluating outcomes on real historical tickets rather than chosen arbitrarily. Note the higher threshold (0.92) here compared to the test's 0.99: real embeddings from a trained model rarely reach near-1.0 similarity even for genuine duplicates, unlike the artificially clean fake vectors used in the test above, which is exactly the kind of calibration point Lesson 9 exists to resolve with real, measured data rather than guesswork.

Common Mistakes

• Assuming pairwise similarity above a threshold is automatically transitive without a proper grouping structure. As covered above, naive grouping logic can miss or incorrectly merge clusters; union-find resolves this correctly regardless of the order pairs are discovered in.
• Running a full O(n²) pairwise comparison on a large, growing collection without a plan to scale. This works fine for hundreds or low thousands of documents and becomes impractical well before a million, at which point an approximate nearest-neighbor index (Lesson 6) is needed to avoid comparing every document to every other document.
• Reusing a threshold tuned on a fake or synthetic test directly in production. As shown above, a threshold that works cleanly against a hand-constructed test vector set does not necessarily reflect the similarity range real embeddings produce for genuinely duplicate real-world text.

Best Practices

• Separate the three concerns explicitly: computing similarity, deciding what counts as "the same," and grouping. find_similar_pairs, the threshold parameter, and UnionFind each do one job, which keeps the application easy to reason about and to adjust independently.
• Report clusters with their similarity scores, not just the grouping. Showing max_similarity alongside each cluster lets a human reviewer judge borderline cases rather than trusting an opaque yes/no duplicate flag.
• Treat automatic action (auto-merging, auto-deleting) as a separate, higher-confidence threshold than the one used for surfacing a review suggestion. A lower threshold is reasonable for "flag this for a human to check"; a much higher, carefully evaluated threshold should gate anything the system does without human confirmation.