Corrective RAG

Quand le retrieval ne suffit pas

Le RAG que vous avez construit repose sur un pari implicite : les chunks les plus similaires à la question sont les bons. Ce pari tient souvent, mais pas toujours. Interrogez votre index sur un sujet qu’il ne couvre pas et FAISS vous rendra malgré tout les cinq vecteurs les moins éloignés, aussi hors sujet soient-ils — un moteur de similarité ne sait pas répondre « je n’ai rien ». Le modèle reçoit alors du contexte inadapté, et comme le prompt lui demande de s’appuyer dessus, il produit une réponse qui a toutes les apparences du sérieux et aucun fondement. C’est le scénario d’hallucination le plus insidieux, parce que la chaîne technique a fonctionné sans erreur.

Le Corrective RAG (CRAG) intercale une étape d’évaluation entre le retrieval et la génération. Un LLM examine chaque document retrouvé et tranche : pertinent ou non. Si le verdict est négatif, le système ne se contente pas de générer avec ce qu’il a — il bascule vers une recherche web pour aller chercher ailleurs l’information qui manque à votre corpus. Le flux compte donc quatre temps : le retrieval classique dans le vector store, le grading de chaque document par rapport à la question, la décision de déclencher ou non la recherche web dès lors qu’au moins un document est jugé non pertinent, puis la génération avec les meilleurs documents disponibles.

Le grader de documents

Tout repose sur le grader. L’astuce consiste à demander au LLM un jugement binaire au format JSON plutôt qu’une appréciation libre : {"score": "yes"} ou {"score": "no"}, rien d’autre. Le paramètre response_format={"type": "json_object"} garantit une sortie parsable, et temperature=0 élimine la variabilité — un même couple question/document doit toujours recevoir le même verdict, sans quoi le comportement de votre pipeline devient irreproductible.

from mistralai.client import Mistral
import json

client = Mistral(api_key=api_key)

def grade_document(question, document):
    """Évaluer si un document est pertinent pour une question."""
    prompt = f"""Vous êtes un évaluateur de pertinence. Étant donné une question
et un document, déterminez si le document contient des informations pertinentes
pour répondre à la question.

Répondez UNIQUEMENT avec un objet JSON : {{"score": "yes"}} ou {{"score": "no"}}

Question : {question}
Document : {document}

Évaluation JSON :"""

    response = client.chat.complete(
        model="mistral-large-latest",
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
        response_format={"type": "json_object"},
    )

    result = json.loads(response.choices[0].message.content)
    return result.get("score", "no") == "yes"

# Test
question = "Quelles sont les étapes du RAG ?"
doc_pertinent = "Le RAG se décompose en retrieval, augmentation et génération."
doc_non_pertinent = "La recette du tiramisu nécessite du mascarpone."

print(grade_document(question, doc_pertinent))      # True
print(grade_document(question, doc_non_pertinent))   # False

Le test de fin de bloc mérite qu’on s’y arrête. La recette de tiramisu illustre exactement ce que la similarité vectorielle laisse passer : si votre corpus ne parle que de cuisine et qu’on l’interroge sur le RAG, ce chunk arrivera en tête du classement faute de concurrence. Le grader, lui, ne compare pas des vecteurs mais lit le contenu, et le rejette.

Appliqué à l’ensemble des chunks retrouvés, ce jugement produit deux informations : la liste filtrée des documents à conserver, et un booléen indiquant s’il faut compléter par le web. Le seuil retenu ici est volontairement strict — un seul document écarté suffit à déclencher le fallback, au motif que le contexte est alors incomplet.

def grade_documents(question, retrieved_chunks):
    """Évaluer et filtrer les documents retrouvés."""
    relevant_docs = []
    irrelevant_count = 0

    for chunk in retrieved_chunks:
        is_relevant = grade_document(question, chunk["text"])
        if is_relevant:
            relevant_docs.append(chunk)
        else:
            irrelevant_count += 1

    need_web_search = irrelevant_count > 0

    print(f"  Pertinents: {len(relevant_docs)}/{len(retrieved_chunks)}")
    print(f"  Recherche web nécessaire: {need_web_search}")

    return relevant_docs, need_web_search

Le filet de sécurité web

Quand les documents locaux ne suffisent pas, le système va chercher ailleurs. Mistral fournit pour cela un outil intégré, web_search, que l’on active à travers l’API Conversations : le modèle lance lui-même la recherche, puis rédige sa réponse en citant ses sources. Ce détail compte, car un simple appel client.chat.complete ne peut pas chercher sur le web : le modèle répondrait de mémoire, avec le risque d’inventer exactement ce que le CRAG cherche à éviter. Et depuis la version 3.0 du SDK, les chat completions refusent d’ailleurs l’outil web_search.

Le point de conception à retenir est le format de retour : la fonction rend une liste de dictionnaires ayant la même forme que vos chunks locaux, avec un champ source valant web_search, et y ajoute les URL citées. Cette uniformité permet aux étapes suivantes d’ignorer complètement l’origine des documents, tout en conservant la trace pour l’utilisateur final.

def web_search_fallback(question):
    """Recherche web de secours quand les documents locaux sont insuffisants."""
    response = client.beta.conversations.start(
        model="mistral-medium-latest",
        inputs=question,
        instructions="Réponds de façon factuelle, en t'appuyant sur les résultats de la recherche web.",
        tools=[{"type": "web_search"}],
        store=False,  # pas besoin de conserver la conversation côté Mistral
    )

    texte, urls = "", []
    for entry in response.outputs:
        if entry.type != "message.output":
            continue  # on ignore l'entrée qui décrit l'exécution de l'outil
        if isinstance(entry.content, str):
            texte += entry.content
            continue
        for chunk in entry.content:
            if chunk.type == "text":
                texte += chunk.text
            elif chunk.type == "tool_reference" and chunk.url:
                urls.append(chunk.url)

    # Retourner comme un "document" web, avec les sources citées
    return [{
        "text": texte,
        "source": "web_search",
        "urls": urls,
    }]

La réponse contient deux types d’entrées : une entrée tool.execution, qui décrit la recherche lancée, et une entrée message.output, dont le contenu alterne des fragments de texte et des références (tool_reference) portant le titre et l’URL des pages consultées. Un administrateur de l’organisation peut restreindre les domaines interrogés par une liste d’autorisation ou d’exclusion ; les résultats filtrés disparaissent sans avertissement, si bien qu’une liste d’autorisation trop étroite peut laisser la recherche sans résultat.

Source officielle : Websearch (documentation Mistral)

Le pipeline assemblé

La classe CorrectiveRAG reprend la structure du pipeline standard en y insérant le grading. Le k=4 par défaut est un peu plus généreux que les trois chunks habituels, ce qui se comprend : puisque le filtrage va en écarter une partie, mieux vaut partir avec de la marge. La méthode ask orchestre les quatre temps et affiche chacun d’eux, ce qui vous permet de suivre en direct le raisonnement du système — combien de documents ont survécu au grading, si le fallback s’est déclenché. Le dictionnaire de retour expose d’ailleurs web_search_used : c’est un indicateur précieux à surveiller en production, car un taux de recours au web anormalement élevé ne signale pas un défaut du CRAG mais une lacune de votre corpus.

class CorrectiveRAG:
    """Pipeline Corrective RAG avec évaluation et fallback web."""

    def __init__(self, chunks, embeddings, index):
        self.chunks = chunks
        self.embeddings = embeddings
        self.index = index

    def retrieve(self, question, k=4):
        """Étape 1 : retrieval standard."""
        response = client.embeddings.create(
            model="mistral-embed",
            inputs=[question],
        )
        query_vec = np.array([response.data[0].embedding], dtype=np.float32)
        distances, indices = self.index.search(query_vec, k)

        results = []
        for dist, idx in zip(distances[0], indices[0]):
            if idx >= 0:
                results.append({
                    "text": self.chunks[idx]["text"],
                    "source": self.chunks[idx].get("source", "local"),
                })
        return results

    def grade(self, question, documents):
        """Étape 2 : évaluer la pertinence."""
        relevant = []
        for doc in documents:
            if grade_document(question, doc["text"]):
                relevant.append(doc)
        need_web = len(relevant) < len(documents)
        return relevant, need_web

    def generate(self, question, documents):
        """Étape 3 : générer la réponse."""
        context = "\n\n---\n\n".join([d["text"] for d in documents])
        prompt = f"""Contexte :\n{context}\n\nQuestion : {question}"""

        response = client.chat.complete(
            model="mistral-large-latest",
            messages=[
                {"role": "system", "content": "Répondez en vous basant sur le contexte."},
                {"role": "user", "content": prompt},
            ],
            temperature=0.1,
        )
        return response.choices[0].message.content

    def ask(self, question, k=4):
        """Pipeline complet."""
        print(f"\n=== Question : {question} ===")

        # 1. Retrieval
        print("1. Retrieval...")
        docs = self.retrieve(question, k=k)

        # 2. Grading
        print("2. Grading...")
        relevant_docs, need_web = self.grade(question, docs)

        # 3. Web search fallback si nécessaire
        if need_web:
            print("3. Fallback web search...")
            web_docs = web_search_fallback(question)
            relevant_docs.extend(web_docs)

        # 4. Génération
        print("4. Génération...")
        answer = self.generate(question, relevant_docs)

        return {
            "answer": answer,
            "sources_used": len(relevant_docs),
            "web_search_used": need_web,
        }

# Utilisation
crag = CorrectiveRAG(chunks, embeddings, index)
result = crag.ask("Comment Mistral implémente-t-il les embeddings ?")

print(f"\nRéponse : {result['answer']}")
print(f"Web search utilisé : {result['web_search_used']}")

Cette fiabilité a un prix qu’il faut assumer en connaissance de cause : avec k=4, chaque question déclenche quatre appels de grading avant même la génération. La latence perçue par l’utilisateur augmente d’autant, et la facture aussi. L’arbitrage se pose différemment selon l’usage — sur un assistant juridique ou médical, la vérification est non négociable ; sur un moteur de recherche interne où l’utilisateur voit les sources et juge lui-même, elle l’est moins.

Points clés à retenir

• Le Corrective RAG ajoute une évaluation de pertinence après le retrieval
• Un LLM sert de “grader” pour juger chaque document : pertinent ou non
• Si des documents sont jugés non pertinents, un fallback vers la recherche web se déclenche
• Cette approche réduit significativement les hallucinations dues à un contexte inadapté
• Le grading ajoute de la latence mais améliore fortement la fiabilité des réponses