Adaptive RAG
Router les questions intelligemment
Toutes les questions n’ont pas la même complexité. Une question factuelle simple ne nécessite pas le même traitement qu’une question nécessitant une synthèse de plusieurs documents. L’Adaptive RAG ajoute un routeur intelligent en amont du pipeline qui analyse la question et choisit la stratégie optimale.
Le principe du routage
Le routeur classifie chaque question entrante et la dirige vers le chemin approprié :
- Vers le vector store : questions liées aux documents indexés
- Vers la recherche web : questions d’actualité ou hors périmètre des documents
- Réponse directe : questions simples que le LLM peut résoudre seul
Implémenter le routeur
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 etes un routeur de questions. Etant donne une question,
determinez la meilleure source d'information.
Vos documents indexés couvrent les sujets suivants : {topics_str}
Regles :
- Si la question concerne les sujets indexes -> "vectorstore"
- Si la question concerne l'actualite ou un sujet non couvert -> "websearch"
- Si la question est tres simple ou de culture generale -> "direct"
Repondez 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}")Pipeline Adaptive RAG complet
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": "Repondez 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]}...")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