Corrective RAG : Le Système qui S'Auto-Corrige pour des Réponses Parfaites

Le RAG naif a un problème que tout le monde connaît mais que peu résolvent : il génère des réponses même quand les documents récupérés ne contiennent pas la bonne information. Corrective RAG (CRAG) et Self-RAG introduisent une couche d'auto-évaluation qui permet au système de détecter les mauvaises récupérations et de se corriger automatiquement. Le résultat : une réduction de 40 à 60% des hallucinations.

TL;DR

• CRAG (Yan et al., 2024) évalue la pertinence des documents récupérés et décide de la stratégie : utiliser, compléter ou remplacer
• Self-RAG (Asai et al., 2024) va plus loin avec des tokens de réflexion pour évaluer chaque étape
• Trois décisions : Correct (utiliser les docs), Ambigu (compléter via web search), Incorrect (fallback complet)
• -40-60% d'hallucinations par rapport au RAG naif
• Surcoût en latence : +1-3 secondes (évaluation de pertinence)
• Implémentable avec LangChain/LangGraph en moins de 100 lignes

Le problème du RAG naif

Le RAG naif fait confiance aveuglément aux documents récupérés. Si le top-k contient des documents non pertinents, le LLM va quand même s'en servir pour générer une réponse, souvent en hallucinent.

Pipeline RAG Naif :
┌──────────┐    ┌──────────────┐    ┌─────────────┐    ┌──────────┐
│ Question │───▶│  Retrieval   │───▶│  Documents  │───▶│   LLM    │
│          │    │  (top-k=5)   │    │ (pertinents │    │ Génère   │
│          │    │              │    │   ou non!)   │    │ quand    │
│          │    │              │    │              │    │ même...  │
└──────────┘    └──────────────┘    └─────────────┘    └──────────┘
                                          ❌
                               Pas de vérification
                               de la pertinence !

Cas typiques de défaillance

Architecture de Corrective RAG (CRAG)

Le paper CRAG (Yan et al., 2024) introduit un évaluateur de pertinence léger qui agit comme un filtre entre le retrieval et la génération.

Le flux de décision

┌──────────────────────────────────────────────────────────────────┐
│                    Pipeline Corrective RAG                        │
├──────────────────────────────────────────────────────────────────┤
│                                                                   │
│  Question ──▶ Retrieval (top-k) ──▶ Évaluation pertinence        │
│                                           │                       │
│                          ┌────────────────┼────────────────┐      │
│                          │                │                │      │
│                          ▼                ▼                ▼      │
│                     ┌─────────┐    ┌───────────┐    ┌──────────┐ │
│                     │ CORRECT │    │  AMBIGU   │    │ INCORRECT│ │
│                     │ Score   │    │  Score    │    │  Score   │ │
│                     │ > 0.7   │    │ 0.3-0.7  │    │  < 0.3   │ │
│                     └────┬────┘    └─────┬─────┘    └────┬─────┘ │
│                          │               │               │       │
│                          ▼               ▼               ▼       │
│                     Utiliser       Compléter par     Web Search   │
│                     les docs       web search        complet     │
│                     récupérés      + docs            (remplacer) │
│                          │               │               │       │
│                          └───────────────┼───────────────┘       │
│                                          ▼                       │
│                                   Knowledge                      │
│                                   Refinement                     │
│                                          │                       │
│                                          ▼                       │
│                                     Génération                   │
│                                     (LLM final)                  │
│                                                                   │
└──────────────────────────────────────────────────────────────────┘

Implémentation avec LangGraph

DEVELOPERpython
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.documents import Document
from typing import TypedDict, List, Literal
import json

class CRAGState(TypedDict):
    question: str
    documents: List[Document]
    relevance_decision: Literal["correct", "ambiguous", "incorrect"]
    web_results: List[Document]
    refined_documents: List[Document]
    generation: str

# 1. Évaluateur de pertinence
def evaluate_relevance(state: CRAGState) -> CRAGState:
    """Évalue la pertinence des documents récupérés."""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

    question = state["question"]
    documents = state["documents"]

    scores = []
    for doc in documents:
        prompt = f"""Évalue la pertinence de ce document pour répondre à la question.

Question : {question}
Document : {doc.page_content[:500]}

Score de pertinence (0.0 à 1.0) :
- 1.0 = parfaitement pertinent, répond directement
- 0.5 = partiellement pertinent
- 0.0 = pas du tout pertinent

Réponds UNIQUEMENT avec un nombre décimal."""

        score = float(llm.invoke(prompt).content.strip())
        scores.append(score)

    avg_score = sum(scores) / len(scores) if scores else 0

    if avg_score > 0.7:
        decision = "correct"
    elif avg_score > 0.3:
        decision = "ambiguous"
    else:
        decision = "incorrect"

    return {
        **state,
        "relevance_decision": decision
    }

# 2. Web search fallback
async def web_search_fallback(state: CRAGState) -> CRAGState:
    """Recherche web quand les documents sont insuffisants."""
    from langchain_community.tools import TavilySearchResults

    search = TavilySearchResults(max_results=5)
    results = await search.ainvoke(state["question"])

    web_docs = [
        Document(
            page_content=r["content"],
            metadata={"source": r["url"], "type": "web"}
        )
        for r in results
    ]

    return {**state, "web_results": web_docs}

# 3. Knowledge refinement
def refine_knowledge(state: CRAGState) -> CRAGState:
    """Affine et fusionne les sources de connaissances."""
    decision = state["relevance_decision"]

    if decision == "correct":
        refined = state["documents"]
    elif decision == "ambiguous":
        refined = state["documents"] + state.get("web_results", [])
    else:  # incorrect
        refined = state.get("web_results", [])

    # Déduplication et scoring
    seen = set()
    unique_docs = []
    for doc in refined:
        content_hash = hash(doc.page_content[:100])
        if content_hash not in seen:
            seen.add(content_hash)
            unique_docs.append(doc)

    return {**state, "refined_documents": unique_docs}

# 4. Génération finale
def generate_response(state: CRAGState) -> CRAGState:
    """Génère la réponse finale avec les documents affinés."""
    llm = ChatOpenAI(model="gpt-4o", temperature=0.1)

    context = "\n\n".join([
        doc.page_content for doc in state["refined_documents"]
    ])

    prompt = f"""Réponds à la question en te basant UNIQUEMENT sur le contexte fourni.
Si le contexte ne contient pas assez d'information, dis-le clairement.

Contexte :
{context}

Question : {state["question"]}

Réponse :"""

    response = llm.invoke(prompt)
    return {**state, "generation": response.content}

# Construction du graphe LangGraph
def route_by_relevance(state: CRAGState) -> str:
    """Route vers le bon noeud selon la décision."""
    if state["relevance_decision"] == "correct":
        return "refine"
    else:
        return "web_search"

workflow = StateGraph(CRAGState)

workflow.add_node("evaluate", evaluate_relevance)
workflow.add_node("web_search", web_search_fallback)
workflow.add_node("refine", refine_knowledge)
workflow.add_node("generate", generate_response)

workflow.set_entry_point("evaluate")
workflow.add_conditional_edges(
    "evaluate",
    route_by_relevance,
    {"refine": "refine", "web_search": "web_search"}
)
workflow.add_edge("web_search", "refine")
workflow.add_edge("refine", "generate")
workflow.add_edge("generate", END)

crag_chain = workflow.compile()

Self-RAG : L'Auto-Réflexion Poussée à l'Extrême

Self-RAG (Asai et al., 2024) va plus loin que CRAG en intégrant des tokens de réflexion directement dans le processus de génération. Le modèle apprend à évaluer chaque étape de son propre raisonnement.

Les quatre tokens de réflexion

Flux Self-RAG

Question: "Quels sont les avantages du RAG par rapport au fine-tuning ?"

Étape 1: [Retrieve] = yes → Lancer le retrieval
Étape 2: Doc 1 récupéré → [IsRel] = relevant ✅
         Doc 2 récupéré → [IsRel] = irrelevant ❌ (filtré)
Étape 3: Génération avec Doc 1
         → [IsSup] = fully supported ✅
         → [IsUse] = 4/5 ✅
Étape 4: Réponse finale validée

Implémentation Self-RAG

DEVELOPERpython
from langchain_openai import ChatOpenAI
from langchain_core.documents import Document
from typing import List, Tuple
from dataclasses import dataclass
from enum import Enum

class RetrieveDecision(Enum):
    YES = "yes"
    NO = "no"
    CONTINUE = "continue"

class RelevanceScore(Enum):
    RELEVANT = "relevant"
    IRRELEVANT = "irrelevant"

class SupportScore(Enum):
    FULLY = "fully_supported"
    PARTIALLY = "partially_supported"
    NONE = "no_support"

@dataclass
class ReflectionResult:
    retrieve: RetrieveDecision
    relevance: dict  # doc_id -> RelevanceScore
    support: SupportScore
    usefulness: int  # 1-5

class SelfRAG:
    def __init__(self, llm_model: str = "gpt-4o"):
        self.critic_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
        self.gen_llm = ChatOpenAI(model=llm_model, temperature=0.1)
        self.retriever = None  # Votre retriever ici

    def should_retrieve(self, question: str) -> RetrieveDecision:
        """Token [Retrieve] : faut-il chercher des documents ?"""
        prompt = f"""Évalue si cette question nécessite une recherche documentaire.

Question : {question}

Réponds avec UN seul mot :
- "yes" : la question nécessite des faits spécifiques
- "no" : la question peut être répondue avec des connaissances générales
- "continue" : la génération en cours est suffisante"""

        result = self.critic_llm.invoke(prompt).content.strip().lower()
        return RetrieveDecision(result)

    def evaluate_relevance(
        self, question: str, documents: List[Document]
    ) -> List[Tuple[Document, RelevanceScore]]:
        """Token [IsRel] : chaque document est-il pertinent ?"""
        evaluated = []
        for doc in documents:
            prompt = f"""Le document suivant est-il pertinent pour la question ?

Question : {question}
Document : {doc.page_content[:500]}

Réponds "relevant" ou "irrelevant"."""

            score = self.critic_llm.invoke(prompt).content.strip().lower()
            evaluated.append((
                doc,
                RelevanceScore(score)
            ))
        return evaluated

    def check_support(
        self, response: str, documents: List[Document]
    ) -> SupportScore:
        """Token [IsSup] : la réponse est-elle supportée ?"""
        context = "\n".join([d.page_content for d in documents])
        prompt = f"""La réponse est-elle supportée par les documents ?

Documents : {context[:2000]}
Réponse : {response}

Réponds avec :
- "fully_supported" : chaque affirmation est dans les documents
- "partially_supported" : certaines affirmations sont supportées
- "no_support" : la réponse n'est pas dans les documents"""

        result = self.critic_llm.invoke(prompt).content.strip().lower()
        return SupportScore(result)

    def rate_usefulness(self, question: str, response: str) -> int:
        """Token [IsUse] : la réponse est-elle utile ?"""
        prompt = f"""Note l'utilité de cette réponse (1 à 5).

Question : {question}
Réponse : {response}

5 = Parfaitement utile et complète
4 = Très utile mais pourrait être plus complète
3 = Moyennement utile
2 = Peu utile
1 = Inutile

Réponds avec UN seul chiffre."""

        return int(self.critic_llm.invoke(prompt).content.strip())

    async def query(self, question: str) -> str:
        """Pipeline Self-RAG complet."""
        # Étape 1: Faut-il récupérer ?
        retrieve_decision = self.should_retrieve(question)

        if retrieve_decision == RetrieveDecision.NO:
            return self.gen_llm.invoke(question).content

        # Étape 2: Récupérer et évaluer la pertinence
        documents = await self.retriever.aget_relevant_documents(question)
        evaluated = self.evaluate_relevance(question, documents)

        relevant_docs = [
            doc for doc, score in evaluated
            if score == RelevanceScore.RELEVANT
        ]

        if not relevant_docs:
            return "Je n'ai pas trouvé d'information pertinente."

        # Étape 3: Générer avec les documents pertinents
        context = "\n\n".join([d.page_content for d in relevant_docs])
        response = self.gen_llm.invoke(
            f"Contexte :\n{context}\n\nQuestion : {question}"
        ).content

        # Étape 4: Vérifier le support et l'utilité
        support = self.check_support(response, relevant_docs)
        usefulness = self.rate_usefulness(question, response)

        if support == SupportScore.NONE or usefulness < 3:
            # Re-générer avec un prompt plus strict
            response = self.gen_llm.invoke(
                f"""Réponds STRICTEMENT à partir du contexte.
Ne fais aucune supposition. Si l'info manque, dis-le.

Contexte :\n{context}\n\nQuestion : {question}"""
            ).content

        return response

Comparaison des variantes RAG

Benchmarks sur des datasets standard

Optimisations pratiques

1. Évaluateur de pertinence rapide

Au lieu d'utiliser un LLM pour évaluer chaque document, entraînez un petit classificateur.

DEVELOPERpython
from sentence_transformers import CrossEncoder

class FastRelevanceEvaluator:
    """Évaluateur rapide basé sur un cross-encoder."""

    def __init__(self, model_name="cross-encoder/ms-marco-MiniLM-L-6-v2"):
        self.model = CrossEncoder(model_name)

    def evaluate_batch(
        self, question: str, documents: List[Document]
    ) -> List[float]:
        """Évalue tous les documents en un seul batch."""
        pairs = [(question, doc.page_content) for doc in documents]
        scores = self.model.predict(pairs)
        return scores.tolist()

    def filter_relevant(
        self,
        question: str,
        documents: List[Document],
        threshold: float = 0.5
    ) -> tuple:
        """Retourne les docs pertinents et la décision."""
        scores = self.evaluate_batch(question, documents)

        relevant = [
            doc for doc, score in zip(documents, scores)
            if score > threshold
        ]

        avg = sum(scores) / len(scores) if scores else 0
        if avg > 0.7:
            decision = "correct"
        elif avg > 0.3:
            decision = "ambiguous"
        else:
            decision = "incorrect"

        return relevant, decision, scores

2. Cache des décisions de pertinence

DEVELOPERpython
import hashlib
from functools import lru_cache

class CachedCRAG:
    """CRAG avec cache des évaluations de pertinence."""

    def __init__(self):
        self.evaluator = FastRelevanceEvaluator()
        self.cache = {}

    def _cache_key(self, question: str, doc_content: str) -> str:
        combined = f"{question}||{doc_content[:200]}"
        return hashlib.md5(combined.encode()).hexdigest()

    def evaluate_with_cache(
        self, question: str, documents: List[Document]
    ) -> List[float]:
        scores = []
        uncached_indices = []
        uncached_docs = []

        for i, doc in enumerate(documents):
            key = self._cache_key(question, doc.page_content)
            if key in self.cache:
                scores.append(self.cache[key])
            else:
                scores.append(None)
                uncached_indices.append(i)
                uncached_docs.append(doc)

        if uncached_docs:
            new_scores = self.evaluator.evaluate_batch(
                question, uncached_docs
            )
            for idx, score in zip(uncached_indices, new_scores):
                key = self._cache_key(
                    question, documents[idx].page_content
                )
                self.cache[key] = score
                scores[idx] = score

        return scores

3. Seuils adaptatifs par domaine

DEVELOPERpython
DOMAIN_THRESHOLDS = {
    "medical": {
        "correct_threshold": 0.85,   # Plus strict
        "ambiguous_threshold": 0.5,
        "min_relevant_docs": 2       # Exige 2+ docs concordants
    },
    "legal": {
        "correct_threshold": 0.80,
        "ambiguous_threshold": 0.45,
        "min_relevant_docs": 2
    },
    "general": {
        "correct_threshold": 0.7,
        "ambiguous_threshold": 0.3,
        "min_relevant_docs": 1
    },
    "ecommerce": {
        "correct_threshold": 0.6,    # Plus tolérant
        "ambiguous_threshold": 0.25,
        "min_relevant_docs": 1
    }
}

Métriques d'évaluation

Pour mesurer l'efficacité de votre implémentation CRAG, suivez ces métriques.

FAQ

CRAG est-il compatible avec tous les retrievers ?

Oui. CRAG est indépendant du retriever utilisé en amont. Il fonctionne avec la recherche vectorielle pure, la recherche hybride, BM25, ou même un retriever basé sur les graphes de connaissances. L'évaluation de pertinence se fait sur les documents retournés, quelle que soit la méthode de récupération.

Comment gérer le cas où le web search ne retourne rien de pertinent non plus ?

Dans ce cas, le système devrait reconnaître honnêtement qu'il ne dispose pas de l'information. Implémentez un second évaluateur sur les résultats web. Si les résultats web sont aussi non pertinents, retournez un message explicite : "Je n'ai pas trouvé d'information fiable pour répondre à cette question." C'est toujours mieux qu'une hallucination.

Quel est le coût supplémentaire de CRAG par rapport au RAG naif ?

L'évaluation de pertinence avec un cross-encoder est quasi gratuite (< 50ms). Si vous utilisez un LLM pour l'évaluation, comptez environ 500 tokens supplémentaires par document évalué, soit environ $0.002-0.005 par requête avec GPT-4o-mini. Le web search fallback via Tavily coûte environ $0.005 par recherche. Au total, CRAG ajoute $0.005-0.01 par requête.

Self-RAG nécessite-t-il un modèle fine-tuné ?

Le paper original de Self-RAG utilise un modèle Llama fine-tuné avec des tokens de réflexion. En pratique, vous pouvez simuler le comportement avec des prompts structurés sur GPT-4o ou Claude, comme montré dans les exemples de code ci-dessus. Les performances sont légèrement inférieures mais restent largement supérieures au RAG naif.

CRAG fonctionne-t-il en temps réel pour un chatbot ?

Oui, avec des optimisations. Utilisez un cross-encoder rapide (MiniLM) plutôt qu'un LLM pour l'évaluation de pertinence. Pré-calculez les seuils pour votre domaine. Implémentez le cache des évaluations. Avec ces optimisations, CRAG ajoute 500ms à 1.5s de latence, ce qui reste acceptable pour un chatbot.

Conclusion

Corrective RAG et Self-RAG représentent une évolution majeure dans la fiabilité des systèmes RAG. En ajoutant une couche d'auto-évaluation, ces approches transforment le RAG d'un pipeline "fire and forget" en un système intelligent capable de questionner ses propres résultats.

Points clés :

1. CRAG ajoute une évaluation de pertinence entre retrieval et génération
2. Self-RAG intègre la réflexion directement dans le processus de génération
3. Le web search fallback est essentiel pour les questions hors périmètre
4. Un cross-encoder rapide permet l'évaluation en temps réel
5. Les seuils adaptatifs par domaine optimisent le rapport précision/recall

Ailog intègre des mécanismes d'auto-correction inspirés de CRAG pour garantir la fiabilité des réponses de ses chatbots. Créez votre compte gratuitement pour découvrir un RAG qui ne se contente pas de répondre, mais qui vérifie avant de parler.

Ressources

• Paper CRAG - "Corrective Retrieval Augmented Generation" (Yan et al., 2024)
• Paper Self-RAG - "Self-RAG: Learning to Retrieve, Generate, and Critique" (Asai et al., 2024)
• LangGraph Documentation - Framework pour les workflows RAG
• Guide Retrieval Fundamentals - Les bases du retrieval
• Détection des hallucinations - Complémentaire à CRAG
• Agentic RAG - L'étape suivante : agents autonomes