Le RAG (Retrieval-Augmented Generation) est la façon la plus pratique de donner à un LLM accès à vos données privées sans fine-tuning. On stocke les documents dans une base vectorielle, on récupère les chunks les plus pertinents à la requête, et on les injecte dans le prompt comme contexte. Ce guide construit le pipeline avec LangChain et FastAPI, puis traite ce qui sépare une démo d'un service qui tient en production.

Les trois étapes : indexation, récupération, génération

• ✓Indexation : charger les documents, les découper en chunks, embedder chaque chunk, stocker les vecteurs. Fait une fois, puis en incrémental quand les documents changent.
• ✓Récupération : embedder la question, lancer une recherche par similarité, retourner les top-k chunks, éventuellement filtrés par metadata.
• ✓Génération : construire un prompt à partir des chunks récupérés, appeler le LLM, retourner la réponse avec ses sources.

Indexation : loaders, découpage, embeddings

Utilisez un loader par type de source, découpez avec RecursiveCharacterTextSplitter pour que les chunks cassent sur les frontières de paragraphe et de phrase, puis embeddez et persistez. La taille de chunk est un compromis : plus petit donne une récupération plus nette mais perd le contexte autour, plus grand garde le contexte mais dilue la correspondance. Partez autour de 800 tokens avec 15 % d'overlap et ajustez ensuite avec votre jeu d'évaluation.

from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma

loader = PyPDFLoader("docs/spec-technique.pdf")
splitter = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=120)
chunks = splitter.split_documents(loader.load())

for chunk in chunks:
    chunk.metadata["source"] = "spec-technique"
    chunk.metadata["visibility"] = "internal"

vectorstore = Chroma.from_documents(
    chunks,
    OpenAIEmbeddings(model="text-embedding-3-small"),
    persist_directory="./chroma_db",
)

Récupération : là où se gagne ou se perd la qualité

L'essentiel des gains de qualité vient de la récupération, pas du modèle. Ajustez k au plus petit nombre de chunks qui contient de façon fiable la réponse. Utilisez MMR quand votre corpus a des passages quasi dupliqués, pour que le contexte soit varié plutôt que cinq copies du même paragraphe. Appliquez des filtres metadata pour le contrôle d'accès et la fraîcheur. Pour les corpus riches en termes exacts (références produit, messages d'erreur), ajoutez un retriever par mots-clés à côté du vectoriel et fusionnez les résultats.

retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 4,
        "fetch_k": 20,
        "filter": {"visibility": "internal"},
    },
)

Génération : un prompt ancré avec LCEL

Construisez la chaîne avec LCEL : create_stuff_documents_chain formate les chunks récupérés dans le prompt, create_retrieval_chain place le retriever devant. Le prompt doit dire au modèle de répondre uniquement à partir du contexte et de dire qu'il ne sait pas sinon, ce qui est le garde-fou le plus efficace contre l'hallucination.

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

prompt = ChatPromptTemplate.from_messages([
    ("system",
     "Réponds à la question en utilisant uniquement le contexte ci-dessous. "
     "Si le contexte ne contient pas la réponse, dis que tu ne sais pas. "
     "Cite la source de chaque fait.\n\nContexte :\n{context}"),
    ("human", "{input}"),
])

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
combine = create_stuff_documents_chain(llm, prompt)
rag_chain = create_retrieval_chain(retriever, combine)

Exposition via FastAPI, en streaming

Un endpoint naïf attend la génération complète avant de répondre, ce qui paraît lent sur une réponse de plusieurs secondes. Streamez les tokens au fur et à mesure avec une StreamingResponse sur chain.astream, et renvoyez les documents sources pour que le client affiche les citations. Gardez une variante non-streaming pour les appelants serveur à serveur qui veulent juste le JSON.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

app = FastAPI()

class Query(BaseModel):
    question: str

@app.post("/ask")
async def ask(body: Query):
    async def tokens():
        async for part in rag_chain.astream({"input": body.question}):
            if answer := part.get("answer"):
                yield answer
    return StreamingResponse(tokens(), media_type="text/plain")

Les préoccupations de production

• ✓Limitez les appels LLM par utilisateur et par clé d'API, avec une réponse 429 claire.
• ✓Mettez en cache les réponses aux questions fréquentes ou identiques, idéalement avec un cache sémantique indexé sur la question embeddée.
• ✓Suivez les tokens et le coût par requête, attribués à un utilisateur ou un tenant, pour que l'échelle ne vous surprenne pas sur la facture.
• ✓Fixez un timeout sur l'appel LLM et une réponse de repli (une page d'aide statique, un formulaire de contact) en cas d'échec.
• ✓Filtrez à la récupération pour le contrôle d'accès et les données personnelles, ne comptez jamais sur le prompt pour garder un document caché.
• ✓Tracez chaque requête (chunks récupérés, prompt, latence, coût) pour pouvoir déboguer une mauvaise réponse après coup.

Évaluer un système RAG

Construisez un jeu de référence de questions avec leur réponse attendue et leur source attendue. Mesurez récupération et génération séparément : est-ce que le bon chunk arrive dans le top-k (récupération), et est-ce que la réponse reste fidèle au contexte (génération). Rejouez tout le jeu à chaque changement de chunking, de récupération ou de prompt, et traitez une régression comme bloquante.

Un pipeline RAG de production n'est pas beaucoup plus de code qu'une démo, mais c'est un état d'esprit différent : récupération ajustée, prompt ancré, streaming, cache, suivi des coûts, et un jeu d'évaluation qui tourne à chaque changement. Mettez cela en place et le système reste fiable à mesure que le corpus et le trafic grandissent.