RAG en production - systèmes d'entreprise

Ton prototype RAG fonctionne à merveille sur ton portable. Puis cent personnes l'utilisent en même temps, la facture d'API grimpe d'heure en heure, une question attend dix secondes, et quand quelque chose casse, personne ne sait à quelle étape. Un campement d'une nuit, ce n'est pas la même chose qu'une base d'expédition permanente.

Construire des systèmes RAG en production demande de prendre en compte quatre choses à la fois : les performances, la scalabilité, le monitoring et la sécurité. C'est comme concevoir un écosystème entier - chaque élément doit fonctionner en harmonie avec les autres.

Architecture d'un RAG en production

Le schéma montre les couches que traverse chaque requête :

1┌─────────────────────────────────────────────────────────────────────┐
2│                    Production RAG Architecture                       │
3├─────────────────────────────────────────────────────────────────────┤
4│                                                                      │
5│  ┌──────────┐    ┌──────────────┐    ┌──────────────┐              │
6│  │  Client  │───▶│ Load Balancer│───▶│  API Gateway │              │
7│  └──────────┘    └──────────────┘    └──────────────┘              │
8│                                               │                      │
9│                                               ▼                      │
10│  ┌────────────────────────────────────────────────────────────┐    │
11│  │                     RAG Service Layer                       │    │
12│  │  ┌────────────┐  ┌────────────┐  ┌────────────┐           │    │
13│  │  │  Retrieval │  │ Augment    │  │ Generation │           │    │
14│  │  │  Service   │  │ Service    │  │ Service    │           │    │
15│  │  └─────┬──────┘  └─────┬──────┘  └─────┬──────┘           │    │
16│  └────────┼───────────────┼───────────────┼──────────────────┘    │
17│           │               │               │                        │
18│  ┌────────▼───────┐  ┌────▼────┐  ┌──────▼──────┐                │
19│  │ Vector Database│  │  Cache  │  │  LLM APIs   │                │
20│  │ (Pinecone/     │  │ (Redis) │  │ (OpenAI/    │                │
21│  │  Qdrant)       │  │         │  │  Anthropic) │                │
22│  └────────────────┘  └─────────┘  └─────────────┘                │
23│                                                                      │
24│  ┌───────────────────────────────────────────────────────────────┐ │
25│  │                    Observability Layer                         │ │
26│  │  Prometheus │ Grafana │ Jaeger │ LangSmith │ Weights & Biases │ │
27│  └───────────────────────────────────────────────────────────────┘ │
28└─────────────────────────────────────────────────────────────────────┘

De haut en bas : load balancer et API Gateway, la couche de service RAG avec ses trois étapes, en dessous la base vectorielle, le cache Redis et les API des modèles, et sous tout cela la couche d'observabilité.

Service RAG avec FastAPI

On écrit le service avec FastAPI, parce qu'il gère async et valide les données tout seul. D'abord les imports et les modèles Pydantic qui décrivent la requête et la réponse :

1from fastapi import FastAPI, HTTPException, BackgroundTasks
2from pydantic import BaseModel
3from typing import Optional
4import asyncio
5import hashlib
6from contextlib import asynccontextmanager
7import uvicorn
8
9# Modèles
10class QueryRequest(BaseModel):
11    question: str
12    top_k: int = 5
13    filters: Optional[dict] = None
14
15class QueryResponse(BaseModel):
16    answer: str
17    sources: list[dict]
18    latency_ms: float

QueryRequest accepte une question, le nombre de documents et des filtres optionnels. FastAPI rejette tout seul une requête sans le champ question.

La classe de service se connecte à trois services. Les clients sont asynchrones, donc attendre un service ne bloque pas les autres requêtes :

1class RAGService:
2    """Service RAG de production."""
3
4    def __init__(self):
5        self.vector_store = None
6        self.llm = None
7        self.cache = None
8
9    async def initialize(self):
10        """Initialise les connexions."""
11        # Vector store
12        from qdrant_client import AsyncQdrantClient
13        self.vector_store = AsyncQdrantClient(host="qdrant", port=6333)
14
15        # Cache
16        import redis.asyncio as redis
17        self.cache = redis.Redis(host="redis", port=6379)
18
19        # LLM
20        from openai import AsyncOpenAI
21        self.llm = AsyncOpenAI()

On crée les connexions dans initialize, car elles ont besoin de await. Les noms d'hôte qdrant et redis sont les noms des services du fichier Docker Compose que tu verras à la fin de la leçon.

Le cœur du service, c'est la méthode query. Elle vérifie d'abord le cache, et ne lance le pipeline complet qu'en cas d'absence :

1    async def query(self, request: QueryRequest) -> QueryResponse:
2        """Traite une requête RAG."""
3        import time
4        start = time.time()
5
6        # 1. Vérifie le cache
7        cache_key = f"rag:{hashlib.sha256(request.question.encode()).hexdigest()}"
8        cached = await self.cache.get(cache_key)
9        if cached:
10            return QueryResponse.model_validate_json(cached)
11
12        # 2. Embedding
13        embedding = await self._get_embedding(request.question)
14
15        # 3. Retrieval
16        docs = await self._retrieve(embedding, request.top_k, request.filters)
17
18        # 4. Génération
19        answer = await self._generate(request.question, docs)
20
21        # 5. Réponse
22        response = QueryResponse(
23            answer=answer,
24            sources=[{"text": d.payload["text"], "score": d.score} for d in docs],
25            latency_ms=(time.time() - start) * 1000
26        )
27
28        # 6. Cache
29        await self.cache.setex(cache_key, 3600, response.model_dump_json())
30
31        return response

On calcule la clé du cache avec SHA-256, et non avec le hash() intégré. Cette correction compte en production : hash() sur les chaînes est randomisé à chaque démarrage du processus, donc deux workers calculeraient des clés différentes pour la même question. setex enregistre la réponse pour 3600 secondes, soit une heure.

Trois méthodes utilitaires s'occupent de l'embedding, de la recherche et de la génération :

1    async def _get_embedding(self, text: str) -> list[float]:
2        """Génère un embedding."""
3        response = await self.llm.embeddings.create(
4            model="text-embedding-3-small",
5            input=text
6        )
7        return response.data[0].embedding
8
9    async def _retrieve(self, embedding, top_k, filters):
10        """Récupère les documents."""
11        response = await self.vector_store.query_points(
12            collection_name="documents",
13            query=embedding,
14            limit=top_k,
15            query_filter=filters
16        )
17        return response.points
18
19    async def _generate(self, question: str, docs) -> str:
20        """Génère une réponse."""
21        context = "\n".join([d.payload["text"] for d in docs])
22
23        response = await self.llm.chat.completions.create(
24            model="gpt-4o-mini",
25            messages=[
26                {"role": "system", "content": f"Context:\n{context}"},
27                {"role": "user", "content": question}
28            ]
29        )
30        return response.choices[0].message.content

La recherche utilise query_points, car la méthode search a disparu des nouvelles versions de qdrant-client. Le texte du document se trouve dans le payload, c'est pourquoi on construit les sources à partir de d.payload["text"].

Pour finir, l'application FastAPI et l'endpoint :

1# App FastAPI
2rag_service = RAGService()
3
4@asynccontextmanager
5async def lifespan(app: FastAPI):
6    await rag_service.initialize()
7    yield
8
9app = FastAPI(title="RAG API", lifespan=lifespan)
10
11@app.post("/query", response_model=QueryResponse)
12async def query(request: QueryRequest):
13    try:
14        return await rag_service.query(request)
15    except Exception as e:
16        raise HTTPException(status_code=500, detail=str(e))

lifespan lance initialize une seule fois, au démarrage de l'application. Chaque erreur devient un code 500, mais en production ne renvoie pas str(e) au client.

Monitoring et observabilité

Sans mesures, tu apprends les pannes par tes utilisateurs. Prometheus collecte trois types de métriques : un compteur, un histogramme et une jauge :

1from prometheus_client import Counter, Histogram, Gauge
2import logging
3import structlog
4
5# Métriques Prometheus
6QUERY_COUNT = Counter(
7    "rag_queries_total",
8    "Total number of RAG queries",
9    ["status"]
10)
11
12QUERY_LATENCY = Histogram(
13    "rag_query_latency_seconds",
14    "Query latency in seconds",
15    buckets=[0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0]
16)
17
18RETRIEVAL_SCORE = Gauge(
19    "rag_retrieval_score",
20    "Average retrieval score"
21)
22
23# Logging structuré
24structlog.configure(
25    processors=[
26        structlog.stdlib.add_log_level,
27        structlog.processors.TimeStamper(fmt="iso"),
28        structlog.processors.JSONRenderer()
29    ]
30)
31logger = structlog.get_logger()

Un Counter ne fait qu'augmenter, et le label status sépare les succès des erreurs. Un Histogram range les temps de réponse dans des intervalles, et structlog écrit les logs en JSON.

Le service surveillé enveloppe le traitement dans une mesure de temps :

1class MonitoredRAGService:
2    """RAG avec monitoring."""
3
4    async def query(self, request: QueryRequest) -> QueryResponse:
5        with QUERY_LATENCY.time():
6            try:
7                response = await self._process_query(request)
8                QUERY_COUNT.labels(status="success").inc()
9
10                # Log
11                logger.info(
12                    "query_completed",
13                    question=request.question[:50],
14                    latency_ms=response.latency_ms,
15                    num_sources=len(response.sources)
16                )
17
18                return response
19            except Exception as e:
20                QUERY_COUNT.labels(status="error").inc()
21                logger.error("query_failed", error=str(e))
22                raise

QUERY_LATENCY.time() mesure la durée du bloc with, et une exception incrémente le compteur d'erreurs puis continue de remonter grâce à raise. _process_query représente ici la logique de la méthode query du service.

Chaînes LangChain

Dans l'écosystème de production, tu croiseras souvent LangChain. Son modèle de base, c'est la chaîne : un prompt relié à un modèle par l'opérateur | et appelé avec la méthode invoke :

1from langchain_core.prompts import ChatPromptTemplate
2from langchain_openai import ChatOpenAI
3
4prompt = ChatPromptTemplate.from_template("Answer the question briefly: {input}")
5chain = prompt | ChatOpenAI(model="gpt-4o-mini")
6
7user_query = "What is a cache in a RAG system?"
8result = chain.invoke({"input": user_query})
9print(result.content)

invoke prend un dictionnaire dont les clés correspondent aux variables du modèle de prompt, ici {input}. Le résultat est un message du modèle, et le texte se trouve dans le champ content.

Tracing avec LangSmith

LangSmith est la plateforme de l'équipe LangChain pour tracer les appels. Le décorateur @traceable enregistre toute la fonction, et trace crée des étapes imbriquées à l'intérieur :

1from langsmith import traceable, trace
2from langsmith.wrappers import wrap_openai
3from openai import OpenAI
4
5# Chaque appel via ce client arrive dans LangSmith comme une étape séparée
6openai_client = wrap_openai(OpenAI())
7
8@traceable(name="RAG Query")
9async def traced_query(question: str) -> str:
10    """Requête RAG avec tracing."""
11
12    # Embedding
13    with trace("embedding", metadata={"model": "text-embedding-3-small"}):
14        embedding = get_embedding(question)
15
16    # Retrieval
17    with trace("retrieval") as span:
18        docs = retrieve(embedding)
19        span.add_metadata({"num_docs": len(docs)})
20
21    # Génération
22    with trace("generation", metadata={"model": "gpt-4o-mini"}):
23        response = generate(question, docs)
24
25    return response

Attention : trace est une fonction séparée du paquet langsmith, pas une méthode de l'objet Client. get_embedding, retrieve et generate sont des fonctions des leçons précédentes.

L'évaluation exécute le système sur un ensemble de questions avec leurs réponses attendues. La fonction traced_query est asynchrone, donc on utilise aevaluate :

1from langsmith import aevaluate
2
3async def target(inputs: dict) -> dict:
4    """Adaptateur : LangSmith passe un dict avec la question et attend un dict avec la réponse."""
5    return {"answer": await traced_query(inputs["question"])}
6
7def contains_expected(outputs: dict, reference_outputs: dict) -> bool:
8    """Évaluateur simple : la réponse contient-elle la phrase attendue ?"""
9    return reference_outputs["answer"].lower() in outputs["answer"].lower()
10
11async def evaluate_rag():
12    """Évalue le système RAG."""
13    # Dataset avec questions et réponses attendues, créé auparavant dans LangSmith
14    results = await aevaluate(
15        target,
16        data="rag-eval-dataset",
17        evaluators=[contains_expected]
18    )
19    return results

Un évaluateur est une fonction ordinaire qui renvoie True ou False. Des métriques comme la faithfulness ou la context precision se calculent de la même façon, seul le juge est un modèle de langage.

Rate Limiting et sécurité

Une API publique sans limites, c'est une invitation aux abus. D'abord un limiteur simple avec une fenêtre d'une minute :

1from fastapi import Depends, HTTPException
2from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
3import time
4
5# Rate limiter
6class RateLimiter:
7    """Rate limiter simple."""
8
9    def __init__(self, requests_per_minute: int = 60):
10        self.rpm = requests_per_minute
11        self.requests: dict[str, list[float]] = {}
12
13    async def check(self, user_id: str) -> bool:
14        now = time.time()
15        minute_ago = now - 60
16
17        if user_id not in self.requests:
18            self.requests[user_id] = []
19
20        # Supprime les anciennes requêtes
21        self.requests[user_id] = [
22            t for t in self.requests[user_id] if t > minute_ago
23        ]
24
25        if len(self.requests[user_id]) >= self.rpm:
26            return False
27
28        self.requests[user_id].append(now)
29        return True
30
31rate_limiter = RateLimiter()
32security = HTTPBearer()

Le limiteur retient les heures des requêtes de chaque utilisateur et supprime celles de plus de 60 secondes. Il garde les données dans la mémoire du processus, donc avec plusieurs instances du service, déplace-le dans Redis.

Maintenant, on branche l'autorisation et le limiteur sur l'endpoint via Depends :

1async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
2    """Vérification du token."""
3    # En production : vérification JWT
4    if credentials.credentials == "invalid":
5        raise HTTPException(status_code=401, detail="Invalid token")
6    return credentials.credentials
7
8async def rate_limit(user_id: str = Depends(verify_token)):
9    """Middleware de rate limiting."""
10    if not await rate_limiter.check(user_id):
11        raise HTTPException(status_code=429, detail="Too many requests")
12    return user_id
13
14@app.post("/query")
15async def query(request: QueryRequest, user: str = Depends(rate_limit)):
16    return await rag_service.query(request)

HTTPBearer extrait le token de l'en-tête Authorization, et verify_token renvoie 401 pour un mauvais token. Le dépassement de la limite se termine par un code 429. Cet endpoint remplace le /query précédent, alors ne garde qu'un seul des deux dans l'application, car FastAPI utilise la première route correspondante.

Déploiement avec Docker

Pour finir, on emballe le service dans un conteneur. Le Dockerfile décrit l'image de l'application :

1# Dockerfile
2FROM python:3.11-slim
3
4WORKDIR /app
5
6# Installe les dépendances
7COPY requirements.txt .
8RUN pip install --no-cache-dir -r requirements.txt
9
10# Copie le code
11COPY . .
12
13# Lance
14CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

On copie d'abord requirements.txt, puis le code, donc une modification du code n'oblige pas à réinstaller les dépendances.

Docker Compose lance toute la base d'un coup : l'API, Qdrant et Redis :

1# docker-compose.yml
2version: '3.8'
3
4services:
5  rag-api:
6    build: .
7    ports:
8      - "8000:8000"
9    environment:
10      - OPENAI_API_KEY=${OPENAI_API_KEY}
11      - QDRANT_HOST=qdrant
12      - REDIS_HOST=redis
13    depends_on:
14      - qdrant
15      - redis
16
17  qdrant:
18    image: qdrant/qdrant:latest
19    ports:
20      - "6333:6333"
21    volumes:
22      - qdrant_data:/qdrant/storage
23
24  redis:
25    image: redis:alpine
26    ports:
27      - "6379:6379"

Compose remplace ${OPENAI_API_KEY} par la valeur de la variable d'environnement de l'hôte, donc la clé n'atterrit jamais dans le dépôt. La clé version est obsolète et ignorée dans le Compose actuel.

La suite du fichier ajoute la couche d'observabilité :

1  prometheus:
2    image: prom/prometheus
3    ports:
4      - "9090:9090"
5    volumes:
6      - ./prometheus.yml:/etc/prometheus/prometheus.yml
7
8  grafana:
9    image: grafana/grafana
10    ports:
11      - "3000:3000"
12
13volumes:
14  qdrant_data:

Prometheus collecte les métriques, et Grafana en tire des graphiques. Le volume qdrant_data permet aux vecteurs de survivre au redémarrage du conteneur.

Bonnes pratiques

Une liste de bonnes pratiques résume toutes les leçons de ce module :

1"""
2Production RAG Best Practices:
3
41. RETRIEVAL
5   - Use hybrid search (vector + keyword)
6   - Implement reranking
7   - Filter by metadata
8
92. CHUNKING
10   - Experiment with chunk sizes
11   - Use overlap
12   - Consider semantic chunking
13
143. CACHING
15   - Cache embeddings
16   - Cache frequent queries
17   - Invalidate on document updates
18
194. MONITORING
20   - Track latency at each stage
21   - Monitor retrieval quality
22   - Alert on anomalies
23
245. SECURITY
25   - Rate limiting
26   - Input sanitization
27   - PII detection and filtering
28
296. SCALABILITY
30   - Horizontal scaling API
31   - Sharding vector database
32   - Async processing
33"""

Je te conseille de commencer par le cache et le monitoring, car ce sont eux qui rapportent le plus vite : le cache réduit les coûts d'API et la latence, et les métriques montrent où chercher un problème. Remarque que les prompts codés en dur ne figurent pas sur la liste : en production, les prompts vivent dans la configuration ou dans un système de versionnement.

Félicitations ! Tu as découvert les systèmes RAG et multi-agents avancés. Dans la prochaine leçon, tu vas découvrir LangGraph - un framework pour construire des workflows d'agents complexes !

Retiens : un RAG en production, c'est une base d'expédition permanente où chaque tente a sa mission et où un garde compte tous ceux qui entrent.