Détection d'ancres pour RAG : détecteurs parallèles, puis un seul appel LLM à la fin.

Détection d'ancres pour RAG : Détecteurs parallèles, puis un seul appel LLM à la fin

La génération augmentée par récupération (RAG) est devenue l'architecture de facto pour ancrer les sorties des grands modèles de langage (LLM) dans des connaissances externes. Cependant, un défi persiste : comment garantir que le contexte récupéré contient des « ancres » fiables et pertinentes avant que le LLM ne génère une réponse. Les approches traditionnelles appellent le LLM plusieurs fois – une fois pour la récupération, une fois pour la vérification, et une fois pour la génération finale – ce qui introduit latence et coût. Un paradigme plus efficace émerge : exécuter des détecteurs légers en parallèle pour identifier les ancres dans les documents récupérés, puis effectuer un seul appel LLM à la fin pour la génération. Cet article explore l'approche de détection d'ancres, fournit une implémentation concrète et montre comment construire un système prêt pour la production à l'aide d'outils open-source.

Comprendre la détection d'ancres dans RAG

Une ancre dans RAG est un morceau de texte récupéré qui sert de base fiable pour la réponse du LLM. Il s'agit d'un fait, d'une statistique, d'une citation ou d'une déclaration logique qui peut être vérifiée et utilisée directement dans la génération. La détection d'ancres est le processus d'identification de ces morceaux avant que le LLM ne voie le contexte. L'idée clé est que de nombreuses ancres potentielles sont bruitées – elles peuvent être non pertinentes, contradictoires ou obsolètes. En exécutant des détecteurs parallèles, vous pouvez filtrer et classer les ancres sans invoquer le LLM, réservant l'appel LLM pour la sortie finale et polie.

Cette approche s'aligne sur les tendances récentes en conception de systèmes d'IA, où des modèles spécialisés et légers (par exemple, modèles d'embedding, classifieurs ou petits transformeurs) gèrent le prétraitement, tandis que les grands modèles génératifs sont utilisés avec parcimonie. Les grands acteurs de l'industrie comme OpenAI, Google et Microsoft ont publié des recherches sur la réduction des appels LLM grâce à des pipelines de prétraitement efficaces, bien que les détails d'implémentation varient. L'approche des détecteurs parallèles est une synthèse pratique de ces idées.

Prérequis

Pour suivre ce guide, vous avez besoin de :

• Python 3.9 ou ultérieur
• Une machine avec au moins 8 Go de RAM (16 Go recommandés pour exécuter plusieurs détecteurs)
• Une familiarité de base avec Python et les outils en ligne de commande
• Les packages Python suivants : `transformers`, `sentence-transformers`, `torch`, `faiss-cpu`, `pandas`, `numpy`, `openai` (ou similaire pour l'accès LLM)

Vous aurez également besoin d'une clé API LLM (par exemple, d'OpenAI) pour l'étape de génération finale. Les détecteurs d'ancres eux-mêmes s'exécutent localement.

Installation étape par étape

1. Créer un environnement virtuel

Commencez par créer un environnement Python isolé pour éviter les conflits de dépendances.

python3 -m venv anchor-rag-env
source anchor-rag-env/bin/activate

Cette commande crée un environnement virtuel nommé `anchor-rag-env` et l'active.

2. Installer les dépendances principales

Installez les bibliothèques nécessaires pour les détecteurs et l'intégration LLM.

pip install transformers sentence-transformers torch faiss-cpu pandas numpy openai

• `transformers` et `sentence-transformers` fournissent des modèles pré-entraînés pour l'encodage et la classification de texte.
• `torch` est le backend PyTorch.
• `faiss-cpu` permet une recherche de similarité rapide pour le classement des ancres.
• `pandas` et `numpy` gèrent la manipulation des données.
• `openai` est utilisé pour l'appel LLM final (remplacez par votre fournisseur LLM choisi).

3. Vérifier l'installation

Testez que l'environnement est correctement configuré en exécutant une petite vérification.

python -c "import sentence_transformers; print('Installation OK')"

Si vous voyez « Installation OK », continuez. Sinon, vérifiez les erreurs (par exemple, pilotes CUDA manquants pour torch – si vous n'avez pas de GPU, les versions CPU seulement suffiront).

Architecture générale

Le système de détection d'ancres fonctionne en trois étapes :

1. **Récupération** – Obtenez un ensemble de documents candidats (par exemple, à partir d'une base de données vectorielle ou d'une recherche web). 2. **Détecteurs parallèles** – Exécutez plusieurs modèles légers simultanément pour noter et filtrer les ancres :

• Un détecteur de pertinence (par exemple, classifieur basé sur BERT)
• Un détecteur de factualité (par exemple, un modèle NLI)
• Un détecteur de redondance (par exemple, similarité cosinus)
• Un détecteur de cohérence des entités (par exemple, chevauchement d'entités nommées)

3. **Appel LLM unique** – Agrégrez les ancres les mieux notées et passez-les comme contexte condensé au LLM pour la génération.

Cette conception minimise les appels LLM tout en maximisant la qualité du contexte d'entrée.

Implémentation : Construction des détecteurs parallèles

Détecteur 1 : Score de pertinence

Le détecteur de pertinence utilise un modèle sentence-transformer pour calculer la similarité cosinus entre la requête et chaque morceau récupéré. Il s'agit d'une étape rapide et non supervisée.

from sentence_transformers import SentenceTransformer, util
import numpy as np

# Charger un modèle d'embedding léger
model = SentenceTransformer('all-MiniLM-L6-v2')

def relevance_detector(query, chunks):
    query_emb = model.encode(query, convert_to_tensor=True)
    chunk_embs = model.encode(chunks, convert_to_tensor=True)
    scores = util.cos_sim(query_emb, chunk_embs)[0].cpu().numpy()
    return scores

Cette fonction retourne un tableau numpy de scores de pertinence (0 à 1) pour chaque morceau. Des scores élevés indiquent un fort alignement sémantique avec la requête.

Détecteur 2 : Vérification de factualité

Un détecteur de factualité utilise un modèle d'inférence en langage naturel (NLI) pour vérifier si chaque morceau contient des déclarations logiquement cohérentes. Nous utilisons un modèle NLI pré-entraîné de Hugging Face.

from transformers import pipeline

nli_pipeline = pipeline("text-classification", model="roberta-large-mnli")

def factuality_detector(chunks):
    # Pour chaque morceau, vérifier s'il est "impliqué" par lui-même (une heuristique pour la cohérence interne)
    scores = []
    for chunk in chunks:
        result = nli_pipeline(chunk[:512])  # Tronquer à la longueur maximale du modèle
        # Utiliser le score "ENTAILMENT" comme proxy pour la factualité
        if result[0]['label'] == 'ENTAILMENT':
            scores.append(result[0]['score'])
        else:
            scores.append(0.0)
    return np.array(scores)

Il s'agit d'une heuristique simplifiée. En production, vous compareriez les morceaux à une base de connaissances de confiance ou utiliseriez un modèle de factualité dédié. Le point clé est qu'il s'exécute en parallèle avec les autres détecteurs.

Détecteur 3 : Filtre de redondance

La redondance est détectée en calculant la similarité cosinus par paire entre les morceaux. Les morceaux trop similaires entre eux sont pénalisés pour éviter les informations en double.

def redundancy_detector(chunks, threshold=0.85):
    chunk_embs = model.encode(chunks, convert_to_tensor=True)
    sim_matrix = util.cos_sim(chunk_embs, chunk_embs).cpu().numpy()
    scores = np.ones(len(chunks))
    for i in range(len(chunks)):
        # Compter combien d'autres morceaux sont très similaires à celui-ci
        redundant_count = np.sum(sim_matrix[i] > threshold) - 1
        scores[i] = 1.0 / (1.0 + redundant_count)  # Pénaliser les morceaux redondants
    return scores

Un morceau unique obtient un score de 1,0 ; un morceau identique à trois autres obtient un score de 0,25.

Détecteur 4 : Cohérence des entités

Ce détecteur garantit que les morceaux contiennent des entités (personnes, lieux, dates) pertinentes pour la requête. Il utilise un modèle simple de reconnaissance d'entités nommées (NER).

from transformers import pipeline

ner_pipeline = pipeline("ner", model="dslim/bert-base-NER", aggregation_strategy="simple")

def entity_coherence_detector(query, chunks):
    query_entities = set([e['word'].lower() for e in ner_pipeline(query)])
    scores = []
    for chunk in chunks:
        chunk_entities = set([e['word'].lower() for e in ner_pipeline(chunk[:512])])
        overlap = len(query_entities & chunk_entities) / max(len(query_entities), 1)
        scores.append(overlap)
    return np.array(scores)

Un score de 1,0 signifie que toutes les entités de la requête apparaissent dans le morceau ; 0,0 signifie aucun chevauchement.

Exécution des détecteurs en parallèle

Pour exécuter tous les détecteurs simultanément, utilisez le module `concurrent.futures` de Python. C'est le cœur du concept de « détecteurs parallèles ».

from concurrent.futures import ThreadPoolExecutor

def run_all_detectors(query, chunks):
    with ThreadPoolExecutor(max_workers=4) as executor:
        future_relevance = executor.submit(relevance_detector, query, chunks)
        future_factuality = executor.submit(factuality_detector, chunks)
        future_redundancy = executor.submit(redundancy_detector, chunks)
        future_entity = executor.submit(entity_coherence_detector, query, chunks)
        
        relevance_scores = future_relevance.result()
        factuality_scores = future_factuality.result()
        redundancy_scores = future_redundancy.result()
        entity_scores = future_entity.result()
    
    # Combiner les scores (somme pondérée simple)
    combined = (0.4 * relevance_scores +
                0.3 * factuality_scores +
                0.2 * redundancy_scores +
                0.1 * entity_scores)
    return combined

Ajustez les poids en fonction de votre cas d'utilisation. Les scores combinés sont utilisés pour sélectionner les k meilleures ancres.

Appel LLM unique à la fin

Après que les détecteurs parallèles ont noté et filtré les morceaux, vous récupérez les k meilleures ancres et les passez au LLM en un seul appel. C'est là que la génération finale a lieu.

import openai

openai.api_key = "votre-clé-api-ici"

def generate_with_anchors(query, top_anchors, llm_model="gpt-4"):
    # Concaténer les ancres en un contexte condensé
    anchor_text = "\n\n".join([f"[Ancre {i+1}] {chunk}" for i, chunk in enumerate(top_anchors)])
    
    system_prompt = "Vous êtes un assistant utile. Utilisez uniquement les ancres fournies pour répondre à la requête. Si les ancres sont insuffisantes, dites-le."
    user_prompt = f"Requête : {query}\n\nAncres :\n{anchor_text}\n\nRéponse :"
    
    response = openai.ChatCompletion.create(
        model=llm_model,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ],
        max_tokens=500,
        temperature=0.3
    )
    return response.choices[0].message['content']

Cette fonction effectue exactement un appel LLM. Le LLM reçoit uniquement les ancres les mieux classées, réduisant l'utilisation de tokens et le risque d'hallucination.

Exemple d'utilisation

Étape 1 : Préparer des données d'exemple

Supposons que vous ayez récupéré trois morceaux d'une base de connaissances.

query = "Quelle est la capitale de la France ?"
chunks = [
    "La France est un pays d'Europe. Sa capitale est Paris, connue pour la Tour Eiffel.",
    "Paris est la capitale de la France et est souvent appelée la Ville Lumière.",
    "Le musée du Louvre à Paris abrite la Joconde."
]

Étape 2 : Exécuter les détecteurs parallèles

combined_scores = run_all_detectors(query, chunks)
top_indices = np.argsort(combined_scores)[-2:][::-1]  # Sélectionner les 2 meilleures ancres
top_anchors = [chunks[i] for i in top_indices]
print("Meilleures ancres :", top_anchors)

La sortie pourrait être :

Meilleures ancres : ['La France est un pays d'Europe. Sa capitale est Paris, connue pour la Tour Eiffel.', 'Paris est la capitale de la France et est souvent appelée la Ville Lumière.']

Étape 3 : Générer la réponse finale

answer = generate_with_anchors(query, top_anchors)
print(answer)

Sortie :

La capitale de la France est Paris.

Un seul appel LLM a été effectué, et il a utilisé uniquement les deux morceaux les plus pertinents, factuels et non redondants.

Considérations sur les performances

• **Latence :** Les détecteurs parallèles s'exécutent en ~0,5 à 2 secondes sur un CPU (selon le nombre de morceaux et la taille du modèle). L'appel LLM ajoute 1 à 3 secondes. Le temps total est généralement inférieur à 5 secondes.
• **Coût :** Vous payez pour un appel LLM par requête au lieu de trois ou quatre. Les détecteurs s'exécutent localement sans coût d'API.
• **Précision :** Le système de score pondéré peut être ajusté. Pour une utilisation spécifique à un domaine, remplacez le modèle NLI générique par un modèle de factualité finement ajusté.
• **Évolutivité :** Utilisez l'indexation FAISS pour de grands ensembles de morceaux (par exemple, des milliers de morceaux) afin d'accélérer la détection de pertinence et de redondance.

Conclusion

La détection d'ancres avec des détecteurs parallèles suivie d'un seul appel LLM est un modèle de conception pragmatique pour les systèmes RAG en production. Elle réduit la latence, diminue les coûts d'API et améliore la qualité des sorties en filtrant le contexte bruité avant que le LLM ne le voie. L'implémentation présentée utilise des modèles légers open-source pour la pertinence, la factualité, la redondance et la cohérence des entités – tous s'exécutant en parallèle. L'appel LLM final génère ensuite une réponse ancrée à partir d'un ensemble d'ancres sélectionnées. Cette approche s'inspire des tendances industrielles vers des pipelines d'IA efficaces en plusieurs étapes, comme le montrent les recherches d'OpenAI, Google et Microsoft. En adoptant ce modèle, vous pouvez construire des systèmes RAG à la fois rapides et fiables, sans sacrifier la puissance générative des grands modèles de langage.