Agentic RAG in Practice: 5 Core Patterns for Building Autonomous Retrieval Agents

Traditional RAG Is No Longer Enough

You carefully built a RAG system, and a user asks, "Which is better for my use case: RAG or fine-tuning?" The system returns a pile of basic RAG introductions and fine-tuning definitions — completely missing the point. This isn't a retrieval accuracy problem; it's that traditional RAG has no reasoning capability. It can only "search what you ask," and when faced with complex questions requiring multi-step reasoning, cross-source synthesis, or self-correction, it simply stalls.

In 2026, Agentic RAG has become the breakthrough. Instead of a passive "retrieve-then-generate" pipeline, it empowers an Agent to autonomously decide: Should I retrieve? Where should I look? Are the results good enough? Should I try a different strategy? This "autonomous decision-making + iterative refinement" paradigm is the correct approach for complex questions.

Core Concepts at a Glance

Concept	Description	Typical Implementation
Agentic RAG	A RAG paradigm where the Agent autonomously decides retrieval strategy	LangGraph StateGraph, LlamaIndex Agent
Autonomous Retrieval	Agent decides whether to retrieve and what to retrieve based on the question	ReAct loop, tool calling
Multi-hop Reasoning	Decomposing complex questions into sub-questions for step-by-step retrieval	Question decomposition, chain retrieval
Self-reflection	Agent evaluates retrieval quality and decides whether to re-search	Document grading, query rewriting
Tool Calling	Agent invokes external tools (search, calculator, database) to augment capabilities	Function Calling, Tool Use
LangGraph	Framework for building stateful Agent workflows	StateGraph, conditional edges, checkpoints
ReAct Pattern	Alternating between Reasoning and Acting	Thought-Action-Observation loop

Problem Analysis: 5 Limitations of Traditional RAG

#	Limitation	Manifestation	Impact
1	Single Retrieval	Only one vector search, no iterative optimization	Incomplete retrieval, missing key information
2	No Reasoning	Cannot decompose complex questions into sub-questions	Poor quality on multi-hop questions
3	No Self-correction	Cannot identify irrelevant documents and retry	Severe hallucinations, off-topic answers
4	No Planning	No retrieval strategy, cannot decide search order	Inefficient, wasted resources
5	Cannot Handle Multi-hop	"Which is better for scenario C: A or B?" — completely unsolvable	Terrible UX, high abandonment rate

The fundamental problem with traditional RAG: it's a stateless pipeline, not an intelligent Agent. The core shift in Agentic RAG is from "pipeline" to "Agent" — giving the retrieval process autonomous decision-making, iterative refinement, and self-correction capabilities.

Pattern 1: Router Agent — Intelligent Dispatch

The simplest Agentic RAG pattern: route queries to different knowledge bases or retrieval strategies based on question type. Much more precise than unified vector search.

from enum import Enum
from typing import Optional


class QueryType(Enum):
    TECHNICAL = "technical"
    BUSINESS = "business"
    COMPARISON = "comparison"
    TUTORIAL = "tutorial"


class RouterAgent:
    def __init__(self, llm, retrievers: dict[str, object]):
        self.llm = llm
        self.retrievers = retrievers

    def classify(self, question: str) -> QueryType:
        prompt = f"""Classify this question into one of:
- technical: code/API/implementation questions
- business: pricing/ROI/strategy questions
- comparison: comparing two or more options
- tutorial: how-to/step-by-step guides

Question: {question}
Type:"""
        result = self.llm.invoke(prompt).strip().lower()
        return QueryType(result)

    def route(self, question: str) -> list[dict]:
        query_type = self.classify(question)
        retriever = self.retrievers.get(query_type.value)
        if not retriever:
            retriever = self.retrievers["technical"]
        docs = retriever.similarity_search(question, k=5)
        return [{"content": d.page_content, "source": d.metadata.get("source", "")} for d in docs]


retrievers = {
    "technical": tech_vector_store,
    "business": biz_vector_store,
    "comparison": compare_vector_store,
    "tutorial": tutorial_vector_store,
}

router = RouterAgent(llm, retrievers)
result = router.route("What are the differences between TiDB and CockroachDB in distributed transactions?")

Router Agents work best when knowledge bases are clearly categorized — simple to implement, immediate results.

Pattern 2: Multi-hop Reasoning Agent

Complex questions require decomposition. For example, "Compare TiDB and CockroachDB in distributed transactions with benchmark data" requires retrieving information about each database separately, then benchmark data, and finally synthesizing an answer.

from langchain.agents import create_react_agent


tools = [
    search_knowledge_base,
    search_web,
    calculate_metrics,
    query_database,
]

agent = create_react_agent(llm, tools, prompt)
result = agent.invoke({
    "input": "Compare TiDB vs CockroachDB in distributed transactions with benchmark data"
})

A more granular multi-hop implementation:

from dataclasses import dataclass
from typing import Optional


@dataclass
class SubQuestion:
    question: str
    answer: Optional[str] = None
    sources: list[str] = None


class MultiHopAgent:
    def __init__(self, llm, retriever):
        self.llm = llm
        self.retriever = retriever

    def decompose(self, question: str) -> list[SubQuestion]:
        prompt = f"""Break down this complex question into 2-4 sub-questions.
Each sub-question should be independently answerable.

Question: {question}

Sub-questions (one per line):"""
        result = self.llm.invoke(prompt)
        lines = [l.strip() for l in result.strip().split("\n") if l.strip()]
        return [SubQuestion(question=line) for line in lines]

    def answer_sub_question(self, sub_q: SubQuestion) -> SubQuestion:
        docs = self.retriever.similarity_search(sub_q.question, k=3)
        context = "\n".join([d.page_content for d in docs])
        answer = self.llm.invoke(
            f"Based on:\n{context}\n\nAnswer: {sub_q.question}"
        )
        sub_q.answer = answer
        sub_q.sources = [d.metadata.get("source", "") for d in docs]
        return sub_q

    def synthesize(self, question: str, sub_answers: list[SubQuestion]) -> str:
        evidence = "\n".join([f"- {sq.question}: {sq.answer}" for sq in sub_answers])
        return self.llm.invoke(
            f"Based on the following evidence, answer the original question.\n\n"
            f"Original: {question}\n\nEvidence:\n{evidence}\n\nAnswer:"
        )

    def run(self, question: str) -> dict:
        sub_questions = self.decompose(question)
        answered = [self.answer_sub_question(sq) for sq in sub_questions]
        final = self.synthesize(question, answered)
        return {
            "answer": final,
            "sub_questions": [
                {"q": sq.question, "a": sq.answer} for sq in answered
            ],
        }


agent = MultiHopAgent(llm, vector_store)
result = agent.run("Which is better for my use case: RAG or fine-tuning? Analyze from cost, effectiveness, and data volume perspectives")

Pattern 3: Self-Reflective Search Agent

After retrieving documents, the Agent evaluates their quality — if not good enough, it rewrites the query and searches again. This is the core capability of Agentic RAG: self-correction.

from dataclasses import dataclass, field
from typing import Optional


@dataclass
class ReflectionState:
    question: str
    documents: list = field(default_factory=list)
    filtered_docs: list = field(default_factory=list)
    answer: str = ""
    iterations: int = 0
    max_iterations: int = 3
    need_more_search: bool = False


class SelfReflectiveAgent:
    def __init__(self, llm, retriever, min_relevant: int = 2):
        self.llm = llm
        self.retriever = retriever
        self.min_relevant = min_relevant

    def retrieve(self, state: ReflectionState) -> dict:
        docs = self.retriever.similarity_search(state.question, k=5)
        return {"documents": docs}

    def grade_documents(self, state: ReflectionState) -> dict:
        filtered = []
        for doc in state.documents:
            score = self.llm.invoke(
                f"Is this document relevant to '{state.question}'? "
                f"Answer yes or no:\n{doc.page_content[:500]}"
            )
            if "yes" in score.lower():
                filtered.append(doc)
        return {
            "filtered_docs": filtered,
            "need_more_search": len(filtered) < self.min_relevant,
        }

    def rewrite_query(self, state: ReflectionState) -> dict:
        better = self.llm.invoke(
            f"Rewrite this question for better search results: {state.question}"
        )
        return {"question": better, "iterations": state.iterations + 1}

    def generate(self, state: ReflectionState) -> dict:
        context = "\n".join([d.page_content for d in state.filtered_docs])
        answer = self.llm.invoke(
            f"Based on:\n{context}\n\nAnswer: {state.question}"
        )
        return {"answer": answer}

    def run(self, question: str) -> dict:
        state = ReflectionState(question=question)
        self.retrieve(state)

        while state.iterations < state.max_iterations:
            self.grade_documents(state)
            if not state.need_more_search:
                break
            self.rewrite_query(state)
            self.retrieve(state)

        self.generate(state)
        return {"answer": state.answer, "iterations": state.iterations}


agent = SelfReflectiveAgent(llm, vector_store)
result = agent.run("How to deploy Agentic RAG in production?")

Pattern 4: Tool-Augmented Agent

The Agent doesn't just retrieve documents — it can also invoke search engines, calculators, databases, and other external tools. This dramatically expands the Agent's capability boundary.

from typing import Annotated


def search_web(query: str) -> str:
    """Search the web for up-to-date information."""
    results = web_search_engine.search(query, max_results=3)
    return "\n".join([r.snippet for r in results])


def query_database(sql: str) -> str:
    """Execute SQL query against the product database."""
    rows = db.execute(sql)
    return str(rows[:20])


def calculate(expression: str) -> str:
    """Safely evaluate a mathematical expression."""
    allowed = set("0123456789+-*/.() ")
    if not all(c in allowed for c in expression):
        return "Error: invalid expression"
    return str(eval(expression))


def search_knowledge_base(query: str) -> str:
    """Search internal knowledge base for documentation."""
    docs = vector_store.similarity_search(query, k=3)
    return "\n".join([d.page_content for d in docs])


tools = [search_web, query_database, calculate, search_knowledge_base]

tool_descriptions = "\n".join(
    [f"- {t.__name__}: {t.__doc__}" for t in tools]
)


class ToolAugmentedAgent:
    def __init__(self, llm, tools: list):
        self.llm = llm
        self.tools = {t.__name__: t for t in tools}

    def run(self, question: str, max_steps: int = 5) -> dict:
        messages = [{"role": "user", "content": question}]
        steps = []

        for _ in range(max_steps):
            response = self.llm.invoke(
                f"Available tools:\n{tool_descriptions}\n\n"
                f"To use a tool, respond: USE: tool_name(arg)\n"
                f"To give final answer, respond: ANSWER: your answer\n\n"
                f"Question: {messages[-1]['content']}"
            )

            if response.startswith("ANSWER:"):
                return {"answer": response[7:].strip(), "steps": steps}

            if response.startswith("USE:"):
                tool_call = response[4:].strip()
                tool_name = tool_call.split("(")[0]
                tool_arg = tool_call.split("(")[1].rstrip(")")
                tool_fn = self.tools.get(tool_name)
                if tool_fn:
                    observation = tool_fn(tool_arg)
                    steps.append({
                        "tool": tool_name,
                        "arg": tool_arg,
                        "result": observation[:200],
                    })
                    messages.append({
                        "role": "assistant",
                        "content": f"Used {tool_name}, got: {observation}",
                    })

        return {"answer": "Max steps reached", "steps": steps}


agent = ToolAugmentedAgent(llm, tools)
result = agent.run("What is our product's monthly active users? How does it compare to competitors?")

Pattern 5: LangGraph Workflow Orchestration

The most complete Agentic RAG implementation — using LangGraph's StateGraph to define a full retrieve-grade-rewrite-generate loop with conditional branching and checkpoints.

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator


class AgentState(TypedDict):
    question: str
    documents: Annotated[list, operator.add]
    answer: str
    iterations: int
    need_more_search: bool


def retrieve(state: AgentState) -> dict:
    docs = vector_store.similarity_search(state["question"], k=5)
    return {"documents": docs}


def grade_documents(state: AgentState) -> dict:
    question = state["question"]
    docs = state["documents"]
    filtered = []
    for doc in docs:
        score = llm.invoke(f"Is this doc relevant to '{question}'? Answer yes/no: {doc.page_content}")
        if "yes" in score.lower():
            filtered.append(doc)
    return {"documents": filtered, "need_more_search": len(filtered) < 2}


def generate(state: AgentState) -> dict:
    context = "\n".join([d.page_content for d in state["documents"]])
    answer = llm.invoke(f"Based on:\n{context}\n\nAnswer: {state['question']}")
    return {"answer": answer}


def decide_to_search(state: AgentState) -> str:
    if state["need_more_search"] and state["iterations"] < 3:
        return "rewrite"
    return "generate"


def rewrite_query(state: AgentState) -> dict:
    better_query = llm.invoke(f"Rewrite this question for better search: {state['question']}")
    return {"question": better_query, "iterations": state["iterations"] + 1}


workflow = StateGraph(AgentState)
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade", grade_documents)
workflow.add_node("generate", generate)
workflow.add_node("rewrite", rewrite_query)

workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "grade")
workflow.add_conditional_edges("grade", decide_to_search, {
    "rewrite": "rewrite",
    "generate": "generate",
})
workflow.add_edge("rewrite", "retrieve")
workflow.add_edge("generate", END)

app = workflow.compile()
result = app.invoke({"question": "Which is better for my use case: RAG or fine-tuning?", "iterations": 0})

LangGraph's advantages: visualizable workflows, checkpoint recovery support, flexible conditional branching, and easy debugging. The top choice for production environments.

Pitfall Guide: 5 Common Traps

#	Trap	Manifestation	Solution
1	Infinite Loop	Agent keeps rewriting queries, never satisfied	Set a hard max_iterations limit (3-5)
2	Over-retrieval	Every question triggers multiple search rounds, latency spikes	Simple questions use fast path; only complex ones go Agentic
3	Tool Hallucination	Agent invents non-existent tool names to call	Strictly validate tool names with a whitelist
4	Context Explosion	Documents from all retrieval rounds stuffed into the prompt	Keep only top-K highest-scored documents per round
5	Grading Bias	LLM document relevance scoring is inconsistent	Use structured output (JSON Schema) to constrain grading format

Troubleshooting: 10 Common Errors

#	Error Message	Cause	Solution
1	`GraphRecursionError: Recursion limit reached`	Agent loop exceeds LangGraph default step limit	Set `recursion_limit` parameter, or optimize conditional edge logic
2	`KeyError: 'documents'`	Missing expected field in State	Check node return values include all required fields
3	`ToolCallingError: Unknown tool`	Agent called an unregistered tool	Check tools list, ensure tool names match descriptions
4	`TokenLimitExceeded`	Accumulated documents from multiple rounds exceed context window	Filter documents in grade node, limit docs passed to generate
5	`ValidationError: State schema mismatch`	State type definition doesn't match actual data	Check TypedDict field types, Annotated usage correctness
6	`TimeoutError: LLM invocation timed out`	Grading/generation LLM response is slow	Set timeout parameter, consider async invocation
7	`ImportError: No module named 'langgraph'`	LangGraph not installed	`pip install langgraph langchain-core`
8	`JSONDecodeError in structured output`	LLM returns non-standard JSON	Use `with_structured_output()` to force structured output
9	`RateLimitError: Too many requests`	Multiple retrieval rounds trigger API rate limiting	Add exponential backoff retry, or switch to local model
10	`AttributeError: 'NoneType' has no attribute`	A node returns None	Ensure every node returns a dict, not None

Advanced Optimization Tips

1. Hybrid Routing: Fast Path for Simple Questions

Not all questions need an Agentic flow. Add a classifier — simple questions go through Naive RAG, complex ones through Agentic RAG. Average latency drops by 60%.

class HybridRouter:
    def __init__(self, llm, naive_rag, agentic_rag):
        self.llm = llm
        self.naive_rag = naive_rag
        self.agentic_rag = agentic_rag

    def is_simple(self, question: str) -> bool:
        result = self.llm.invoke(
            f"Is this a simple factual question? Answer yes/no: {question}"
        )
        return "yes" in result.lower()

    def run(self, question: str) -> dict:
        if self.is_simple(question):
            return self.naive_rag.query(question)
        return self.agentic_rag.invoke({"question": question, "iterations": 0})

2. Checkpoint Recovery: Don't Lose Long-Running Workflows

LangGraph supports checkpoints — the Agent can resume from its last state after interruption.

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app = workflow.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke({"question": "Analyze our tech stack choices", "iterations": 0}, config=config)

3. Parallel Retrieval: Search Multiple Knowledge Sources Simultaneously

import asyncio


async def parallel_retrieve(question: str, stores: list) -> list:
    tasks = [store.asimilarity_search(question, k=3) for store in stores]
    results = await asyncio.gather(*tasks)
    return [doc for batch in results for doc in batch]

4. Cache Hot Queries

from functools import lru_cache


@lru_cache(maxsize=1000)
def cached_retrieve(query_hash: str, question: str) -> list:
    return vector_store.similarity_search(question, k=5)

5. Observability: Trace Every Agent Decision Step

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("agentic_rag")


def traced_node(func):
    def wrapper(state):
        logger.info(f"Node: {func.__name__}, Input: {str(state)[:200]}")
        result = func(state)
        logger.info(f"Node: {func.__name__}, Output: {str(result)[:200]}")
        return result
    return wrapper

Comparison: Agentic RAG vs Naive RAG vs Advanced RAG vs Modular RAG

Dimension	Naive RAG	Advanced RAG	Modular RAG	Agentic RAG
Retrieval Strategy	Single vector search	Query rewriting + hybrid retrieval	Pluggable retrieval modules	Agent autonomously decides retrieval
Reasoning	✗	✗	✗	✓ Multi-hop reasoning
Self-correction	✗	△ Re-ranking	✗	✓ Reflection + rewriting
Tool Calling	✗	✗	△	✓ Multi-tool collaboration
Workflow	Linear pipeline	Linear + preprocessing	Modular pipeline	Stateful graph
Implementation Complexity	★☆☆	★★☆	★★★	★★★★
Latency	~1s	~2s	~2s	~3-8s
Accuracy	60-70%	75-85%	80-88%	85-95%
Best For	Simple FAQ	Standard Q&A	Customized scenarios	Complex multi-hop questions

More ★ = higher implementation complexity; ✓ supported △ partially supported ✗ not supported; latency and accuracy are reference values

Recommended Online Tools

JSON Formatter — Format Agent State and retrieval result JSON structures, essential for debugging
Hash Calculator — Compute MD5/SHA hashes for document deduplication and query caching
Curl to Code — Convert LLM API debugging curl commands to Python code

Summary & Outlook

"The future of RAG is not better retrieval — it's smarter agents that know when and how to retrieve."

Agentic RAG is transitioning from "experimental architecture" to "production standard." Key trends for 2026:

Native Agent Support: LangGraph, LlamaIndex, and other frameworks treat Agentic RAG as a first-class citizen
Multi-Agent Collaboration: Multiple specialized Agents working together on complex retrieval tasks
Adaptive Strategy: Agents automatically select retrieval depth based on question complexity
Enhanced Explainability: Every retrieval decision has a clear reasoning chain for traceability
Cost Optimization: Intelligent routing + caching brings Agentic RAG costs close to Naive RAG

The principle for choosing an Agentic RAG approach: start with Router Agent, upgrade incrementally. First use routing to solve knowledge base dispatch, then add self-reflection to improve retrieval quality, and finally orchestrate the full workflow with LangGraph. Don't build the most complex version first — get the Agent running, then make it smarter.