Pipeline RAG de A à Z : Construire, Évaluer et Déployer (2026)

Architecture d'ensemble

Un système RAG en production comporte deux phases distinctes qui s'exécutent à des moments différents :

Pipeline d'indexation (une seule fois, ou lors de mises à jour)

Documents bruts → Chargement → Nettoyage → Chunking → Embedding → Stockage en base vectorielle

Pipeline de requête (à chaque demande utilisateur)

Requête utilisateur → Embedding → Récupération Top-K → Construction du prompt → LLM → Réponse

L'insight clé : la qualité des embeddings et la conception des chunks sont figées au moment de l'indexation. Un document mal découpé ne peut pas être rattrapé au moment de la requête — c'est pourquoi ce tutoriel consacre un temps important à ces étapes initiales.

Étape 1 : Configuration de l'environnement

Python 3.11+ requis. Installez les dépendances dans un environnement virtuel :

# Création de l'environnement isolé
python -m venv .venv
source .venv/bin/activate  # Windows : .venv\Scripts\activate

# Stack RAG principale
pip install langchain langchain-openai langchain-community langchain-chroma

# Traitement de documents
pip install pypdf unstructured[pdf] python-docx

# Bases vectorielles
pip install chromadb              # Local, open source
pip install pinecone              # Cloud managé (optionnel)

# Framework d'évaluation
pip install ragas datasets

# Déploiement
pip install fastapi uvicorn mangum  # mangum = adaptateur ASGI pour Lambda

# Utilitaires
pip install python-dotenv tiktoken

Créez un fichier .env pour vos clés :

# .env
OPENAI_API_KEY=sk-...             # Ou utilisez Ollama pour des modèles locaux gratuits
PINECONE_API_KEY=...              # Seulement si vous utilisez Pinecone

# Pour Ollama (gratuit, modèles locaux) :
# Installez depuis https://ollama.ai, puis :
# ollama pull nomic-embed-text   (embeddings 768 dimensions)
# ollama pull llama3.2           (modèle 8B, rapide)

Étape 2 : Configuration de la base vectorielle

Option A : ChromaDB (Local / Docker)

ChromaDB est le meilleur point de départ — gratuit, fonctionne en mode intégré ou serveur, aucune configuration cloud nécessaire.

# chroma_setup.py
import chromadb
from chromadb.config import Settings

# Mode intégré (persisté sur disque, processus unique)
client = chromadb.PersistentClient(
    path="./chroma_data",
    settings=Settings(anonymized_telemetry=False)
)

# Créer (ou récupérer) la collection
collection = client.get_or_create_collection(
    name="documents",
    metadata={"hnsw:space": "cosine"}  # Similarité cosinus pour la recherche sémantique
)

print(f"Collection prête : {collection.name}")
print(f"Documents indexés : {collection.count()}")

Pour un serveur partagé entre processus ou conteneurs :

# Démarrer ChromaDB comme serveur standalone :
# docker run -p 8000:8000 chromadb/chroma

# Connexion depuis Python :
import chromadb

client = chromadb.HttpClient(host="localhost", port=8000)
collection = client.get_or_create_collection("documents")

Option B : Pinecone (Cloud, passage à l'échelle)

Pinecone excelle avec des millions de documents ou quand vous avez besoin de réplication managée.

# pinecone_setup.py
from pinecone import Pinecone, ServerlessSpec

pc = Pinecone(api_key="votre-api-key")

# Créer l'index (1536 dims = OpenAI text-embedding-3-small)
# Pour nomic-embed-text : dimension=768
if "rag-docs" not in pc.list_indexes().names():
    pc.create_index(
        name="rag-docs",
        dimension=1536,
        metric="cosine",
        spec=ServerlessSpec(cloud="aws", region="eu-west-1")  # EU pour RGPD
    )

index = pc.Index("rag-docs")
print(index.describe_index_stats())
# Sortie : {'dimension': 1536, 'total_vector_count': 0, ...}

Note coût : Pinecone Serverless facture 0,096 USD par million de lectures et 2 USD/Go/mois de stockage. Une base de 10 000 documents coûte environ 2-5 EUR/mois. Pour moins d'1M de documents, ChromaDB sur un VPS à 5 EUR/mois est plus économique.

Étape 3 : Chargement et stratégie de chunking

Le chunking est la décision de conception la plus impactante d'un système RAG. Des chunks trop petits perdent le contexte ; trop grands, ils diluent la précision de récupération.

Chargement de plusieurs formats de documents

# document_loader.py
from langchain_community.document_loaders import (
    PyPDFLoader,
    UnstructuredWordDocumentLoader,
    DirectoryLoader,
)


def charger_documents(repertoire: str = "./docs") -> list:
    """Charge tous les documents d'un répertoire en détectant automatiquement le format."""
    loaders = {
        "**/*.pdf": PyPDFLoader,
        "**/*.docx": UnstructuredWordDocumentLoader,
    }

    tous_docs = []
    for pattern, loader_cls in loaders.items():
        loader = DirectoryLoader(
            repertoire,
            glob=pattern,
            loader_cls=loader_cls,
            show_progress=True,
        )
        docs = loader.load()
        tous_docs.extend(docs)
        print(f"Chargé {len(docs)} pages depuis les fichiers {pattern}")

    # Ajout de métadonnées pour le filtrage ultérieur
    for doc in tous_docs:
        doc.metadata["ingested_at"] = "2026-04-09"

    print(f"\nTotal : {len(tous_docs)} pages chargées")
    return tous_docs


docs = charger_documents("./docs")
# Sortie :
# Chargé 45 pages depuis les fichiers **/*.pdf
# Chargé 12 pages depuis les fichiers **/*.docx
# Total : 57 pages chargées

Stratégie 1 : Découpage récursif par caractères (baseline)

from langchain.text_splitter import RecursiveCharacterTextSplitter

# Bon défaut pour la plupart des types de documents
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,       # Taille cible en caractères
    chunk_overlap=200,     # Chevauchement pour préserver le contexte aux frontières
    length_function=len,
    separators=[
        "\n\n",          # Préférer les sauts de paragraphe
        "\n",             # Puis les sauts de ligne
        ". ",              # Puis les fins de phrase
        " ",               # Puis les mots
        "",                # Repli sur les caractères
    ],
)

chunks = splitter.split_documents(docs)
print(f"Découpé en {len(chunks)} chunks")
print(f"Taille moyenne : {sum(len(c.page_content) for c in chunks) // len(chunks)} caractères")

# Sortie :
# Découpé en 341 chunks
# Taille moyenne : 847 caractères

Stratégie 2 : Chunking sémantique (meilleur recall)

Le chunking sémantique découpe sur les frontières de sujets détectées par similarité d'embedding, plutôt que sur des comptages de caractères fixes. Il améliore le context recall de 15-25% car le contenu lié reste ensemble.

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

semantic_splitter = SemanticChunker(
    embeddings,
    breakpoint_threshold_type="percentile",   # Découpe quand la similarité passe sous le 95e percentile
    breakpoint_threshold_amount=95,
)

semantic_chunks = semantic_splitter.split_documents(docs)
print(f"Chunks sémantiques : {len(semantic_chunks)}")
print(f"Taille moyenne : {sum(len(c.page_content) for c in semantic_chunks) // len(semantic_chunks)} caractères")

# Sortie (taille variable, alignée sur les sujets) :
# Chunks sémantiques : 198
# Taille moyenne : 1423 caractères

# Compromis : ~2x plus de tokens à embedder, mais bien meilleure qualité de récupération

Tableau comparatif des stratégies de chunking

Type de document	Taille recommandée	Chevauchement	Splitter
Documentation technique / API	500-800 caractères	100-150	Récursif
Documents juridiques / contrats	1500-2000 caractères	300-400	Récursif (phrase)
Articles de recherche	Par sujet	N/A	Sémantique
FAQ support client	Une Q&R par chunk	0	Personnalisé (séparation Q:)
Fichiers de code	Fonction / classe	0-50	RecursiveCharacter (code)

Étape 4 : Embedding et indexation vectorielle

# indexer.py
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
import os

# Initialiser le modèle d'embedding
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",  # 1536 dims, 0,02 USD par million de tokens
    # Alternative : text-embedding-3-large (3072 dims, précision supérieure, 5x le coût)
)

PERSIST_DIR = "./chroma_data"

if os.path.exists(PERSIST_DIR) and os.listdir(PERSIST_DIR):
    print("Chargement de la base vectorielle existante...")
    vectorstore = Chroma(
        persist_directory=PERSIST_DIR,
        embedding_function=embeddings,
        collection_name="documents",
    )
else:
    print(f"Indexation de {len(chunks)} chunks...")
    vectorstore = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=PERSIST_DIR,
        collection_name="documents",
        collection_metadata={"hnsw:space": "cosine"},
    )
    print("Indexation terminée.")

print(f"Base vectorielle prête : {vectorstore._collection.count()} vecteurs")
# Sortie : Base vectorielle prête : 341 vecteurs

Avec les embeddings locaux Ollama (gratuit, aucun envoi de données) :

# D'abord télécharger le modèle :
# ollama pull nomic-embed-text

from langchain_community.embeddings import OllamaEmbeddings

embeddings = OllamaEmbeddings(model="nomic-embed-text")  # 768 dims, gratuit, local
# Le reste du code d'indexation est identique

Étape 5 : Chaîne RAG et génération

# rag_chain.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableParallel

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

# MMR (Maximal Marginal Relevance) réduit les chunks dupliqués
retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 5,         # Retourner 5 chunks
        "fetch_k": 20,  # Considérer 20 candidats, sélectionner les 5 plus diversifiés
        "lambda_mult": 0.7,  # 0 = diversité maximale, 1 = pertinence maximale
    },
)

SYSTEM_PROMPT = """Vous êtes un assistant utile. Répondez à la question de l'utilisateur en utilisant UNIQUEMENT le contexte ci-dessous.
Si le contexte ne contient pas assez d'informations, dites "Je n'ai pas assez d'informations pour répondre à cela."
Ne fabriquez pas d'informations et ne faites pas appel à des connaissances extérieures.

Contexte :
{context}"""

prompt = ChatPromptTemplate.from_messages([
    ("system", SYSTEM_PROMPT),
    ("human", "{question}"),
])


def formater_docs(docs: list) -> str:
    """Formate les documents récupérés avec attribution des sources."""
    parties = []
    for i, doc in enumerate(docs, 1):
        source = doc.metadata.get("source", "inconnu")
        page = doc.metadata.get("page", "")
        label = f"[{i}] {source}" + (f" p.{page}" if page else "")
        parties.append(f"{label}\n{doc.page_content}")
    return "\n\n---\n\n".join(parties)


# Chaîne avec suivi des sources
rag_chain_avec_sources = RunnableParallel(
    answer=(
        {"context": retriever | formater_docs, "question": RunnablePassthrough()}
        | prompt
        | llm
        | StrOutputParser()
    ),
    sources=(retriever),
)

# Requête
resultat = rag_chain_avec_sources.invoke("Quelle est la politique de remboursement ?")
print(f"Réponse :\n{resultat['answer']}\n")
print("Sources :")
for doc in resultat["sources"]:
    print(f"  - {doc.metadata.get('source')} (p.{doc.metadata.get('page', '?')})")

# Sortie attendue :
# Réponse :
# Selon la politique de remboursement (page 4), les clients peuvent demander
# un remboursement complet dans les 30 jours suivant l'achat si le produit
# est inutilisé et dans son emballage d'origine.
#
# Sources :
#   - conditions_generales.pdf (p.4)
#   - faq.pdf (p.12)

Étape 6 : Tests de qualité de récupération

Avant d'utiliser RAGAS, testez manuellement votre retrieveur pour détecter les problèmes évidents de configuration. Cela prend 10 minutes et capture 80% des problèmes.

# retrieval_test.py
from typing import NamedTuple


class TestRecuperation(NamedTuple):
    requete: str
    mots_cles_attendus: list[str]  # Mots devant apparaître dans les chunks récupérés
    k_minimum: int = 3             # Nombre minimum de chunks pertinents attendus


TESTS_RECUPERATION = [
    TestRecuperation(
        requete="Quelle est la politique de remboursement ?",
        mots_cles_attendus=["remboursement", "retour", "jours"],
        k_minimum=2,
    ),
    TestRecuperation(
        requete="Comment réinitialiser mon mot de passe ?",
        mots_cles_attendus=["mot de passe", "réinitialiser", "email"],
        k_minimum=1,
    ),
    TestRecuperation(
        requete="Quels modes de paiement sont acceptés ?",
        mots_cles_attendus=["paiement", "carte bancaire", "virement"],
        k_minimum=2,
    ),
]


def executer_tests(retriever, tests: list[TestRecuperation]) -> dict:
    """Exécute les tests et rapporte les résultats."""
    resultats = {"reussi": 0, "echoue": 0, "details": []}

    for test in tests:
        docs = retriever.invoke(test.requete)
        texte_combine = " ".join(d.page_content.lower() for d in docs)

        mots_trouves = {kw: kw.lower() in texte_combine for kw in test.mots_cles_attendus}
        tous_trouves = all(mots_trouves.values())
        assez_de_docs = len(docs) >= test.k_minimum

        reussi = tous_trouves and assez_de_docs
        resultats["reussi" if reussi else "echoue"] += 1
        resultats["details"].append({
            "requete": test.requete,
            "reussi": reussi,
            "chunks_recuperes": len(docs),
            "mots_trouves": mots_trouves,
        })

    return resultats


rapport = executer_tests(retriever, TESTS_RECUPERATION)

for detail in rapport["details"]:
    statut = "PASS" if detail["reussi"] else "FAIL"
    print(f"[{statut}] {detail['requete']}")
    if not detail["reussi"]:
        manquants = [k for k, v in detail["mots_trouves"].items() if not v]
        print(f"       Mots clés manquants : {manquants}")
        print(f"       Chunks récupérés : {detail['chunks_recuperes']}")

print(f"\nRésultats : {rapport['reussi']}/{len(TESTS_RECUPERATION)} tests réussis")

Étape 7 : Évaluation avec RAGAS

RAGAS (Retrieval Augmented Generation Assessment) mesure quatre dimensions critiques pour la production. Contrairement aux tests manuels, RAGAS utilise une approche LLM-as-judge pour évaluer à grande échelle.

Métrique	Ce qu'elle mesure	Cible production
Faithfulness	La réponse est ancrée dans le contexte (pas d'hallucination)	> 0,85
Answer Relevancy	La réponse traite réellement la question posée	> 0,80
Context Precision	Les chunks récupérés sont pertinents (pas de bruit)	> 0,75
Context Recall	Toutes les informations pertinentes ont été récupérées	> 0,70

Construction du dataset d'évaluation

# evaluation_dataset.py
from datasets import Dataset

donnees_evaluation = {
    "question": [
        "Quelle est la politique de remboursement pour les produits numériques ?",
        "Combien de temps prend la livraison en Europe ?",
        "Puis-je utiliser le produit à des fins commerciales ?",
        "En quelles langues est disponible le support client ?",
        "Y a-t-il une période d'essai gratuite ?",
    ],
    "ground_truth": [
        "Les produits numériques ne sont pas remboursables sauf en cas de problème technique vérifié par notre équipe support.",
        "La livraison standard en Europe prend 7 à 14 jours ouvrés. La livraison express prend 3 à 5 jours ouvrés.",
        "Oui, l'utilisation commerciale est autorisée avec les licences Professional et Enterprise.",
        "Le support client est disponible en français, anglais, espagnol et allemand.",
        "Oui, tous les plans incluent 14 jours d'essai gratuit avec accès complet, sans carte bancaire requise.",
    ],
    "contexts": [],
    "answer": [],
}

eval_dataset = Dataset.from_dict(donnees_evaluation)
print(f"Dataset d'évaluation : {len(eval_dataset)} questions")

Exécution de l'évaluation RAGAS

# run_evaluation.py
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision, context_recall
from ragas.llms import LangchainLLMWrapper
from ragas.embeddings import LangchainEmbeddingsWrapper
from langchain_openai import ChatOpenAI, OpenAIEmbeddings


def preparer_dataset(dataset, retriever, chain):
    """Génère les réponses et collecte les contextes pour chaque question."""
    contexts_list = []
    answers_list = []

    for question in dataset["question"]:
        docs = retriever.invoke(question)
        contexts_list.append([doc.page_content for doc in docs])
        answers_list.append(chain.invoke(question))

    dataset = dataset.add_column("contexts", contexts_list)
    dataset = dataset.add_column("answer", answers_list)
    return dataset


eval_ready = preparer_dataset(eval_dataset, retriever, rag_chain_avec_sources["answer"])

ragas_llm = LangchainLLMWrapper(ChatOpenAI(model="gpt-4o-mini"))
ragas_embeddings = LangchainEmbeddingsWrapper(OpenAIEmbeddings())

results = evaluate(
    dataset=eval_ready,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
    llm=ragas_llm,
    embeddings=ragas_embeddings,
)

print("\n=== Résultats de l'évaluation RAGAS ===")
print(f"Faithfulness :      {results['faithfulness']:.3f}  (cible : >0,85)")
print(f"Answer Relevancy :  {results['answer_relevancy']:.3f}  (cible : >0,80)")
print(f"Context Precision : {results['context_precision']:.3f}  (cible : >0,75)")
print(f"Context Recall :    {results['context_recall']:.3f}  (cible : >0,70)")

# Sortie typique d'un système bien configuré :
# === Résultats de l'évaluation RAGAS ===
# Faithfulness :      0.912  (cible : >0,85)
# Answer Relevancy :  0.847  (cible : >0,80)
# Context Precision : 0.783  (cible : >0,75)
# Context Recall :    0.741  (cible : >0,70)

Diagnostic et amélioration des scores faibles

def diagnostiquer_echecs_ragas(results, seuil=0.75):
    """Imprime des recommandations actionnables pour chaque métrique en échec."""
    metriques = {
        "faithfulness": {
            "score": results["faithfulness"],
            "corrections": [
                "Renforcez le prompt système : 'Répondez UNIQUEMENT en utilisant le contexte.'",
                "Réduisez temperature à 0 pour des réponses déterministes et ancrées",
                "Ajoutez une étape de vérification post-génération",
            ],
        },
        "answer_relevancy": {
            "score": results["answer_relevancy"],
            "corrections": [
                "Améliorez la réécriture des requêtes pour les questions ambiguës",
                "Ajustez le prompt pour exiger de répondre à la question précise",
                "Vérifiez si les réponses hors-sujet viennent de chunks non pertinents récupérés",
            ],
        },
        "context_precision": {
            "score": results["context_precision"],
            "corrections": [
                "Réduisez k — moins de chunks mais de meilleure qualité améliore la précision",
                "Ajoutez des filtres de métadonnées pour restreindre la portée de recherche",
                "Essayez search_type='mmr' pour réduire les chunks dupliqués ou bruités",
            ],
        },
        "context_recall": {
            "score": results["context_recall"],
            "corrections": [
                "Augmentez k pour récupérer plus de chunks candidats",
                "Améliorez le chunking — les grands chunks peuvent diviser le contenu pertinent",
                "Utilisez le chunking sémantique pour préserver les frontières de sujets",
                "Ajoutez une expansion de requête (générer plusieurs formulations)",
            ],
        },
    }

    print("\n=== Rapport de diagnostic ===")
    for metrique, data in metriques.items():
        if data["score"] < seuil:
            print(f"\nÉCHEC : {metrique} = {data['score']:.3f}")
            print("Corrections recommandées :")
            for correction in data["corrections"]:
                print(f"  • {correction}")

diagnostiquer_echecs_ragas(results)

Étape 8 : Déploiement

Option A : Docker Compose (Local / VPS)

Empaquetez l'API RAG comme service FastAPI aux côtés de ChromaDB. Fonctionne de manière identique en développement, sur un VPS, ou dans un orchestrateur de conteneurs.

# app/main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_chroma import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
import os

app = FastAPI(title="API RAG", version="1.0.0")


class RequeteQuery(BaseModel):
    question: str
    k: int = 5


class ReponseQuery(BaseModel):
    answer: str
    sources: list[dict]
    latency_ms: float


@app.on_event("startup")
async def startup():
    global retriever, chain

    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vectorstore = Chroma(
        host=os.getenv("CHROMA_HOST", "chroma"),  # Nom du service Docker
        port=int(os.getenv("CHROMA_PORT", "8000")),
        collection_name="documents",
        embedding_function=embeddings,
    )

    retriever = vectorstore.as_retriever(search_type="mmr", search_kwargs={"k": 5})
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

    prompt = ChatPromptTemplate.from_messages([
        ("system", "Répondez UNIQUEMENT en utilisant le contexte ci-dessous.\n\nContexte :\n{context}"),
        ("human", "{question}"),
    ])

    chain = (
        {"context": retriever | (lambda docs: "\n\n".join(d.page_content for d in docs)),
         "question": RunnablePassthrough()}
        | prompt | llm | StrOutputParser()
    )


@app.post("/query", response_model=ReponseQuery)
async def query(request: RequeteQuery):
    import time
    start = time.time()

    try:
        docs = retriever.invoke(request.question)
        answer = chain.invoke(request.question)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

    return ReponseQuery(
        answer=answer,
        sources=[
            {"source": d.metadata.get("source", ""), "page": d.metadata.get("page")}
            for d in docs
        ],
        latency_ms=(time.time() - start) * 1000,
    )


@app.get("/health")
async def health():
    return {"status": "ok"}

# docker-compose.yml
version: "3.9"

services:
  chroma:
    image: chromadb/chroma:latest
    ports:
      - "8000:8000"
    volumes:
      - chroma_data:/chroma/chroma
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/api/v1/heartbeat"]
      interval: 10s
      timeout: 5s
      retries: 3

  rag_api:
    build: .
    ports:
      - "8080:8080"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - CHROMA_HOST=chroma
      - CHROMA_PORT=8000
    depends_on:
      chroma:
        condition: service_healthy
    command: uvicorn app.main:app --host 0.0.0.0 --port 8080

volumes:
  chroma_data:

# Démarrage
docker-compose up --build

# Test de l'API
curl -X POST http://localhost:8080/query \
  -H "Content-Type: application/json" \
  -d '{"question": "Quelle est la politique de remboursement ?"}'

# Réponse :
# {
#   "answer": "Les remboursements sont disponibles dans les 30 jours...",
#   "sources": [{"source": "conditions.pdf", "page": 4}],
#   "latency_ms": 1247.3
# }

Option B : AWS Lambda (Serverless)

Pour un déploiement serverless, utilisez Mangum pour adapter FastAPI au format d'événement Lambda, et Pinecone comme base vectorielle.

# lambda_handler.py
from mangum import Mangum
from app.main import app  # Application FastAPI ci-dessus

# Mangum adapte FastAPI pour Lambda + API Gateway
handler = Mangum(app, lifespan="off")

# Étapes de déploiement :
# 1. Empaqueter les dépendances dans un Lambda layer ou image conteneur
# 2. Définir les variables d'environnement : OPENAI_API_KEY, PINECONE_API_KEY
# 3. Utiliser Pinecone (ChromaDB sur Lambda est complexe sans EFS)
# 4. Mémoire minimum : 1024MB (les opérations vectorielles nécessitent de la RAM)
# 5. Timeout : 30 secondes (les requêtes LLM peuvent être lentes)

# serverless.yml (Serverless Framework)
service: rag-api

provider:
  name: aws
  runtime: python3.11
  region: eu-west-1       # EU pour la conformité RGPD
  memorySize: 1024        # MB — les opérations vectorielles nécessitent de la RAM
  timeout: 30             # Secondes — prévoir le cold start + génération LLM
  environment:
    OPENAI_API_KEY: ${env:OPENAI_API_KEY}
    PINECONE_API_KEY: ${env:PINECONE_API_KEY}
    VECTOR_STORE: pinecone

functions:
  api:
    handler: lambda_handler.handler
    events:
      - httpApi:
          path: /{proxy+}
          method: ANY

# Déploiement :
# npm install -g serverless
# serverless deploy --stage prod

Estimation des coûts Lambda : 1 000 requêtes RAG/jour × 30s × 1024MB ≈ 2 EUR/mois en compute. Pinecone ajoute ~2 EUR/mois pour un petit index. Total : ~4-7 EUR/mois pour une API RAG serverless en production servant 30 000 requêtes/mois.

Checklist d'optimisation des performances

Cache des embeddings : Utilisez CacheBackedEmbeddings avec Redis pour éviter de ré-embedder des requêtes identiques — économise 60-80% sur les coûts d'API embedding en production
Récupération asynchrone : Utilisez retriever.ainvoke() et llm.ainvoke() pour des I/O non-bloquantes dans FastAPI — supporte 3-5x plus de requêtes concurrentes
Indexation par lots : Pour plus de 10 000 documents, utilisez vectorstore.add_documents() en lots de 100 pour éviter les limites de débit
Réduire k en premier : Passer de k=10 à k=4 divise par deux les tokens du prompt et améliore généralement la précision — c'est le levier le moins coûteux
Modèles de génération plus petits : gpt-4o-mini coûte 30x moins que gpt-4o avec 85-90% de la qualité pour les tâches de récupération factuelle

Prochaines étapes

Recherche hybride : Combinez BM25 et recherche sémantique avec EnsembleRetriever — améliore la précision de 20-30% sur les requêtes à correspondance exacte
Reranking : Ajoutez un reranker Cohere ou cross-encoder après la récupération pour rescorer les chunks — améliore systématiquement la qualité des réponses pour ~0,001 USD de coût supplémentaire par requête
RAG multi-modal : Étendez aux images et tableaux avec la vision GPT-4o ou l'extraction de tables Unstructured
RAG agentique : Utilisez LangGraph pour construire un agent de récupération qui décide quand chercher, quoi chercher, et quand il a suffisamment de contexte

Pour une formation professionnelle structurée sur ces sujets :

RAG et Agents en Production (3 jours intensifs) : patterns RAG avancés, agents LangGraph, optimisation des bases vectorielles, CI/CD pour pipelines IA
Claude API pour Développeurs (2 jours) : construire des systèmes RAG avec la fenêtre contextuelle 200K tokens de Claude et le raisonnement étendu

Foire aux questions

Quelle est la différence entre ChromaDB et Pinecone pour RAG ?

ChromaDB est une base vectorielle open source gratuite qui fonctionne en local (ou dans Docker). Elle convient parfaitement au développement, aux datasets de petite à moyenne taille (<10M vecteurs) et aux déploiements sensibles à la confidentialité. Pinecone est un service cloud managé avec mise à l'échelle automatique, facturation serverless (~0,096 USD par million de lectures) et réplication intégrée — idéal pour les systèmes de production avec des millions de documents ou les équipes sans expertise infrastructure. Vous pouvez développer avec ChromaDB et migrer vers Pinecone sans changer votre code LangChain.

Quels scores RAGAS viser avant de passer en production ?

Les benchmarks industrie pour les systèmes RAG en production : Faithfulness > 0,85 (la réponse est ancrée dans le contexte récupéré), Answer Relevancy > 0,80 (la réponse traite bien la question), Context Precision > 0,75 (les chunks récupérés sont pertinents), Context Recall > 0,70 (suffisamment de contexte pertinent est récupéré). Si un score est sous le seuil, diagnostiquez : context recall bas → augmentez k ou améliorez les embeddings ; faithfulness bas → renforcez le prompt système ; answer relevancy bas → affinez la réécriture des requêtes.

Comment choisir la taille des chunks ? Y a-t-il une formule ?

Pas de formule universelle, mais une heuristique pratique : commencez avec 1000 caractères / 200 de chevauchement et mesurez. Pour des requêtes courtes (<5 mots), les petits chunks (500 chars) récupèrent avec plus de précision. Pour des questions complexes multi-phrases, les grands chunks (1500-2000) préservent le contexte de raisonnement. Le chunking sémantique (découpage sur les limites de sujets plutôt que sur les comptages de caractères) surpasse systématiquement le découpage à taille fixe de 15-25% sur le context recall.

Puis-je exécuter le pipeline RAG complet localement sans coûts API ?

Oui. Utilisez Ollama pour l'inférence LLM locale (llama3.2 ou mistral) et les embeddings locaux (nomic-embed-text), plus ChromaDB comme base vectorielle. Tout est gratuit. Lancez `ollama pull llama3.2` et `ollama pull nomic-embed-text`, puis remplacez les clients OpenAI par OllamaEmbeddings et ChatOllama dans LangChain. Sur un MacBook Pro M2, attendez 15-20 tokens/sec pour llama3.2 8B. Pour le déploiement Docker, ajoutez un service Ollama au fichier compose et pointez votre service RAG vers lui.

Quels sont les coûts réels d'un RAG serverless sur AWS Lambda ?

Estimation pour 30 000 requêtes/mois : Lambda (1024MB, 5s de durée moyenne) ≈ 2 EUR/mois ; Pinecone (index < 1M vecteurs) ≈ 2 EUR/mois ; OpenAI gpt-4o-mini (2000 tokens/requête) ≈ 3 EUR/mois. Total : ~7 EUR/mois. Le cold start Lambda ajoute 800ms-2s — utilisez 1-2 instances de concurrence provisionnée (~15 EUR/mois) pour les chemins critiques en latence.