Aller au contenu principal

Self-RAG

Mis à jour le 29 juillet 2026

L’auto-réflexion dans le RAG

Le Corrective RAG contrôle ce qui entre dans le prompt, mais rien de ce qui en sort. Or un contexte impeccable ne garantit pas une bonne réponse : le modèle peut très bien ajouter une précision qui ne figure nulle part dans les documents, ou produire un texte exact mais qui esquive la question posée. Le Self-RAG (Self-Reflective RAG) referme cette brèche en ajoutant deux vérifications après la génération. La première interroge le fondement factuel — la réponse s’appuie-t-elle réellement sur les documents fournis ? La seconde interroge l’utilité — répond-elle effectivement à ce qui a été demandé ? Si l’une échoue, le système boucle et réessaie, ce qui en fait un mécanisme d’auto-correction plutôt qu’un simple filtre.

Le flux étend donc celui du Corrective RAG. On retrouve le retrieval puis le grading des documents à l’identique, suivis de la génération. Viennent ensuite les deux nouveaux noeuds : le hallucination check, qui confronte la réponse aux documents, et l’answer check, qui la confronte à la question. En cas d’échec, le système retourne à la génération ou repart chercher de la matière sur le web, selon la vérification qui a cédé.

Vérifier le fondement factuel

Le premier vérificateur reçoit deux choses : l’ensemble des documents concaténés et la réponse produite. Sa mission n’est pas de juger si la réponse est vraie dans l’absolu, mais si elle est entièrement fondée sur ce qui lui a été donné. La nuance est essentielle : une affirmation exacte mais absente des documents doit être rejetée, parce qu’elle vient de la mémoire d’entraînement du modèle et non de votre corpus — et rien ne garantit que la suivante sera exacte. On retrouve le même format de sortie que le grader, un JSON binaire à température nulle.

import json
from mistralai import Mistral

client = Mistral(api_key=api_key)

def check_hallucination(documents, generation):
    """Vérifier si la génération est fondée sur les documents."""
    docs_text = "\n\n".join([d["text"] for d in documents])

    prompt = f"""Vous êtes un évaluateur de fondement factuel. Étant donné un ensemble
de documents et une réponse générée, déterminez si la réponse est entièrement
fondée sur les informations contenues dans les documents.

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

Documents :
{docs_text}

Réponse générée :
{generation}

É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("grounded", "no") == "yes"

Vérifier l’utilité

Le second vérificateur ne voit ni les documents ni le retrieval : seulement la question et la réponse. Ce champ de vision restreint est délibéré, car il isole un défaut que le premier contrôle laisse passer. Imaginez qu’on demande combien de dimensions produit un modèle d’embeddings et que la réponse décrive longuement à quoi servent les embeddings, sans jamais donner le chiffre. Tout y est fondé sur les documents, le premier vérificateur dira « oui », et l’utilisateur repartira sans son information. Le second dira « non », et déclenchera une nouvelle tentative.

def check_answer_usefulness(question, generation):
    """Vérifier si la réponse répond effectivement à la question."""
    prompt = f"""Vous êtes un évaluateur de qualité de réponse. Étant donné une question
et une réponse, déterminez si la réponse répond de manière utile et complète a la question.

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

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

É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("useful", "no") == "yes"

La boucle d’auto-correction

La classe SelfRAG reprend le retrieval et le grading connus, puis les enveloppe dans une boucle de tentatives bornée par max_retries. La logique de cette boucle est plus subtile qu’il n’y paraît, car les deux échecs ne reçoivent pas le même traitement. Une hallucination détectée relance simplement la génération avec le même contexte : le problème vient du modèle, qui a débordé, et un nouveau tirage suffit souvent. Une réponse jugée inutile déclenche en revanche un web search supplémentaire avant de réessayer : ici le problème vient du contexte, trop pauvre pour permettre une réponse complète, et régénérer sans rien ajouter donnerait le même résultat.

Notez enfin la sortie de secours après la boucle. Quand toutes les tentatives sont épuisées, le système ne lève pas d’exception : il retourne la dernière génération accompagnée de ses drapeaux grounded et useful. C’est un choix de conception assumé — mieux vaut rendre une réponse imparfaite mais étiquetée comme telle, que votre interface pourra signaler à l’utilisateur, qu’une erreur brute.

import numpy as np

class SelfRAG:
    """Pipeline Self-RAG avec auto-correction."""

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

    def retrieve(self, question, k=4):
        """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)

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

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

    def generate(self, question, documents):
        """Générer une 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": "user", "content": prompt}],
            temperature=0.2,
        )
        return response.choices[0].message.content

    def ask(self, question, k=4):
        """Pipeline Self-RAG complet avec auto-correction."""
        print(f"\n=== Self-RAG : {question[:60]}... ===")

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

        # 2. Grading
        print("2. Grading des documents...")
        relevant_docs, need_web = self.grade_documents(question, docs)

        if need_web:
            print("   -> Fallback web search")
            web_docs = web_search_fallback(question)
            relevant_docs.extend(web_docs)

        # 3. Boucle de génération avec auto-correction
        for attempt in range(self.max_retries + 1):
            print(f"3. Génération (tentative {attempt + 1})...")
            generation = self.generate(question, relevant_docs)

            # 4. Vérification des hallucinations
            print("4. Vérification hallucinations...")
            is_grounded = check_hallucination(relevant_docs, generation)

            if not is_grounded:
                print("   -> HALLUCINATION détectée, nouvelle tentative...")
                continue

            # 5. Vérification de l'utilité
            print("5. Vérification utilité...")
            is_useful = check_answer_usefulness(question, generation)

            if not is_useful:
                print("   -> Réponse NON UTILE, fallback web search...")
                web_docs = web_search_fallback(question)
                relevant_docs.extend(web_docs)
                continue

            # Succès
            print("   -> Réponse validée !")
            return {
                "answer": generation,
                "attempts": attempt + 1,
                "grounded": True,
                "useful": True,
            }

        # Si toutes les tentatives échouent
        print("   -> Nombre max de tentatives atteint")
        return {
            "answer": generation,
            "attempts": self.max_retries + 1,
            "grounded": is_grounded,
            "useful": is_useful,
        }

# Utilisation
self_rag = SelfRAG(chunks, embeddings, index, max_retries=2)
result = self_rag.ask("Quels sont les avantages du RAG par rapport au fine-tuning ?")

print(f"\nRéponse : {result['answer']}")
print(f"Tentatives : {result['attempts']}")
print(f"Fondée : {result['grounded']}, Utile : {result['useful']}")

Ce que coûte la fiabilité

Chaque garde-fou est un appel LLM de plus, et l’addition monte vite. La fonction ci-dessous la rend explicite en décomposant les appels par étape : un grading par document, puis une génération, une vérification de fondement et une vérification d’utilité, chacune répétée à chaque nouvelle tentative. Avec quatre documents et une seule reprise, on arrive à dix appels pour une unique question — là où le RAG de base en demandait deux. Faites tourner cette estimation avec vos propres valeurs avant de fixer k et max_retries : c’est le calcul qui vous dira si votre budget et vos exigences de latence supportent le Self-RAG sur l’ensemble du trafic, ou seulement sur les questions les plus sensibles.

def estimate_self_rag_cost(n_docs=4, n_retries=1):
    """Estimer le nombre d'appels LLM pour une question Self-RAG."""
    calls = {
        "grading": n_docs,           # 1 appel par document
        "generation": 1 + n_retries, # 1 initial + retries
        "hallucination_check": 1 + n_retries,
        "answer_check": 1 + n_retries,
    }
    total = sum(calls.values())
    print(f"Appels LLM estimés : {total}")
    for step, count in calls.items():
        print(f"  {step}: {count}")
    return total

estimate_self_rag_cost(n_docs=4, n_retries=1)
# Appels LLM estimés : 10

Points clés à retenir

  • Le Self-RAG ajoute deux vérifications post-génération : hallucinations et utilité
  • Si la réponse n’est pas fondée sur les documents, le système regénère
  • Si la réponse n’est pas utile, le système effectue un web search et réessaie
  • Le nombre de tentatives (retries) doit être limité pour contrôler la latence
  • Le Self-RAG est plus coûteux (davantage d’appels LLM) mais plus fiable