RAG agentique correctif : un assistant documentaire qui se relit et refuse d'halluciner

Préparer le retriever

On suppose une base déjà vectorisée (chunks de la doc). Le retriever est notre porte d’entrée documentaire.

from langchain_core.vectorstores import InMemoryVectorStore
from langchain.embeddings import init_embeddings

embeddings = init_embeddings("openai:text-embedding-3-small")
store = InMemoryVectorStore.from_documents(chunks_doc, embeddings)
retriever = store.as_retriever(search_kwargs={"k": 4})

Nœud recuperer

def recuperer(etat: Etat) -> dict:
    docs = retriever.invoke(etat["question"])
    return {"documents": [d.page_content for d in docs]}

Juge n°1 — la pertinence des documents

On évalue chaque document : pertinent ou non. On ne garde que les bons. C’est la correction d’entrée : on refuse de générer sur du bruit.

from pydantic import BaseModel, Field
from langchain.chat_models import init_chat_model

modele = init_chat_model("openai:gpt-4o-mini")

class NotePertinence(BaseModel):
    pertinent: bool = Field(description="Le document aide-t-il à répondre à la question ?")

def noter_documents(etat: Etat) -> dict:
    juge = modele.with_structured_output(NotePertinence)
    retenus = []
    for doc in etat["documents"]:
        v = juge.invoke(f"Question : {etat['question']}\nDocument : {doc}\n"
                        f"Ce document est-il pertinent ? Sois strict.")
        if v.pertinent:
            retenus.append(doc)
    return {"documents": retenus}

L’aiguillage : générer ou reformuler ?

S’il reste des documents pertinents, on génère. Sinon, la récupération a échoué : on reformule la question (sauf si on a épuisé nos tentatives).

from typing import Literal
from langgraph.graph import END

def decider(etat: Etat) -> Literal["generer", "reformuler", "abandonner"]:
    if etat["documents"]:
        return "generer"
    if etat["tentatives"] >= 2:           # garde-fou anti-boucle
        return "abandonner"
    return "reformuler"

Nœud reformuler

On réécrit la requête pour la rendre plus efficace côté recherche vectorielle, puis on relancera la récupération.

def reformuler(etat: Etat) -> dict:
    meilleure = modele.invoke(
        f"Reformule cette question pour une recherche documentaire plus efficace "
        f"(synonymes, termes techniques explicites) :\n{etat['question']}"
    )
    return {"question": meilleure.content, "tentatives": etat["tentatives"] + 1}

Nœud generer

La génération classique du RAG : répondre uniquement à partir des documents retenus.

def generer(etat: Etat) -> dict:
    contexte = "\n\n".join(etat["documents"])
    reponse = modele.invoke(
        f"Réponds à la question UNIQUEMENT à partir du contexte. "
        f"Si le contexte ne suffit pas, dis-le.\n\n"
        f"Contexte :\n{contexte}\n\nQuestion : {etat['question_origine']}"
    )
    return {"generation": reponse.content}

Juges n°2 et n°3 — ancrage et réponse effective

Après génération, deux vérifications. Ancrage : la réponse est-elle justifiée par les documents (pas d’hallucination) ? Pertinence : répond-elle vraiment à la question d’origine ?

class NoteAncrage(BaseModel):
    ancre: bool = Field(description="La réponse est-elle entièrement justifiée par les documents ?")

class NoteReponse(BaseModel):
    repond: bool = Field(description="La réponse traite-t-elle bien la question posée ?")

def verifier(etat: Etat) -> Literal["ok", "regenerer", "reformuler", "abandonner"]:
    # 1) ancrage (anti-hallucination)
    ancrage = modele.with_structured_output(NoteAncrage).invoke(
        f"Documents :\n{chr(10).join(etat['documents'])}\n\nRéponse : {etat['generation']}\n"
        f"La réponse est-elle entièrement justifiée par les documents ?"
    )
    if not ancrage.ancre:
        return "regenerer" if etat["tentatives"] < 2 else "abandonner"

    # 2) la réponse traite-t-elle la question ?
    rep = modele.with_structured_output(NoteReponse).invoke(
        f"Question : {etat['question_origine']}\nRéponse : {etat['generation']}\n"
        f"La réponse traite-t-elle la question ?"
    )
    if rep.repond:
        return "ok"
    return "reformuler" if etat["tentatives"] < 2 else "abandonner"

Câbler le graphe correctif

On relie tout, en refermant les boucles correctives — et l’abandonner qui produit un refus honnête.

from langgraph.graph import StateGraph, START

def abandonner(etat: Etat) -> dict:
    return {"generation": "Je n'ai pas trouvé d'information fiable pour répondre. "
                          "Je préfère ne pas inventer — souhaitez-vous une escalade ?"}

g = StateGraph(Etat)
g.add_node("recuperer", recuperer)
g.add_node("noter_documents", noter_documents)
g.add_node("reformuler", reformuler)
g.add_node("generer", generer)
g.add_node("abandonner", abandonner)

g.add_edge(START, "recuperer")
g.add_edge("recuperer", "noter_documents")
g.add_conditional_edges("noter_documents", decider, {
    "generer": "generer", "reformuler": "reformuler", "abandonner": "abandonner",
})
g.add_edge("reformuler", "recuperer")             # ← la boucle corrective
g.add_conditional_edges("generer", verifier, {
    "ok": END, "regenerer": "generer", "reformuler": "reformuler", "abandonner": "abandonner",
})
g.add_edge("abandonner", END)

rag = g.compile()