Comment implémenter Corrective RAG en 2026

#!/bin/bash
pip install langchain langchain-openai langchain-community faiss-cpu python-dotenv
mkdir -p data embeddings
cp .env.example .env  # Ajoutez OPENAI_API_KEY=sk-...
echo "Setup terminé. Lancez python main.py"

import os
from dotenv import load_dotenv
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader

load_dotenv()

# Charger et splitter docs (ex: data/doc1.txt à doc5.txt)
docs = []
for i in range(1, 6):
    loader = TextLoader(f"data/doc{i}.txt")
    docs.extend(loader.load())

splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
chunks = splitter.split_documents(docs)

# Embed et stocker
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = FAISS.from_documents(chunks, embeddings)
vectorstore.save_local("embeddings/faiss_index")
print("Index créé :", len(vectorstore.index_to_docstore_id))

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# Charger vectorstore
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = FAISS.load_local("embeddings/faiss_index", embeddings, allow_dangerous_deserialization=True)
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

llm = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_template("""
Réponds à la question en te basant uniquement sur le contexte fourni.
Contexte: {context}
Question: {question}
Réponse:"""
)

chain = ({"context": retriever, "question": lambda x: x} | prompt | llm | StrOutputParser())
response = chain.invoke("Qu'est-ce que le RAG ?")
print(response)

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from typing import List
from langchain_core.documents import Document

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

grader_prompt = ChatPromptTemplate.from_template("""
Tu évalues la pertinence d'un chunk pour la question.
Classe en: relevant | irrelevant | ambiguous.
Justifie brièvement.

Question: {question}
Chunk: {chunk}
Classement:"""
)

grader_chain = grader_prompt | llm.with_config(output_parser=StrOutputParser())

def grade_documents(question: str, docs: List[Document]) -> List[Document]:
    relevant_docs = []
    for doc in docs:
        grade = grader_chain.invoke({"question": question, "chunk": doc.page_content})
        if "relevant" in grade.lower():
            relevant_docs.append(doc)
    return relevant_docs

# Test
docs = retriever.invoke("Avantages du RAG")
graded = grade_documents("Avantages du RAG", docs)
print(f"Chunks gardés: {len(graded)}/{len(docs)}")

from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
from typing import List

class RelevanceGrade(BaseModel):
    binary_relevance: str = Field(description="relevant|irrelevant")

parser = PydanticOutputParser(pydantic_object=RelevanceGrade)

grader_prompt = ChatPromptTemplate.from_template("""
{format_instructions}
Question: {question}
Chunk: {chunk}""") \
    .partial(format_instructions=parser.get_format_instructions())

grader = (grader_prompt | llm | parser)

class CorrectiveRetriever:
    def __init__(self, retriever):
        self.initial_retriever = retriever

    def retrieve(self, query: str) -> List[Document]:
        docs = self.initial_retriever.invoke(query)
        filtered = []
        for doc in docs:
            grade = grader.invoke({"question": query, "chunk": doc.page_content})
            if grade.binary_relevance == "relevant":
                filtered.append(doc)
        if len(filtered) < 2:
            # Fallback: re-query raffinée
            refined_query = query + " détails techniques"
            docs = self.initial_retriever.invoke(refined_query)
            filtered.extend([d for d in docs[:2]])
        return filtered[:4]

corrective_retriever = CorrectiveRetriever(retriever)

prompt = ChatPromptTemplate.from_template("""
Contexte corrigé: {context}
Question: {question}
Réponse précise:"""
)

chain = ({"context": lambda x: corrective_retriever.retrieve(x["question"]), "question": RunnablePassthrough()} | prompt | llm | StrOutputParser())

response = chain.invoke("Différences RAG vs fine-tuning ?")
print("Réponse Corrective RAG:", response)

# Comparaison avec standard
std_response = chain_standard.invoke("Différences RAG vs fine-tuning ?")  # De rag_standard.py
importé
print("Réponse Standard:", std_response)