Anomalie- und Duplikaterkennung

Lernziele

• Erkennung von anomalen Texten innerhalb eines Korpus
• Identifikation semantischer Duplikate
• Aufbau einer Deduplizierungspipeline

Anomalieerkennung

Die beiden Anwendungen in dieser Lektion beruhen auf derselben Intuition, angewendet auf entgegengesetzte Enden der Ähnlichkeitsskala. Ein Text gilt als „anomal”, wenn er semantisch von allem anderen im Korpus entfernt ist; ein Text ist ein Duplikat, wenn er zu nah liegt. Embeddings ermöglichen es, diesen Abstand in beiden Fällen zu messen. Der Großteil der Arbeit besteht darin, die Schwellenwerte zu kalibrieren.

Der erste Ansatz erstellt den Korpus-Zentroid – den Durchschnitt aller Vektoren, der „das übliche Thema dieses Korpus” repräsentiert – und misst, wie weit sich jeder Text davon entfernt. Die Schwelle ist absichtlich kein absoluter Wert, sondern ein Prozentsatz: Sie können nicht im Voraus wissen, wie weit eine Anomalie entfernt liegt, aber Sie wissen, dass Sie die 5 % atypischsten Texte untersuchen möchten.

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]:
    """Erkennt anomale Texte anhand des Abstands zum Zentroid."""
    response = client.embeddings.create(
        input=texts, model=model
    )
    embeddings = np.array(
        [d.embedding for d in sorted(response.data, key=lambda x: x.index)]
    )

    # Korpus-Zentroid
    centroid = embeddings.mean(axis=0)
    centroid /= np.linalg.norm(centroid)

    # Abstand jedes Textes zum Zentroid
    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

# Beispiel
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]}...")

Das Beispiel-Korpus erzählt die Geschichte: fünf Sätze zur Personalrichtlinie und ein SQL-Injection-Versuch. Kein Keyword-Filter hätte den Injection-Versuch ohne eine vorab geschriebene Regel erfasst; der Abstand zum Zentroid flaggt ihn ohne jegliche vorherige Antizipation. Dies ist das Hauptargument für die Verwendung dieser Methode zur Überwachung eines Stroms eingehender Dokumente.

Der Zentroid hat eine Schwäche: Er geht davon aus, dass sich das Korpus um ein einziges Thema dreht. Wenn Ihre Dokumente drei sehr unterschiedliche Bereiche abdecken, fällt der Durchschnittspunkt in ein semantisches Vakuum, und alle Texte erscheinen gleich weit entfernt. Der k-nearest-neighbors-Ansatz umgeht dies, indem er jeden Text basierend auf seiner unmittelbaren Nachbarschaft statt auf dem gesamten Satz beurteilt: Ein Dokument einer Minderheitsgruppe bleibt normal, solange es Nachbarn hat.

def detect_anomalies_knn(
    texts: list[str],
    k: int = 3,
    percentile_threshold: float = 90,
    model: str = "text-embedding-3-large"
) -> list[dict]:
    """Erkennt Anomalien anhand des Abstands zu den k nächsten Nachbarn."""
    response = client.embeddings.create(
        input=texts, model=model
    )
    embeddings = np.array(
        [d.embedding for d in sorted(response.data, key=lambda x: x.index)]
    )

    # Ähnlichkeitsmatrix
    sim_matrix = embeddings @ embeddings.T

    # Für jeden Text: Durchschnitt der k nächsten Nachbarn
    isolation_scores = []
    for i in range(len(texts)):
        sims = sim_matrix[i].copy()
        sims[i] = -1  # Selbst ausschließen
        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

Erkennung semantischer Duplikate

An der anderen Ende der Skala liegen Texte, die zu nah beieinander liegen. Zwei Einträge können Duplikate sein, auch wenn sie unterschiedliche Wörter verwenden – „How do I reset my password?” und „Forgot password, what do I do?” haben nur drei gemeinsame Wörter – und genau das wird von Hashing oder String-Vergleichen niemals erfasst. Embeddings erkennen sie.

Die Funktion berechnet die vollständige Ähnlichkeitsmatrix und gibt alle Paare oberhalb der Schwelle zurück. Achten Sie auf die Kosten: Dieser Vergleich ist quadratisch, was für einige tausend Texte sofort erfolgt, aber für eine Million Texte unpraktisch wird – Sie müssen dann einen approximierten Index verwenden, wie er bei Vektordatenbanken behandelt wird.

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

    # Ähnlichkeitsmatrix
    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)

# Beispiel
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']}")

Der Aufruf verwendet hier eine Schwelle von 0,85 statt dem Standardwert 0,90, da Reformulierungen derselben Support-Frage oft deutlich unter 0,90 liegen. Dies ist die Einstellung, bei der Sie die meiste Zeit zur Feinabstimmung benötigen: Setzen Sie sie zu hoch, und Sie lassen Duplikate durch; setzen Sie sie zu niedrig, und Sie vereinigen „how do I change my password” mit „how do I change my email address”, zwei unterschiedliche Fragen, die sich sehr ähnlich ansehen.

Deduplizierungspipeline

Das Finden von Paaren reicht nicht aus, um ein Korpus zu bereinigen. Wenn A B ähnelt und B C ähnelt, gehören alle drei derselben Gruppe an, auch wenn A und C nie direkt oberhalb der Schwelle verglichen wurden. Transitive Gruppierung ist die eigentliche Arbeit, und die Union-Find-Struktur erledigt dies in einem Durchlauf: Jedes erkannte Paar verschmilzt zwei Mengen und ergibt am Ende vollständig gebildete Duplikatgruppen.

Sie müssen immer noch wählen, welches Mitglied überlebt. Die hier verwendete Regel – der kürzeste Text – funktioniert gut für Support-Fragen, bei denen die kürzeste Formulierung oft die kanonischste ist. Für ein Dokumentenkorpus werden Sie wahrscheinlich das Gegenteil wünschen, oder den neuesten: Dies ist eine geschäftliche Entscheidung, kein technisches Detail. Das zurückgegebene Wörterbuch behält die Liste der gelöschten Texte bei, damit ein Mensch die Grenzfälle vor der Genehmigung überprüfen kann.

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

    def deduplicate(self, texts: list[str]) -> dict:
        """Entfernt semantische Duplikate."""
        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 zur Gruppierung von Duplikaten
        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

        # Ähnliche Paare finden
        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)

        # Ein Repräsentant pro Gruppe behalten
        groups = {}
        for i in range(len(texts)):
            root = find(i)
            if root not in groups:
                groups[root] = []
            groups[root].append(i)

        # Eindeutige Texte auswählen (kürzester in jeder Gruppe)
        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)
        }

# Nutzung
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']}")

Angewendet upstream eines RAG-Indexes hat diese Deduplizierung eine direkte Auswirkung auf die Qualität: Sie verhindert, dass die fünf für eine Frage zurückgegebenen Ergebnisse fünf Versionen desselben Passages sind, was dem Generator jegliche zusätzlichen Informationen entziehen würde.

Zusammenfassung

• Abstand zum Zentroid oder k-NN erkennt anomale Texte
• Kosinusähnlichkeit oberhalb einer Schwelle identifiziert Duplikate
• Union-Find gruppiert transitive Duplikate effizient
• Kalibrieren Sie Schwellenwerte auf Ihren Daten (typischerweise 0,85–0,95 für Duplikate)