Adaptive RAG
Mis à jour le 29 juillet 2026
Router les questions intelligemment
Le Self-RAG applique le même traitement à toutes les questions, et c’est précisément son défaut. Demandez-lui combien font deux et deux : il encodera la requête, fouillera l’index, gradera quatre documents sans rapport, basculera sur le web, générera, vérifiera. Une dizaine d’appels LLM pour une question qu’un modèle résout instantanément. Toutes les questions n’ont pas la même complexité, et une question factuelle simple ne nécessite pas le même traitement qu’une question exigeant la synthèse de plusieurs documents. L’Adaptive RAG ajoute donc un routeur intelligent en amont du pipeline : il analyse la question et choisit la stratégie optimale avant d’engager le moindre traitement coûteux.
Le routeur dispose de trois destinations. Les questions portant sur les sujets réellement couverts par vos documents partent vers le vector store, c’est le chemin RAG complet. Les questions d’actualité ou hors du périmètre indexé partent vers la recherche web, puisque interroger l’index ne rendrait de toute façon que du bruit. Enfin les questions simples ou de culture générale reçoivent une réponse directe du modèle, sans aucun retrieval.
Implémenter le routeur
Le point délicat est que le routeur doit connaître le périmètre de votre index pour décider. La solution retenue est aussi simple qu’efficace : on lui passe la liste des sujets couverts, indexed_topics, qu’il compare à la question. Les trois règles sont énoncées explicitement dans le prompt, et le format JSON contraint garantit une valeur exploitable. Le result.get("route", "vectorstore") mérite l’attention : en cas de sortie inattendue, le défaut retombe sur le vector store, c’est-à-dire sur le comportement le plus sûr, celui qui s’appuie sur vos données plutôt que sur le web ou sur la mémoire du modèle.
import json
from mistralai import Mistral
client = Mistral(api_key=api_key)
def route_question(question, indexed_topics):
"""Router une question vers la meilleure stratégie."""
topics_str = ", ".join(indexed_topics)
prompt = f"""Vous êtes un routeur de questions. Étant donné une question,
déterminez la meilleure source d'information.
Vos documents indexés couvrent les sujets suivants : {topics_str}
Règles :
- Si la question concerne les sujets indexés -> "vectorstore"
- Si la question concerne l'actualité ou un sujet non couvert -> "websearch"
- Si la question est très simple ou de culture générale -> "direct"
Répondez UNIQUEMENT avec un objet JSON : {{"route": "vectorstore"}} ou {{"route": "websearch"}} ou {{"route": "direct"}}
Question : {question}
Route JSON :"""
response = client.chat.complete(
model="mistral-large-latest",
messages=[{"role": "user", "content": prompt}],
temperature=0,
response_format={"type": "json_object"},
)
result = json.loads(response.choices[0].message.content)
return result.get("route", "vectorstore")
# Tester le routeur
topics = ["embeddings", "RAG", "Mistral AI", "recherche vectorielle", "FAISS"]
tests = [
"Comment fonctionne mistral-embed ?", # -> vectorstore
"Quel est le cours de l'action Apple aujourd'hui ?", # -> websearch
"Combien font 2 + 2 ?", # -> direct
"Quelles sont les étapes du RAG ?", # -> vectorstore
"Qui a gagné la Coupe du Monde 2026 ?", # -> websearch
]
for q in tests:
route = route_question(q, topics)
print(f" [{route:12s}] {q}")Le jeu de tests illustre chaque cas de figure et vous servira de garde-fou. Une question sur mistral-embed tombe en plein dans les sujets déclarés, elle va au vector store. Le cours d’une action ou le résultat d’une compétition changent chaque jour : aucun index statique ne peut les contenir, direction le web. Deux plus deux ne demande ni corpus ni recherche. Reprenez ce bloc avec vos propres sujets et une quinzaine de questions représentatives de votre trafic réel : c’est la manière la plus économique de vérifier que la liste topics est assez précise, car un routeur qui envoie tout au vector store n’apporte rien, et un routeur qui répond en direct sur vos sujets métier est franchement dangereux.
Le pipeline complet
AdaptiveRAG rassemble tout ce que vous avez construit depuis trois leçons. Chaque route dispose de sa propre méthode d’exécution, et la méthode ask se lit comme un aiguillage. La route direct retourne immédiatement, avec une température de 0.3 puisqu’il n’y a pas de contexte à respecter, et une liste de sources vide qui signale honnêtement à l’utilisateur que rien ne fonde cette réponse. La route websearch génère à partir des seuls résultats web. La route vectorstore déroule en revanche l’artillerie complète : grading des documents à la manière du Corrective RAG, fallback web si aucun ne survit, puis boucle de génération avec les vérifications de fondement et d’utilité héritées du Self-RAG, regroupées ici dans grade_and_correct.
Une différence de seuil mérite d’être relevée par rapport à la leçon sur le Corrective RAG. Le fallback web ne se déclenche plus dès qu’un document est écarté, mais seulement si relevant_docs est vide. Ce choix est plus tolérant : quand deux chunks sur quatre restent pertinents, il est raisonnable de générer avec eux plutôt que d’aller chercher sur le web, ce qui économise un appel et réduit la latence. À vous d’arbitrer selon l’exigence de votre domaine.
import numpy as np
class AdaptiveRAG:
"""Pipeline Adaptive RAG avec routage intelligent."""
def __init__(self, chunks, embeddings, index, topics):
self.chunks = chunks
self.embeddings = embeddings
self.index = index
self.topics = topics
def route(self, question):
"""Déterminer la stratégie optimale."""
return route_question(question, self.topics)
def retrieve_from_vectorstore(self, question, k=4):
"""Retrieval depuis le vector store local."""
response = client.embeddings.create(
model="mistral-embed", inputs=[question]
)
query_vec = np.array([response.data[0].embedding], dtype=np.float32)
distances, indices = self.index.search(query_vec, k)
return [
{"text": self.chunks[idx]["text"], "source": "vectorstore"}
for dist, idx in zip(distances[0], indices[0]) if idx >= 0
]
def search_web(self, question):
"""Recherche web pour les questions hors périmètre."""
return web_search_fallback(question)
def direct_answer(self, question):
"""Réponse directe sans retrieval."""
response = client.chat.complete(
model="mistral-large-latest",
messages=[{"role": "user", "content": question}],
temperature=0.3,
)
return response.choices[0].message.content
def generate_with_context(self, question, documents):
"""Génération avec contexte (RAG classique)."""
context = "\n\n---\n\n".join([d["text"] for d in documents])
prompt = f"Contexte :\n{context}\n\nQuestion : {question}"
response = client.chat.complete(
model="mistral-large-latest",
messages=[
{"role": "system", "content": "Répondez en vous basant sur le contexte fourni."},
{"role": "user", "content": prompt},
],
temperature=0.1,
)
return response.choices[0].message.content
def grade_and_correct(self, question, documents, generation):
"""Vérifications Self-RAG (hallucinations + utilité)."""
is_grounded = check_hallucination(documents, generation)
if not is_grounded:
return None, "hallucination"
is_useful = check_answer_usefulness(question, generation)
if not is_useful:
return None, "not_useful"
return generation, "ok"
def ask(self, question, max_retries=2):
"""Pipeline Adaptive RAG complet."""
print(f"\n{'='*60}")
print(f"Question : {question}")
# 1. Routage
route = self.route(question)
print(f"Route choisie : {route}")
# 2. Exécuter selon la route
if route == "direct":
answer = self.direct_answer(question)
return {"answer": answer, "route": "direct", "sources": []}
elif route == "websearch":
docs = self.search_web(question)
answer = self.generate_with_context(question, docs)
return {"answer": answer, "route": "websearch", "sources": ["web"]}
else: # vectorstore
docs = self.retrieve_from_vectorstore(question)
# Grading des documents
relevant_docs = []
for doc in docs:
if grade_document(question, doc["text"]):
relevant_docs.append(doc)
# Fallback web si pas assez de documents pertinents
if not relevant_docs:
print(" Aucun document pertinent, fallback web...")
relevant_docs = self.search_web(question)
# Boucle de génération avec vérification
for attempt in range(max_retries + 1):
generation = self.generate_with_context(question, relevant_docs)
answer, status = self.grade_and_correct(
question, relevant_docs, generation
)
if status == "ok":
return {
"answer": answer,
"route": "vectorstore",
"attempts": attempt + 1,
"sources": [d.get("source", "") for d in relevant_docs],
}
elif status == "not_useful":
web_docs = self.search_web(question)
relevant_docs.extend(web_docs)
# Retourner la dernière génération même si imparfaite
return {
"answer": generation,
"route": "vectorstore",
"attempts": max_retries + 1,
"sources": [d.get("source", "") for d in relevant_docs],
}
# Utilisation
adaptive_rag = AdaptiveRAG(chunks, embeddings, index, topics=topics)
questions = [
"Comment créer un index FAISS avec mistral-embed ?",
"Quel temps fait-il à Paris demain ?",
"Combien de dimensions a codestral-embed ?",
]
for q in questions:
result = adaptive_rag.ask(q)
print(f"\nRoute: {result['route']}")
print(f"Réponse: {result['answer'][:200]}...")Les trois questions d’exemple font défiler l’ensemble des chemins en une exécution, et le champ route du résultat vous dit lequel a été emprunté. C’est cette valeur qu’il faut journaliser en production : la distribution des routes sur votre trafic réel est le meilleur indicateur de santé de votre système, révélant aussi bien un index trop étroit qu’un routeur mal calibré.
Points clés à retenir
- L’Adaptive RAG route chaque question vers la stratégie optimale (vectorstore, web, direct)
- Le routeur utilise un LLM avec structured output pour classifier les questions
- Les questions sur les sujets indexés vont au vector store, le reste au web
- Le routage réduit la latence et les coûts en évitant le RAG complet pour les questions simples
- L’Adaptive RAG combine les idées du Corrective RAG et du Self-RAG avec le routage