Implémenter un système RAG avec LangChain : Guide complet du prototype à la production

Documents sources
    ↓
Document Loaders (chargement)
    ↓
Text Splitters (découpage)
    ↓
Embeddings (vectorisation)
    ↓
Vector Store (stockage)
    ↓
Retriever (recherche)
    ↓
LLM + Prompt (génération)
    ↓
Réponse finale

from langchain_community.document_loaders import PyPDFLoader

# Chargement d'un PDF
loader = PyPDFLoader("documentation.pdf")
documents = loader.load()

# Chaque document contient page_content et metadata
print(f"Nombre de pages : {len(documents)}")
print(f"Contenu première page : {documents[0].page_content[:200]}")
print(f"Métadonnées : {documents[0].metadata}")

from langchain_community.document_loaders import WebBaseLoader

# Chargement d'une ou plusieurs pages
loader = WebBaseLoader([
    "https://naileru.com/ia/rag/",
    "https://naileru.com/ia/embedding/"
])
documents = loader.load()

from langchain_community.document_loaders import TextLoader, UnstructuredMarkdownLoader

# Texte brut
text_loader = TextLoader("notes.txt", encoding="utf-8")

# Markdown avec structure préservée
md_loader = UnstructuredMarkdownLoader("readme.md")

documents = text_loader.load() + md_loader.load()

# Notion
from langchain_community.document_loaders import NotionDirectoryLoader
loader = NotionDirectoryLoader("./notion_export")

# Google Drive
from langchain_community.document_loaders import GoogleDriveLoader
loader = GoogleDriveLoader(folder_id="your_folder_id")

# Confluence
from langchain_community.document_loaders import ConfluenceLoader
loader = ConfluenceLoader(url="https://your-domain.atlassian.net")

from langchain_community.document_loaders import DirectoryLoader

# Tous les fichiers .txt d'un dossier
loader = DirectoryLoader(
    "./docs",
    glob="**/*.txt",
    show_progress=True
)
documents = loader.load()

from langchain.text_splitter import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,          # Taille des chunks en caractères
    chunk_overlap=200,        # Chevauchement pour conserver le contexte
    length_function=len,
    separators=["\n\n", "\n", " ", ""]  # Ordre de priorité des séparateurs
)

splits = text_splitter.split_documents(documents)
print(f"Documents originaux : {len(documents)}")
print(f"Chunks créés : {len(splits)}")

from langchain.text_splitter import TokenTextSplitter

# Découpage basé sur les tokens réels du modèle
token_splitter = TokenTextSplitter(
    chunk_size=500,      # En tokens
    chunk_overlap=50
)

splits = token_splitter.split_documents(documents)

# Pour le code Python
from langchain.text_splitter import PythonCodeTextSplitter
python_splitter = PythonCodeTextSplitter(chunk_size=500)

# Pour le Markdown (respecte les titres)
from langchain.text_splitter import MarkdownHeaderTextSplitter
md_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=[
        ("#", "Header 1"),
        ("##", "Header 2"),
        ("###", "Header 3"),
    ]
)

# Pour HTML (respecte les balises)
from langchain.text_splitter import HTMLHeaderTextSplitter
html_splitter = HTMLHeaderTextSplitter(
    headers_to_split_on=[("h1", "Header 1"), ("h2", "Header 2")]
)

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(
    model="text-embedding-3-large",  # Meilleur modèle
    dimensions=1536                   # Dimension des vecteurs
)

# Test
vector = embeddings.embed_query("Qu'est-ce que le RAG ?")
print(f"Dimension du vecteur : {len(vector)}")  # 1536

from langchain_community.embeddings import HuggingFaceEmbeddings

# Modèle français optimisé
embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2",
    model_kwargs={'device': 'cpu'},  # ou 'cuda' si GPU
    encode_kwargs={'normalize_embeddings': True}
)

from langchain_community.vectorstores import FAISS

# Création de la base
vectorstore = FAISS.from_documents(
    documents=splits,
    embedding=embeddings
)

# Sauvegarde locale
vectorstore.save_local("./faiss_index")

# Rechargement
vectorstore = FAISS.load_local(
    "./faiss_index",
    embeddings,
    allow_dangerous_deserialization=True  # Nécessaire pour FAISS
)

from langchain_community.vectorstores import Chroma

# Création avec persistance
vectorstore = Chroma.from_documents(
    documents=splits,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# Auto-sauvegardé, rechargement automatique
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings
)

from langchain_community.vectorstores import Pinecone
import pinecone

# Configuration
pinecone.init(api_key="your-api-key", environment="us-west1-gcp")

# Création de l'index
index_name = "rag-knowledge-base"
if index_name not in pinecone.list_indexes():
    pinecone.create_index(
        name=index_name,
        dimension=1536,
        metric="cosine"
    )

# Indexation
vectorstore = Pinecone.from_documents(
    splits,
    embeddings,
    index_name=index_name
)

# K documents les plus similaires
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 4}  # Retourne 4 documents
)

# Test de recherche
docs = retriever.get_relevant_documents("Comment fonctionne le RAG ?")
for i, doc in enumerate(docs):
    print(f"\n--- Document {i+1} ---")
    print(doc.page_content[:200])
    print(f"Score: {doc.metadata.get('score', 'N/A')}")

retriever = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={
        "k": 10,                # Chercher dans 10 candidats
        "score_threshold": 0.7  # Garder seulement ceux > 0.7
    }
)

retriever = vectorstore.as_retriever(
    search_type="mmr",  # Maximum Marginal Relevance
    search_kwargs={
        "k": 4,
        "fetch_k": 20,      # Chercher dans 20 candidats
        "lambda_mult": 0.5  # 0=diversité max, 1=similarité max
    }
)

# Recherche uniquement dans des sources spécifiques
retriever = vectorstore.as_retriever(
    search_kwargs={
        "k": 4,
        "filter": {
            "source": "documentation.pdf",
            "page": {"$gte": 10, "$lte": 50}  # Pages 10-50
        }
    }
)

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain_openai import ChatOpenAI

# Compresseur intelligent
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
compressor = LLMChainExtractor.from_llm(llm)

# Retriever avec compression
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=retriever
)

# Les documents retournés sont automatiquement condensés
docs = compression_retriever.get_relevant_documents(
    "Quelle est la différence entre LoRA et QLoRA ?"
)

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain

# 1. LLM
llm = ChatOpenAI(model="gpt-4", temperature=0)

# 2. Prompt pour le RAG
prompt = ChatPromptTemplate.from_template("""
Réponds à la question en te basant sur le contexte suivant.
Si tu ne trouves pas la réponse dans le contexte, dis "Je ne trouve pas cette information dans les documents fournis".
Cite toujours tes sources en mentionnant le document d'origine.

Contexte:
{context}

Question: {input}

Réponse:""")

# 3. Chaîne de combinaison de documents
document_chain = create_stuff_documents_chain(llm, prompt)

# 4. Chaîne RAG complète
retrieval_chain = create_retrieval_chain(retriever, document_chain)

# 5. Utilisation
response = retrieval_chain.invoke({
    "input": "Qu'est-ce que le fine-tuning ?"
})

print(response["answer"])
print("\n--- Sources ---")
for doc in response["context"]:
    print(f"- {doc.metadata}")

from langchain.chains import create_history_aware_retriever
from langchain_core.prompts import MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage

# Prompt pour reformuler la question avec historique
contextualize_q_prompt = ChatPromptTemplate.from_messages([
    ("system", """Reformule la dernière question de l'utilisateur en une question autonome,
    en prenant en compte l'historique de la conversation."""),
    MessagesPlaceholder("chat_history"),
    ("human", "{input}")
])

# Retriever conscient de l'historique
history_aware_retriever = create_history_aware_retriever(
    llm, retriever, contextualize_q_prompt
)

# Chaîne RAG avec historique
qa_prompt = ChatPromptTemplate.from_messages([
    ("system", "Réponds en te basant sur le contexte:\n\n{context}"),
    MessagesPlaceholder("chat_history"),
    ("human", "{input}")
])

document_chain = create_stuff_documents_chain(llm, qa_prompt)
rag_chain = create_retrieval_chain(history_aware_retriever, document_chain)

# Conversation
chat_history = []

# Tour 1
response1 = rag_chain.invoke({
    "input": "Qu'est-ce que LoRA ?",
    "chat_history": chat_history
})
print(response1["answer"])

# Mise à jour de l'historique
chat_history.extend([
    HumanMessage(content="Qu'est-ce que LoRA ?"),
    AIMessage(content=response1["answer"])
])

# Tour 2 (utilise le contexte)
response2 = rag_chain.invoke({
    "input": "Quels sont ses avantages ?",  # "ses" fait référence à LoRA
    "chat_history": chat_history
})
print(response2["answer"])

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever

# Retriever par mots-clés (BM25)
bm25_retriever = BM25Retriever.from_documents(splits)
bm25_retriever.k = 4

# Retriever sémantique
semantic_retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

# Combinaison hybride
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, semantic_retriever],
    weights=[0.3, 0.7]  # 30% mots-clés, 70% sémantique
)

# Meilleure précision, surtout pour les termes techniques exacts
docs = ensemble_retriever.get_relevant_documents("API REST authentication")

from langchain.retrievers.multi_query import MultiQueryRetriever

multi_query_retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(),
    llm=ChatOpenAI(model="gpt-4o-mini", temperature=0)
)

# Génère automatiquement 3-5 variantes de la question
docs = multi_query_retriever.get_relevant_documents(
    "Comment personnaliser un modèle d'IA ?"
)
# → Génère : "fine-tuning", "adapter un LLM", "entraînement personnalisé", etc.

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore

# Store pour les documents parents
store = InMemoryStore()

# Splitter pour petits chunks (recherche)
child_splitter = RecursiveCharacterTextSplitter(chunk_size=200)

# Splitter pour parents (contexte)
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=1000)

# Retriever intelligent
retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=store,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter
)

retriever.add_documents(documents)

# Recherche précise, contexte complet
docs = retriever.get_relevant_documents("authentication JWT")

from langchain.retrievers.self_query.base import SelfQueryRetriever
from langchain.chains.query_constructor.base import AttributeInfo

# Description des métadonnées disponibles
metadata_field_info = [
    AttributeInfo(
        name="source",
        description="Le fichier source du document",
        type="string"
    ),
    AttributeInfo(
        name="page",
        description="Le numéro de page",
        type="integer"
    ),
]

document_content_description = "Documentation technique sur l'IA"

# Retriever auto-filtrant
retriever = SelfQueryRetriever.from_llm(
    llm=ChatOpenAI(model="gpt-4o-mini", temperature=0),
    vectorstore=vectorstore,
    document_contents=document_content_description,
    metadata_field_info=metadata_field_info
)

# Question avec filtre implicite
docs = retriever.get_relevant_documents(
    "Parle-moi du fine-tuning dans le fichier documentation.pdf"
)
# → Ajoute automatiquement le filtre source="documentation.pdf"

import asyncio
from langchain.vectorstores import Chroma

async def index_documents_batch(documents, batch_size=100):
    """Indexation par batches pour gérer de gros volumes"""
    for i in range(0, len(documents), batch_size):
        batch = documents[i:i+batch_size]
        await vectorstore.aadd_documents(batch)
        print(f"Indexé {i+batch_size}/{len(documents)} documents")

# Utilisation
await index_documents_batch(splits, batch_size=50)

from langchain.cache import InMemoryCache
from langchain.globals import set_llm_cache

# Cache les réponses identiques
set_llm_cache(InMemoryCache())

# Première fois : appel API
response1 = rag_chain.invoke({"input": "Qu'est-ce que le RAG ?"})

# Deuxième fois : récupération du cache (instantané)
response2 = rag_chain.invoke({"input": "Qu'est-ce que le RAG ?"})

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

# LLM principal
primary_llm = ChatOpenAI(model="gpt-4", timeout=10)

# LLM de secours
fallback_llm = ChatAnthropic(model="claude-sonnet-4.5")

# Chaîne avec fallback automatique
llm = primary_llm.with_fallbacks([fallback_llm])

# Si GPT-4 échoue ou timeout, utilise Claude
rag_chain = create_retrieval_chain(retriever, llm)

from langchain.callbacks import StdOutCallbackHandler

class CustomCallback(StdOutCallbackHandler):
    def on_retriever_end(self, documents, **kwargs):
        print(f"\n✓ Retriever: {len(documents)} documents trouvés")

    def on_llm_start(self, serialized, prompts, **kwargs):
        print(f"\n✓ LLM: Génération en cours...")

    def on_llm_end(self, response, **kwargs):
        tokens = response.llm_output.get('token_usage', {})
        print(f"✓ LLM: {tokens.get('total_tokens', 0)} tokens utilisés")

# Utilisation
response = rag_chain.invoke(
    {"input": "Qu'est-ce que le fine-tuning ?"},
    config={"callbacks": [CustomCallback()]}
)

# Solution 1 : Augmenter le seuil de similarité
retriever = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"score_threshold": 0.8}  # Plus strict
)

# Solution 2 : Améliorer le prompt
prompt = ChatPromptTemplate.from_template("""
Réponds UNIQUEMENT si tu trouves l'information dans le contexte.
Si le contexte ne contient pas la réponse, réponds exactement :
"Je ne trouve pas cette information dans les documents."

Contexte: {context}
Question: {input}
""")

# Solution 1 : Réduire k
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

# Solution 2 : Chunks plus petits
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,  # Au lieu de 1000
    chunk_overlap=50
)

# Solution 3 : Compression
from langchain.retrievers import ContextualCompressionRetriever
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=retriever
)

# Solution 1 : Splitter sémantique
from langchain.text_splitter import SpacyTextSplitter
splitter = SpacyTextSplitter(chunk_size=1000)

# Solution 2 : Overlap plus important
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=300  # 30% de recouvrement
)

# Solution 3 : Parent Document Retriever (voir plus haut)

# Solution 1 : Base vectorielle plus rapide
vectorstore = FAISS.from_documents(splits, embeddings)  # Plus rapide que Chroma

# Solution 2 : Embeddings plus légers
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# Solution 3 : Réduire k
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

# Solution 4 : LLM plus rapide pour la génération
llm = ChatOpenAI(model="gpt-4o-mini")  # Au lieu de gpt-4

from langchain.evaluation import load_evaluator

# 1. Pertinence du retrieval
retrieval_evaluator = load_evaluator("criteria", criteria="relevance")

# 2. Qualité de la réponse
qa_evaluator = load_evaluator("qa")

# 3. Cohérence avec les sources
coherence_evaluator = load_evaluator("criteria", criteria="coherence")

test_cases = [
    {
        "question": "Qu'est-ce que LoRA ?",
        "expected_answer": "LoRA est une technique de fine-tuning...",
        "expected_source": "fine-tuning.md"
    },
    # ... plus de cas
]

# Évaluation automatique
for case in test_cases:
    result = rag_chain.invoke({"input": case["question"]})
    score = qa_evaluator.evaluate_strings(
        prediction=result["answer"],
        reference=case["expected_answer"]
    )
    print(f"Question: {case['question']}")
    print(f"Score: {score['score']}")

# 1. Cache les embeddings pour éviter la recomputation
from langchain.embeddings import CacheBackedEmbeddings
from langchain.storage import LocalFileStore

store = LocalFileStore("./embedding_cache")
cached_embeddings = CacheBackedEmbeddings.from_bytes_store(
    embeddings, store, namespace="openai-embeddings"
)

# 2. Utiliser un LLM moins cher pour les tâches simples
cheap_llm = ChatOpenAI(model="gpt-4o-mini")  # 60x moins cher que GPT-4
expensive_llm = ChatOpenAI(model="gpt-4")

# Routing intelligent
def get_llm(question):
    if len(question) < 50 and "?" in question:
        return cheap_llm
    return expensive_llm

# 3. Limiter le contexte envoyé au LLM
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})  # Au lieu de 5