Pourquoi votre RAG échoue (et comment le corriger)

Votre RAG fonctionne en démo mais échoue en production. Les réponses sont hors sujet, incomplètes, ou pire, inventées. Vous n'êtes pas seul : 70% des projets RAG n'atteignent jamais la production selon les retours d'expérience de la communauté ML.

Ce guide identifie les causes réelles d'échec et propose des solutions concrètes pour chacune.

Cause 1 : Vos données sont le problème

Symptômes

• Le RAG trouve des documents mais les réponses sont incorrectes
• Les mêmes questions donnent des réponses contradictoires
• L'information existe mais n'est jamais retrouvée

Diagnostic

def diagnose_data_quality(documents: List[Document]):
    issues = []

    for doc in documents:
        # Documents trop courts (< 100 mots)
        if len(doc.content.split()) < 100:
            issues.append(f"Document trop court: {doc.id}")

        # Documents dupliqués
        if is_near_duplicate(doc, documents):
            issues.append(f"Doublon détecté: {doc.id}")

        # Contenu obsolète
        if doc.last_updated < datetime.now() - timedelta(days=365):
            issues.append(f"Contenu obsolète: {doc.id}")

        # Métadonnées manquantes
        if not doc.metadata.get("source") or not doc.metadata.get("title"):
            issues.append(f"Métadonnées incomplètes: {doc.id}")

    return issues

Problèmes courants et solutions

Solution : Pipeline de nettoyage

class DataCleaner:
    def clean(self, raw_content: str) -> str:
        content = raw_content

        # 1. Supprimer le HTML/Markdown superflu
        content = self.strip_formatting(content)

        # 2. Normaliser les espaces et sauts de ligne
        content = self.normalize_whitespace(content)

        # 3. Supprimer les headers/footers répétitifs
        content = self.remove_boilerplate(content)

        # 4. Détecter et corriger l'encodage
        content = self.fix_encoding(content)

        # 5. Valider la qualité minimale
        if not self.is_quality_sufficient(content):
            raise LowQualityContentError(content)

        return content

    def is_quality_sufficient(self, content: str) -> bool:
        words = content.split()
        # Au moins 50 mots
        if len(words) < 50:
            return False
        # Ratio mots uniques > 30%
        if len(set(words)) / len(words) < 0.3:
            return False
        return True

Cause 2 : Votre chunking est inadapté

Symptômes

• Les réponses sont incomplètes (information coupée)
• Le contexte manque de cohérence
• Trop de chunks non pertinents remontent

Le problème du chunking naïf

text = "La fonction authenticate() vérifie les credentials. Elle retourne"

Stratégies de chunking

Solution : Chunking sémantique

from langchain.text_splitter import RecursiveCharacterTextSplitter

class SemanticChunker:
    def __init__(self, embedding_model):
        self.embedder = embedding_model
        self.base_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200,
            separators=["\n\n", "\n", ". ", "! ", "? ", "; ", ", ", " "]
        )

    def chunk(self, document: str) -> List[str]:
        # 1. Split initial
        raw_chunks = self.base_splitter.split_text(document)

        # 2. Fusionner les chunks sémantiquement proches
        merged_chunks = self.merge_similar_chunks(raw_chunks)

        # 3. Ajouter le contexte parent
        contextualized = self.add_context(merged_chunks, document)

        return contextualized

    def merge_similar_chunks(self, chunks: List[str], threshold=0.85) -> List[str]:
        """Fusionne les chunks consécutifs très similaires"""
        embeddings = self.embedder.embed(chunks)
        merged = []
        current = chunks[0]

        for i in range(1, len(chunks)):
            similarity = cosine_similarity(embeddings[i-1], embeddings[i])
            if similarity > threshold:
                current += " " + chunks[i]
            else:
                merged.append(current)
                current = chunks[i]

        merged.append(current)
        return merged

    def add_context(self, chunks: List[str], full_doc: str) -> List[str]:
        """Ajoute un résumé du document parent à chaque chunk"""
        summary = self.summarize(full_doc)
        return [f"[Context: {summary}]\n\n{chunk}" for chunk in chunks]

Cause 3 : Vos embeddings ne capturent pas le sens

Symptômes

• Recherches par mots-clés fonctionnent mieux que la recherche vectorielle
• Documents similaires ont des scores très différents
• La recherche rate des synonymes évidents

Diagnostic des embeddings

def diagnose_embeddings(queries: List[str], expected_docs: List[str], vector_store):
    issues = []

    for query, expected in zip(queries, expected_docs):
        results = vector_store.search(query, top_k=10)
        result_ids = [r.id for r in results]

        if expected not in result_ids:
            # Document attendu pas dans le top 10
            issues.append({
                "query": query,
                "expected": expected,
                "got": result_ids[:3],
                "issue": "Document pertinent non trouvé"
            })
        elif result_ids.index(expected) > 3:
            # Document attendu trop bas dans le ranking
            issues.append({
                "query": query,
                "expected": expected,
                "rank": result_ids.index(expected),
                "issue": "Ranking incorrect"
            })

    return issues

Problèmes d'embeddings courants

Solution : Choix et adaptation du modèle

from sentence_transformers import SentenceTransformer

model = SentenceTransformer("intfloat/multilingual-e5-large")

from sentence_transformers import InputExample, losses

train_examples = [
    InputExample(texts=["erreur connexion OAuth", doc_oauth_troubleshooting]),
    InputExample(texts=["configurer SSL Spring", doc_ssl_configuration]),
    # ... plus d'exemples de votre domaine
]

model.fit(
    train_objectives=[(train_dataloader, losses.MultipleNegativesRankingLoss(model))],
    epochs=3,
    warmup_steps=100
)

Cause 4 : Votre retrieval manque de précision

Symptômes

• Beaucoup de documents retournés mais peu pertinents
• Le bon document est noyé dans le bruit
• Résultats très différents pour des requêtes similaires

Le problème du top-K fixe

results = vector_store.search(query, top_k=5)

Solutions de retrieval avancées

class AdaptiveRetriever:
    def __init__(self, vector_store, keyword_store, reranker):
        self.vector = vector_store
        self.keyword = keyword_store
        self.reranker = reranker

    def retrieve(self, query: str, min_docs=3, max_docs=10) -> List[Document]:
        # 1. Recherche hybride large
        vector_results = self.vector.search(query, top_k=max_docs * 2)
        keyword_results = self.keyword.search(query, top_k=max_docs * 2)

        # 2. Fusion RRF
        candidates = self.rrf_fusion(vector_results, keyword_results)

        # 3. Reranking
        reranked = self.reranker.rerank(query, candidates)

        # 4. Filtrage adaptatif par score
        threshold = self.compute_adaptive_threshold(reranked)
        filtered = [doc for doc in reranked if doc.score > threshold]

        # 5. Garantir un minimum de résultats
        if len(filtered) < min_docs:
            filtered = reranked[:min_docs]

        return filtered[:max_docs]

    def compute_adaptive_threshold(self, docs: List[Document]) -> float:
        """Seuil dynamique basé sur la distribution des scores"""
        scores = [doc.score for doc in docs]
        mean = sum(scores) / len(scores)
        std = (sum((s - mean) ** 2 for s in scores) / len(scores)) ** 0.5

        # Garder les docs à plus de 0.5 std au-dessus de la moyenne
        return mean + 0.5 * std

Ajouter un reranker

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]) -> List[Document]:
        # Scorer chaque paire query-document
        pairs = [(query, doc.content[:512]) for doc in documents]
        scores = self.model.predict(pairs)

        # Attacher les scores et trier
        for doc, score in zip(documents, scores):
            doc.rerank_score = float(score)

        return sorted(documents, key=lambda d: d.rerank_score, reverse=True)

Impact mesuré du reranker :

• Precision@5 sans reranker : 62%
• Precision@5 avec reranker : 84%
• Latence ajoutée : 50-80ms

Cause 5 : Votre prompt est mal conçu

Symptômes

• Le LLM ignore le contexte fourni
• Réponses génériques malgré des documents spécifiques
• Hallucinations fréquentes

Prompts problématiques

prompt = f"Réponds à cette question: {query}\nDocuments: {docs}"

prompt = f"Voici des documents:\n{docs}\n\nQuestion: {query}"

prompt = f"""Tu es un assistant utile et bienveillant. Tu dois toujours
être poli et respectueux. Tu peux utiliser les documents suivants mais
tu peux aussi utiliser tes connaissances générales. N'hésite pas à
demander des clarifications si nécessaire.

Documents: {docs}

Question: {query}"""

Solution : Prompt structuré et contraignant

class PromptBuilder:
    SYSTEM = """Tu es un assistant technique qui répond UNIQUEMENT à partir des documents fournis.

RÈGLES STRICTES:
1. Utilise UNIQUEMENT les informations des documents ci-dessous
2. Si l'information n'est pas dans les documents, réponds "Je n'ai pas trouvé cette information dans la documentation."
3. Cite tes sources avec [1], [2], etc.
4. Sois concis et précis
5. N'invente JAMAIS d'information"""

    def build(self, query: str, documents: List[Document]) -> str:
        # Formater les documents avec numéros
        docs_formatted = []
        for i, doc in enumerate(documents, 1):
            docs_formatted.append(
                f"[{i}] Source: {doc.metadata.get('title', 'Unknown')}\n"
                f"{doc.content[:1500]}"
            )

        context = "\n\n---\n\n".join(docs_formatted)

        return f"""{self.SYSTEM}

DOCUMENTS DE RÉFÉRENCE:
{context}

QUESTION DE L'UTILISATEUR:
{query}

RÉPONSE (avec citations [1], [2], etc.):"""

Techniques anti-hallucination

class HallucinationGuard:
    def __init__(self, llm):
        self.llm = llm

    def generate_with_verification(self, query: str, docs: List[Document]) -> str:
        # 1. Générer la réponse
        response = self.generate(query, docs)

        # 2. Extraire les affirmations factuelles
        claims = self.extract_claims(response)

        # 3. Vérifier chaque affirmation contre les documents
        verified_claims = []
        for claim in claims:
            if self.verify_claim(claim, docs):
                verified_claims.append(claim)
            else:
                # Marquer comme non vérifié ou supprimer
                verified_claims.append(f"[Non vérifié] {claim}")

        # 4. Reconstruire la réponse
        return self.rebuild_response(verified_claims)

    def verify_claim(self, claim: str, docs: List[Document]) -> bool:
        """Vérifie si une affirmation est supportée par les documents"""
        prompt = f"""L'affirmation suivante est-elle supportée par les documents?

Affirmation: {claim}

Documents:
{self.format_docs(docs)}

Réponds uniquement par OUI ou NON."""

        verdict = self.llm.generate(prompt, max_tokens=5)
        return "OUI" in verdict.upper()

Cause 6 : Votre infrastructure ne tient pas la charge

Symptômes

• Latence acceptable en dev, catastrophique en prod
• Timeouts fréquents
• Coûts qui explosent

Goulots d'étranglement typiques

Solutions d'optimisation

class OptimizedRAG:
    def __init__(self):
        # 1. Cache multi-niveaux
        self.query_cache = LRUCache(maxsize=10000)
        self.embedding_cache = LRUCache(maxsize=50000)
        self.response_cache = TTLCache(maxsize=5000, ttl=3600)

        # 2. Batch processing pour embeddings
        self.embedding_batch = BatchProcessor(
            max_batch_size=32,
            max_wait_ms=50
        )

        # 3. Connection pooling
        self.vector_store = VectorStore(
            connection_pool_size=20,
            max_retries=3
        )

    async def answer(self, query: str) -> str:
        # Check response cache
        cache_key = self.hash_query(query)
        if cache_key in self.response_cache:
            return self.response_cache[cache_key]

        # Parallel retrieval + embedding
        embedding_task = self.get_or_compute_embedding(query)
        normalized_query = self.normalize_query(query)

        embedding = await embedding_task
        docs = await self.vector_store.search_async(embedding)

        # Streaming response
        response = await self.llm.generate_stream(query, docs)

        self.response_cache[cache_key] = response
        return response

    async def get_or_compute_embedding(self, text: str):
        if text in self.embedding_cache:
            return self.embedding_cache[text]

        embedding = await self.embedding_batch.add(text)
        self.embedding_cache[text] = embedding
        return embedding

Métriques à monitorer

from prometheus_client import Histogram, Counter, Gauge

embedding_latency = Histogram('rag_embedding_seconds', 'Embedding latency')
retrieval_latency = Histogram('rag_retrieval_seconds', 'Retrieval latency')
rerank_latency = Histogram('rag_rerank_seconds', 'Reranking latency')
generation_latency = Histogram('rag_generation_seconds', 'LLM generation latency')

cache_hits = Counter('rag_cache_hits_total', 'Cache hit count', ['cache_type'])
empty_results = Counter('rag_empty_results_total', 'Queries with no results')
hallucination_detected = Counter('rag_hallucinations_total', 'Detected hallucinations')

vector_store_connections = Gauge('rag_vectorstore_connections', 'Active connections')
llm_queue_size = Gauge('rag_llm_queue_size', 'Pending LLM requests')

Checklist de debugging

Quand votre RAG échoue, suivez cette checklist :

1. Vérifier les données

• [ ] Documents bien indexés ? (count, sample)
• [ ] Pas de doublons massifs ?
• [ ] Métadonnées présentes ?
• [ ] Contenu nettoyé ?

2. Vérifier le retrieval

• [ ] La requête problématique retourne des documents ?
• [ ] Les documents retournés sont pertinents ?
• [ ] Le reranker améliore-t-il le ranking ?

3. Vérifier le prompt

• [ ] Le contexte est bien formaté ?
• [ ] Les instructions sont claires ?
• [ ] Le LLM utilise bien le contexte ?

4. Vérifier l'infrastructure

• [ ] Latences par composant normales ?
• [ ] Pas d'erreurs dans les logs ?
• [ ] Resources suffisantes ?

Conclusion

Les échecs RAG ont presque toujours des causes identifiables :

1. Données : nettoyez, dédupliquez, enrichissez
2. Chunking : adaptez la stratégie au contenu
3. Embeddings : choisissez le bon modèle, fine-tunez si nécessaire
4. Retrieval : hybride + reranking systématique
5. Prompt : structuré, contraignant, avec anti-hallucination
6. Infrastructure : cache, batch, monitoring

Le RAG parfait n'existe pas du premier coup. Instrumentez, mesurez, itérez. Chaque amélioration de 5% de précision se traduit par une meilleure expérience utilisateur.

Pour l'architecture complète d'un RAG production-ready : RAG en production : architecture simple qui fonctionne vraiment

Pour une introduction au RAG : RAG en 2025 : définition, architecture et cas d'usage