Adaptive RAG
Route questions intelligently
Self-RAG applies the same treatment to every question, and that is precisely its flaw. Ask it what two plus two equals: it will encode the query, scour the index, grade four irrelevant documents, switch to the web, generate, and verify. That is a dozen LLM calls for a question a model solves instantly. Not all questions have the same complexity, and a simple factual question does not require the same treatment as one demanding synthesis across multiple documents. Adaptive RAG therefore adds an intelligent router upstream of the pipeline: it analyzes the question and selects the optimal strategy before engaging in any costly processing.
The router has three destinations. Questions about topics actually covered by your documents go to the vector store; this is the full RAG path. Questions about current events or outside the indexed scope go to web search, since querying the index would only return noise. Finally, simple or general knowledge questions receive a direct answer from the model, with no retrieval at all.
Implement the router
The tricky part is that the router must know the scope of your index to make a decision. The chosen solution is as simple as it is effective: pass it the list of covered topics, indexed_topics, which it compares against the question. The three rules are stated explicitly in the prompt, and the constrained JSON format guarantees a usable value. The result.get("route", "vectorstore") deserves attention: in case of unexpected output, the default falls back to the vector store. This is the safest behavior, relying on your data rather than the web or the model’s memory.
import json
from mistralai.client import Mistral
client = Mistral(api_key=api_key)
def route_question(question, indexed_topics):
"""Route a question to the best strategy."""
topics_str = ", ".join(indexed_topics)
prompt = f"""You are a question router. Given a question,
determine the best source of information.
Your indexed documents cover the following topics: {topics_str}
Rules:
- If the question concerns indexed topics -> "vectorstore"
- If the question concerns current events or an uncovered topic -> "websearch"
- If the question is very simple or general knowledge -> "direct"
Respond ONLY with a JSON object: {{"route": "vectorstore"}} or {{"route": "websearch"}} or {{"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")
# Test the router
topics = ["embeddings", "RAG", "Mistral AI", "vector search", "FAISS"]
tests = [
"How does mistral-embed work?", # -> vectorstore
"What is Apple's stock price today?", # -> websearch
"How much is 2 + 2?", # -> direct
"What are the steps of RAG?", # -> vectorstore
"Who won the 2026 World Cup?", # -> websearch
]
for q in tests:
route = route_question(q, topics)
print(f" [{route:12s}] {q}")The test set illustrates each scenario and will serve as your safeguard. A question about mistral-embed falls squarely within the declared topics, so it goes to the vector store. Stock prices or competition results change daily: no static index can contain them, so they go to the web. Two plus two requires neither a corpus nor search. Reuse this block with your own topics and about fifteen questions representative of your actual traffic: this is the most economical way to verify that the topics list is accurate enough. A router that sends everything to the vector store adds no value, and a router that answers directly on your business topics is downright dangerous.
The complete pipeline
AdaptiveRAG brings together everything you have built over the last three lessons. Each route has its own execution method, and the ask method reads like a switch statement. The direct route returns immediately, with a temperature of 0.3 since there is no context to adhere to, and an empty sources list that honestly signals to the user that nothing underpins this answer. The websearch route generates from web results alone. The vectorstore route, however, deploys the full artillery: document grading in the style of Corrective RAG, web fallback if none survive, then a generation loop with grounding and usefulness checks inherited from Self-RAG, grouped here in grade_and_correct.
One threshold difference is worth noting compared to the Corrective RAG lesson. The web fallback no longer triggers as soon as a document is discarded, but only if relevant_docs is empty. This choice is more tolerant: when two out of four chunks remain relevant, it is reasonable to generate with them rather than go to the web, which saves a call and reduces latency. You must arbitrate based on your domain’s requirements.
import numpy as np
class AdaptiveRAG:
"""Adaptive RAG pipeline with intelligent routing."""
def __init__(self, chunks, embeddings, index, topics):
self.chunks = chunks
self.embeddings = embeddings
self.index = index
self.topics = topics
def route(self, question):
"""Determine the optimal strategy."""
return route_question(question, self.topics)
def retrieve_from_vectorstore(self, question, k=4):
"""Retrieval from the local vector store."""
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):
"""Web search for out-of-scope questions."""
return web_search_fallback(question)
def direct_answer(self, question):
"""Direct answer without 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):
"""Generation with context (classic RAG)."""
context = "\n\n---\n\n".join([d["text"] for d in documents])
prompt = f"Context:\n{context}\n\nQuestion: {question}"
response = client.chat.complete(
model="mistral-large-latest",
messages=[
{"role": "system", "content": "Answer based on the provided context."},
{"role": "user", "content": prompt},
],
temperature=0.1,
)
return response.choices[0].message.content
def grade_and_correct(self, question, documents, generation):
"""Self-RAG checks (hallucinations + usefulness)."""
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):
"""Complete Adaptive RAG pipeline."""
print(f"\n{'='*60}")
print(f"Question: {question}")
# 1. Routing
route = self.route(question)
print(f"Chosen route: {route}")
# 2. Execute based on 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)
# Document grading
relevant_docs = []
for doc in docs:
if grade_document(question, doc["text"]):
relevant_docs.append(doc)
# Web fallback if not enough relevant documents
if not relevant_docs:
print(" No relevant documents, web fallback...")
relevant_docs = self.search_web(question)
# Generation loop with verification
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)
# Return the last generation even if imperfect
return {
"answer": generation,
"route": "vectorstore",
"attempts": max_retries + 1,
"sources": [d.get("source", "") for d in relevant_docs],
}
# Usage
adaptive_rag = AdaptiveRAG(chunks, embeddings, index, topics=topics)
questions = [
"How do I create a FAISS index with mistral-embed?",
"What is the weather in Paris tomorrow?",
"How many dimensions does codestral-embed have?",
]
for q in questions:
result = adaptive_rag.ask(q)
print(f"\nRoute: {result['route']}")
print(f"Answer: {result['answer'][:200]}...")The three example questions walk through all paths in a single run, and the route field in the result tells you which one was taken. This is the value you should log in production: the distribution of routes across your real traffic is the best health indicator for your system, revealing both an index that is too narrow and a poorly calibrated router.
Key takeaways
- Adaptive RAG routes each question to the optimal strategy (vectorstore, web, direct)
- The router uses an LLM with structured output to classify questions
- Questions about indexed topics go to the vector store; the rest go to the web
- Routing reduces latency and costs by avoiding full RAG for simple questions
- Adaptive RAG combines ideas from Corrective RAG and Self-RAG with routing