How to Implement Corrective RAG in 2026

#!/bin/bash
pip install langchain langchain-openai langchain-community faiss-cpu sentence-transformers
pip install langchain-experimental

# Set the OpenAI API key
export OPENAI_API_KEY="votre-cle-api-ici"

# Verification
echo "Installation terminée. Vérifiez OPENAI_API_KEY."

from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
import os

# Documents d'exemple (contexte : docs sur l'IA)
docs_content = """
L'Intelligence Artificielle Générative (GenAI) repose sur des modèles comme GPT-4.
Les embeddings convertissent le texte en vecteurs pour la similarité cosinus.
RAG combine retrieval vectoriel et génération pour réduire les hallucinations.
FAISS est un index vectoriel rapide pour des millions de chunks.
"""

loader = TextLoader.from_text(docs_content)
documents = loader.load()

text_splitter = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=50)
splits = text_splitter.split_documents(documents)

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = FAISS.from_documents(splits, embeddings)

retriever = vectorstore.as_retriever(search_kwargs={"k": 4})
print("Vector store prêt avec", len(splits), "chunks.")

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_experimental.rag import RetrievalGrader
from langchain_core.runnables import RunnablePassthrough

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Prompt pour grader la retrieval
grader_prompt = ChatPromptTemplate.from_template(
    """Tu évalues les chunks récupérés pour la question: {question}

Chunks: {documents}

Verdict (oui/non) si les chunks sont pertinents et exhaustifs:"""
)

grader = RetrievalGrader.from_llm(llm, grader_prompt)

# Retrieval basique + grading
def corrective_retrieve(query):
    docs = retriever.invoke(query)
    grade = grader.invoke({"question": query, "documents": docs})
    if "non" in grade["verdict"].lower():
        # Correction: re-retrieve avec query augmentée
        corrected_query = query + " (détails précis s'il vous plaît)"
        docs = retriever.invoke(corrected_query)
    return docs

print("Corrective retriever prêt.")

from langchain_experimental.rag import ResponseGrader

# Prompt RAG pour génération
rag_prompt = ChatPromptTemplate.from_template(
    """Réponds à {question} en te basant UNIQUEMENT sur le contexte.

Contexte: {context}

Réponse:"""
)

stuff_chain = create_stuff_documents_chain(llm, rag_prompt)

# Response Grader
response_grader_prompt = ChatPromptTemplate.from_template(
    """Question: {question}
Réponse: {response}
Contexte: {documents}

La réponse est-elle fidèle au contexte ? (oui/non):"""
)
response_grader = ResponseGrader.from_llm(llm, response_grader_prompt)

# Chaîne complète corrective
def generate_corrected_response(query):
    docs = corrective_retrieve(query)
    response = stuff_chain.invoke({"question": query, "context": docs})
    grade = response_grader.invoke({"question": query, "response": response, "documents": docs})
    if "non" in grade["verdict"].lower():
        # Correction: régénérer avec instruction anti-hallucination
        rag_prompt_strict = rag_prompt.partial(system="Pas d'invention. Si incertain, dis 'données manquantes'.")
        stuff_chain_strict = create_stuff_documents_chain(llm, rag_prompt_strict)
        response = stuff_chain_strict.invoke({"question": query, "context": docs})
    return response

print("Response corrector prêt.")

def full_crags_pipeline(query):
    print(f"Query: {query}")
    docs = corrective_retrieve(query)
    print(f"Chunks corrigés ({len(docs)}):", [doc.page_content[:50] for doc in docs])
    response = generate_corrected_response(query)
    print(f"Réponse finale: {response['answer']}")
    return response

# Test expert
query_test = "Qu'est-ce que FAISS dans le contexte RAG ?"
result = full_crags_pipeline(query_test)

query_bad = "Quelle est la capitale de la France selon FAISS ?"  # Test hallucination
result_bad = full_crags_pipeline(query_bad)

from langchain_community.tools.tavily_search import TavilySearchResults

# Ajout fallback web si retrieval échoue (nécessite TAVILY_API_KEY)
search = TavilySearchResults(max_results=3)

def corrective_retrieve_advanced(query):
    docs = retriever.invoke(query)
    grade = grader.invoke({"question": query, "documents": docs})
    if "non" in grade["verdict"].lower():
        web_docs = search.invoke(query)
        docs.extend(web_docs)  # Fusion chunks + web
    return docs

print("Fallback web activé pour cas extrêmes.")