Multi-agent RAG pipelines

1 · One agent with tools vs a multi-agent RAG pipeline

Picture the moment this lesson exists for. You built a RAG agent in M3, it worked, and then the requests piled on: "also search the wiki, not just the docs"; "stop making things up — check the answer first"; "the judging step is burning our expensive model on trivial relevance calls." So you kept bolting instructions onto one system prompt until it read like a job description for five people. It half-plans, half-retrieves, half-judges, half-writes, half-verifies — and does none of them reliably. That bloated, do-everything prompt is the pain a multi-agent RAG pipeline cures.

The fix is the oldest trick in engineering: divide the labour. Think of a newsroom instead of a lone blogger. A blogger researches, writes, and fact-checks the same piece alone — fine for a small blog, but it doesn't scale and nobody's checking the writer's own claims. A newsroom splits those into an editor who assigns the story, reporters who gather sources, a copy editor who cuts the weak material, a writer who drafts, and a fact-checker with the authority to send a piece back. Each role is simpler, individually good at its job, separately measurable, and — crucially — the fact-checker is not the same person who wrote the story. A multi-agent RAG pipeline is that newsroom: router, retriever, grader, synthesizer, verifier, each a small focused agent handing work down the line through a shared notepad.

The common mistake — and it is common — is reaching for the pipeline too early. "Multi-agent" sounds sophisticated, so people split a perfectly good single agent into five just to have an impressive diagram. Every agent you add is another prompt to maintain, more tokens, more latency, and one more component that can misbehave — you have multiplied your surface area for bugs. The honest rule: M3's single agent with a retrieve→grade→rewrite loop is the right default. Split into a pipeline only when a specific stage genuinely needs its own prompt, its own model tier, or its own evaluation gate — the signals in the table below. If you can't point at which stage needs to be separate and why, you don't need a pipeline yet.

In M3 a single agent owns the whole flow: it decides when to retrieve, grades what came back, and answers. That is the right default. A multi-agent RAG pipeline is different — it splits the work across separate, specialized agents that each do one job well and hand off through a shared state object. You reach for it only when a single agent's one prompt is doing too many jobs at once.

2 · The pipeline roles — a canonical decomposition

Here is the good news that saves you a design meeting. You do not have to invent the roles from a blank page — the same five keep showing up in every serious RAG pipeline because they map to the five distinct jobs hidden inside "answer this question from documents." Learn them once and you'll recognize them in any framework, any codebase, any vendor's diagram: router, retriever, grader, synthesizer, verifier. Different names sometimes (planner, re-ranker, critic), same jobs.

The analogy that makes them stick is a restaurant kitchen taking an order. The router is the expediter reading the ticket and deciding which stations it goes to. The retriever is the line cooks pulling ingredients from the walk-in — maybe several at once from different fridges. The grader is the cook who inspects and throws out anything wilted before it hits the pan — junk in equals junk out. The synthesizer is the chef plating the dish from the good ingredients only. And the verifier is the head chef who inspects the plate before it leaves the pass and can send it back to be redone. Notice the pattern: work flows one direction, each station does one thing, and the last station has a veto.

The subtle point people miss is why the grader and the verifier both exist — they look like the same "quality check" and the instinct is to merge them to save an agent. Resist it. They guard opposite ends of the pipeline. The grader guards the input: it protects the synthesizer from irrelevant retrieved chunks so the answer is written from clean context. The verifier guards the output: it protects the user from a fluent, confident answer the context never actually supported — a hallucination. An answer can pass the grader (the chunks were relevant) and still fail the verifier (the draft claimed something not in those chunks). Collapsing them is exactly the "the writer fact-checks their own story" mistake from §1, and it is the single most valuable reason to run a pipeline at all. The table below is your reference card: each agent, its one job, and precisely what it reads from and writes to the shared state.

A multi-agent RAG pipeline is a small assembly line of specialists. Each agent reads the shared state, does one job, and writes its result back for the next stage. The canonical five:

3 · Orchestration & shared state

You have five specialists. Now: how do they actually pass work to each other? This is the question that separates a pipeline that you can debug at 2am from one that becomes an untraceable tangle of who-called-whom. The wiring matters as much as the agents, because a bug in a five-agent system is rarely "the synthesizer is wrong" — it's "the grader dropped everything and the synthesizer never noticed." How the agents communicate decides whether you can see that.

Think of the two coordination styles as two ways an office team works. One team shares a single whiteboard: everyone reads the current state off it, does their bit, and writes their result back for the next person. Anyone walking past can glance at the board and know exactly where the work stands. The other team fires emails at each other — flexible and great when the flow is a free-for-all, but reconstructing "what happened" means digging through everyone's inbox. The whiteboard is the blackboard / shared state pattern (one State object every agent reads and writes — what LangGraph's StateGraph gives you); the emails are message passing (AutoGen-style, M2). For a RAG pipeline, whose work flows mostly in one direction, the whiteboard wins: the code reads top-to-bottom like a data flow, and the state object is your debugging view.

Now the part that turns a straight line into a real pipeline, and the part beginners most often get dangerously wrong: the back-edges. A RAG pipeline is not purely one-directional. The grader might look at what it kept, decide the context is too thin, and send control back to the retriever to try again with a rewritten query — that is M3's retrieve→grade loop, now living as an edge between two separate stages. The verifier might reject a draft as unfaithful and send it back to re-synthesize or re-retrieve. So the pipeline is really a small state machine: mostly forward, with a couple of loops. The common — and costly — mistake is to add a back-edge and forget to bound it. An ungroundable question then loops retrieve→grade→retrieve forever, or until your token bill does the stopping for you. Every back-edge must carry a shared counter (the tries field in the lab below) so it can spin only a fixed number of times before the pipeline gives up honestly. A loop without a bound isn't a feature; it's an outage waiting for the wrong input.

How do the agents talk? Two common shapes. In a blackboard / shared scratchpad, every agent reads and writes one state object — simple, inspectable, and what LangGraph's StateGraph gives you. In message passing, agents send messages to each other (AutoGen-style, M2). For a linear-ish RAG pipeline the shared state object is usually the cleaner choice — the pipeline reads like a data flow.

The interesting part is conditional routing: the grader may decide the context is too weak and send control back to retrieve (the M3 loop, now a stage-to-stage edge); the verifier may reject the draft and send back to synthesize or re-retrieve. That is the whole pipeline — a state machine with a couple of back-edges, every one of them bounded.

Lab M5.1 · The orchestration, offline & runnable

You do not need a cluster or an API key to get the orchestration logic right. Here the whole pipeline runs on deterministic fakes — a hash-based fake embed/retrieve, a term-overlap grader, a template synthesizer, a claim-support verifier — so the control flow (routing, the bounded loop, the send-back) is genuinely exercisable. Swap the fakes for real models later; the wiring is what this lab pins down.

pipeline.py"""Multi-agent RAG pipeline — ORCHESTRATION only, fully offline & deterministic.
No API key, no cluster: fake embed/retrieve + grader/synth/verify agents + the loop.
Run: python3 pipeline.py"""
from dataclasses import dataclass, field

# ---- shared state (the blackboard every agent reads/writes) ----------------
@dataclass
class State:
    question: str
    plan: str = ""
    sources: tuple = ()
    candidates: list = field(default_factory=list)   # retriever output
    context: list = field(default_factory=list)       # grader-kept chunks
    draft: str = ""
    cites: list = field(default_factory=list)
    verdict: str = ""
    tries: int = 0
    trace: list = field(default_factory=list)

CORPUS = {
    "refund":  "Refunds are accepted within 30 days of purchase.",
    "reset":   "Reset your password in Settings -> Security.",
    "shipping":"Standard shipping takes 3-5 business days.",
}

def _score(query, text):            # deterministic fake "relevance": prefix-term overlap
    q = query.lower().split(); t = text.lower().split()
    # count query words that prefix-match some doc word ("refund" ~ "refunds")
    return sum(any(w.startswith(qw) or qw.startswith(w) for w in t) for qw in q)

# ---- the specialized agents (each does ONE job) ----------------------------
def router(s: State) -> State:
    s.plan = "lookup"; s.sources = ("vector",)
    s.trace.append(("router", s.plan)); return s

def retriever(s: State) -> State:                     # pull candidates from the index
    s.candidates = [(k, v, _score(s.question, v)) for k, v in CORPUS.items()]
    s.tries += 1
    s.trace.append(("retriever", f"try#{s.tries} pulled {len(s.candidates)}")); return s

def grader(s: State, keep_min=1) -> State:            # drop junk, keep the relevant
    s.context = [(k, v) for k, v, sc in s.candidates if sc >= keep_min]
    s.trace.append(("grader", f"kept {len(s.context)}/{len(s.candidates)}")); return s

def synthesizer(s: State) -> State:                   # grounded, cited draft
    if s.context:
        s.draft = " ".join(v for _, v in s.context)
        s.cites = [k for k, _ in s.context]
    else:
        s.draft = ""; s.cites = []
    s.trace.append(("synth", f"draft={bool(s.draft)} cites={s.cites}")); return s

def verifier(s: State) -> State:                      # faithful to context? every
    ctx = " ".join(v for _, v in s.context).lower()   # answer word must appear in ctx
    unsupported = [w for w in s.draft.lower().split() if w not in ctx]
    s.verdict = "faithful" if s.draft and not unsupported else "unfaithful"
    s.trace.append(("verifier", s.verdict)); return s

# ---- orchestrator: the state machine with BOUNDED back-edges ---------------
def run(question, max_tries=3):
    s = State(question=question)
    s = router(s)
    while True:
        s = retriever(s)
        s = grader(s)
        if not s.context and s.tries < max_tries:     # weak retrieval -> loop back
            s.question += " policy"                    # (a real router would rewrite)
            continue
        s = synthesizer(s)
        s = verifier(s)
        if s.verdict == "faithful":
            return {"answer": s.draft, "cites": s.cites, "trace": s.trace}
        if s.tries >= max_tries:                        # bound the send-back loop
            return {"answer": "I don't have enough grounded information to answer.",
                    "cites": [], "trace": s.trace}
        # unfaithful but budget left -> send back to retrieve for more context

if __name__ == "__main__":
    import json
    print(json.dumps(run("what is the refund window"), indent=2))
    print(json.dumps(run("who is the president"), indent=2))   # ungroundable -> refuse

{
  "answer": "Refunds are accepted within 30 days of purchase.",
  "cites": ["refund"],
  "trace": [["router","lookup"],["retriever","try#1 pulled 3"],
            ["grader","kept 1/3"],["synth","draft=True cites=['refund']"],
            ["verifier","faithful"]]
}
{
  "answer": "I don't have enough grounded information to answer.",
  "cites": [], "trace": [ ... router/retriever/grader loop x3 ... ]
}

The same shape in a real framework (needs the lib / API key)

The offline lab above is the control flow. A real deployment swaps the fakes for models and expresses the graph in a framework. Here is the LangGraph shape — needs pip install langgraph and a model/API key; the node functions are the same agents:

graph_langgraph.py# needs: pip install langgraph  + a model/API key. Node fns = the agents above.
from langgraph.graph import StateGraph, START, END

g = StateGraph(dict)                       # dict state == the blackboard
g.add_node("router", router)
g.add_node("retriever", retriever)
g.add_node("grader", grader)
g.add_node("synth", synthesizer)
g.add_node("verify", verifier)

g.add_edge(START, "router")
g.add_edge("router", "retriever")
g.add_edge("retriever", "grader")
# grader decides: enough context -> synth, else loop back to retriever (bounded in state)
g.add_conditional_edges("grader",
    lambda s: "synth" if s["context"] or s["tries"] >= 3 else "retriever",
    {"synth": "synth", "retriever": "retriever"})
g.add_edge("synth", "verify")
# verifier decides: faithful -> END, else send back to retriever (bounded)
g.add_conditional_edges("verify",
    lambda s: END if s["verdict"] == "faithful" or s["tries"] >= 3 else "retriever",
    {END: END, "retriever": "retriever"})
app = g.compile()
# CrewAI (needs: pip install crewai + key) models the same as role-based agents with a
# manager process; see M1. AutoGen (M2) would use message-passing between the agents.

4 · Failure modes & control

Every agent you added in §2 was a feature. Every agent you added is also a new way to fail. That is the tax nobody prints on the multi-agent brochure: more moving parts means more moving parts that can jam. A single agent has a small blast radius — it does its one prompt and returns. A five-agent pipeline with a couple of loops has feedback, budget, and hand-off failures a single agent simply cannot have. This section is the honest counterweight to §1's enthusiasm: before you ship a pipeline, you must know the ways it breaks and have a control for each.

The mental model is a factory line with a rework loop. On a well-run line, a station that finds a defect sends the part back to be redone — good. But without limits, three specific disasters follow, and they're worth feeling before you meet them in a table. Loop forever: two stations bounce a part back and forth (grader rejects → retrieve → grader rejects again) and the line never produces anything. Cost blow-up: every trip through a station costs money, and a five-station line run three times is fifteen expensive operations — a single user question can quietly cost fifteen model calls. Deadlock by disagreement: the grader keeps passing a part the verifier keeps rejecting, and the two fight forever because neither has final authority. And the quiet fourth one, no graceful give-up: when the line genuinely can't build the part, a bad factory ships a broken one anyway — the pipeline invents an answer rather than admitting defeat.

The controls all share one idea: someone has to be allowed to stop. The beginner instinct is to trust the agents to sort it out among themselves — they won't, because each is optimizing its own narrow job. So you impose limits from outside. Bound every back-edge with a shared iteration cap (the tries counter from §3). Bound total spend with a call/token budget, and shift cheap work to a cheap model — router and grader don't need your strongest, most expensive model; the synthesizer does. Give the verifier final authority to end a run so a grader/verifier disagreement resolves instead of thrashing, and log the disagreement so you can tune it later. And make the give-up path a real, deliberate output: on budget exhaustion, return an explicit "I don't have enough grounded information to answer" — never a fabricated one. An honest "I don't know" is a feature; a confident wrong answer is the worst thing a RAG system can do. The table names each failure and its control.

A multi-agent pipeline has more moving parts than a single agent, so it has more ways to go wrong. The four that bite in production:

Lab M5.2 · A control / budget guard

One small object enforces the bounds: a cap on iterations and on total model calls, shared across all agents. It is offline and deterministic — the point is the accounting, not the model behind each call.

budget.py"""A shared budget guard for the pipeline. Offline & deterministic.
Bounds BOTH the loop (iterations) and the spend (model calls). Run: python3 budget.py"""
class Budget:
    def __init__(self, max_iters=3, max_calls=12):
        self.max_iters = max_iters; self.max_calls = max_calls
        self.iters = 0; self.calls = 0

    def spend(self, n=1):                 # every agent call goes through here
        self.calls += n
        if self.calls > self.max_calls:
            raise BudgetExceeded(f"call budget blown: {self.calls}/{self.max_calls}")

    def next_iter(self) -> bool:          # False -> stop looping, give up gracefully
        self.iters += 1
        return self.iters <= self.max_iters

class BudgetExceeded(Exception):
    pass

def guarded_pipeline(question, budget: Budget):
    """Model the loop with the guard deciding when to stop."""
    trace = []
    while budget.next_iter():
        budget.spend(2)                   # e.g. retriever + grader call
        grounded = "refund" in question.lower()   # fake: only 'refund' is groundable
        trace.append((budget.iters, budget.calls, "hit" if grounded else "miss"))
        if grounded:
            budget.spend(2)               # synth + verify
            return {"answer": "Refunds within 30 days.", "cost": budget.calls, "trace": trace}
        question += " policy"             # rewrite & retry within budget
    # loop exhausted WITHOUT a grounded hit -> the honest fallback
    return {"answer": "I don't know — no grounded context found.",
            "cost": budget.calls, "trace": trace}

if __name__ == "__main__":
    print(guarded_pipeline("what is the refund window", Budget()))
    print(guarded_pipeline("what is the meaning of life", Budget()))
    try:
        guarded_pipeline("x", Budget(max_iters=99, max_calls=3))
    except BudgetExceeded as e:
        print("STOPPED:", e)

{'answer': 'Refunds within 30 days.', 'cost': 4, 'trace': [(1, 2, 'hit')]}
{'answer': "I don't know — no grounded context found.", 'cost': 6, 'trace': [(1,2,'miss'),(2,4,'miss'),(3,6,'miss')]}
STOPPED: call budget blown: 4/3

5 · Evaluating the whole pipeline

Imagine the pipeline's grounded-answer rate drops from 90% to 70% overnight. Something regressed. Which agent? If your only measurement is "did the final answer come out right," you're standing in front of a five-station assembly line that just started producing defects and you have no idea which station broke. You'll spend a day guessing. This section is about never being in that position — and it hinges on one insight beginners routinely miss: evaluating a multi-agent pipeline is not one number, it is two layers of numbers.

Here's the analogy. A car has to pass an end-to-end test — does it drive from A to B safely? — but a mechanic diagnosing a problem also reads per-component gauges: tire pressure, oil level, brake wear. The end-to-end test tells you that something is wrong; the gauges tell you what. A pipeline is identical. The end-to-end metric is the grounded-answer rate: given a real question, is the final answer both grounded and correct? That's the number your lead cares about, the one on the dashboard. But it's the per-stage metrics — retriever recall@k, grader precision, verifier catch-rate — that tell you which agent to fix when the end-to-end number drops. Measure only the end-to-end number and you're the driver who knows the car is sluggish but can't pop the hood.

The trap here is subtle and specific to pipelines, so name it plainly: measuring retrieval recall alone and calling it done. Recall@k is a great retriever gauge, and it's tempting to stop there because M3 and Chapter 3 leaned on it. But a pipeline's whole selling point over M3 was the verifier — and a verifier that rubber-stamps everything is worse than useless, because it makes a single-agent RAG feel safe while adding latency and cost. So the most important pipeline-specific metric is the verifier catch-rate: deliberately feed it drafts you know are ungrounded and measure the fraction it rejects. That number is the honest answer to "does this pipeline actually guard against hallucination, or does it just look like it does?" You reuse the FA5 machinery per stage rather than reinventing it — recall@k for the retriever, faithfulness scoring for the verifier — and layer the end-to-end grounded-answer rate on top. The table splits the two layers so you always know which question each number is answering.

FA5 covers RAG evaluation — retrieval recall@k, faithfulness, and the eval-set scorecard. Do not repeat that work; reuse it per stage. A multi-agent pipeline needs two layers of evaluation:

The end-to-end number is the one leaders care about, but it's the per-stage numbers that tell you which agent to fix when it drops. The most important pipeline-specific metric is the verifier catch-rate: feed it drafts you know are ungrounded and measure how many it rejects — that is the difference between a pipeline that guards hallucination and one that just looks like it does.

Lab M5.3 · Recall@k + a faithfulness gate, offline

Two offline evals that run without a judge model. recall_at_k reuses the FA5 set-overlap idea for the retriever stage; faithfulness_gate is the end-to-end check the verifier stage performs. Deterministic — swap in a real judge later.

eval.py"""Pipeline evaluation, offline & deterministic (no judge model needed).
Per-stage recall@k for the retriever + an end-to-end faithfulness gate. Run: python3 eval.py"""

# ---- per-stage: retriever recall@k (FA5 set-overlap idea) ------------------
def recall_at_k(retrieved_ids, gold_ids, k):
    top = set(retrieved_ids[:k]); gold = set(gold_ids)
    return len(top & gold) / len(gold) if gold else 0.0

# ---- end-to-end: faithfulness gate (what the verifier agent enforces) ------
import re
STOP = {"is","are","the","a","an","of","in","to","and","our","within"}

def _words(text):                    # lowercase, strip punctuation ("days." -> "days")
    return re.findall(r"[a-z0-9]+", text.lower())

def faithfulness(answer, context):
    """Fraction of answer content-words supported by the retrieved context."""
    ctx = set(w for c in context for w in _words(c)) - STOP
    ans = [w for w in _words(answer) if w not in STOP]
    if not ans:
        return 0.0
    return sum(w in ctx for w in ans) / len(ans)

def faithfulness_gate(answer, context, threshold=0.8):
    score = faithfulness(answer, context)
    return {"score": round(score, 2),
            "verdict": "pass" if score >= threshold else "FAIL (ungrounded)"}

def evaluate_pipeline(cases):
    """cases: list of (retrieved_ids, gold_ids, answer, context)."""
    recalls, passes = [], 0
    for rid, gid, ans, ctx in cases:
        recalls.append(recall_at_k(rid, gid, k=3))
        if faithfulness_gate(ans, ctx)["verdict"] == "pass":
            passes += 1
    return {"mean_recall@3": round(sum(recalls)/len(recalls), 2),
            "grounded_answer_rate": round(passes/len(cases), 2)}

if __name__ == "__main__":
    print(faithfulness_gate("Refunds are accepted within 30 days.",
                            ["Refunds are accepted within 30 days of purchase."]))
    print(faithfulness_gate("The CEO is Bob.",                       # ungrounded
                            ["Refunds are accepted within 30 days."]))
    cases = [
        (["refund","reset","x"], ["refund"], "Refunds within 30 days.",
         ["Refunds are accepted within 30 days."]),
        (["a","b","c"], ["shipping"], "The CEO is Bob.",             # bad retrieval + ungrounded
         ["Refunds are accepted within 30 days."]),
    ]
    print(evaluate_pipeline(cases))

{'score': 1.0, 'verdict': 'pass'}
{'score': 0.33, 'verdict': 'FAIL (ungrounded)'}
{'mean_recall@3': 0.5, 'grounded_answer_rate': 0.5}

6 · Production shape (optional)

The offline lab runs on your laptop with fakes. Production is where the fakes become real models, real users, and a real bill — and where a design that looked clean in a diagram meets the unglamorous questions of where it runs, how you watch it, and what happens when it's wrong on a high-stakes answer. The reassuring news: you have already learned most of this. A multi-agent RAG pipeline is a set of stages, and stages deploy exactly like any other LLM service you've shipped in this course. There is no exotic new infrastructure here — just a handful of notes specific to the fact that you now have several agents instead of one.

The instinct to worry about is treating the pipeline as one indivisible black box in production. Don't. Its greatest operational strength is the same thing that made it worth building: it comes pre-divided into stages, and each stage is separately deployable, separately scalable, and separately observable. Heavy retrieval/fusion can be its own scalable service on the infra you know (K7). Each agent gets its own trace (I4) so the per-stage metrics from §5 become real dashboards, not aspirations. Cost tiers from §4 map straight onto deployment — cheap model for router/grader, strong for synthesizer/verifier, budget guard enforcing the ceiling.

Two production concerns deserve special care because they're easy to skip and expensive to skip. First, the human-in-the-loop gate: for high-stakes answers, a verifier FAIL (or a low-confidence pass) should route to a person, not auto-answer — the pipeline's honesty about uncertainty is only useful if something acts on it (L5's HITL pattern). Second, and most under-appreciated: every retrieved chunk is untrusted input. More retrieval hops means more surface for prompt injection — a poisoned document in your index can carry instructions the synthesizer might obey. Treat retrieved text as data, never as commands, and keep any destructive tool actions gated behind their own checks (T1). A pipeline that fuses six sources has six times the injection surface of a single lookup; the checklist below is how you deploy it without shipping that risk to your users.

Where does this run? The pipeline is a set of stages, so it maps cleanly onto the deployment patterns you already know. A few production notes specific to multi-agent RAG:

Production checklist

1. Runs as stages — deploy on the same infra as any LLM service (K7); heavy retrieval/fusion can be its own scalable service.
2. Observability per agent — trace each stage separately (I4); the per-stage metrics from §5 are your dashboards.
3. Human-in-the-loop gate — for high-stakes answers, route a verifier FAIL (or a low-confidence pass) to a human instead of auto-answering (L5's HITL pattern).
4. Cost tiers — cheap model for router/grader, strong model for synthesizer/verifier; the budget guard from §4 enforces the ceiling.
5. Treat retrieved text as untrusted — more retrieval hops = more prompt-injection surface (T1); keep destructive actions gated.

Common pitfalls

def grade(query, candidates, keep_min=1):
    q = set(query.lower().split())
    scored = [(cid, text, len(q & set(text.lower().split()))) for cid, text in candidates]
    kept = [(cid, text) for cid, text, sc in scored if sc >= keep_min]
    # best first
    kept.sort(key=lambda ct: -len(q & set(ct[1].lower().split())))
    return kept

cands = [("a", "Refunds are accepted within 30 days"),
         ("b", "Our office is in Berlin"),
         ("c", "Refund requests must include the order id")]
print(grade("how long is the refund window", cands))
# -> keeps a and c (mention refund), drops b

CORPUS = {"refund": "Refunds within 30 days.", "reset": "Reset in Settings."}

def retrieve(query):
    return [(k, v) for k, v in CORPUS.items()
            if any(w in query.lower() for w in k.split())]

def grade(query, cands, keep_min=1):
    q = set(query.lower().split())
    return [(k, v) for k, v in cands if len(q & set(v.lower().split())) >= keep_min or k in query.lower()]

def loop(query, max_tries=3):
    for attempt in range(1, max_tries + 1):
        context = grade(query, retrieve(query))
        if context:
            return f"answer from {[k for k,_ in context]} (try {attempt})"
        query += " refund"                 # rewrite & retry
    return "give up: no grounded context (would say 'I don't know')"

print(loop("what is the refund window"))
print(loop("money back"))            # rewritten -> refund
print(loop("what is the weather"))    # ungroundable -> give up

STOP = {"is","are","the","a","an","of","within","our"}

def faithfulness(draft, context):
    ctx = set(w for c in context for w in c.lower().split()) - STOP
    ans = [w for w in draft.lower().split() if w not in STOP]
    return sum(w in ctx for w in ans) / len(ans) if ans else 0.0

def verify(draft, context, threshold=0.8):
    return "faithful" if faithfulness(draft, context) >= threshold else "unfaithful"

def synth_and_verify(context, bad_synth, max_tries=2):
    for attempt in range(1, max_tries + 1):
        # attempt 1 uses a (possibly bad) synth; later attempts stick to context
        draft = bad_synth if attempt == 1 else " ".join(context)
        if verify(draft, context) == "faithful":
            return f"ship: '{draft}' (try {attempt})"
    return "give up: could not produce a faithful draft -> 'I don't know'"

ctx = ["Refunds are accepted within 30 days."]
print(synth_and_verify(ctx, bad_synth="Refunds are accepted within 30 days."))  # faithful
print(synth_and_verify(["Shipping takes 3-5 days."], bad_synth="The CEO is Bob."))  # rejected

from collections import defaultdict

def rrf(rankings, k=60):
    """rankings: list of ranked id-lists (best first). Returns fused id list."""
    score = defaultdict(float)
    for ranked in rankings:
        for rank, doc_id in enumerate(ranked):     # rank 0 = best
            score[doc_id] += 1.0 / (k + rank + 1)
    return sorted(score, key=lambda d: -score[d])

vector_hits = ["d1", "d2", "d3"]     # source A's ranking
graph_hits  = ["d3", "d1", "d4"]     # source B disagrees
fused = rrf([vector_hits, graph_hits])
print(fused)   # d1 & d3 rank well in BOTH -> float to the top
# feed the fused list into the grader from Exercise 1

from collections import defaultdict

class Budget:
    def __init__(self, max_iters=3, max_calls=12):
        self.max_iters, self.max_calls = max_iters, max_calls
        self.iters = 0; self.by_stage = defaultdict(int)
    @property
    def calls(self): return sum(self.by_stage.values())
    def spend(self, stage, n=1):
        self.by_stage[stage] += n
        if self.calls > self.max_calls:
            raise RuntimeError(f"budget blown: {self.calls}/{self.max_calls}")
    def next_iter(self):
        self.iters += 1
        return self.iters <= self.max_iters

def run(question, budget):
    while budget.next_iter():
        budget.spend("router"); budget.spend("retriever"); budget.spend("grader")
        if "refund" in question.lower():
            budget.spend("synth"); budget.spend("verifier")
            return {"answer": "Refunds within 30 days.",
                    "spend": budget.calls, "per_stage": dict(budget.by_stage)}
        question += " policy"
    return {"answer": "I don't know.", "spend": budget.calls,
            "per_stage": dict(budget.by_stage)}

print(run("refund window", Budget()))
print(run("meaning of life", Budget(max_iters=2)))
try:
    run("x", Budget(max_calls=2))
except RuntimeError as e:
    print("STOPPED:", e)

from collections import defaultdict

def rrf(rankings, k=60):
    s = defaultdict(float)
    for r in rankings:
        for i, d in enumerate(r): s[d] += 1/(k+i+1)
    return sorted(s, key=lambda d: -s[d])

INDEX = {"refund": "Refunds within 30 days.", "ship": "Shipping 3-5 days."}
WEB   = {"refund": "Blog: refunds honored in 30 days."}

class Budget:
    def __init__(self, max_iters=2, max_calls=14):
        self.max_iters, self.max_calls, self.iters = max_iters, max_calls, 0
        self.by_stage = defaultdict(int)
    @property
    def calls(self): return sum(self.by_stage.values())
    def spend(self, stage, n=1):
        self.by_stage[stage]+=n
        if self.calls>self.max_calls: raise RuntimeError("SLO cost ceiling hit")
    def next_iter(self):
        self.iters+=1; return self.iters<=self.max_iters

STOP={"is","are","the","a","of","within","in"}
def faithful(draft, ctx):
    c=set(w for t in ctx for w in t.lower().split())-STOP
    a=[w for w in draft.lower().split() if w not in STOP]
    return (sum(w in c for w in a)/len(a) if a else 0)>=0.7

def assistant(question, budget):
    trace=[]
    while budget.next_iter():
        budget.spend("router")
        key=next((k for k in INDEX if k in question.lower()), None)
        # multi-source retrieve + fuse
        budget.spend("retriever", 2)
        idx=[key] if key in INDEX else []
        web=[key] if key in WEB else []
        fused=rrf([idx, web]) if (idx or web) else []
        budget.spend("grader")
        ctx=[INDEX.get(d) or WEB.get(d) for d in fused if (INDEX.get(d) or WEB.get(d))]
        trace.append((budget.iters, "hit" if ctx else "miss", budget.calls))
        if not ctx:
            question+=" policy"; continue
        budget.spend("synth"); draft=" ".join(ctx)
        budget.spend("verifier")
        if faithful(draft, ctx):
            return {"answer": draft, "cost": budget.calls,
                    "per_stage": dict(budget.by_stage), "trace": trace}
        # unfaithful -> loop back within budget
    return {"answer": "I don't know — could not ground the answer.",
            "cost": budget.calls, "per_stage": dict(budget.by_stage), "trace": trace}

import json
print(json.dumps(assistant("what is the refund window", Budget()), indent=2))
print(json.dumps(assistant("who won the election", Budget()), indent=2))

# needs the lib / an API key -- production mapping (verify against current docs):
# LangGraph: StateGraph(dict) with nodes router/retriever/grader/synth/verify;
#   conditional edges enforce the bounded loop; compile() -> app. (L4)
# CrewAI:   role-based agents (researcher/grader/writer/critic) + a manager process;
#   the budget becomes a max_iter / step cap. (M1)
# Enforce the SLO with the Budget object; gate deploys on end-to-end grounded-answer
# rate AND verifier catch-rate from a held-out eval set (FA5 + section 5).

Knowledge check