Agentic RAG: Adaptive and Corrective Retrieval
Source: mortalapps.com- Agentic RAG is an advanced retrieval paradigm that replaces static, single-shot vector search with dynamic, multi-step agentic reasoning, query routing, and self-correction loops.
- It solves the problem of retrieval failure, where irrelevant, incomplete, or hallucinated context is passed directly to the generation model.
- In production, it ensures high-fidelity context assembly by dynamically routing queries based on complexity and validating retrieved documents before generation.
- Implementing this pattern reduces downstream hallucination rates and prevents system failures when handling out-of-domain queries.
- The primary tradeoff is increased latency and token cost due to multiple LLM evaluation steps prior to final generation.
Why This Matters
Traditional Retrieval-Augmented Generation (RAG) relies on a fragile, single-shot assumption: that a single vector database query will return the exact context required to answer any prompt. In production, this assumption fails. Queries are often ambiguous, multi-faceted, or completely out-of-domain, leading to irrelevant context retrieval and downstream hallucinations. Agentic RAG solves this structural limitation by transforming retrieval from a static pipeline into an active, closed-loop decision process. This approach was invented because production systems require deterministic quality guarantees that simple vector search cannot provide.
If you ignore agentic RAG patterns in complex enterprise environments, your system will suffer from silent retrieval failures. The LLM will confidently generate incorrect answers based on noisy, irrelevant, or incomplete context, leading to severe compliance and operational risks. By contrast, implementing adaptive routing and corrective evaluation loops allows the system to assess query complexity upfront, evaluate the relevance of retrieved documents, and trigger alternative search strategies (such as web search or multi-hop knowledge graph queries) when vector search fails.
You should use agentic RAG when building systems that query heterogeneous data sources, handle highly variable user intent, or operate in high-stakes domains like finance, legal, or healthcare where accuracy is non-negotiable. For simple, single-domain Q&A over homogeneous document sets, standard hybrid search is a more cost-effective and lower-latency alternative.
Core Concepts
To build and operate an Agentic RAG system, you must understand the following core concepts that govern its decision-making loops:
- Adaptive Routing: The process of dynamically selecting the optimal retrieval strategy or data source based on an initial analysis of the user's query.
- Corrective Retrieval: An iterative loop that evaluates the relevance of retrieved documents and triggers corrective actions (such as query reformulation or external search) if the initial results are insufficient.
- Document Grading: A classification step where retrieved chunks are evaluated (usually by a lightweight LLM or cross-encoder) to determine if they contain information relevant to the query.
- Query Reformulation: The algorithmic rewriting of a user query to improve its retrieval performance, often by resolving ambiguities, expanding acronyms, or breaking down multi-part questions.
- Fallback Search: An alternative retrieval mechanism (e.g., web search, SQL database, or knowledge graph) triggered when primary vector databases yield low-confidence results.
Comparison of RAG Paradigms
| Strategy | Latency Overhead | Implementation Complexity | Best For |
|---|---|---|---|
| Standard RAG | Low (No intermediate steps) | Low | Homogeneous data, simple queries |
| Adaptive RAG | Medium (1 routing step) | Medium | Heterogeneous data sources, variable query complexity |
| Corrective RAG (CRAG) | Medium (grading + rewrite) | High | High-accuracy domains, noisy vector spaces |
| Self-RAG | High (multiple reflection steps) | Very High | Creative generation, multi-hop reasoning |
How It Works
The internal lifecycle of an Agentic RAG system replaces the linear "Retrieve -> Synthesize" pipeline with a stateful, iterative graph. The workflow consists of five distinct phases.
Phase 1: Query Analysis and Adaptive Routing
The lifecycle begins when a user query enters the system. Instead of directly querying a vector database, an adaptive router (typically a fine-tuned classifier or a structured LLM call) analyzes the query's intent, complexity, and domain. The router decides whether the query requires simple retrieval, complex multi-hop reasoning, external web search, or no retrieval at all.
Phase 2: Multi-Source Retrieval
Based on the router's decision, the system dispatches queries to the selected data stores. This could involve querying a local vector database, executing a hybrid BM25 search, or querying a structured SQL database. If the query is complex, it may be decomposed into sub-queries executed in parallel.
Phase 3: Corrective Document Grading
Once documents are retrieved, they are not passed directly to the generator. A document grader evaluates each retrieved chunk against the query. The grader outputs a binary or scalar score representing relevance. If a document is graded as relevant, it is kept in the context pool. If it is graded as irrelevant, it is discarded.
Phase 4: Fallback and Query Reformulation
If the context pool is empty or falls below a predefined relevance threshold, the corrective loop is triggered. The system reformulates the query to resolve ambiguities and executes a fallback search (e.g., Google Search API, Bing Web Search, or a broader knowledge graph query). The newly retrieved documents are graded and added to the context pool.
Phase 5: Synthesis and Generation
The finalized, graded context pool is assembled into a structured prompt. The generator LLM synthesizes the final answer. If the generator supports self-reflection, it evaluates its own output against the retrieved context to ensure no hallucinations were introduced during synthesis.
Architecture
The architecture of an Agentic RAG system is structured as a state machine or directed acyclic graph (DAG) managed by an orchestrator. The primary components include the Query Router, the Retrieval Engine, the Document Grader, the Query Reformulator, the Fallback Search Engine, and the Response Generator.
Execution starts at the Query Router, which receives the raw user query. The router outputs a routing decision, sending the query along one of several paths to the Retrieval Engine (which interfaces with Vector DBs or SQL databases). The retrieved documents flow into the Document Grader.
The Document Grader evaluates the documents and updates the global state with a list of validated context chunks and a relevance status. If the status is "insufficient", the state transitions to the Query Reformulator, which rewrites the query and sends it to the Fallback Search Engine (e.g., an external search API). The newly retrieved documents are graded and merged into the context pool.
Finally, the execution path leads to the Response Generator, which consumes the validated context pool and the original query to produce the final output. The state machine terminates here, returning the response to the client.
Decision Framework: When to Deploy Agentic RAG
Engineers must balance the trade-offs between retrieval accuracy, latency, and cost. Standard RAG is sufficient when the document corpus is clean, static, and homogeneous, and when user queries map directly to single document chunks. Agentic RAG is required when:
- The query space is highly heterogeneous (e.g., some queries need database lookups, others need semantic search, others need external web search).
- Document quality is highly variable or contains contradictory information.
- The cost of a hallucination or an incorrect answer is extremely high (e.g., medical dosage guidelines, financial compliance).
Adaptive Routing Strategies: Semantic vs. LLM-Based
Routing can be implemented using different mechanisms, each with distinct trade-offs:
- LLM-Based Routing (Structured Outputs): Using tools like Pydantic or instructor to force an LLM to output a JSON object containing the target route. This is highly flexible but adds 100-300ms of latency.
- Semantic Router (Vector-Based): Comparing the user query embedding against pre-embedded "route prototypes" using cosine similarity. This is extremely fast (<20ms) and cost-effective, but struggles with complex, multi-intent queries.
- Zero-Shot Classifiers: Using smaller, specialized models (e.g., DeBERTa) to classify the query into predefined categories.
Designing the Corrective Loop: Document Grading and Fallback
The core of Corrective RAG (CRAG) is the grading step. A naive implementation uses a prompt asking "Is this document relevant to the query? Yes or No." In production, this is unreliable.
A robust grader uses structured outputs to evaluate three dimensions:
- Semantic Relevance: Does the document address the core concepts of the query?
- Factuality/Authority: Is the document from a trusted source, and is it free of obvious formatting errors?
- Completeness: Does the document contain enough information to construct a full answer?
If the cumulative score of the retrieved documents falls below a configurable threshold (e.g., 0.7 out of 1.0), the system must execute a fallback. The fallback query should be stripped of conversational filler and expanded with synonyms or temporal context (e.g., appending the current year if looking for recent events).
Mitigating Latency and Token Overhead
Because Agentic RAG introduces multiple LLM calls (routing, grading, reformulation, generation) before returning a response, it can easily exceed acceptable SLA limits. To mitigate this:
- Parallelize Grading: Grade all retrieved documents concurrently using asynchronous LLM calls or a batch inference endpoint.
- Speculative Decoding and Small Models: Use smaller, highly optimized models (e.g., 8B parameters) for routing and grading, reserving the larger frontier model (e.g., 70B+ parameters) solely for final synthesis.
- Caching: Implement semantic caching on the router and document grader to bypass LLM calls for common or identical queries.
Code Example
import os
import asyncio
import logging
from typing import List, Literal
from pydantic import BaseModel, Field
from openai import AsyncOpenAI
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("AgenticRAG")
# Initialize client using environment variables
api_key = os.environ.get("OPENAI_API_KEY")
if not api_key:
raise ValueError("OPENAI_API_KEY environment variable is required")
client = AsyncOpenAI(api_key=api_key)
# Define schemas for structured outputs
class RoutingDecision(BaseModel):
route: Literal["vector_db", "web_search", "direct_response"] = Field(
description="The target data source based on query complexity and domain."
)
reasoning: str = Field(description="Explanation for the routing decision.")
class DocumentGrade(BaseModel):
is_relevant: bool = Field(
description="True if the document contains information directly relevant to the query, False otherwise."
)
confidence: float = Field(description="Confidence score between 0.0 and 1.0.")
# Mock retrieval functions for demonstration
async def mock_vector_db_retrieve(query: str) -> List[str]:
logger.info(f"Retrieving from Vector DB for query: {query}")
await asyncio.sleep(0.1) # Simulate network latency
return [
"The company's Q4 revenue grew by 15% year-over-year, driven by cloud services.",
"The coffee machine in the breakroom is currently broken and awaiting parts."
]
async def mock_web_search(query: str) -> List[str]:
logger.info(f"Executing fallback web search for query: {query}")
await asyncio.sleep(0.2) # Simulate web search latency
return [
"Latest news: The company announced a new CEO on January 15th, 2026."
]
# Core Agentic RAG Pipeline
class AgenticRAGPipeline:
def __init__(self, model_name: str = "gpt-4o-mini"):
self.model_name = model_name
async def route_query(self, query: str) -> RoutingDecision:
try:
response = await client.beta.chat.completions.parse(
model=self.model_name,
messages=[
{"role": "system", "content": "You are an expert query router. Analyze the query and route it to the most appropriate data source."},
{"role": "user", "content": query}
],
response_format=RoutingDecision,
temperature=0.0
)
return response.choices[0].message.parsed
except Exception as e:
logger.error(f"Routing failed: {e}. Defaulting to vector_db.")
return RoutingDecision(route="vector_db", reasoning="Fallback due to error.")
async def grade_document(self, query: str, doc: str) -> DocumentGrade:
try:
response = await client.beta.chat.completions.parse(
model=self.model_name,
messages=[
{"role": "system", "content": "Grade the relevance of the document to the user query. Be strict."},
{"role": "user", "content": f"Query: {query}
Document: {doc}"}
],
response_format=DocumentGrade,
temperature=0.0
)
return response.choices[0].message.parsed
except Exception as e:
logger.error(f"Grading failed: {e}. Defaulting to irrelevant.")
return DocumentGrade(is_relevant=False, confidence=0.0)
async def run(self, query: str) -> str:
# Step 1: Route the query
decision = await self.route_query(query)
logger.info(f"Route Decision: {decision.route} | Reason: {decision.reasoning}")
if decision.route == "direct_response":
return "This query does not require external context. How else can I help you?"
context_pool = []
# Step 2: Retrieve based on route
if decision.route == "vector_db":
docs = await mock_vector_db_retrieve(query)
# Step 3: Grade documents in parallel
grading_tasks = [self.grade_document(query, doc) for doc in docs]
grades = await asyncio.gather(*grading_tasks)
for doc, grade in zip(docs, grades):
if grade.is_relevant:
context_pool.append(doc)
logger.info(f"Document kept (Confidence: {grade.confidence})")
else:
logger.info("Document discarded as irrelevant.")
# Step 4: Corrective action if context is insufficient
if not context_pool or decision.route == "web_search":
logger.warning("No relevant context found. Triggering corrective fallback search.")
fallback_docs = await mock_web_search(query)
context_pool.extend(fallback_docs)
# Step 5: Final Generation
context_str = "
".join(context_pool)
try:
response = await client.chat.completions.create(
model=self.model_name,
messages=[
{"role": "system", "content": "Answer the query using ONLY the provided context. If the context is empty, state that you do not know."},
{"role": "user", "content": f"Context:
{context_str}
Query: {query}"}
],
temperature=0.0
)
return response.choices[0].message.content
except Exception as e:
logger.error(f"Generation failed: {e}")
return "An error occurred during response generation."
# Execution block
async def main():
pipeline = AgenticRAGPipeline()
query = "What was the company's revenue growth in Q4?"
result = await pipeline.run(query)
print(f"
Final Result:
{result}")
if __name__ == "__main__":
asyncio.run(main())INFO:AgenticRAG:Route Decision: vector_db | Reason: The query asks about company revenue growth, which is typically stored in internal financial documents. INFO:AgenticRAG:Retrieving from Vector DB for query: What was the company's revenue growth in Q4? INFO:AgenticRAG:Document kept (Confidence: 1.0) INFO:AgenticRAG:Document discarded as irrelevant. Final Result: The company's Q4 revenue grew by 15% year-over-year.