Détection d'anomalies et de duplicates

Objectifs

• Détecter les textes anormaux dans un corpus
• Identifier les duplicates sémantiques
• Construire un pipeline de déduplication

Détection d’anomalies

Les deux applications de cette leçon partagent la même intuition, appliquée aux deux extrémités de l’échelle de similarité. Un texte est « anormal » s’il est sémantiquement éloigné de tous les autres dans le corpus ; un texte est un duplicate s’il en est trop proche. Les embeddings permettent de mesurer cette distance dans les deux cas, et le travail consiste surtout à calibrer les seuils.

La première approche construit le centroïde du corpus — la moyenne de tous les vecteurs, qui matérialise « le sujet dont parle habituellement ce corpus » — puis mesure à quel point chaque texte s’en écarte. Le seuil n’est volontairement pas une valeur absolue mais un percentile : on ne sait pas d’avance à quelle distance se trouve une anomalie, en revanche on sait qu’on veut examiner les 5 % les plus atypiques.

from openai import OpenAI
import numpy as np

client = OpenAI()

def detecter_anomalies(
    textes: list[str],
    seuil_percentile: float = 95,
    model: str = "text-embedding-3-large"
) -> list[dict]:
    """Détecte les textes anormaux par distance au centroïde."""
    response = client.embeddings.create(
        input=textes, model=model
    )
    embeddings = np.array(
        [d.embedding for d in sorted(response.data, key=lambda x: x.index)]
    )

    # Centroïde du corpus
    centroide = embeddings.mean(axis=0)
    centroide /= np.linalg.norm(centroide)

    # Distance de chaque texte au centroïde
    similarites = embeddings @ centroide
    seuil = np.percentile(similarites, 100 - seuil_percentile)

    anomalies = []
    for i, sim in enumerate(similarites):
        if sim < seuil:
            anomalies.append({
                "index": i,
                "texte": textes[i],
                "similarite": float(sim),
                "est_anomalie": True
            })

    return anomalies

# Exemple
corpus = [
    "Notre politique de congés prévoit 25 jours par an",
    "Les congés doivent être posés 2 semaines à l'avance",
    "Le solde de congés est visible sur le portail RH",
    "Les congés exceptionnels sont accordés sur justificatif",
    "SELECT * FROM users WHERE admin=1; DROP TABLE users;--",  # Anomalie !
    "La politique de télétravail autorise 3 jours par semaine",
]

anomalies = detecter_anomalies(corpus)
for a in anomalies:
    print(f"  ANOMALIE [{a['similarite']:.3f}] : {a['texte'][:60]}...")

Le corpus d’exemple est éloquent : cinq phrases de politique RH et une tentative d’injection SQL. Aucun filtre par mots-clés n’aurait attrapé celle-ci sans une règle écrite à l’avance ; la distance au centroïde la fait ressortir sans qu’on ait rien anticipé. C’est le principal argument de la méthode pour surveiller un flux de documents entrants.

Le centroïde a cependant une faiblesse : il suppose que le corpus tourne autour d’un thème unique. Si vos documents couvrent trois domaines très différents, le point moyen tombe dans un vide sémantique et tous les textes en paraissent également éloignés. L’approche par k plus proches voisins contourne cela en jugeant chaque texte sur son voisinage immédiat plutôt que sur l’ensemble : un document qui appartient à un groupe minoritaire reste normal, tant qu’il a des voisins.

def detecter_anomalies_knn(
    textes: list[str],
    k: int = 3,
    seuil_percentile: float = 90,
    model: str = "text-embedding-3-large"
) -> list[dict]:
    """Détecte les anomalies par distance aux k voisins les plus proches."""
    response = client.embeddings.create(
        input=textes, model=model
    )
    embeddings = np.array(
        [d.embedding for d in sorted(response.data, key=lambda x: x.index)]
    )

    # Matrice de similarité
    sim_matrix = embeddings @ embeddings.T

    # Pour chaque texte, moyenne des k plus proches voisins
    scores_isolation = []
    for i in range(len(textes)):
        sims = sim_matrix[i].copy()
        sims[i] = -1  # Exclure soi-même
        top_k_sims = np.sort(sims)[-k:]
        scores_isolation.append(top_k_sims.mean())

    seuil = np.percentile(scores_isolation, 100 - seuil_percentile)

    anomalies = []
    for i, score in enumerate(scores_isolation):
        if score < seuil:
            anomalies.append({
                "index": i,
                "texte": textes[i],
                "score_isolation": float(score)
            })

    return anomalies

Détection de duplicates sémantiques

À l’autre bout de l’échelle se trouvent les textes trop proches. Deux entrées peuvent être des duplicates même avec des mots différents — « comment réinitialiser mon mot de passe ? » et « mot de passe oublié, comment faire ? » ne partagent que trois mots courants —, et c’est exactement ce qu’un hachage ou une comparaison de chaînes ne verra jamais. Les embeddings les détectent.

La fonction calcule la matrice complète des similarités et remonte toutes les paires au-dessus du seuil. Attention au coût : cette comparaison est quadratique, ce qui reste instantané sur quelques milliers de textes mais devient impraticable sur un million — il faut alors passer par un index approché comme ceux vus avec les bases vectorielles.

def trouver_duplicates(
    textes: list[str],
    seuil: float = 0.90,
    model: str = "text-embedding-3-large"
) -> list[dict]:
    """Trouve les paires de textes sémantiquement identiques."""
    response = client.embeddings.create(
        input=textes, model=model
    )
    embeddings = np.array(
        [d.embedding for d in sorted(response.data, key=lambda x: x.index)]
    )

    # Matrice de similarité
    sim_matrix = embeddings @ embeddings.T

    duplicates = []
    for i in range(len(textes)):
        for j in range(i + 1, len(textes)):
            if sim_matrix[i][j] >= seuil:
                duplicates.append({
                    "texte_a": textes[i],
                    "texte_b": textes[j],
                    "index_a": i,
                    "index_b": j,
                    "similarite": float(sim_matrix[i][j])
                })

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

# Exemple
textes = [
    "Comment réinitialiser mon mot de passe ?",
    "Je n'arrive pas à me connecter à mon compte",
    "Mot de passe oublié, comment faire ?",           # Duplicate de [0]
    "Quelle est la politique de remboursement ?",
    "Impossible d'accéder à mon espace personnel",    # Duplicate de [1]
    "Comment obtenir un remboursement ?",              # Duplicate de [3]
]

dupes = trouver_duplicates(textes, seuil=0.85)
print(f"{len(dupes)} paires de duplicates trouvées :")
for d in dupes:
    print(f"  [{d['similarite']:.3f}]")
    print(f"    A: {d['texte_a']}")
    print(f"    B: {d['texte_b']}")

L’appel passe ici un seuil de 0,85 plutôt que la valeur par défaut de 0,90, parce que des reformulations d’une même question de support restent nettement en dessous de 0,90. C’est le réglage sur lequel vous passerez le plus de temps : trop haut, vous laissez passer les doublons ; trop bas, vous fusionnez « comment changer mon mot de passe » et « comment changer mon adresse email », deux questions distinctes qui se ressemblent beaucoup.

Pipeline de déduplication

Détecter des paires ne suffit pas à nettoyer un corpus. Si A ressemble à B et B à C, les trois appartiennent au même groupe, même si A et C ne se sont jamais comparés directement au-dessus du seuil. Le regroupement transitif est le vrai travail, et la structure Union-Find le fait en un passage : chaque paire détectée fusionne deux ensembles, et l’on obtient à la fin des groupes de duplicates entièrement constitués.

Reste à choisir quel membre survit. La règle retenue ici — le texte le plus court — convient bien à des questions de support, où la formulation la plus concise est souvent la plus canonique. Sur un corpus documentaire, vous voudrez probablement l’inverse, ou le plus récent : c’est une décision métier, pas un détail technique. Le dictionnaire renvoyé conserve la liste des textes supprimés, de sorte qu’un humain puisse vérifier la casse avant de valider.

class Deduplicateur:
    def __init__(self, seuil: float = 0.90):
        self.client = OpenAI()
        self.seuil = seuil

    def dedupliquer(self, textes: list[str]) -> dict:
        """Supprime les duplicates sémantiques."""
        response = self.client.embeddings.create(
            input=textes, model="text-embedding-3-large"
        )
        embeddings = np.array(
            [d.embedding for d in sorted(response.data, key=lambda x: x.index)]
        )

        # Union-Find pour regrouper les duplicates
        parent = list(range(len(textes)))

        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

        # Trouver les paires similaires
        sim_matrix = embeddings @ embeddings.T
        for i in range(len(textes)):
            for j in range(i + 1, len(textes)):
                if sim_matrix[i][j] >= self.seuil:
                    union(i, j)

        # Garder un représentant par groupe
        groupes = {}
        for i in range(len(textes)):
            root = find(i)
            if root not in groupes:
                groupes[root] = []
            groupes[root].append(i)

        # Sélectionner les textes uniques (le plus court de chaque groupe)
        uniques = []
        supprimes = []
        for root, membres in groupes.items():
            representant = min(membres, key=lambda i: len(textes[i]))
            uniques.append(textes[representant])
            for m in membres:
                if m != representant:
                    supprimes.append(textes[m])

        return {
            "uniques": uniques,
            "supprimes": supprimes,
            "n_original": len(textes),
            "n_deduplique": len(uniques),
            "n_supprimes": len(supprimes)
        }

# Utilisation
dedup = Deduplicateur(seuil=0.85)
resultat = dedup.dedupliquer(textes)
print(f"Avant : {resultat['n_original']} textes")
print(f"Après : {resultat['n_deduplique']} textes")
print(f"Supprimés : {resultat['n_supprimes']}")

Appliquée en amont d’un index RAG, cette déduplication a un effet direct sur la qualité : elle empêche que les cinq résultats retournés à une question soient cinq versions du même passage, ce qui priverait le générateur de toute information complémentaire.

Résumé

• La distance au centroïde ou aux k-NN détecte les textes anormaux
• La similarité cosinus au-dessus d’un seuil identifie les duplicates
• L’Union-Find regroupe efficacement les duplicates transitifs
• Calibrez les seuils sur vos données (0.85-0.95 typiquement pour les duplicates)