Agentic RAG: adding reasoning steps to retrieval

Agentic RAG extends retrieval-augmented generation by inserting an autonomous reasoning layer between the user query and the final answer. Instead of a single embedding lookup followed by generation, an agent plans retrieval steps, decomposes complex questions, evaluates intermediate results, and decides whether to search again, reformulate, or synthesize. This shifts RAG from a fixed pipeline to a dynamic control flow where the model drives its own information gathering.

How agentic RAG differs from standard RAG

Standard RAG follows a linear sequence: embed the query, retrieve top-k chunks, stuff them into a prompt, generate. The retrieval strategy is static — same embedding model, same index, same k — regardless of question complexity. If the top-k misses a critical document, the answer hallucinates or fails.

Agentic RAG introduces a planner that can:

• Rewrite or decompose the original query into sub-queries
• Choose different retrieval tools (vector search, keyword search, SQL, API calls)
• Inspect retrieved chunks for relevance and completeness
• Decide to re-retrieve with modified parameters
• Synthesize across multiple retrieval rounds before answering

The control flow becomes a loop rather than a pipeline. Each iteration produces evidence that the agent evaluates against the original intent.

Core components

Planner / controller

The planner is typically an LLM prompted with a system prompt that defines available tools and a reasoning framework (ReAct, plan-and-solve, or custom). It outputs structured actions: search, retrieve, evaluate, synthesize.

# Simplified planner prompt structure
PLANNER_SYSTEM = """
You are a research agent. Given a user question, you can:
1. search(query: str) -> List[Document]
2. evaluate(docs: List[Document], question: str) -> Assessment
3. synthesize(docs: List[Document], question: str) -> Answer

Return JSON: {"action": "search|evaluate|synthesize", "args": {...}}
"""

Retrieval tools

The agent needs access to multiple retrieval modalities. A minimal set:

from dataclasses import dataclass
from abc import ABC, abstractmethod
from typing import List

@dataclass
class Document:
    content: str
    metadata: dict
    score: float

class Retriever(ABC):
    @abstractmethod
    def retrieve(self, query: str, k: int = 10) -> List[Document]:
        pass

class VectorRetriever(Retriever):
    def __init__(self, index, embedder):
        self.index = index
        self.embedder = embedder
    
    def retrieve(self, query: str, k: int = 10) -> List[Document]:
        vec = self.embedder.embed(query)
        return self.index.search(vec, k)

class KeywordRetriever(Retriever):
    def __init__(self, bm25_index):
        self.bm25 = bm25_index
    
    def retrieve(self, query: str, k: int = 10) -> List[Document]:
        return self.bm25.search(query, k)

class HybridRetriever(Retriever):
    def __init__(self, retrievers: List[Retriever], weights: List[float]):
        self.retrievers = retrievers
        self.weights = weights
    
    def retrieve(self, query: str, k: int = 10) -> List[Document]:
        # Reciprocal rank fusion or weighted score merge
        all_results = []
        for r, w in zip(self.retrievers, self.weights):
            docs = r.retrieve(query, k * 2)
            for d in docs:
                d.score *= w
            all_results.extend(docs)
        # Deduplicate and rerank
        return self._merge_and_rerank(all_results, k)

Evaluation / critic

After each retrieval, the agent judges whether the evidence suffices. This prevents wasted rounds and catches gaps early.

EVALUATOR_SYSTEM = """
Assess if the retrieved documents answer the question.
Return JSON: {"sufficient": bool, "gaps": List[str], "next_queries": List[str]}
"""

Memory and state

The agent maintains a scratchpad of prior queries, retrieved documents, and evaluations. This avoids redundant searches and enables multi-hop reasoning.

@dataclass
class AgentState:
    original_question: str
    sub_questions: List[str]
    retrieved_docs: List[Document]
    evaluations: List[dict]
    iteration: int = 0
    max_iterations: int = 5

Why it matters

Single-pass retrieval fails on questions requiring:

• Multi-hop reasoning: “Which CEO founded the company that acquired the startup where the author of ‘Attention Is All You Need’ worked in 2018?” — needs chained lookups across entities.
• Comparative analysis: “Compare the latency claims in the 2023 and 2024 technical blogs for Model X” — requires retrieving two distinct documents and aligning claims.
• Negation or exclusion: “Find case studies of RAG implementations that do not use vector databases” — standard similarity search retrieves the opposite.
• Ambiguous or underspecified queries: “How do I optimize the thing?” — the agent must clarify or explore multiple interpretations.

Agentic RAG trades latency and token cost for correctness on these classes of problems. The overhead is real: each reasoning step consumes tokens and adds round-trips. But for high-value queries — legal research, technical troubleshooting, financial analysis — the cost is justified.

Concrete example: debugging a production incident

Consider an on-call engineer asking: “Why did the payment service latency spike at 3 AM yesterday?”

Standard RAG retrieves chunks mentioning “payment service,” “latency,” “spike,” “3 AM.” Likely misses the root cause because the relevant logs, config changes, and deployment records live in different systems with different vocabularies.

Agentic RAG executes:

Iteration 1: search("payment service latency spike 3 AM yesterday")
  → Retrieves generic monitoring docs, no specific incident
  → Evaluation: insufficient, gaps=["specific incident", "deployments", "config changes"]
  → Next queries: ["payment service deployment 3 AM yesterday", "config changes payment service last 24h", "datadog alerts payment service 3 AM"]

Iteration 2: search("payment service deployment 3 AM yesterday")
  → Retrieves deploy log: v2.4.1 rolled out 02:55 AM
  → search("config changes payment service last 24h")
  → Retrieves feature flag flip: "new_retry_policy" enabled 02:58 AM
  → Evaluation: sufficient evidence found
  → Synthesize: "Latency spike correlates with v2.4.1 deploy and new_retry_policy flag enable at ~3 AM. The retry policy increased tail latency under load. Recommend rollback or flag disable."

The agent discovered the causal chain by iterating across deployment logs, feature flag audit trails, and monitoring data — each requiring different query formulations.

Common misconceptions

“Agentic RAG just means using an LLM to rewrite queries”

Query rewriting is one tactic. The defining characteristic is autonomous control flow — the model decides whether to retrieve, what to retrieve, how many times, and when to stop. A fixed pipeline that always rewrites once then retrieves is not agentic.

“More iterations always improve quality”

Each iteration adds latency and error propagation risk. Poorly designed evaluators loop indefinitely or chase irrelevant tangents. Hard limits (max iterations, token budgets) and strong evaluation prompts are essential. Diminishing returns hit fast — most production systems cap at 3-5 rounds.

“You need a complex multi-agent framework”

A single LLM with a well-structured prompt, tool definitions, and a loop can implement agentic RAG. Frameworks (LangGraph, AutoGen, crewAI) help with orchestration, observability, and state management, but they’re not prerequisites. Start with a simple loop; extract to a framework when the logic warrants it.

“Agentic RAG replaces the need for good indexes”

Garbage in, garbage out still applies. If your vector index misses relevant documents, no amount of reasoning retrieves them. Agentic RAG mitigates index weaknesses by trying alternative queries and retrieval modes, but it cannot conjure documents that don’t exist in the corpus. Invest in hybrid search, metadata filtering, and chunking strategies first.

“It works out of the box on any domain”

The planner needs domain knowledge to generate useful sub-queries and evaluate evidence. A generic “you are a research agent” prompt fails on specialized domains (legal, biomedical, proprietary codebases). You must inject domain-specific retrieval heuristics, evaluation criteria, and synthesis templates into the system prompt or few-shot examples.

Implementation considerations

Latency management

Parallelize independent sub-queries. If the planner emits three search actions with no dependencies, run them concurrently.

import asyncio

async def execute_parallel_searches(queries: List[str], retriever: Retriever) -> List[List[Document]]:
    tasks = [asyncio.to_thread(retriever.retrieve, q) for q in queries]
    return await asyncio.gather(*tasks)

Token budgeting

Track cumulative tokens across iterations. Reserve budget for the final synthesis. Truncate or summarize older retrieved chunks when context fills.

def manage_context(docs: List[Document], max_tokens: int, tokenizer) -> List[Document]:
    total = sum(len(tokenizer.encode(d.content)) for d in docs)
    if total <= max_tokens:
        return docs
    # Keep highest-scored docs, summarize or drop rest
    docs.sort(key=lambda d: d.score, reverse=True)
    kept, running = [], 0
    for d in docs:
        t = len(tokenizer.encode(d.content))
        if running + t <= max_tokens:
            kept.append(d)
            running += t
        else:
            # Optionally summarize instead of dropping
            break
    return kept

Observability

Log every iteration: query, retrieved doc IDs, scores, evaluation output, decision. This is the only way to debug why the agent missed something or looped uselessly. Structure logs for querying:

{
  "trace_id": "abc-123",
  "iteration": 2,
  "action": "search",
  "query": "config changes payment service last 24h",
  "retrieved_count": 8,
  "top_scores": [0.92, 0.87, 0.81],
  "evaluation": {"sufficient": false, "gaps": ["deployment correlation"]},
  "next_action": "search"
}

Fallback behavior

When the agent exhausts iterations without sufficient evidence, fall back to a standard RAG pass over all accumulated documents, or escalate to human. Never return a confident hallucination.

def run_agentic_rag(question: str, agent: Agent, fallback_retriever: Retriever) -> Answer:
    state = AgentState(original_question=question)
    while state.iteration < state.max_iterations:
        action = agent.plan(state)
        if action.type == "synthesize":
            return agent.synthesize(state)
        elif action.type == "search":
            docs = execute_search(action.query)
            state.retrieved_docs.extend(docs)
            state.evaluations.append(agent.evaluate(docs, question))
            state.iteration += 1
        else:
            break
    # Fallback: best-effort synthesis from all gathered evidence
    return agent.synthesize(state) if state.retrieved_docs else fallback_answer(question)

When to use agentic RAG

Use it when:

• Questions routinely require multi-source synthesis
• Query intent is ambiguous or underspecified
• False negatives from single-pass retrieval are costly
• You have engineering capacity to build and maintain the control loop

Stick with standard RAG when:

• Queries are simple lookups (“What is our refund policy?”)
• Latency budget is tight (< 2s end-to-end)
• Corpus is narrow and well-indexed
• Team lacks bandwidth for prompt engineering and evaluation loops

Agentic RAG is not a universal upgrade — it’s a targeted tool for queries that break the single-pass assumption. Build the simple version first, measure where it fails, then add agency only where the ROI justifies the complexity.