Technical Tutorials

Les systèmes de génération augmentée par récupération (RAG) ne sont pas des entités statiques ; leur survie dépend de la qualité de leur couche de récupération. L'une des menaces les plus subtiles mais destructrices pour cette couche est la dérive des embeddings. À mesure que votre corpus grandit, que les versions de modèles sont mises à jour ou que la sémantique des documents évolue, les relations géométriques entre vos vecteurs stockés peuvent se dégrader, entraînant des récupérations non pertinentes et des réponses hallucinées. Ce guide explore des stratégies techniques pour maintenir la cohérence des vecteurs dans des environnements dynamiques.

Comprendre la dérive des embeddings

La dérive des embeddings se produit lorsque la représentation de nouveaux documents ou de documents mis à jour dans l'espace vectoriel diverge de celle des documents précédemment stockés, même si leur signification sémantique reste similaire. Cela peut se produire en raison de :

Mises à jour de modèles : La mise à niveau d'une version A à une version B de Sentence-Transformer modifie l'ensemble de l'espace vectoriel.
Dérive du corpus : Les nouveaux documents introduisent de nouveaux vocabulaires ou concepts, déplaçant le centroïde de l'espace vectoriel.
Modifications du prétraitement : La modification des stratégies de découpage ou des techniques de normalisation change la manière dont le texte est mappé vers des vecteurs.

Stratégie 1 : Index vectoriels versionnés

La solution la plus robuste consiste à traiter les index vectoriels comme des versions de code. Ne remplacez jamais les vecteurs existants. Au lieu de cela, créez de nouvelles partitions d'index pour chaque version de modèle ou de prétraitement.


import chromadb
from datetime import datetime

# Initialiser le client
client = chromadb.PersistentClient(path="./chroma_db")

def create_versioned_index(collection_name, model_version, embedding_fn):
    """
    Crée une nouvelle collection pour une version spécifique du modèle.
    Les données existantes ne sont PAS modifiées, garantissant la cohérence pour les requêtes héritées.
    """
    versioned_name = f"{collection_name}_v{model_version}"
    
    # Vérifier si la collection existe
    if versioned_name not in client.list_collections():
        # Configurer la similarité cosinus pour des métriques de distance cohérentes
        client.create_collection(
            name=versioned_name,
            metadata={"hnsw:space": "cosine", "model_version": model_version, "created_at": datetime.utcnow().isoformat()}
        )
        print(f"Nouvel index créé : {versioned_name}")
    else:
        print(f"L'index {versioned_name} existe déjà.")
    
    return client.get_collection(versioned_name)

# Exemple d'utilisation
# col_v1 = create_versioned_index("docs", "1.0.0", lambda x: embed_v1(x))
# col_v2 = create_versioned_index("docs", "2.0.0", lambda x: embed_v2(x))

Stratégie 2 : Points d'ancrage et normalisation

Si vous devez utiliser un index unique (par exemple, en raison de contraintes de ressources), implémentez un mécanisme de « détection de dérive » à l'aide de points d'ancrage. Sélectionnez un petit ensemble stable de documents de référence (ancres) qui représentent des concepts fondamentaux. Calculez régulièrement la similarité entre ces ancres et leurs embeddings actuels par rapport à leurs embeddings historiques.


import numpy as np

def check_drift(anchor_embeddings_current, anchor_embeddings_history, threshold=0.05):
    """
    Compare la similarité cosinus des points d'ancrage entre les embeddings actuels et historiques.
    """
    # Normaliser les vecteurs
    current_norm = anchor_embeddings_current / np.linalg.norm(anchor_embeddings_current, axis=1, keepdims=True)
    history_norm = anchor_embeddings_history / np.linalg.norm(anchor_embeddings_history, axis=1, keepdims=True)
    
    # Calculer la similarité cosinus par paires
    similarities = np.dot(current_norm, history_norm.T)
    
    # Similarité moyenne sur toutes les ancres
    avg_similarity = np.mean(similarities)
    
    if avg_similarity < (1 - threshold):
        raise Exception(f"Dérive des embeddings détectée : Similarité moyenne {avg_similarity:.4f} inférieure au seuil {1-threshold}")
    else:
        print(f"Vérification de la dérive réussie : Similarité moyenne {avg_similarity:.4f}")

# Exemple :
# current_anchors = get_embeddings(anchor_texts, model_version="v2")
# history_anchors = get_embeddings(anchor_texts, model_version="v1")
# check_drift(current_anchors, history_anchors)

Stratégie 3 : Pipelines de ré-embedding continus

Intégrez le ré-embedding dans votre pipeline CI/CD. Lorsqu'une mise à jour de modèle est déclenchée, planifiez une tâche en arrière-plan pour ré-encoder uniquement le delta (nouveaux documents/documents modifiés) dans la nouvelle version de l'index. Utilisez une stratégie d'écriture double pendant la période de transition, où les requêtes sont servies à partir des anciens et nouveaux index avec des scores pondérés, en déplaçant progressivement le poids vers le nouvel index à mesure que la confiance augmente.

Conclusion

La dérive des embeddings est une conséquence inévitable des systèmes RAG dynamiques, mais elle est gérable. En adoptant des index versionnés, en implémentant une détection de dérive basée sur des ancres et en automatisant les pipelines de ré-embedding, vous pouvez vous assurer que votre couche de récupération reste cohérente et fiable. Traitez votre espace vectoriel avec la même rigueur que votre schéma de base de données — surveillez-le, versionnez-le et testez-le avant le déploiement.