Agentic RAG

When to Go Agentic

Agentic adds latency and cost; use when static cannot recover from retrieval failure or needs multiple steps.

ReAct with Retrieval Tool

from langchain_core.tools import tool
from langchain_anthropic import ChatAnthropic
from langgraph.prebuilt import create_react_agent

@tool
def search_kb(query: str) -> str:
    """Search the internal knowledge base. Returns top passages."""
    docs = retriever.invoke(query)
    return "\n\n".join(f"[{d.metadata['id']}] {d.page_content}" for d in docs[:5])

@tool
def search_web(query: str) -> str:
    """Search the open web. Use for recent events or things not in the KB."""
    return tavily_client.search(query, max_results=5)

llm = ChatAnthropic(model="claude-sonnet-4-5-20250929")
agent = create_react_agent(llm, tools=[search_kb, search_web])

result = agent.invoke({"messages": [("human", "What changed in our refund policy since the 2024 update?")]})

Agent decides when and what to search; handles multi-hop automatically via tool-loop.

Self-RAG (self-reflection tokens)

Generate, self-critique retrieval, critique answer, regenerate if needed.

from typing import Literal
from pydantic import BaseModel, Field

class RetrievalDecision(BaseModel):
    needs_retrieval: Literal["yes", "no"]
    reason: str

class Grade(BaseModel):
    score: Literal["relevant", "irrelevant"]
    reason: str

class Support(BaseModel):
    support: Literal["fully_supported", "partially_supported", "not_supported"]
    reason: str

decider = llm.with_structured_output(RetrievalDecision)
grader = llm.with_structured_output(Grade)
supporter = llm.with_structured_output(Support)

def self_rag(question: str, max_retries: int = 2):
    decision = decider.invoke(f"Does answering '{question}' require retrieval from the KB?")
    if decision.needs_retrieval == "no":
        return llm.invoke(question).content

    for attempt in range(max_retries + 1):
        docs = retriever.invoke(question)
        relevant = [d for d in docs
                    if grader.invoke(f"Q: {question}\nPassage: {d.page_content}").score == "relevant"]
        if not relevant and attempt < max_retries:
            question = rewrite(question)  # query rewrite
            continue
        answer = generate_answer(question, relevant)
        sup = supporter.invoke(f"Answer: {answer}\nContext: {[d.page_content for d in relevant]}")
        if sup.support == "fully_supported":
            return answer
        if attempt == max_retries:
            return answer + "\n\n(Partial support only.)"
    return "I could not find enough information."

Paper: Asai et al., 2023. The structured critics are the core contribution.

Corrective RAG (CRAG)

Grade retrieval confidence; on low confidence, rewrite the query and fall back to web search.

from langgraph.graph import StateGraph, END
from typing import TypedDict

class CRAGState(TypedDict):
    question: str
    docs: list
    web_docs: list
    grade: str
    answer: str

def retrieve_node(s: CRAGState) -> CRAGState:
    s["docs"] = retriever.invoke(s["question"])
    return s

def grade_node(s: CRAGState) -> CRAGState:
    grades = [grader.invoke(f"Q:{s['question']}\nPassage:{d.page_content}").score for d in s["docs"]]
    relevant = sum(1 for g in grades if g == "relevant")
    if relevant >= 2:
        s["grade"] = "correct"
    elif relevant >= 1:
        s["grade"] = "ambiguous"
    else:
        s["grade"] = "incorrect"
    s["docs"] = [d for d, g in zip(s["docs"], grades) if g == "relevant"]
    return s

def web_fallback_node(s: CRAGState) -> CRAGState:
    rewritten = rewrite(s["question"])
    s["web_docs"] = tavily_client.search(rewritten, max_results=5)
    return s

def generate_node(s: CRAGState) -> CRAGState:
    ctx = s["docs"] + s.get("web_docs", [])
    s["answer"] = generate_answer(s["question"], ctx)
    return s

def route_after_grade(s: CRAGState) -> str:
    return "generate" if s["grade"] == "correct" else "web_fallback"

g = StateGraph(CRAGState)
g.add_node("retrieve", retrieve_node)
g.add_node("grade", grade_node)
g.add_node("web_fallback", web_fallback_node)
g.add_node("generate", generate_node)
g.set_entry_point("retrieve")
g.add_edge("retrieve", "grade")
g.add_conditional_edges("grade", route_after_grade,
                        {"generate": "generate", "web_fallback": "web_fallback"})
g.add_edge("web_fallback", "generate")
g.add_edge("generate", END)
crag = g.compile()

result = crag.invoke({"question": "What is our refund window for enterprise contracts?"})

Web search is the correction leg; works when the KB is narrower than user demand.

Adaptive RAG

Classifier decides retrieval path up front — skip retrieval for chit-chat, single-hop for factual, multi-hop for research.

from typing import Literal
from pydantic import BaseModel

class RouteDecision(BaseModel):
    path: Literal["no_retrieval", "single_hop", "multi_hop", "web"]
    reason: str

router = llm.with_structured_output(RouteDecision)

def adaptive_rag(q: str):
    d = router.invoke(f"Classify: no_retrieval | single_hop | multi_hop | web\n\nQuery: {q}")
    if d.path == "no_retrieval": return llm.invoke(q).content
    docs = (tavily_client.search(q, max_results=5) if d.path == "web"
            else retriever.invoke(q) if d.path == "single_hop"
            else multi_hop_retrieve(q, hops=3))
    return generate_answer(q, docs)

Multi-Hop Retrieval

Iteratively retrieve, synthesize a sub-question, retrieve again. Cap hops (3-5) to bound cost; stop on explicit DONE, query repetition, or hop limit.

def multi_hop_retrieve(q: str, hops: int = 3) -> list:
    accumulated, current_q = [], q
    for _ in range(hops):
        docs = retriever.invoke(current_q)
        accumulated.extend(docs)
        ctx = "\n\n".join(d.page_content for d in docs)
        follow_up = llm.invoke(
            f"Given partial context, what sub-question refines '{q}'? Reply 'DONE' if sufficient.\n\n{ctx}"
        ).content.strip()
        if follow_up == "DONE" or follow_up == current_q:
            break
        current_q = follow_up
    seen, out = set(), []
    for d in accumulated:
        key = d.metadata.get("id") or hash(d.page_content)
        if key not in seen: seen.add(key); out.append(d)
    return out

Plan-and-Execute RAG

Make the plan once up front, execute steps, synthesize. Lower latency variance than ReAct loops but worse recovery from surprises.

from pydantic import BaseModel
from typing import Literal

class Step(BaseModel):
    action: Literal["retrieve", "compute", "answer"]
    input: str

class Plan(BaseModel):
    steps: list[Step]

planner = llm.with_structured_output(Plan)

def plan_and_execute(q: str):
    plan = planner.invoke(f"Break into 2-5 steps (retrieve/compute/answer): {q}")
    scratch = []
    for step in plan.steps:
        if step.action == "retrieve":
            docs = retriever.invoke(step.input)
            scratch.append({"q": step.input, "r": [d.page_content for d in docs[:3]]})
        elif step.action == "compute":
            scratch.append({"q": step.input, "r": llm.invoke(
                f"Compute: {step.input}\nCtx: {scratch}").content})
    return llm.invoke(f"Using this scratchpad, answer: {q}\n\n{scratch}").content

Guardrails on Agent Loops

Every agentic flow must bound tool calls, tokens, and wall-clock time. LangGraph's recursion_limit caps graph iterations; combine with asyncio.wait_for.

import asyncio

MAX_TOOL_CALLS = 8
HARD_TIMEOUT_S = 30

async def bounded_invoke(app, inputs):
    try:
        return await asyncio.wait_for(
            app.ainvoke(inputs, config={"recursion_limit": MAX_TOOL_CALLS}),
            timeout=HARD_TIMEOUT_S,
        )
    except asyncio.TimeoutError:
        return {"answer": "Timed out; returning partial context.", "partial": True}

LangGraph additionally provides checkpointing (MemorySaver, Postgres), streaming events, and human-in-the-loop pauses — use a persistent checkpointer in production so you can replay agent traces.

Anti-Patterns

Production Checklist

• Routing classifier decides retrieval path (adaptive RAG)
• Grader on retrieval confidence (CRAG)
• Web fallback wired behind low-confidence branch
• Hop / recursion limits enforced
• Hard wall-clock timeout per request
• Haiku for graders/routers, Sonnet for synthesis
• Structured output for every graded decision (Pydantic schemas)
• LangGraph checkpointer persistent (Postgres/Redis) in production
• LangSmith or OpenTelemetry tracing on every node
• Streaming of intermediate events to UI
• Fallback to static RAG on agent error
• Cost per query dashboarded — agentic paths cost 3-10x static