RAG Adaptatif : L'IA qui Choisit Automatiquement la Meilleure Stratégie

Pourquoi utiliser un marteau pour toutes les vis ? Le RAG fixe applique la même stratégie de retrieval à chaque requête, qu'elle soit simple ou complexe. Le RAG adaptatif analyse d'abord la question, puis choisit automatiquement la meilleure approche : pas de retrieval du tout, retrieval simple, ou retrieval multi-étapes. Résultat : une précision proche d'une stratégie multi-étapes systématique, mais avec bien moins d'appels de recherche (1,08 étape en moyenne contre 4,69), et des coûts réduits sur les questions simples.

TL;DR

• RAG adaptatif route dynamiquement chaque requête vers la stratégie optimale
• Trois routes : no-retrieval (LLM direct), single-step retrieval, multi-step retrieval
• Classificateur : LLM léger ou modèle entraîné qui analyse la complexité de la requête
• Précision proche du multi-étapes systématique avec ~4× moins d'étapes de retrieval — 1,08 contre 4,69 en moyenne (Adaptive-RAG, Jeong et al., 2024)
• Moins d'appels de retrieval en évitant la recherche inutile sur les questions simples
• Compatible avec CRAG, Self-RAG et d'autres techniques avancées

Le problème du RAG fixe

Le RAG fixe traite toutes les requêtes de la même manière. C'est un gaspillage.

RAG Fixe (One-size-fits-all) :

"Bonjour !" ──────────────▶ Retrieval + Génération  ❌ Inutile
"Quelle heure est-il ?" ──▶ Retrieval + Génération  ❌ Inutile
"Votre politique retour ?" ▶ Retrieval + Génération  ✅ Pertinent
"Comparez les plans A et B
 puis recommandez selon
 mon profil PME" ──────────▶ Retrieval + Génération  ⚠️ Insuffisant
                                                        (besoin multi-step)

Les trois types de requêtes

*Proportions typiques observées sur des chatbots e-commerce et support.

Architecture du RAG Adaptatif

┌──────────────────────────────────────────────────────────────────────┐
│                     Pipeline RAG Adaptatif                            │
├──────────────────────────────────────────────────────────────────────┤
│                                                                       │
│  Requête utilisateur                                                  │
│         │                                                             │
│         ▼                                                             │
│  ┌─────────────────┐                                                 │
│  │  Classificateur  │  Analyse : complexité, besoin factuel,          │
│  │  de requête      │  domaine, type de réponse attendue              │
│  └────────┬────────┘                                                 │
│           │                                                           │
│     ┌─────┼─────────────────┐                                        │
│     │     │                 │                                        │
│     ▼     ▼                 ▼                                        │
│  ┌──────┐ ┌──────────┐ ┌────────────────┐                           │
│  │ No   │ │ Single   │ │ Multi-step     │                           │
│  │Retr. │ │ Step     │ │ Retrieval      │                           │
│  │      │ │ Retrieval│ │                │                           │
│  │ LLM  │ │ Query→   │ │ Décomposition  │                           │
│  │direct│ │ Retrieve │ │ → Sous-queries │                           │
│  │      │ │ → Generate│ │ → Multi-retrieve│                           │
│  │      │ │          │ │ → Synthèse     │                           │
│  └──┬───┘ └────┬─────┘ └───────┬────────┘                           │
│     │          │               │                                     │
│     └──────────┴───────────────┘                                     │
│                    │                                                  │
│                    ▼                                                  │
│             Réponse finale                                           │
│                                                                       │
└──────────────────────────────────────────────────────────────────────┘

Le Classificateur de Requêtes

Le coeur du RAG adaptatif est le classificateur qui détermine la stratégie optimale pour chaque requête.

Approche 1 : Classificateur LLM (zero-shot)

La méthode la plus simple : utiliser un LLM léger pour classifier.

DEVELOPERpython
from langchain_openai import ChatOpenAI
from typing import Literal
from pydantic import BaseModel, Field

class QueryClassification(BaseModel):
    """Classification d'une requête pour le routing RAG."""
    complexity: Literal["simple", "factual", "complex"] = Field(
        description="Niveau de complexité de la requête"
    )
    reasoning: str = Field(
        description="Explication de la classification"
    )
    needs_retrieval: bool = Field(
        description="La requête nécessite-t-elle des documents ?"
    )
    estimated_steps: int = Field(
        description="Nombre d'étapes de retrieval nécessaires",
        ge=0, le=5
    )

class QueryClassifier:
    def __init__(self):
        self.llm = ChatOpenAI(
            model="gpt-4o-mini",
            temperature=0
        ).with_structured_output(QueryClassification)

    def classify(self, query: str) -> QueryClassification:
        prompt = f"""Classifie cette requête pour déterminer la stratégie de retrieval optimale.

Requête : "{query}"

Règles de classification :
- "simple" : salutations, questions générales sans besoin de docs spécifiques
  → needs_retrieval = false, estimated_steps = 0
- "factual" : question directe avec une réponse factuelle précise
  → needs_retrieval = true, estimated_steps = 1
- "complex" : comparaison, analyse multi-aspects, nécessite plusieurs sources
  → needs_retrieval = true, estimated_steps = 2-5"""

        return self.llm.invoke(prompt)

# Usage
classifier = QueryClassifier()

result = classifier.classify("Bonjour, comment allez-vous ?")
# complexity="simple", needs_retrieval=False, estimated_steps=0

result = classifier.classify("Quel est le prix du plan Enterprise ?")
# complexity="factual", needs_retrieval=True, estimated_steps=1

result = classifier.classify(
    "Comparez les avantages des plans Pro et Enterprise "
    "pour une PME de 200 employés dans le secteur santé"
)
# complexity="complex", needs_retrieval=True, estimated_steps=3

Approche 2 : Classificateur entraîné (plus rapide)

Pour la production, un petit modèle entraîné est plus rapide et moins coûteux.

DEVELOPERpython
from sentence_transformers import SentenceTransformer
from sklearn.ensemble import GradientBoostingClassifier
import numpy as np
import joblib

class TrainedQueryClassifier:
    """Classificateur entraîné pour le routing RAG."""

    def __init__(self, model_path: str = None):
        self.encoder = SentenceTransformer(
            "all-MiniLM-L6-v2"
        )
        if model_path:
            self.classifier = joblib.load(model_path)
        else:
            self.classifier = GradientBoostingClassifier(
                n_estimators=100,
                max_depth=5
            )

    def train(self, queries: list, labels: list):
        """Entraîne le classificateur sur des exemples labellisés."""
        embeddings = self.encoder.encode(queries)

        # Features supplémentaires
        features = []
        for q, emb in zip(queries, embeddings):
            extra = [
                len(q),                          # Longueur
                q.count("?"),                     # Nombre de questions
                len(q.split()),                   # Nombre de mots
                int(any(w in q.lower() for w in  # Mots de comparaison
                    ["compare", "versus", "différence", "avantage"])),
                int(any(w in q.lower() for w in  # Mots multi-step
                    ["puis", "ensuite", "aussi", "également"])),
            ]
            features.append(np.concatenate([emb, extra]))

        X = np.array(features)
        self.classifier.fit(X, labels)

    def predict(self, query: str) -> str:
        """Prédit la catégorie d'une requête."""
        emb = self.encoder.encode([query])[0]
        extra = [
            len(query),
            query.count("?"),
            len(query.split()),
            int(any(w in query.lower() for w in
                ["compare", "versus", "différence", "avantage"])),
            int(any(w in query.lower() for w in
                ["puis", "ensuite", "aussi", "également"])),
        ]
        X = np.array([np.concatenate([emb, extra])])
        return self.classifier.predict(X)[0]

# Entraînement
training_data = [
    ("Bonjour", "simple"),
    ("Merci beaucoup", "simple"),
    ("Quel est le prix ?", "factual"),
    ("Délai de livraison ?", "factual"),
    ("Comparez les plans A et B pour une PME", "complex"),
    ("Expliquez les différences entre RAG et fine-tuning", "complex"),
    # ... des centaines d'exemples
]

queries, labels = zip(*training_data)
classifier = TrainedQueryClassifier()
classifier.train(list(queries), list(labels))

Le Router Adaptatif

Le router connecte le classificateur aux différentes stratégies de retrieval.

DEVELOPERpython
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.documents import Document
from typing import TypedDict, List, Optional

class AdaptiveRAGState(TypedDict):
    question: str
    classification: str
    sub_questions: Optional[List[str]]
    documents: List[Document]
    generation: str

# Route 1 : No Retrieval (LLM direct)
def direct_llm_response(state: AdaptiveRAGState) -> AdaptiveRAGState:
    """Répond directement sans retrieval."""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
    response = llm.invoke(state["question"]).content
    return {**state, "generation": response}

# Route 2 : Single-Step Retrieval
async def single_step_retrieval(state: AdaptiveRAGState) -> AdaptiveRAGState:
    """Retrieval standard en une étape."""
    # Retrieval
    docs = await retriever.aget_relevant_documents(state["question"])

    # Génération
    llm = ChatOpenAI(model="gpt-4o", temperature=0.1)
    context = "\n\n".join([d.page_content for d in docs])

    response = llm.invoke(
        f"Contexte :\n{context}\n\nQuestion : {state['question']}"
    ).content

    return {**state, "documents": docs, "generation": response}

# Route 3 : Multi-Step Retrieval
async def decompose_question(state: AdaptiveRAGState) -> AdaptiveRAGState:
    """Décompose une question complexe en sous-questions."""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

    prompt = f"""Décompose cette question complexe en sous-questions simples
et indépendantes qui, une fois répondues, permettront de répondre
à la question principale.

Question : {state["question"]}

Retourne une liste JSON de sous-questions (max 4).
Exemple : ["Sous-question 1 ?", "Sous-question 2 ?"]"""

    result = llm.invoke(prompt).content
    import json
    sub_questions = json.loads(result)

    return {**state, "sub_questions": sub_questions}

async def multi_step_retrieval(state: AdaptiveRAGState) -> AdaptiveRAGState:
    """Récupère des documents pour chaque sous-question."""
    all_docs = []

    for sub_q in state["sub_questions"]:
        docs = await retriever.aget_relevant_documents(sub_q)
        all_docs.extend(docs)

    # Déduplication
    seen = set()
    unique_docs = []
    for doc in all_docs:
        key = hash(doc.page_content[:100])
        if key not in seen:
            seen.add(key)
            unique_docs.append(doc)

    return {**state, "documents": unique_docs}

async def synthesize_response(state: AdaptiveRAGState) -> AdaptiveRAGState:
    """Synthétise une réponse à partir de multiples sources."""
    llm = ChatOpenAI(model="gpt-4o", temperature=0.1)
    context = "\n\n".join([d.page_content for d in state["documents"]])

    prompt = f"""Réponds à cette question complexe en synthétisant
les informations de multiples sources.

Question principale : {state["question"]}
Sous-questions traitées : {state["sub_questions"]}

Contexte (multiples sources) :
{context}

Fournis une réponse structurée et complète."""

    response = llm.invoke(prompt).content
    return {**state, "generation": response}

# Construction du workflow adaptatif
def classify_and_route(state: AdaptiveRAGState) -> str:
    """Classifie et route vers la bonne stratégie."""
    classifier = QueryClassifier()
    result = classifier.classify(state["question"])

    if result.complexity == "simple":
        return "direct"
    elif result.complexity == "factual":
        return "single_step"
    else:
        return "decompose"

workflow = StateGraph(AdaptiveRAGState)

workflow.add_node("direct", direct_llm_response)
workflow.add_node("single_step", single_step_retrieval)
workflow.add_node("decompose", decompose_question)
workflow.add_node("multi_retrieve", multi_step_retrieval)
workflow.add_node("synthesize", synthesize_response)

workflow.set_conditional_entry_point(
    classify_and_route,
    {
        "direct": "direct",
        "single_step": "single_step",
        "decompose": "decompose"
    }
)

workflow.add_edge("direct", END)
workflow.add_edge("single_step", END)
workflow.add_edge("decompose", "multi_retrieve")
workflow.add_edge("multi_retrieve", "synthesize")
workflow.add_edge("synthesize", END)

adaptive_rag = workflow.compile()

Comparaison : RAG Fixe vs RAG Adaptatif

Benchmarks (Jeong et al., 2024)

Sur les 6 datasets de question-réponse évalués dans le papier (SQuAD, Natural Questions, TriviaQA, MuSiQue, HotpotQA, 2WikiMultiHopQA), avec FLAN-T5-XL (3B), les résultats moyens sont :

L'enseignement clé : l'Adaptive-RAG atteint quasiment la précision de l'approche multi-étapes systématique (46,9 vs 48,9 en F1) tout en n'utilisant en moyenne que 1,08 étape de retrieval contre 4,69 — soit un coût de recherche divisé par plus de quatre. Il dépasse nettement le single-step (44,3 F1) et double presque la précision du no-retrieval (21,1 F1).

Arbre de décision détaillé

Requête entrante
│
├── Contient-elle des mots de salutation/politesse ?
│   ├── Oui → Est-ce UNIQUEMENT une salutation ?
│   │   ├── Oui → Route : NO RETRIEVAL
│   │   └── Non → Continuer l'analyse ↓
│   └── Non → Continuer l'analyse ↓
│
├── La requête concerne-t-elle un fait précis ?
│   ├── Oui → Un seul fait demandé ?
│   │   ├── Oui → Route : SINGLE-STEP
│   │   └── Non → Plusieurs faits liés ?
│   │       ├── Oui → Route : MULTI-STEP (2 étapes)
│   │       └── Non → Route : MULTI-STEP (3+ étapes)
│   └── Non → Continuer l'analyse ↓
│
├── La requête demande-t-elle une comparaison ?
│   ├── Oui → Route : MULTI-STEP (décomposition + synthèse)
│   └── Non → Continuer l'analyse ↓
│
├── La requête demande-t-elle une analyse ou recommandation ?
│   ├── Oui → Route : MULTI-STEP (analyse contextuelle)
│   └── Non → Route : SINGLE-STEP (par défaut)
│
└── Fallback → Route : SINGLE-STEP

Optimisations avancées

1. Cache de classification

DEVELOPERpython
from functools import lru_cache
import hashlib

class CachedClassifier:
    """Classificateur avec cache pour les requêtes similaires."""

    def __init__(self):
        self.classifier = QueryClassifier()
        self.cache = {}

    def _normalize(self, query: str) -> str:
        """Normalise la requête pour le cache."""
        return query.lower().strip().rstrip("?!.")

    def classify(self, query: str) -> str:
        normalized = self._normalize(query)
        key = hashlib.md5(normalized.encode()).hexdigest()

        if key not in self.cache:
            self.cache[key] = self.classifier.classify(query)

        return self.cache[key]

2. Feedback loop pour améliorer le classificateur

DEVELOPERpython
class AdaptiveClassifierWithFeedback:
    """Classificateur qui s'améliore avec les retours utilisateur."""

    def __init__(self):
        self.classifier = TrainedQueryClassifier()
        self.feedback_buffer = []

    def record_feedback(
        self,
        query: str,
        predicted_route: str,
        was_helpful: bool,
        response_time_ms: int
    ):
        """Enregistre le feedback pour ré-entraînement."""
        self.feedback_buffer.append({
            "query": query,
            "predicted": predicted_route,
            "helpful": was_helpful,
            "latency": response_time_ms
        })

        # Ré-entraîner toutes les 100 feedbacks
        if len(self.feedback_buffer) >= 100:
            self._retrain()

    def _retrain(self):
        """Ré-entraîne le classificateur avec les feedbacks."""
        corrections = []
        for fb in self.feedback_buffer:
            if not fb["helpful"]:
                # Déduire la bonne route
                if fb["predicted"] == "simple" and not fb["helpful"]:
                    corrections.append((fb["query"], "factual"))
                elif fb["predicted"] == "factual" and not fb["helpful"]:
                    corrections.append((fb["query"], "complex"))

        if corrections:
            queries, labels = zip(*corrections)
            self.classifier.train(list(queries), list(labels))

        self.feedback_buffer = []

3. Combinaison avec CRAG

Le RAG adaptatif peut être combiné avec Corrective RAG pour une double couche de qualité.

DEVELOPERpython
async def adaptive_corrective_pipeline(question: str) -> str:
    """Pipeline combinant RAG adaptatif et correctif."""
    # Étape 1 : Classification
    classification = classifier.classify(question)

    if classification.complexity == "simple":
        return direct_llm(question)

    # Étape 2 : Retrieval adaptatif
    if classification.complexity == "factual":
        docs = await single_step_retrieve(question)
    else:
        sub_questions = decompose(question)
        docs = await multi_step_retrieve(sub_questions)

    # Étape 3 : Évaluation corrective (CRAG)
    relevance = evaluate_relevance(question, docs)

    if relevance == "incorrect":
        docs = await web_search_fallback(question)
    elif relevance == "ambiguous":
        web_docs = await web_search_fallback(question)
        docs = docs + web_docs

    # Étape 4 : Génération
    return generate_response(question, docs)

Métriques de suivi

FAQ

Comment déterminer les seuils du classificateur ?

Commencez avec les seuils par défaut (simple < 0.3, factuel 0.3-0.7, complexe > 0.7), puis ajustez en fonction de vos données. Analysez 500 requêtes manuellement, labellisez-les, et mesurez la précision du classificateur. Ajustez les seuils pour minimiser les erreurs les plus coûteuses (une question complexe routée vers "simple" est pire que l'inverse).

Le RAG adaptatif ajoute-t-il de la latence ?

Le classificateur lui-même ajoute 100-300ms (LLM) ou 10-50ms (modèle entraîné). C'est largement compensé par les économies sur les requêtes simples qui évitent le retrieval. En moyenne, le RAG adaptatif est plus rapide que le RAG fixe car 20-35% des requêtes sont traitées sans retrieval.

Comment gérer les requêtes ambiguës entre deux catégories ?

Deux stratégies : (1) en cas de doute, choisir la route la plus riche (préférer "factual" à "simple", "complex" à "factual") pour éviter les réponses incomplètes ; (2) implémenter un seuil de confiance et router vers la catégorie supérieure si la confiance est inférieure à 80%.

Le RAG adaptatif fonctionne-t-il pour les chatbots multi-tours ?

Oui, et c'est même encore plus pertinent. Dans un chat multi-tours, les messages de suivi ("Et pour le plan Enterprise ?") sont souvent simples et ne nécessitent qu'une modification du contexte précédent, pas un nouveau retrieval complet. Le classificateur peut prendre en compte l'historique de conversation pour optimiser le routing.

Peut-on ajouter plus de trois routes ?

Absolument. Des routes supplémentaires courantes incluent : "SQL query" (pour les requêtes sur des données structurées), "API call" (pour les actions en temps réel), "human escalation" (pour les sujets sensibles). Le principe reste le même : classifier puis router vers le handler optimal.

Conclusion

Le RAG adaptatif est une optimisation essentielle pour tout système RAG en production. En adaptant la stratégie de retrieval à la complexité de chaque requête, il offre simultanément de meilleures performances et des coûts réduits.

Points clés :

1. Classifier avant de retriever permet d'économiser du temps et de l'argent
2. Trois routes de base couvrent la majorité des cas d'usage
3. Un classificateur entraîné est plus rapide qu'un LLM pour le routing
4. Combiner avec CRAG offre une double couche de qualité
5. Le feedback loop améliore continuellement la précision du routing

Ailog utilise le routing adaptatif pour optimiser chaque interaction avec ses chatbots. Testez gratuitement et constatez la différence entre un RAG qui réfléchit et un RAG qui exécute aveuglément.

Ressources

• Paper Adaptive RAG - "Adaptive-RAG: Learning to Adapt Retrieval-Augmented Large Language Models through Question Complexity" (Jeong et al., 2024)
• LangGraph Documentation - Routing et workflows
• Query Routing RAG - Guide complémentaire sur le routing
• Corrective RAG - Auto-correction des résultats
• Retrieval Fundamentals - Les bases du retrieval