Adaptive RAG: AI That Automatically Picks the Best Strategy

Why use a hammer for every screw? Fixed RAG applies the same retrieval strategy to every query, whether simple or complex. Adaptive RAG analyzes the question first, then automatically chooses the best approach: no retrieval at all, simple retrieval, or multi-step retrieval. Result: accuracy close to an always-multi-step strategy, but with far fewer search calls (1.08 steps on average versus 4.69), and reduced costs on simple questions.

TL;DR

• Adaptive RAG dynamically routes each query to the optimal strategy
• Three routes: no-retrieval (direct LLM), single-step retrieval, multi-step retrieval
• Classifier: lightweight LLM or trained model that analyzes query complexity
• Accuracy close to the always-multi-step approach with ~4x fewer retrieval steps — 1.08 versus 4.69 on average (Adaptive-RAG, Jeong et al., 2024)
• Fewer retrieval calls by avoiding unnecessary search on simple questions
• Compatible with CRAG, Self-RAG, and other advanced techniques

The Problem with Fixed RAG

Fixed RAG treats all queries the same way. It's wasteful.

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

"Hello!" ──────────────────▶ Retrieval + Generation  ❌ Unnecessary
"What time is it?" ────────▶ Retrieval + Generation  ❌ Unnecessary
"Your return policy?" ─────▶ Retrieval + Generation  ✅ Relevant
"Compare plans A and B
 then recommend based on
 my SMB profile" ──────────▶ Retrieval + Generation  ⚠️ Insufficient
                                                        (needs multi-step)

The Three Types of Queries

*Typical proportions observed on e-commerce and support chatbots.

Adaptive RAG Architecture

┌──────────────────────────────────────────────────────────────────────┐
│                     Adaptive RAG Pipeline                             │
├──────────────────────────────────────────────────────────────────────┤
│                                                                       │
│  User Query                                                           │
│         │                                                             │
│         ▼                                                             │
│  ┌─────────────────┐                                                 │
│  │  Query           │  Analysis: complexity, factual need,            │
│  │  Classifier      │  domain, expected response type                 │
│  └────────┬────────┘                                                 │
│           │                                                           │
│     ┌─────┼─────────────────┐                                        │
│     │     │                 │                                        │
│     ▼     ▼                 ▼                                        │
│  ┌──────┐ ┌──────────┐ ┌────────────────┐                           │
│  │ No   │ │ Single   │ │ Multi-step     │                           │
│  │Retr. │ │ Step     │ │ Retrieval      │                           │
│  │      │ │ Retrieval│ │                │                           │
│  │ LLM  │ │ Query→   │ │ Decomposition  │                           │
│  │direct│ │ Retrieve │ │ → Sub-queries  │                           │
│  │      │ │ → Generate│ │ → Multi-retrieve│                           │
│  │      │ │          │ │ → Synthesis    │                           │
│  └──┬───┘ └────┬─────┘ └───────┬────────┘                           │
│     │          │               │                                     │
│     └──────────┴───────────────┘                                     │
│                    │                                                  │
│                    ▼                                                  │
│             Final Response                                           │
│                                                                       │
└──────────────────────────────────────────────────────────────────────┘

The Query Classifier

The heart of adaptive RAG is the classifier that determines the optimal strategy for each query.

Approach 1: LLM Classifier (zero-shot)

The simplest method: use a lightweight LLM to classify.

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

class QueryClassification(BaseModel):
    """Query classification for RAG routing."""
    complexity: Literal["simple", "factual", "complex"] = Field(
        description="Query complexity level"
    )
    reasoning: str = Field(
        description="Classification explanation"
    )
    needs_retrieval: bool = Field(
        description="Does the query need documents?"
    )
    estimated_steps: int = Field(
        description="Number of retrieval steps needed",
        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"""Classify this query to determine the optimal retrieval strategy.

Query: "{query}"

Classification rules:
- "simple": greetings, general questions without need for specific docs
  → needs_retrieval = false, estimated_steps = 0
- "factual": direct question with a precise factual answer
  → needs_retrieval = true, estimated_steps = 1
- "complex": comparison, multi-aspect analysis, requires multiple sources
  → needs_retrieval = true, estimated_steps = 2-5"""

        return self.llm.invoke(prompt)

# Usage
classifier = QueryClassifier()

result = classifier.classify("Hello, how are you?")
# complexity="simple", needs_retrieval=False, estimated_steps=0

result = classifier.classify("What's the Enterprise plan price?")
# complexity="factual", needs_retrieval=True, estimated_steps=1

result = classifier.classify(
    "Compare the advantages of Pro and Enterprise plans "
    "for a 200-employee SMB in the healthcare sector"
)
# complexity="complex", needs_retrieval=True, estimated_steps=3

Approach 2: Trained Classifier (faster)

For production, a small trained model is faster and cheaper.

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

class TrainedQueryClassifier:
    """Trained classifier for RAG routing."""

    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):
        """Train the classifier on labeled examples."""
        embeddings = self.encoder.encode(queries)

        # Additional features
        features = []
        for q, emb in zip(queries, embeddings):
            extra = [
                len(q),                          # Length
                q.count("?"),                     # Number of questions
                len(q.split()),                   # Word count
                int(any(w in q.lower() for w in  # Comparison words
                    ["compare", "versus", "difference", "advantage"])),
                int(any(w in q.lower() for w in  # Multi-step words
                    ["then", "also", "additionally", "furthermore"])),
            ]
            features.append(np.concatenate([emb, extra]))

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

    def predict(self, query: str) -> str:
        """Predict the category of a query."""
        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", "difference", "advantage"])),
            int(any(w in query.lower() for w in
                ["then", "also", "additionally", "furthermore"])),
        ]
        X = np.array([np.concatenate([emb, extra])])
        return self.classifier.predict(X)[0]

The Adaptive Router

The router connects the classifier to the different retrieval strategies.

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 (direct LLM)
def direct_llm_response(state: AdaptiveRAGState) -> AdaptiveRAGState:
    """Respond directly without 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:
    """Standard single-step retrieval."""
    docs = await retriever.aget_relevant_documents(state["question"])

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

    response = llm.invoke(
        f"Context:\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:
    """Decompose a complex question into sub-questions."""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

    prompt = f"""Decompose this complex question into simple,
independent sub-questions that, once answered, will allow
answering the main question.

Question: {state["question"]}

Return a JSON list of sub-questions (max 4).
Example: ["Sub-question 1?", "Sub-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:
    """Retrieve documents for each sub-question."""
    all_docs = []

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

    # Deduplication
    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:
    """Synthesize a response from multiple sources."""
    llm = ChatOpenAI(model="gpt-4o", temperature=0.1)
    context = "\n\n".join([d.page_content for d in state["documents"]])

    prompt = f"""Answer this complex question by synthesizing
information from multiple sources.

Main question: {state["question"]}
Sub-questions addressed: {state["sub_questions"]}

Context (multiple sources):
{context}

Provide a structured and complete answer."""

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

# Build adaptive workflow
def classify_and_route(state: AdaptiveRAGState) -> str:
    """Classify and route to the right strategy."""
    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()

Comparison: Fixed RAG vs Adaptive RAG

Benchmarks (Jeong et al., 2024)

Across the 6 QA datasets evaluated in the paper (SQuAD, Natural Questions, TriviaQA, MuSiQue, HotpotQA, 2WikiMultiHopQA), with FLAN-T5-XL (3B), the averaged results are:

Key takeaway: Adaptive-RAG nearly matches the accuracy of the always-multi-step approach (46.9 vs 48.9 F1) while using only 1.08 retrieval steps on average versus 4.69 — cutting search cost by more than 4x. It clearly outperforms single-step (44.3 F1) and almost doubles the accuracy of no-retrieval (21.1 F1).

Detailed Decision Tree

Incoming query
│
├── Contains greeting/politeness words?
│   ├── Yes → Is it ONLY a greeting?
│   │   ├── Yes → Route: NO RETRIEVAL
│   │   └── No → Continue analysis ↓
│   └── No → Continue analysis ↓
│
├── Does the query concern a specific fact?
│   ├── Yes → Only one fact requested?
│   │   ├── Yes → Route: SINGLE-STEP
│   │   └── No → Multiple related facts?
│   │       ├── Yes → Route: MULTI-STEP (2 steps)
│   │       └── No → Route: MULTI-STEP (3+ steps)
│   └── No → Continue analysis ↓
│
├── Does the query ask for a comparison?
│   ├── Yes → Route: MULTI-STEP (decomposition + synthesis)
│   └── No → Continue analysis ↓
│
├── Does the query ask for analysis or recommendation?
│   ├── Yes → Route: MULTI-STEP (contextual analysis)
│   └── No → Route: SINGLE-STEP (default)
│
└── Fallback → Route: SINGLE-STEP

Advanced Optimizations

1. Classification Cache

DEVELOPERpython
from functools import lru_cache
import hashlib

class CachedClassifier:
    """Classifier with cache for similar queries."""

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

    def _normalize(self, query: str) -> str:
        """Normalize the query for caching."""
        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 to Improve the Classifier

DEVELOPERpython
class AdaptiveClassifierWithFeedback:
    """Classifier that improves with user feedback."""

    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
    ):
        """Record feedback for retraining."""
        self.feedback_buffer.append({
            "query": query,
            "predicted": predicted_route,
            "helpful": was_helpful,
            "latency": response_time_ms
        })

        # Retrain every 100 feedbacks
        if len(self.feedback_buffer) >= 100:
            self._retrain()

    def _retrain(self):
        """Retrain the classifier with feedback."""
        corrections = []
        for fb in self.feedback_buffer:
            if not fb["helpful"]:
                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. Combination with CRAG

Adaptive RAG can be combined with Corrective RAG for a double quality layer.

DEVELOPERpython
async def adaptive_corrective_pipeline(question: str) -> str:
    """Pipeline combining adaptive and corrective RAG."""
    # Step 1: Classification
    classification = classifier.classify(question)

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

    # Step 2: Adaptive retrieval
    if classification.complexity == "factual":
        docs = await single_step_retrieve(question)
    else:
        sub_questions = decompose(question)
        docs = await multi_step_retrieve(sub_questions)

    # Step 3: Corrective evaluation (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

    # Step 4: Generation
    return generate_response(question, docs)

Monitoring Metrics

FAQ

How do you determine the classifier thresholds?

Start with default thresholds (simple < 0.3, factual 0.3-0.7, complex > 0.7), then adjust based on your data. Manually analyze 500 queries, label them, and measure classifier accuracy. Adjust thresholds to minimize the most costly errors (a complex question routed to "simple" is worse than the reverse).

Does Adaptive RAG add latency?

The classifier itself adds 100-300ms (LLM) or 10-50ms (trained model). This is more than offset by savings on simple queries that avoid retrieval. On average, Adaptive RAG is faster than Fixed RAG because 20-35% of queries are processed without retrieval.

How do you handle queries that are ambiguous between two categories?

Two strategies: (1) when in doubt, choose the richer route (prefer "factual" over "simple", "complex" over "factual") to avoid incomplete answers; (2) implement a confidence threshold and route to the higher category if confidence is below 80%.

Does Adaptive RAG work for multi-turn chatbots?

Yes, and it's even more relevant. In multi-turn chat, follow-up messages ("And for the Enterprise plan?") are often simple and only need a modification of the previous context, not a full new retrieval. The classifier can take conversation history into account to optimize routing.

Can you add more than three routes?

Absolutely. Common additional routes include: "SQL query" (for structured data queries), "API call" (for real-time actions), "human escalation" (for sensitive topics). The principle remains the same: classify then route to the optimal handler.

Conclusion

Adaptive RAG is an essential optimization for any production RAG system. By adapting the retrieval strategy to each query's complexity, it simultaneously delivers better performance and reduced costs.

Key takeaways:

1. Classify before retrieving saves time and money
2. Three base routes cover the majority of use cases
3. A trained classifier is faster than an LLM for routing
4. Combining with CRAG provides a double quality layer
5. The feedback loop continuously improves routing accuracy

Ailog uses adaptive routing to optimize every chatbot interaction. Try it free and see the difference between a RAG that thinks and a RAG that blindly executes.

Resources

• Adaptive RAG Paper - "Adaptive-RAG: Learning to Adapt Retrieval-Augmented Large Language Models through Question Complexity" (Jeong et al., 2024)
• LangGraph Documentation - Routing and workflows
• Query Routing RAG - Complementary routing guide
• Corrective RAG - Self-correcting results
• Retrieval Fundamentals - Retrieval basics