agentic-rag

What this skill does

# Agentic RAG

## When to Go Agentic

| Signal | Static RAG | Agentic RAG |
|---|---|---|
| Known single-hop questions | Yes | No |
| Unknown retrieval depth | No | Yes |
| Multi-source (docs + web + DB) | No | Yes |
| Need to detect retrieval failure and recover | No | Yes |
| Latency budget < 2s | Yes | No |
| Audit trail / explainability | Harder | Natural |

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

## ReAct with Retrieval Tool

```python
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.

```python
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.

```python
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.

```python
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.

```python
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.

```python
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`.

```python
import asyncio

MAX_TOOL_CALLS = 8
HARD_TIMEOUT_S = 30

async def bounded_invoke(app, inputs):
    try:
        return await asyncio.wa