RAG en production : architecture simple qui fonctionne vraiment

La plupart des tutoriels RAG vous montrent un prototype qui fonctionne en 20 lignes de code. Puis vous déployez en production et tout s'effondre : latence excessive, réponses incohérentes, coûts qui explosent. Le problème n'est pas le RAG, c'est l'architecture sous-dimensionnée.

Cet article présente une architecture RAG pragmatique, testée en production, qui équilibre performance, fiabilité et coûts.

Architecture de référence

┌─────────────────────────────────────────────────────────────────┐
│                        UTILISATEUR                              │
└─────────────────────────────┬───────────────────────────────────┘
                              │ Question
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                      API GATEWAY                                │
│              (Rate limiting, Auth, Cache)                       │
└─────────────────────────────┬───────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                    QUERY PROCESSOR                              │
│         (Reformulation, Expansion, Classification)              │
└─────────────────────────────┬───────────────────────────────────┘
                              │
              ┌───────────────┼───────────────┐
              ▼               ▼               ▼
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│  VECTOR SEARCH  │ │  KEYWORD SEARCH │ │  METADATA       │
│  (Embeddings)   │ │  (BM25/Elastic) │ │  FILTERS        │
└────────┬────────┘ └────────┬────────┘ └────────┬────────┘
         │                   │                   │
         └───────────────────┼───────────────────┘
                             ▼
┌─────────────────────────────────────────────────────────────────┐
│                      RERANKER                                   │
│              (Cross-encoder, Scoring)                           │
└─────────────────────────────┬───────────────────────────────────┘
                              │ Top-K documents
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                    LLM GENERATOR                                │
│           (Prompt + Context → Response)                         │
└─────────────────────────────┬───────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                    POST-PROCESSING                              │
│         (Validation, Citations, Formatting)                     │
└─────────────────────────────────────────────────────────────────┘

Les 5 composants essentiels

1. Query Processor : ne jamais utiliser la question brute

La question de l'utilisateur est rarement optimale pour la recherche. Le query processor la transforme :

class QueryProcessor:
    def __init__(self, llm_client):
        self.llm = llm_client

    def process(self, raw_query: str) -> ProcessedQuery:
        # 1. Classification de l'intention
        intent = self.classify_intent(raw_query)

        # 2. Reformulation pour la recherche
        search_query = self.reformulate(raw_query)

        # 3. Extraction des filtres métadonnées
        filters = self.extract_filters(raw_query)

        # 4. Expansion avec synonymes/concepts liés
        expanded_terms = self.expand_query(search_query)

        return ProcessedQuery(
            original=raw_query,
            search_query=search_query,
            intent=intent,
            filters=filters,
            expanded_terms=expanded_terms
        )

    def reformulate(self, query: str) -> str:
        prompt = f"""Reformule cette question pour optimiser la recherche documentaire.
        Garde les termes techniques importants.
        Question: {query}
        Reformulation:"""
        return self.llm.generate(prompt, max_tokens=100)

Exemple de transformation :

• Entrée : "Comment faire pour que mon app Spring soit plus rapide ?"
• Reformulation : "optimisation performance Spring Boot latence throughput"
• Filtres extraits : {technology: "Spring Boot"}

2. Recherche hybride : vecteurs + mots-clés

La recherche vectorielle seule rate souvent des résultats pertinents. Combinez-la avec une recherche par mots-clés :

class HybridRetriever:
    def __init__(self, vector_store, keyword_store):
        self.vector_store = vector_store  # Qdrant, Pinecone, Weaviate
        self.keyword_store = keyword_store  # Elasticsearch, OpenSearch

    def retrieve(self, query: ProcessedQuery, top_k: int = 20) -> List[Document]:
        # Recherche vectorielle
        vector_results = self.vector_store.search(
            query_embedding=self.embed(query.search_query),
            top_k=top_k,
            filters=query.filters
        )

        # Recherche par mots-clés (BM25)
        keyword_results = self.keyword_store.search(
            query=query.search_query,
            expanded_terms=query.expanded_terms,
            top_k=top_k,
            filters=query.filters
        )

        # Fusion des résultats (Reciprocal Rank Fusion)
        fused = self.rrf_fusion(vector_results, keyword_results)

        return fused[:top_k]

    def rrf_fusion(self, *result_lists, k=60) -> List[Document]:
        """Reciprocal Rank Fusion - combine plusieurs rankings"""
        scores = defaultdict(float)

        for results in result_lists:
            for rank, doc in enumerate(results):
                scores[doc.id] += 1 / (k + rank + 1)

        sorted_docs = sorted(scores.items(), key=lambda x: x[1], reverse=True)
        return [self.get_doc(doc_id) for doc_id, _ in sorted_docs]

Pourquoi hybride ?

3. Reranker : le filtre de qualité

Le retriever ramène 20-50 documents. Le reranker les classe par pertinence réelle :

from sentence_transformers import CrossEncoder

class Reranker:
    def __init__(self, model_name="cross-encoder/ms-marco-MiniLM-L-12-v2"):
        self.model = CrossEncoder(model_name)

    def rerank(self, query: str, documents: List[Document], top_k: int = 5) -> List[Document]:
        # Préparer les paires (query, document)
        pairs = [(query, doc.content) for doc in documents]

        # Scorer chaque paire
        scores = self.model.predict(pairs)

        # Trier par score décroissant
        ranked = sorted(
            zip(documents, scores),
            key=lambda x: x[1],
            reverse=True
        )

        return [doc for doc, score in ranked[:top_k]]

Impact du reranker :

• Sans reranker : précision ~60%
• Avec reranker : précision ~85%
• Latence ajoutée : 50-100ms

4. Prompt structuré : contexte + instructions claires

Le prompt du LLM doit être précis et structuré :

class PromptBuilder:
    SYSTEM_PROMPT = """Tu es un assistant technique expert.
Tu réponds UNIQUEMENT à partir des documents fournis.
Si l'information n'est pas dans les documents, dis-le clairement.
Cite tes sources avec [1], [2], etc."""

    def build_prompt(self, query: str, documents: List[Document]) -> str:
        # Formater les documents avec numéros de référence
        context = self.format_context(documents)

        return f"""{self.SYSTEM_PROMPT}

DOCUMENTS:
{context}

QUESTION: {query}

RÉPONSE (avec citations):"""

    def format_context(self, documents: List[Document]) -> str:
        formatted = []
        for i, doc in enumerate(documents, 1):
            formatted.append(f"[{i}] {doc.title}\n{doc.content[:1500]}")
        return "\n\n".join(formatted)

Règles du prompt :

• Limiter le contexte (4-8 documents max)
• Demander explicitement les citations
• Interdire l'invention d'informations
• Spécifier le format de réponse attendu

5. Post-processing : validation et formatage

Ne faites jamais confiance au LLM aveuglément :

class ResponseProcessor:
    def process(self, response: str, documents: List[Document]) -> FinalResponse:
        # 1. Extraire les citations
        citations = self.extract_citations(response)

        # 2. Valider que les citations existent
        valid_citations = self.validate_citations(citations, documents)

        # 3. Détecter les hallucinations potentielles
        hallucination_score = self.detect_hallucination(response, documents)

        # 4. Formater la réponse finale
        formatted = self.format_response(response, valid_citations)

        return FinalResponse(
            content=formatted,
            citations=valid_citations,
            confidence=1 - hallucination_score,
            sources=[doc.metadata for doc in documents]
        )

    def detect_hallucination(self, response: str, documents: List[Document]) -> float:
        """Score de 0 (fiable) à 1 (hallucination probable)"""
        # Vérifier que les faits clés sont dans les documents
        facts = self.extract_facts(response)
        verified = sum(1 for f in facts if self.fact_in_docs(f, documents))
        return 1 - (verified / len(facts)) if facts else 0.5

Choix techniques recommandés

Vector Store

Recommandation : Qdrant pour la production (performance + coût), pgvector pour démarrer simple.

Modèle d'embeddings

Recommandation : OpenAI text-embedding-3-small (rapport qualité/prix), BGE-large pour self-hosted.

LLM pour génération

Recommandation : GPT-4o-mini pour 90% des cas, GPT-4o pour les requêtes complexes.

Erreurs courantes à éviter

Erreur 1 : Chunks trop grands ou trop petits

chunker = RecursiveCharacterTextSplitter(chunk_size=16000)

chunker = RecursiveCharacterTextSplitter(chunk_size=400)

chunker = RecursiveCharacterTextSplitter(
    chunk_size=2000,  # ~500 tokens
    chunk_overlap=400,  # 20% overlap
    separators=["\n\n", "\n", ". ", " "]
)

Erreur 2 : Pas de métadonnées sur les chunks

chunk = {"content": "La méthode retourne null si..."}

chunk = {
    "content": "La méthode retourne null si...",
    "metadata": {
        "source": "docs/api-reference.md",
        "title": "UserService.findById()",
        "section": "Return Values",
        "last_updated": "2025-01-15",
        "language": "fr",
        "doc_type": "api_doc"
    }
}

Erreur 3 : Ignorer le cache

def answer(query):
    embedding = embed(query)  # 100ms
    docs = search(embedding)   # 50ms
    response = llm(query, docs)  # 2000ms
    return response

class CachedRAG:
    def __init__(self):
        self.embedding_cache = LRUCache(maxsize=10000)
        self.response_cache = TTLCache(maxsize=1000, ttl=3600)

    def answer(self, query: str) -> str:
        # Cache niveau 1 : réponse complète
        cache_key = self.hash_query(query)
        if cache_key in self.response_cache:
            return self.response_cache[cache_key]

        # Cache niveau 2 : embeddings
        if query not in self.embedding_cache:
            self.embedding_cache[query] = self.embed(query)
        embedding = self.embedding_cache[query]

        docs = self.search(embedding)
        response = self.llm(query, docs)

        self.response_cache[cache_key] = response
        return response

Erreur 4 : Pas de fallback

def answer(query):
    docs = search(query)
    if not docs:
        return "Je ne sais pas"  # Frustrant pour l'utilisateur

def answer(query):
    # Tentative 1 : recherche précise
    docs = search(query, threshold=0.8)

    # Tentative 2 : recherche élargie
    if len(docs) < 3:
        docs = search(query, threshold=0.5)

    # Tentative 3 : recherche par mots-clés seuls
    if len(docs) < 3:
        docs = keyword_search(query)

    # Fallback : réponse honnête avec suggestions
    if not docs:
        return suggest_alternatives(query)

    return generate(query, docs)

Métriques de monitoring

Métriques essentielles

metrics = {
    # Latence
    "retrieval_latency_p50": 45,   # ms
    "retrieval_latency_p99": 120,  # ms
    "generation_latency_p50": 1800,  # ms
    "total_latency_p50": 2100,  # ms

    # Qualité
    "retrieval_precision": 0.82,  # % docs pertinents
    "answer_relevance": 0.78,     # Score 0-1
    "hallucination_rate": 0.05,   # % réponses avec hallucinations

    # Usage
    "cache_hit_rate": 0.35,       # % requêtes servies par cache
    "empty_retrieval_rate": 0.02, # % recherches sans résultat

    # Coûts
    "cost_per_query": 0.003,      # $ par requête moyenne
    "tokens_per_query": 2500,     # tokens moyens consommés
}

Dashboard de monitoring

from prometheus_client import Histogram, Counter, Gauge

retrieval_latency = Histogram(
    'rag_retrieval_latency_seconds',
    'Latency of document retrieval',
    buckets=[0.01, 0.05, 0.1, 0.25, 0.5, 1.0]
)

generation_latency = Histogram(
    'rag_generation_latency_seconds',
    'Latency of LLM generation',
    buckets=[0.5, 1.0, 2.0, 5.0, 10.0]
)

queries_total = Counter(
    'rag_queries_total',
    'Total number of RAG queries',
    ['status', 'cache_hit']
)

retrieval_quality = Gauge(
    'rag_retrieval_precision',
    'Precision of retrieved documents'
)

Estimation des coûts

Pour 10 000 requêtes/jour

Optimisations coût

• Cache agressif : -30% de requêtes LLM
• Embeddings locaux (BGE) : -100% coût embeddings
• Reranker local : -100% coût reranking
• Total optimisé : ~$180/mois

Conclusion

Une architecture RAG en production n'est pas un prototype agrandi. Elle nécessite :

1. Query processing pour optimiser les recherches
2. Recherche hybride pour couvrir tous les types de requêtes
3. Reranking pour filtrer le bruit
4. Prompts structurés pour des réponses fiables
5. Post-processing pour valider et formater

Commencez simple avec cette architecture de référence, puis optimisez en fonction de vos métriques. Le RAG parfait n'existe pas du premier coup, mais avec les bons composants et le bon monitoring, vous pouvez itérer rapidement vers une solution fiable.

Pour comprendre pourquoi certains RAG échouent, consultez notre prochain article : Pourquoi votre RAG échoue (et comment le corriger)

Pour une introduction au RAG : Pourquoi tout le monde parle de RAG en 2025 ?