Self-RAG

L’auto-réflexion dans le RAG

Le Self-RAG (Self-Reflective RAG) va plus loin que le Corrective RAG. Non seulement il vérifie la pertinence des documents, mais il ajoute deux vérifications supplémentaires après la génération :

1. Vérification des hallucinations : la réponse est-elle réellement fondée sur les documents fournis ?
2. Vérification de l’utilité : la réponse répond-elle effectivement à la question posée ?

Si l’une de ces vérifications échoue, le système boucle et réessaie, créant un mécanisme d’auto-correction.

Architecture du Self-RAG

Le flux étend le Corrective RAG avec deux nouveaux noeuds :

1. Retrieval puis Grading (comme en Corrective RAG)
2. Génération de la réponse
3. Hallucination Check : la réponse est-elle supportée par les documents ?
4. Answer Check : la réponse est-elle utile pour la question ?
5. Si échec : retour à la génération ou au web search

Le vérificateur d’hallucinations

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 etes un evaluateur de fondement factuel. Etant donne un ensemble
de documents et une reponse generee, determinez si la reponse est entierement
fondee sur les informations contenues dans les documents.

Repondez UNIQUEMENT avec un objet JSON : {{"grounded": "yes"}} ou {{"grounded": "no"}}

Documents :
{docs_text}

Reponse generee :
{generation}

Evaluation 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"

Le vérificateur d’utilité

def check_answer_usefulness(question, generation):
    """Vérifier si la réponse répond effectivement à la question."""
    prompt = f"""Vous etes un evaluateur de qualite de reponse. Etant donne une question
et une reponse, determinez si la reponse repond de maniere utile et complete a la question.

Repondez UNIQUEMENT avec un objet JSON : {{"useful": "yes"}} ou {{"useful": "no"}}

Question : {question}
Reponse : {generation}

Evaluation 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"

Pipeline Self-RAG complet

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']}")

Coût et latence

Le Self-RAG effectue plusieurs appels LLM par question (grading + génération + vérifications). Voici comment estimer le coût :

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