Agentic RAG Knowledge Assistant

The brief advanced

You demo the bot and it dazzles. "What's our refund window?" — instant, cited, correct. Everyone nods, you ship it. Then the first real support ticket lands: "The customer bought under the 2023 policy but is asking about a product we discontinued — does the old warranty still apply?" Your bot retrieves the current policy, sees nothing about 2023 or discontinued products, and either shrugs or — worse — confidently makes something up. The demo questions and the real questions were never the same species.

Here's the pattern behind that failure, and it's the whole reason this project exists. A single-shot retriever is like sending an intern to the filing cabinet with one search term and no permission to come back. If the answer lives across three folders, or the first search term was slightly wrong, the intern returns with a partial folder and you get a partial (or invented) answer. A good researcher, by contrast, searches, reads what came back, notices what's missing, and searches again with a sharper query — looping until the evidence actually covers the question. That loop is the entire difference, and it's what "agentic" means here: not a smarter model, but a model given the authority to search more than once and to judge its own results.

Why does this matter for a design brief specifically? Because the temptation is to "fix" bad answers by swapping in a bigger model or a fancier embedding. Neither helps when the real problem is that the system only looked once. The common mistake teams make is treating retrieval quality as a model problem when it's a control-flow problem — the fix is a loop with a self-check, not a better one-shot. Everything below designs that loop.

"Our chatbot answers easy questions but falls apart on real ones." Single-shot RAG fails when the answer needs several lookups, when the first retrieval misses, or when the question spans multiple sources. An agentic RAG assistant plans its search, checks its own results, and iterates — so it handles the multi-hop, compound questions that a one-shot retriever gets wrong.

1 · Discovery — where does plain RAG break? advanced

Before you build the loop, earn the right to build it. Agentic RAG is more expensive and slower than plain RAG — every extra retrieval is another round-trip and every self-grade is another model call. If you wrap the whole loop around questions that a single lookup already answered, you've added latency and cost and gotten nothing back. So the first design move isn't to build; it's to find the questions where a plain retriever actually breaks, and confirm there are enough of them to justify the machinery.

Think of it like deciding whether to hire a research analyst versus using a search box. For "what's the capital of France," the search box wins — instant, free, right. You only want the analyst for questions that need several steps: cross-referencing sources, chasing a lead the first search surfaced, reconciling documents that disagree. The table below is a triage of exactly that: which question shapes give agentic retrieval real leverage (multi-hop, missed-first-retrieval, cross-source) and which are a waste of it. Read the ⭐ ratings as "how much does the loop buy me here" — and notice the bottom row is a deliberate warning, not an oversight.

The mistake to avoid: assuming "agentic is better, so use it everywhere." The honest design isn't loop-everything — it's route. Cheap single-fact questions go straight to plain RAG; only the hard shapes enter the loop. Discovery is where you gather the evidence that this split is worth engineering, and it's what your evaluation in §6 will later have to prove.

2 · Architecture advanced

The moment you say "search again if the first result is bad," you've stopped drawing a pipeline and started drawing a loop — and loops are where systems get away from you. A straight line always terminates. A loop can spin forever, and if each spin costs an API call, "forever" costs money. So the architecture's real job isn't just wiring boxes together; it's deciding where the loop is allowed to go around again, and — just as importantly — where it must stop.

Picture the four moving parts as roles on a tiny research team. Plan is the lead who breaks a fat question into searchable sub-questions. Retrieve is the runner who goes to the shelf and brings back documents. Grade is the skeptical editor who reads what came back and asks one question — "does this actually answer what was asked, yes or no?" And the two exits are the editor's verdict: publish with citations, or refuse to publish because the evidence isn't there. The diagram below is exactly this team, drawn as a cycle with that skeptic in the middle.

• The self-grader is the load-bearing wall. Remove it and you don't have agentic RAG — you have plain RAG that happens to retrieve twice. The grade is what turns the arrow back.
• The loop-back is the whole point, and the whole danger. "Insufficient → reformulate → search again" is the behavior you're paying for, but it's also the behavior that can run away. It needs a leash (covered next in §3).
• Abstain is a first-class outcome, not an error. "I don't have enough to answer" is a correct result when the evidence genuinely isn't there — designing it in is what stops the system from hallucinating under pressure.

The common mistake here is to draw the happy path (question → answer) beautifully and treat the loop-back and the abstain exit as afterthoughts. They're the opposite: the happy path is what plain RAG already gave you, and these two edges are the entire reason this project earns the word "agentic."

A loop, not a line. The agent plans (decomposing the question into sub-queries), retrieves from the best source, then grades whether the context actually answers the question. If yes → answer with citations. If no → reformulate and search again. If it can't find evidence after a bounded number of tries → abstain. This is ReAct (E1) applied to retrieval.

3 · Risk & safety model advanced

The scariest bug in an agentic RAG system isn't a crash — it's a $400 invoice for a question nobody could answer. A retrieval loop that can decide to search again is, by construction, a loop that can decide to search again forever: the grade keeps coming back "insufficient," the agent keeps reformulating, and each pass burns tokens on a question the corpus simply doesn't contain the answer to. Nothing errors. It just spins, and you find out when the bill arrives. The autonomy you gave the loop in §2 is precisely the thing that has to be constrained here.

The mental model that keeps this straight: an autonomous loop is a dog off the leash. Off-leash is what makes it useful — it can chase the answer down paths you didn't script. But a dog off the leash near a road needs a fence and a whistle, or one day it doesn't come back. The controls in the table below are that fence: a hard cap on iterations (it can search at most N times, then it must stop), a token/cost budget (it can spend at most this much, then it must stop), and a graceful degrade path ("best effort, with a caveat") for when it hits either limit. None of these make the agent smarter — they make its failure bounded.

Read the table as two colors of risk. The 🔴 rows — answering without evidence, hallucinated citations — are trust failures: the system lies, and one confident wrong answer poisons user trust in every right one. The 🟠 rows — runaway loops, drifting reformulations — are resource failures: the system wastes money or wanders off-topic. The common mistake is to obsess over the trust failures (they're embarrassing) and forget the resource ones (they're just expensive) — but in production it's usually the unbounded loop that pages you at 3 a.m. Design both leashes in from the start; retrofitting a cost cap onto a live loop is a bad afternoon.

4 · The retrieval loop, in code advanced

Every diagram you've seen so far collapses into about ten lines of Python — and it's worth seeing that collapse happen, because it kills a myth. Newcomers assume "agentic" means a heavyweight framework, a graph engine, some orchestration layer. It doesn't. The core of an agentic RAG system is a for loop with a range bound and an if that breaks early. If you can read a loop that stops after N tries unless it hits a good result sooner, you can read this whole system.

The shape to hold in your head as you read the snippet: it's a search-until-satisfied loop with a hard ceiling, exactly like retrying a flaky network call — try, check the result, retry with a tweak if it failed, but never more than a fixed number of times. Four verbs do the work — plan the sub-queries, retrieve the hits, judge_sufficiency on what came back, and reformulate when it's not enough — and the range(max_steps) is the leash from §3 made literal: the loop cannot run more than max_steps times, so "forever" is impossible by construction.

Watch one thing especially: the loop has two ways out and they mean opposite things. Returning early with a cited answer is success — the evidence cleared the bar. Falling out the bottom to abstain is honest failure — the leash ran out before the evidence did. A loop with only the success exit is the classic beginner bug: it either loops forever or, worse, returns a confident answer built on evidence it never actually verified. The abstain line is short, easy to skip, and the most important line in the file.

class _Any:
    '''stands in for any undefined demo value; supports call/attr/index/
    iteration and basic arithmetic (as 0.7) so demo snippets run.'''
    def __call__(self, *a, **k): return _Any()
    def __getattr__(self, k): return _Any()
    def __getitem__(self, k): return _Any()
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __contains__(self, o): return True
    def __enter__(self, *a): return _Any()
    def __exit__(self, *a): return False
    def __float__(self): return 0.7
    def __int__(self): return 1
    def __lt__(self, o): return True
    def __gt__(self, o): return False
    def __le__(self, o): return True
    def __ge__(self, o): return False
    def __add__(self, o): return o
    def __radd__(self, o): return o
    def __bool__(self): return True
    def __repr__(self): return 'demo'
    def __str__(self): return 'demo'
def abstain(*a, **k):  # demo stub
    return _Any()
def generate_answer(*a, **k):  # demo stub
    return _Any()
def judge_sufficiency(*a, **k):  # demo stub
    return _Any()
def next_query(*a, **k):  # demo stub
    return _Any()
def plan(*a, **k):  # demo stub
    return _Any()
def reformulate(*a, **k):  # demo stub
    return _Any()
def retrieve(*a, **k):  # demo stub
    return _Any()

agentic_rag.py (shape)def answer(question, max_steps=4):
    subqs = plan(question)                 # decompose (E1)
    context = []
    for step in range(max_steps):     # the leash
        hits = retrieve(next_query(subqs, context))   # P9 engine
        context += hits
        grade = judge_sufficiency(question, context)  # self-critique
        if grade.sufficient:
            return generate_answer(question, context)  # cited
        subqs = reformulate(question, grade.gap)      # try again
    return abstain(question, context)   # honest "not enough evidence"

5 · Tool surface advanced

Give the agent a web-search tool "just in case," and one day it answers an internal-policy question by quoting a random blog it found online. That's the trap of the tool surface: every capability you hand the agent is a door it might walk through, and the ones that reach outside your trusted data are doors to answers you can't vouch for. So the tool surface isn't a menu of "everything we could wire up" — it's a deliberately short, deliberately typed list of what the agent is allowed to do, with the risky doors clearly marked.

A useful frame: think of the tools as the ingredients an agent is permitted to cook with, sorted by how much you trust the source. Your internal knowledge base and read-only database are the pantry — vetted, owned, safe to serve. The open web is the stuff a stranger left on the doorstep — sometimes exactly what you needed, sometimes spoiled, always to be labeled before it goes on the plate. The table below sorts the surface this way: 🟢 read-only internal sources you trust by default, and 🟠 the web tool that's fenced off with "prefer internal, label web." And notice judge_sufficiency sits in the list too — the self-grader is itself a tool, the one that decides when to stop reaching for the others.

Two design instincts to build here. First, read-only is a feature, not a limitation: a retrieval agent that can only read can't corrupt your data no matter how badly it reasons, which shrinks the blast radius of every other bug on this page. Second, the common mistake is adding tools because they're impressive rather than because a question needs them — every extra tool is another choice the agent can get wrong and another attack surface. Keep the surface small; make each entry earn its place against a real question shape from §1.

6 · Evaluation advanced

"It feels smarter" is not a metric, and it's how good agentic RAG systems get quietly killed. You built the loop, it demos well, everyone's impressed — and then someone asks "is it actually better than the plain RAG we already had, or just slower?" If you can't answer with a number, the honest answer is you don't know, and the loop's extra latency and cost become impossible to defend. Evaluation is where you convert a feeling into evidence.

The trap unique to this system is that a single accuracy number lies. Imagine a version that answers every multi-hop question perfectly but also confidently answers questions it should have abstained on — high accuracy, terrible in production. Or one that abstains constantly to stay "safe" — never wrong, never useful. You have to measure the loop along its own axes, and the table below lays them out: correctness on the hard multi-hop set (where plain RAG fails and this should win), faithfulness and citation-correctness (does every claim trace to retrieved evidence, or did it garnish?), abstention-correctness (does it say "I don't know" exactly when it should — not more, not less), and steps/cost per answer (is the loop looping more than the question needed?).

But the one measurement that decides whether this project should exist is the last row: uplift versus plain RAG. This is the researcher-versus-search-box test made quantitative — run both systems on the same question set and compare. If the loop only ties plain RAG, you're paying for machinery that buys nothing, and the right call is to route those questions to the cheaper path. The common mistake is skipping the baseline entirely and reporting the agentic system's numbers in isolation; a number with no baseline can't tell you if the loop earned its keep. Always measure against the thing it's supposed to beat.

7 · Phased rollout expert

The failure mode here is building the whole loop, shipping it, watching it misbehave, and having no idea which piece is at fault. Was it the planner splitting the question badly? The grader passing weak evidence? The reformulation drifting off-topic? When everything went live at once, you can't tell — you have to debug the whole machine as one opaque unit. Phased rollout is the antidote: turn the pieces on one at a time so that when quality moves, you know exactly which piece moved it.

Think of it as adding one instrument to a band and listening before adding the next. Phase 1 is plain RAG alone — one retrieve, one answer, with citations and an abstain path. That's not a throwaway; it's your baseline, the number every later phase has to beat (this is the uplift comparison from §6, built in from day one). Phase 2 adds just the grader and a single retry — the smallest possible taste of "agentic" — and you measure whether one self-check and one re-search actually lifts the hard questions. Only in Phase 3 do you turn on the full loop: planning, bounded iterations, multiple sources, the cost budget, and the routing that sends easy questions to the cheap path.

The discipline this enforces is that every rung earns the next one. If Phase 2's grader-plus-retry doesn't beat the Phase 1 baseline, you've learned something cheap and early — maybe your questions don't actually need the loop, or your grader isn't discriminating. The common mistake is skipping straight to Phase 3 because it's the "real" system; you get a system that's better or worse than the baseline and no way to attribute why. And the "Never" rung is not a phase but a permanent boundary — no answer without passing the sufficiency gate, no loop without a hard cap and budget — the two invariants from §3 that hold in every phase, including the ones you haven't built yet.

Skills & course map expert

How to use this page expert

Do the steps in order. terminal = commands to run; file = create that file with the exact contents. Expected output follows each command.

Step 1 · Project folder expert

terminalmkdir -p agentic-rag/tests
cd agentic-rag

# you are now inside agentic-rag/

Step 2 · Virtual environment expert

terminalpython3 -m venv .venv
source .venv/bin/activate

terminalpy -m venv .venv
.venv\Scripts\Activate.ps1

(.venv) /Users/you/agentic-rag $

Step 3 · Install dependencies expert

terminalpip install pytest "anthropic>=0.40"
pip freeze > requirements.txt

Successfully installed anthropic-0.69.0 pytest-8.3.4 ...

Step 4 · The pieces the loop needs (as swappable functions) expert

Create pieces.py. The loop depends on three functions: retrieve, judge (the grader), and generate. Here we give real (mock-backed) versions and make them injectable so tests can script them. Paste the whole file.

pieces.py"""The retrieve / judge / generate pieces the loop orchestrates."""
import os
from dataclasses import dataclass

# A tiny corpus: each 'doc' is (id, text). The loop learns to combine facts.
CORPUS = {
    "cur": "The current refund policy allows 30 days.",
    "old": "The 2024 refund policy allowed 14 days.",
}


@dataclass
class Grade:
    sufficient: bool
    gap: str = ""


def retrieve(query: str) -> list[str]:
    """Return doc texts whose id-keyword appears in the query. Deliberately
    naive so a first query finds only part of a multi-hop answer."""
    hits = []
    if "current" in query or "refund" in query: hits.append(CORPUS["cur"])
    if "2024" in query or "old" in query or "previous" in query:
        hits.append(CORPUS["old"])
    return hits


def judge(question: str, context: list[str]) -> Grade:
    """Sufficiency grader. Mock rule: a 'compare' question needs BOTH the
    current AND the 2024 fact. Real version calls the model (Step 8)."""
    if os.environ.get("USE_REAL_API") == "1":
        return _real_judge(question, context)
    joined = " ".join(context)
    if "compare" in question.lower():
        has_cur = "current" in joined
        has_old = "2024" in joined
        if has_cur and has_old:
            return Grade(True)
        return Grade(False, gap="the 2024 policy" if has_cur else "the current policy")
    return Grade(bool(context))       # simple question: any context suffices


def generate(question: str, context: list[str]) -> str:
    if os.environ.get("USE_REAL_API") == "1":
        return _real_generate(question, context)
    return " | ".join(context) + " [grounded]"   # mock answer


def _real_judge(question, context):
    import anthropic, json
    client = anthropic.Anthropic()
    msg = client.messages.create(model="claude-opus-4-8", max_tokens=150,
        system="Reply JSON {\"sufficient\":bool,\"gap\":str}. Does CONTEXT fully "
               "answer QUESTION? If not, name what's missing in gap.",
        messages=[{"role":"user","content":f"QUESTION:{question}\nCONTEXT:{context}"}])
    d = json.loads(msg.content[0].text)
    return Grade(d["sufficient"], d.get("gap", ""))


def _real_generate(question, context):
    import anthropic
    client = anthropic.Anthropic()
    msg = client.messages.create(model="claude-opus-4-8", max_tokens=400,
        system="Answer only from the context; cite it.",
        messages=[{"role":"user","content":f"Context:{context}\nQ:{question}"}])
    return msg.content[0].text

pieces.py"""The retrieve / judge / generate pieces the loop orchestrates."""
import os
from dataclasses import dataclass

# A tiny corpus: each 'doc' is (id, text). The loop learns to combine facts.
CORPUS = {
    "cur": "The current refund policy allows 30 days.",
    "old": "The 2024 refund policy allowed 14 days.",
}


@dataclass
class Grade:
    sufficient: bool
    gap: str = ""


def retrieve(query: str) -> list[str]:
    """Return doc texts whose id-keyword appears in the query. Deliberately
    naive so a first query finds only part of a multi-hop answer."""
    hits = []
    if "current" in query or "refund" in query: hits.append(CORPUS["cur"])
    if "2024" in query or "old" in query or "previous" in query:
        hits.append(CORPUS["old"])
    return hits


def judge(question: str, context: list[str]) -> Grade:
    """Sufficiency grader. Mock rule: a 'compare' question needs BOTH the
    current AND the 2024 fact. Real version calls the model (Step 8)."""
    if os.environ.get("USE_REAL_API") == "1":
        return _real_judge(question, context)
    joined = " ".join(context)
    if "compare" in question.lower():
        has_cur = "current" in joined
        has_old = "2024" in joined
        if has_cur and has_old:
            return Grade(True)
        return Grade(False, gap="the 2024 policy" if has_cur else "the current policy")
    return Grade(bool(context))       # simple question: any context suffices


def generate(question: str, context: list[str]) -> str:
    if os.environ.get("USE_REAL_API") == "1":
        return _real_generate(question, context)
    return " | ".join(context) + " [grounded]"   # mock answer


def _real_judge(question, context):
    import openai, json
    client = openai.OpenAI()
    resp = client.responses.create(model="gpt-5.5", max_output_tokens=150,
        instructions="Reply JSON {\"sufficient\":bool,\"gap\":str}. Does CONTEXT fully "
               "answer QUESTION? If not, name what's missing in gap.",
        input=f"QUESTION:{question}\nCONTEXT:{context}")
    d = json.loads(resp.output_text)
    return Grade(d["sufficient"], d.get("gap", ""))


def _real_generate(question, context):
    import openai
    client = openai.OpenAI()
    resp = client.responses.create(model="gpt-5.5", max_output_tokens=400,
        instructions="Answer only from the context; cite it.",
        input=f"Context:{context}\nQ:{question}")
    return resp.output_text

Step 5 · The bounded loop expert

Create loop.py. It plans, retrieves, grades, reformulates from the gap, and stops at MAX_STEPS — abstaining if evidence never suffices.

loop.py"""The agentic retrieval loop: retrieve -> grade -> reformulate, bounded."""
from pieces import retrieve as _retrieve, judge as _judge, generate as _generate

MAX_STEPS = 4
ABSTAIN = "I couldn't find enough to answer confidently."


def answer(question, retrieve=_retrieve, judge=_judge, generate=_generate):
    """Run the loop. The three pieces are injectable so tests can script
    them. Returns dict with answer, steps used, and the trace."""
    subquery = question
    context: list[str] = []
    trace: list[dict] = []
    for step in range(MAX_STEPS):        # the leash
        hits = retrieve(subquery)
        for h in hits:
            if h not in context:
                context.append(h)
        grade = judge(question, context)
        trace.append({"step": step, "subquery": subquery,
                      "hits": len(hits), "sufficient": grade.sufficient})
        if grade.sufficient:
            return {"answer": generate(question, context),
                    "steps": step + 1, "trace": trace, "abstained": False}
        subquery = f"{question} (need: {grade.gap})"   # reformulate
    return {"answer": ABSTAIN, "steps": MAX_STEPS,
            "trace": trace, "abstained": True}


if __name__ == "__main__":
    r = answer("compare the refund policy to the 2024 version")
    for t in r["trace"]:
        print(t)
    print("ANSWER:", r["answer"])
    print("steps:", r["steps"])

Step 6 · Run it — watch the loop reason expert

terminalpython loop.py

{'step': 0, 'subquery': 'compare the refund policy to the 2024 version', 'hits': 1, 'sufficient': False}
{'step': 1, 'subquery': 'compare the refund policy to the 2024 version (need: the 2024 policy)', 'hits': 1, 'sufficient': True}
ANSWER: The current refund policy allows 30 days. | The 2024 refund policy allowed 14 days. [grounded]
steps: 2

Step 7 · Tests (no key — scripted pieces) expert

Create tests/test_loop.py. By injecting scripted retrieve/judge/generate, we test the loop's control flow deterministically — including a grader that never says sufficient (to prove the cap + abstention).

tests/test_loop.py"""Offline loop tests with scripted pieces — no key."""
from loop import answer, MAX_STEPS, ABSTAIN
from pieces import Grade


def test_answers_when_first_retrieval_suffices():
    r = answer("q", retrieve=lambda q: ["fact"],
               judge=lambda question, ctx: Grade(True),
               generate=lambda question, ctx: "done")
    assert r["steps"] == 1 and not r["abstained"]


def test_reformulates_then_answers():
    calls = {"n": 0}
    def judge(question, ctx):
        calls["n"] += 1
        return Grade(True) if calls["n"] >= 2 else Grade(False, gap="more")
    r = answer("q", retrieve=lambda q: ["f"], judge=judge,
               generate=lambda question, ctx: "done")
    assert r["steps"] == 2 and not r["abstained"]


def test_stops_at_max_steps():
    r = answer("q", retrieve=lambda q: ["f"],
               judge=lambda question, ctx: Grade(False, gap="x"),
               generate=lambda question, ctx: "done")
    assert r["steps"] == MAX_STEPS


def test_abstains_when_never_sufficient():
    r = answer("q", retrieve=lambda q: ["f"],
               judge=lambda question, ctx: Grade(False, gap="x"),
               generate=lambda question, ctx: "done")
    assert r["abstained"] and r["answer"] == ABSTAIN


def test_trace_records_every_step():
    r = answer("q", retrieve=lambda q: ["f"],
               judge=lambda question, ctx: Grade(False, gap="x"),
               generate=lambda question, ctx: "done")
    assert len(r["trace"]) == MAX_STEPS


def test_real_multihop_uses_two_steps():
    # uses the built-in mock pieces (no key)
    r = answer("compare the refund policy to the 2024 version")
    assert r["steps"] == 2 and not r["abstained"]

terminalpython -m pytest tests/ -v

tests/test_loop.py::test_answers_when_first_retrieval_suffices PASSED
tests/test_loop.py::test_reformulates_then_answers PASSED
tests/test_loop.py::test_stops_at_max_steps PASSED
tests/test_loop.py::test_abstains_when_never_sufficient PASSED
tests/test_loop.py::test_trace_records_every_step PASSED
tests/test_loop.py::test_real_multihop_uses_two_steps PASSED

6 passed in 0.05s

Step 8 · Go live with Claude (optional) expert

terminalexport ANTHROPIC_API_KEY="sk-ant-your-key-here"
export USE_REAL_API=1
python loop.py

# now judge() and generate() call Claude; the loop logic is unchanged

Troubleshooting — every error you might hit expert

CORPUS = {"cur": "The current refund policy allows 30 days.",
          "old": "The 2024 refund policy allowed 14 days."}
ABSTAIN = "I couldn't find enough to answer confidently."

def retrieve(query):
    hits = []
    if "current" in query or "refund" in query: hits.append(CORPUS["cur"])
    if "2024" in query or "old" in query:        hits.append(CORPUS["old"])
    return hits

def plain_rag(question):
    ctx = retrieve(question)
    if not ctx:
        return ABSTAIN                       # never guess
    return " ".join(ctx) + " [grounded]"

print(plain_rag("what is the refund policy"))    # answers from 'cur'
print(plain_rag("what is your uptime SLA"))       # abstains

from dataclasses import dataclass

@dataclass
class Grade:
    sufficient: bool
    gap: str = ""

def judge(question, context):
    joined = " ".join(context)
    if "compare" in question.lower():        # needs both policies
        has_cur = "current" in joined
        has_old = "2024" in joined
        if has_cur and has_old:
            return Grade(True)
        return Grade(False, gap="the 2024 policy" if has_cur else "the current policy")
    return Grade(bool(context))

print(judge("compare refund to 2024", ["The current refund policy allows 30 days."]))
# Grade(sufficient=False, gap='the 2024 policy')

from dataclasses import dataclass
@dataclass
class Grade:
    sufficient: bool; gap: str = ""
CORPUS = {"cur": "current: 30 days", "old": "2024: 14 days"}
MAX_STEPS, ABSTAIN = 4, "couldn't find enough"

def retrieve(q):
    h = []
    if "current" in q or "refund" in q: h.append(CORPUS["cur"])
    if "2024" in q or "old" in q:        h.append(CORPUS["old"])
    return h
def judge(question, ctx):
    j = " ".join(ctx)
    if "compare" in question.lower():
        return Grade(True) if ("current" in j and "2024" in j) else \
               Grade(False, "the 2024 policy" if "current" in j else "the current policy")
    return Grade(bool(ctx))

def answer(question):
    subquery, context, trace = question, [], []
    for step in range(MAX_STEPS):                 # THE LEASH
        for h in retrieve(subquery):
            if h not in context: context.append(h)
        g = judge(question, context)
        trace.append({"step": step, "sufficient": g.sufficient})
        if g.sufficient:
            return {"answer": " | ".join(context), "steps": step + 1, "trace": trace}
        subquery = f"{question} (need: {g.gap})"   # REFORMULATE
    return {"answer": ABSTAIN, "steps": MAX_STEPS, "trace": trace}

print(answer("compare the refund policy to the 2024 version"))

import json
from dataclasses import dataclass

@dataclass
class Grade:
    sufficient: bool
    gap: str = ""

def real_judge(question, context):
    import anthropic
    client = anthropic.Anthropic()                    # reads ANTHROPIC_API_KEY
    msg = client.messages.create(
        model="claude-opus-4-8",
        max_tokens=150,
        system='Reply ONLY JSON: {"sufficient": bool, "gap": str}.',
        messages=[{"role": "user",
                   "content": f"QUESTION: {question}\nCONTEXT: {context}"}],
    )
    d = json.loads(msg.content[0].text)
    return Grade(bool(d["sufficient"]), d.get("gap", ""))
# Same Grade return type as the stub -> the loop code doesn't change.
#
# OpenAI drop-in (needs OPENAI_API_KEY): swap the client + call, parse the same JSON:
#   import openai; client = openai.OpenAI()
#   resp = client.responses.create(model="gpt-5.5", max_output_tokens=150,
#       instructions='Reply ONLY JSON: {"sufficient": bool, "gap": str}.',
#       input=f"QUESTION: {question}\nCONTEXT: {context}")
#   d = json.loads(resp.output_text)

def evaluate(cases, plain_fn, agent_fn):
    plain_correct = agent_correct = total_steps = 0
    for q, correct in cases:
        if plain_fn(q) == correct: plain_correct += 1
        r = agent_fn(q)
        if r["answer"] == correct: agent_correct += 1
        total_steps += r["steps"]
    n = len(cases)
    return {"plain_acc": plain_correct/n, "agent_acc": agent_correct/n,
            "uplift": (agent_correct - plain_correct)/n,
            "avg_steps": total_steps/n}

# stubbed fns: plain misses multi-hop, agent solves in 2 steps, easy in 1
cases = [("multi-hop", "AB"), ("easy", "A")]
plain = lambda q: "A" if q == "easy" else "wrong"
agent = lambda q: {"answer": "AB" if q=="multi-hop" else "A",
                   "steps": 2 if q=="multi-hop" else 1}
print(evaluate(cases, plain, agent))   # uplift on multi-hop, avg_steps low

MAX_STEPS = 4
MAX_COST = 0.05          # $ per answer ceiling
COST_PER_STEP = 0.015    # retrieve + judge + generate, illustrative

def answer_bounded(retrieve, judge, generate, question):
    context, spent = [], 0.0
    for step in range(MAX_STEPS):
        spent += COST_PER_STEP
        if spent > MAX_COST:                        # cost leash trips first
            return {"answer": "abstain (cost budget hit)", "steps": step, "spent": spent}
        context += retrieve(question)
        if judge(question, context):
            return {"answer": generate(question, context),
                    "steps": step + 1, "spent": round(spent, 3)}
    return {"answer": "abstain (step cap hit)", "steps": MAX_STEPS, "spent": spent}

# never-sufficient judge -> whichever leash is tighter stops it
print(answer_bounded(lambda q: ["x"], lambda q,c: False, lambda q,c: "ans", "q"))

agentic-rag/
├─ .venv/
├─ requirements.txt
├─ pieces.py          (retrieve, judge, generate — mock + real)
├─ loop.py            (the bounded retrieve->grade->reformulate loop)
└─ tests/
   └─ test_loop.py    (6 offline tests)

Knowledge check check yourself