Multi-Agent Research Crew
A team of specialized agents — a planner, several researchers, a critic, and a writer — collaborating on a research task the way a human team would. Where the Deep Research Agent (Project 6) is one agent looping, this is role-specialized division of labor: each agent does one job well, and an orchestrator coordinates them.
- an Anthropic API key (
ANTHROPIC_API_KEY) +pip install anthropic
What this project teaches you to design
- Role specialization: planner, researcher(s), critic, writer — each with one job.
- An orchestration pattern (sequential, hierarchical, or debate) and why.
- Inter-agent communication and shared state without chaos.
- A critic/verifier agent that catches errors before they reach the output.
The brief advanced
"One agent doing everything is a jack of all trades — we want a team of specialists." Complex research benefits from division of labor: someone plans the questions, several people dig into sources in parallel, a skeptic pokes holes, and a writer synthesizes. A multi-agent crew mirrors that — specialized roles produce higher-quality, more thorough output than a single generalist agent on genuinely hard tasks.
1 · Discovery — do you actually need a crew? advanced
| Signal | Points to |
|---|---|
| Task splits into distinct expertises / parallel sub-tasks | ⭐⭐⭐ multi-agent helps |
| Quality benefits from an independent critic/verifier | ⭐⭐⭐ add a critic role |
| Sub-tasks can run in parallel to save wall-clock | ⭐⭐ parallel researchers |
| Task is linear and a single agent handles it fine | ⭐ don't — one agent is cheaper & simpler (Project 6) |
2 · Architecture advanced
This diagram shows how the crew is wired: one coordinator at the top, a row of specialists in the middle, and the final report at the bottom. The arrows show work flowing down and results flowing back.
- The purple box at the top is the Orchestrator — it plans the work and coordinates everyone else. It's the only agent that talks to all the others.
- The middle row holds the specialists: Researcher A, B, and C each dig into a different set of sources (they can run in parallel), and the pink Critic / verifier box independently checks their claims.
- The arrows fanning out from the orchestrator are it handing tasks to each specialist; the arrows curving back down toward the bottom are their results flowing to the Writer.
- The green Writer → report box at the bottom synthesizes the verified findings into the final cited report — it receives what survived the critic, not raw research.
In short: Read it top-to-bottom: orchestrator plans, researchers gather, the critic filters, the writer synthesizes. That top-down flow is exactly the sequential pipeline the code on this page builds.
An orchestrator plans the work and coordinates specialized agents: several researchers investigate different sources (often in parallel), a critic independently verifies claims and flags weak evidence, and a writer synthesizes the verified findings into a cited report. Each agent has a narrow role and prompt — specialization is what raises quality over a single generalist.
3 · Risk & safety model advanced
| Risk | Control |
|---|---|
| 🔴 Errors compounding across agents (bad input → confident bad output) | A dedicated critic/verifier agent; every claim traced to a source; the writer only uses verified findings |
| 🔴 Runaway cost — many agents, many calls, loops | Hard step/turn caps per agent; a total token budget (E2); bounded delegation depth |
| 🟠 Agents talking in circles / no convergence | Clear termination conditions; an orchestrator that ends the round, not open-ended chat |
| 🟠 Role confusion / duplicated work | Crisp role prompts + task assignment; shared state so agents see what's done |
| 🟠 Hallucinated citations in the final report | Critic verifies citations against retrieved sources before the writer synthesizes (Project 7) |
4 · Orchestration patterns advanced
| Pattern | How it works | Best for |
|---|---|---|
| Sequential | plan → research → critique → write, in order | Clear pipeline; most research tasks |
| Hierarchical | A manager agent delegates to workers and integrates | Dynamic task decomposition |
| Parallel | Researchers run concurrently, then merge | Independent sub-questions; save wall-clock |
| Debate / critique | Agents argue/critique to surface errors | High-stakes accuracy; reducing hallucination |
crew.py (shape — CrewAI-style)planner = Agent(role="Planner", goal="break the question into sub-queries")
researcher= Agent(role="Researcher",goal="gather & cite evidence", tools=[search])
critic = Agent(role="Critic", goal="verify claims, flag weak evidence")
writer = Agent(role="Writer", goal="synthesize a cited report")
crew = Crew(agents=[planner, researcher, critic, writer],
process="sequential", max_steps=20) # the leash
This is a sketch (not the file you build later) showing the shape of a crew in CrewAI style: you declare each agent with a role and a one-line goal, then hand the whole team to a Crew that runs them. Read it as "here are four specialists and how they're wired together".
- Each
Agent(...)line creates one specialist. Theroleis its job title and thegoalis its single responsibility in plain English — planner splits the question, researcher gathers evidence, critic checks it, writer synthesizes. - Only the
researchergetstools=[search]— it's the one that needs to look things up. The others reason over what they're handed. Giving each agent only the tools it needs is a safety habit. Crew(agents=[...], process="sequential", ...)assembles the team and says run them in order: plan, then research, then critique, then write.max_steps=20is the leash — a hard cap on how many turns the crew may take, so it can't loop forever and run up cost. The comment calls this out.
Try this: This is the mental model for the rest of the page. As you build agents.py and crew.py, match each real function back to one of these four roles.
5 · Agent / tool surface advanced
| Role | Tools | Risk |
|---|---|---|
| Orchestrator | Task assignment, termination control | 🟢 coordination |
| Researcher | search / retrieval (Project 9), read sources | 🟢 read-only |
| Critic | Verify claim vs source; score confidence | 🟢 the safety gate |
| Writer | Synthesize; format; cite | 🟢 produces a draft |
| Any write/publish | Export report to a doc/ticket | 🟠 gated — human-approved |
6 · Evaluation expert
| Eval | Measures |
|---|---|
| Report quality & faithfulness | Accuracy, coverage, every claim cited (LLM-as-judge + spot human, Ch 5) |
| Critic effectiveness | Does the critic actually catch injected/planted errors? |
| Uplift vs single agent | Prove the crew beats Project 6 on hard tasks — the whole justification |
| Cost & steps per task | Total tokens/calls — is the quality worth the multi-agent tax? |
| Convergence rate | % of runs that terminate cleanly within the caps |
7 · Phased rollout expert
Skills & course map expert
| Skill | Learn it in |
|---|---|
| Multi-agent orchestration (CrewAI) | M1 |
| Conversational multi-agent (AutoGen) | M2 |
| The single-agent research loop | Project 6 |
| Verified retrieval & citations | Project 7 · Ch 3 |
| State, cycles & iteration caps | L5 |
| Quality/cost evals, budgets | Ch 5 · E2 |
By the end you will have
- Role-specialized agents (planner, researcher, critic, writer), each one job.
- A sequential orchestrator with a hard step cap.
- A critic that verifies claims and drops unsupported ones before the writer.
- Six tests, including a planted-false-claim test proving the critic works.
How to use this page expert
Steps in order. terminal = run it; file = create it with the exact contents.
Step 1 · Folder + venv expert
terminalmkdir -p research-crew/tests
cd research-crew
python3 -m venv .venv
source .venv/bin/activate # Windows: .venv\Scripts\Activate.ps1
pip install pytest "anthropic>=0.40"
pip freeze > requirements.txt
(.venv) ... Successfully installed anthropic-0.69.0 pytest-8.3.4
Before any agent code, you build an isolated workspace. A virtual environment (venv) is a private copy of Python just for this project, so the packages you install here can't clash with anything else on your machine. These are terminal commands — type them one line at a time.
mkdir -p research-crew/testsmakes the project folder and atestssub-folder in one go.cd research-crewsteps into it so every later command runs in the right place.python3 -m venv .venvcreates the private Python in a hidden.venvfolder.source .venv/bin/activateswitches your terminal to use it — you'll see(.venv)appear at the start of the prompt. (Windows uses the.ps1line instead.)pip install pytest "anthropic>=0.40"downloads the two libraries you need: pytest to run the tests and anthropic for the optional real model call later.pip freeze > requirements.txtwrites the exact versions you installed into a file, so anyone (including future-you) can recreate the same setup.
What the output means: The Successfully installed anthropic-0.69.0 pytest-8.3.4 line confirms both libraries landed in your venv. The (.venv) prefix means the environment is active.
Try this: If you close the terminal and come back, re-run source .venv/bin/activate first — the venv isn't active until you activate it in each new terminal.
Step 2 · The agents (roles as functions) expert
Create agents.py. Each role is a function that takes a task and a model_call (injectable), so tests can script exact behaviour. The default mock_call makes it run with no key.
research-crew/agents.py
agents.py"""Role-specialized agents. Each is one job; model_call is injectable."""
from dataclasses import dataclass
@dataclass
class Finding:
claim: str
source: str # the text the claim is supposed to come from
def mock_call(role: str, task):
"""Deterministic stand-in for an LLM, keyed by role. No API key."""
if role == "planner":
return ["What is our churn?", "What caused it?"]
if role == "researcher":
# returns a supported finding for each sub-query
return [Finding(claim=f"Answer to: {task}",
source=f"Answer to: {task} (source text)")]
if role == "writer":
return "REPORT: " + "; ".join(f.claim for f in task)
raise ValueError(f"unknown role {role}")
def plan(question, call=mock_call):
return call("planner", question)
def research(subquery, call=mock_call):
return call("researcher", subquery)
def write(verified_findings, call=mock_call):
return call("writer", verified_findings)
This file defines the crew's roles as plain functions and a fake LLM so the whole thing runs offline with no API key. The trick that makes it testable: every role takes a call argument, so a test can swap in its own scripted "model".
@dataclass class Findingdefines a tiny record with two fields: aclaim(something the researcher asserts) and thesourcetext it supposedly came from. Keeping the source attached is what lets the critic check claims later.mock_call(role, task)is a pretend model: given arolestring it returns fixed, predictable output. Planner returns two sub-questions; researcher returns aFindingwhose claim does appear in its source (so it's honest); writer joins the claims into a report string.plan,research, andwriteare thin wrappers. Each just callscall("<role>", ...). Thecall=mock_calldefault means they use the fake model unless you pass a real one.- Because
callis a parameter, a test can inject a fake researcher that lies — a claim whose words are NOT in its source — to prove the critic catches it (that happens in Step 4).
What the output means: Nothing prints yet — this file only defines the roles and the mock. The next file wires them into an orchestrator that actually runs them.
Try this: Read mock_call's researcher branch: the claim string is a substring of the source string. That's deliberate — an honest finding. Later you'll see a dishonest one where it isn't.
call in means a test can inject a "researcher" that lies, so you can prove the critic catches it (Step 4).Step 3 · The critic + the orchestrator expert
Create crew.py. The critic verifies each finding (claim must be supported by its source); the orchestrator runs plan → research → critique → write under a step cap, and the writer only ever sees verified findings.
research-crew/crew.py
crew.py"""Sequential crew with a step cap and a verifying critic."""
from agents import plan, research, write, mock_call, Finding
MAX_STEPS = 12
def critique(findings) -> list:
"""Keep only findings whose claim is supported by its source text.
This is the crew's defence against compounding errors."""
verified = []
for f in findings:
# 'supported' = the claim's words appear in its cited source
if f.claim.lower() in f.source.lower():
verified.append(f)
return verified
def run_crew(question, call=mock_call):
steps = 0
subqueries = plan(question, call); steps += 1
findings = []
for q in subqueries:
if steps >= MAX_STEPS:
break # the leash across all agents
findings += research(q, call); steps += 1
verified = critique(findings); steps += 1 # the gate
report = write(verified, call); steps += 1
return {"report": report, "verified": verified,
"dropped": len(findings) - len(verified), "steps": steps}
if __name__ == "__main__":
r = run_crew("Why did churn rise?")
print("REPORT:", r["report"])
print("verified:", len(r["verified"]), "dropped:", r["dropped"],
"steps:", r["steps"])
This is the heart of the project: the critic that verifies findings and the orchestrator that runs the four roles in order under a step cap. The single most important rule here — the writer only ever sees findings the critic approved.
MAX_STEPS = 12is the hard cap on total agent turns across the whole run — the leash that prevents runaway cost or infinite loops.critique(findings)is the safety gate. For each finding it checks whether the claim's text actually appears inside its source text (if f.claim.lower() in f.source.lower()). Supported claims are kept; unsupported ones are silently dropped. This is a simple stand-in for a real verifier.run_crew(question, call)runs the pipeline in order:planturns the question into sub-queries, theforloop callsresearchon each (counting a step each time andbreaking if it hitsMAX_STEPS), thencritiquefilters, thenwritegets only theverifiedlist.- It returns a dict summarising the run: the
report, theverifiedfindings, how many weredropped, and the totalsteps. Theif __name__ == "__main__"block runs a demo when you execute the file directly.
What the output means: Running it prints the report plus a one-line summary — e.g. all findings verified, 0 dropped, 5 steps — proving the pipeline ran end to end within the cap.
Try this: Notice write(verified, call) is passed verified, never the raw findings. That one choice is what stops an unverified claim from ever reaching the report.
critique() drops any claim not supported by its source — and the writer only receives verified findings. This is the crew's main safety property, so it gets a dedicated planted-error test next.Step 4 · Run it, then test it (no key) expert
terminalpython crew.py
REPORT: REPORT: Answer to: What is our churn?; Answer to: What caused it?
verified: 2 dropped: 0 steps: 5
Now create the tests, including one where a researcher lies (claim not in its source) — the critic must drop it.
research-crew/tests/test_crew.py
tests/test_crew.py"""Offline crew tests with scripted agents — no key."""
from agents import Finding
from crew import run_crew, critique, MAX_STEPS
def test_planner_produces_subqueries():
r = run_crew("why did churn rise?")
assert r["steps"] > 0 and r["report"].startswith("REPORT:")
def test_supported_claims_pass_critique():
good = [Finding(claim="net 30", source="payment is net 30 days")]
assert len(critique(good)) == 1
def test_critic_drops_planted_false_claim():
# claim is NOT in its source -> must be dropped
liar = [Finding(claim="revenue tripled", source="payment is net 30 days")]
assert len(critique(liar)) == 0
def test_writer_only_sees_verified():
mixed = [Finding(claim="net 30", source="net 30 terms"),
Finding(claim="fake", source="unrelated text")]
assert len(critique(mixed)) == 1
def test_step_cap_not_exceeded():
r = run_crew("why did churn rise?")
assert r["steps"] <= MAX_STEPS
def test_report_contains_only_verified_claims():
# build a crew where one researcher lies via a scripted call
def call(role, task):
if role == "planner": return ["q1"]
if role == "researcher":
return [Finding(claim="true fact", source="a true fact appears here"),
Finding(claim="lie", source="nothing relevant")]
if role == "writer": return "; ".join(f.claim for f in task)
r = run_crew("q", call=call)
assert "lie" not in r["report"] and r["dropped"] == 1
Six tests that prove the crew behaves — all offline, no API key, because the roles accept an injectable call. pytest treats every function named test_* as a test; assert means "this must be true, or fail".
test_planner_produces_subqueriesruns the whole crew and checks it took some steps and produced a report — a basic end-to-end smoke test.test_supported_claims_pass_critiquefeeds the critic an honest finding (claim"net 30"is inside"payment is net 30 days") and asserts it survives.test_critic_drops_planted_false_claimis the mirror: a planted lie whose claim isn't in its source must be dropped — this is the key safety test.test_writer_only_sees_verifiedmixes one good and one bad finding and asserts only one passes.test_step_cap_not_exceededchecks the run stays withinMAX_STEPS.test_report_contains_only_verified_claimsdefines its own scriptedcallwhere the researcher returns one true finding and one lie, then asserts the word"lie"never appears in the final report and exactly one claim was dropped — an end-to-end proof the critic protects the output.
What the output means: All six report PASSED and pytest prints 6 passed. The planted-lie test passing is the headline: the critic really does catch a fabricated claim.
Try this: Temporarily break the critic in crew.py (e.g. always append(f)) and re-run — the two planted-lie tests should turn red. That's your safety net doing its job.
terminalpython -m pytest tests/ -v
tests/test_crew.py::test_planner_produces_subqueries PASSED
tests/test_crew.py::test_supported_claims_pass_critique PASSED
tests/test_crew.py::test_critic_drops_planted_false_claim PASSED
tests/test_crew.py::test_writer_only_sees_verified PASSED
tests/test_crew.py::test_step_cap_not_exceeded PASSED
tests/test_crew.py::test_report_contains_only_verified_claims PASSED
6 passed in 0.04s
| Test | Proves |
|---|---|
| planner produces sub-queries | the crew runs end to end |
| supported claims pass | real findings survive the critic |
| critic drops planted false claim | the critic actually catches errors — the key safety property |
| writer only sees verified | unsupported claims can't reach the report |
| step cap not exceeded | the leash holds — no runaway cost |
| report excludes the lie | end-to-end: a lying researcher's claim never gets written |
Step 5 · Go live with Claude / CrewAI (optional) expert
Replace mock_call with a real model call — the orchestration is unchanged. A minimal real caller:
real_call.pyimport anthropic
client = anthropic.Anthropic()
ROLES = {
"planner": "Break the question into 2-4 sub-queries. One per line.",
"researcher": "Answer with facts and quote the source text you used.",
"writer": "Write a short cited report from the findings.",
}
def real_call(role, task):
msg = client.messages.create(model="claude-opus-4-8", max_tokens=600,
system=ROLES[role],
messages=[{"role":"user","content": str(task)}])
return msg.content[0].text
# then: run_crew(question, call=real_call) (after setting ANTHROPIC_API_KEY)
This shows how to go live: swap the fake mock_call for a real one that talks to Claude. Crucially, the orchestrator doesn't change at all — only the function that produces text is different.
client = anthropic.Anthropic()creates the API client; it reads your key from theANTHROPIC_API_KEYenvironment variable automatically, so the key never appears in code.ROLESis a dictionary mapping each role name to a system prompt — a one-line instruction that turns the same model into a planner, a researcher, or a writer. Specialization here is just a different prompt per role.real_call(role, task)sends one message: it picks the system prompt withROLES[role], passes the task as the user message, and returns the model's text (msg.content[0].text).- The final comment shows the payoff:
run_crew(question, call=real_call)— the exact same orchestrator, now driven by a real model instead of the mock.
Try this: Because the mock and the real caller share the same (role, task) shape, you can flip between offline testing and live runs by changing one argument. That separation — logic vs. model — is the pattern to remember.
Troubleshooting — every error you might hit expert
| What you see | What it means & the fix |
|---|---|
ModuleNotFoundError: crew / agents | Run pytest from inside research-crew/. |
| Critic passes the planted lie | The support check must compare claim vs its own source; confirm critique() uses f.source. |
| Crew never finishes | Confirm MAX_STEPS is checked in the researcher loop, not per-agent. |
| Report contains dropped claims | Ensure write() receives verified, not the raw findings. |
| Real mode: researcher won't quote sources | Strengthen the researcher role prompt to require a quoted source span. |
🪜 Practice ladder beginner → industry
Six graded exercises, easy to real-world. Try each before opening its solution.
Context: A research crew is a set of single-job agents. The first slice defines the shared data shape and one role, with the model call injected so the whole crew runs offline and deterministically in tests.
Your task: Define a Finding dataclass (claim, source) and a plan(question) role that returns 2-4 sub-queries, taking the model call as an injectable parameter.
Requirements:
Findingcarries a claim and its sourceplan()returns a small list of sub-queries- The model call is a parameter, defaulting to a scripted mock
- The demo runs with no API key
- The injectable-call seam is what makes every later rung testable
💡 Hint: Make call a parameter on every role; a scripted mock_call keyed on the role name lets the crew run without a model.
Show solution
Design. Each agent is one job and every role takes an injectable call so tests "
"use a scripted stand-in. That seam is the whole reason the crew is testable without a key.
from dataclasses import dataclass
@dataclass
class Finding:
claim: str
source: str
def mock_call(role, task):
if role == "planner":
return ["What is our churn?", "What caused it?"]
raise ValueError(role)
def plan(question, call=mock_call):
return call("planner", question)
print(plan("why did churn rise?")) # ['What is our churn?', 'What caused it?']
Context: The crew's defence against compounding errors is a critic. A researcher can be wrong, but a claim survives only if its own cited source actually contains it — a pure filter, no model needed.
Your task: Build critique(findings) that keeps only findings whose claim text appears in its cited source, and prove a planted false claim is dropped.
Requirements:
- Supported = the claim's words appear in the cited source
- The critic is a pure function with no model call
- A true claim backed by its source is kept
- A fabricated claim citing an unrelated source is dropped
- Show the liar is removed while the good finding survives
💡 Hint: Case-insensitive containment of the claim in its source is enough offline; this one filter is what stops a single bad researcher poisoning the report.
Show solution
Design. The critic is a pure filter — no model needed. 'Supported' = the claim's words appear " "in the source it cites. This is the single most valuable agent: it stops one bad researcher poisoning the " "report.
from dataclasses import dataclass
@dataclass
class Finding:
claim: str
source: str
def critique(findings):
return [f for f in findings if f.claim.lower() in f.source.lower()]
good = Finding("net 30", "payment is net 30 days")
liar = Finding("revenue tripled", "payment is net 30 days")
kept = critique([good, liar])
print(len(kept), kept[0].claim) # 1 net 30 -- liar dropped
Context: An orchestrator wires the roles together, but an unbounded loop can burn budget. A single step leash — one counter across all agents — caps total work regardless of how sub-queries fan out.
Your task: Wire plan → research → critique → write into run_crew with a MAX_STEPS cap shared across all agents, returning the report and how many findings were dropped.
Requirements:
- Steps increment on every agent action, not per-agent
- The research loop breaks when the shared cap is hit
- Findings pass through the critic before the writer
- The result includes the report and a dropped count
- The cap bounds work even when planning fans out widely
💡 Hint: One counter, checked before each research call and incremented on every action, is the leash; it must be global, not reset per role.
Show solution
Design. The leash is a single counter incremented on every agent action and checked before " "each research call. It caps total work regardless of how sub-queries fan out.
from dataclasses import dataclass
@dataclass
class Finding:
claim: str
source: str
MAX_STEPS = 12
def mock_call(role, task):
if role == "planner": return ["q1", "q2"]
if role == "researcher":
return [Finding(f"Answer to {task}", f"Answer to {task} (src)")]
if role == "writer":
return "REPORT: " + "; ".join(f.claim for f in task)
def critique(fs): return [f for f in fs if f.claim.lower() in f.source.lower()]
def run_crew(question, call=mock_call):
steps = 0
subs = call("planner", question); steps += 1
findings = []
for q in subs:
if steps >= MAX_STEPS: break # the leash across all agents
findings += call("researcher", q); steps += 1
verified = critique(findings); steps += 1
report = call("writer", verified); steps += 1
return {"report": report, "dropped": len(findings)-len(verified),
"steps": steps}
print(run_crew("why did churn rise?"))
Context: The critical invariant is that the writer never sees an unverified claim, even when a researcher lies. Correctness here is a property proven by construction, not an output you eyeball.
Your task: Inject a scripted call where the researcher returns a claim absent from its source, and assert the false claim appears nowhere in the final report.
Requirements:
- Only the critic's verified output feeds the writer
- The scripted researcher returns a claim its source doesn't support
- The critic gate runs before
write - Assert the fabricated claim is absent from the report
- The report is empty (or reduced) precisely because the lie was dropped
💡 Hint: Gate before the writer: pass only critique(findings) into write, then assert the lie's text is not in the returned report.
Show solution
Design. Correctness here is a property, not an output: verified findings are the only "
"input to write. Prove it by construction — feed a liar and check the report.
from dataclasses import dataclass
@dataclass
class Finding:
claim: str
source: str
def critique(fs): return [f for f in fs if f.claim.lower() in f.source.lower()]
def run_crew(question, call):
subs = call("planner", question)
findings = []
for q in subs: findings += call("researcher", q)
verified = critique(findings) # gate BEFORE writer
return call("writer", verified)
def lying_call(role, task):
if role == "planner": return ["q"]
if role == "researcher":
return [Finding("revenue tripled", "payment is net 30 days")] # lie
if role == "writer":
return "REPORT: " + "; ".join(f.claim for f in task)
report = run_crew("q", lying_call)
print(report) # REPORT: (empty -- lie dropped)
assert "revenue tripled" not in report
print("invariant holds")
Context: Production crews need graceful degradation. A shared budget debited per call fails closed when exhausted, and a loop breaker stops when two consecutive rounds add no new verified findings.
Your task: Add a shared token budget the crew debits per call and a dead-loop breaker that stops after two consecutive rounds with no new verified findings.
Requirements:
- A budget object debits per call and refuses when it can't cover the next
- On exhaustion the crew returns partial results, not a crash
- Progress is measured as growth in the verified set
- Two stalled rounds stop the crew early with a reason
- The stop reason (budget / no_progress / done) is reported
💡 Hint: Charge the budget before each call and return what you have on empty; track a stall counter that resets whenever the verified set grows and trips at two.
Show solution
Design. Budget is a debit account raised before each call; on empty, return partial results " "rather than crash. Progress = growth in the verified set; two stalls -> stop early and report what you " "have. This is 'graceful degradation', the production default.
class Budget:
def __init__(self, total): self.left = total
def charge(self, n):
if self.left < n: return False
self.left -= n; return True
def crew_with_budget(rounds, budget, cost=100):
verified, stalls = [], 0
for r in rounds: # r = list of (claim, source)
if not budget.charge(cost):
return {"verified": verified, "stopped": "budget"}
before = len(verified)
verified += [c for (c, s) in r if c.lower() in s.lower()]
if len(verified) == before:
stalls += 1
if stalls >= 2:
return {"verified": verified, "stopped": "no_progress"}
else:
stalls = 0
return {"verified": verified, "stopped": "done"}
good = [("net 30", "net 30 terms")]
bad = [("x", "unrelated")]
print(crew_with_budget([good, bad, bad], Budget(1000))) # stopped no_progress
print(crew_with_budget([good]*20, Budget(300))) # stopped budget
Context: Shipping means swapping in a real model and confronting real research: sources disagree. A reconciliation step keeps the more-trusted source on a conflict and flags it for human review rather than silently dropping it.
Your task: Wire a real call via the Anthropic SDK (needs a key) and add reconciliation that, when two verified findings disagree on a topic, keeps the higher source tier and flags the conflict.
Requirements:
- The real call uses the Anthropic SDK with role-specific system prompts
- The offline crew stays intact as the no-key fallback
- Verified findings are grouped by topic
- On disagreement, the higher-trust source tier wins
- Conflicts are recorded for review, never silently dropped
💡 Hint: Keep the offline crew unchanged — the real model is just another call; reconciliation groups by a topic key and resolves ties by a source-tier map.
Show solution
Design. Keep the offline crew intact; the real model is just another call. "
"Reconciliation groups verified findings by topic key; on disagreement, higher source tier wins and the "
"conflict is recorded (never silently dropped) so a human can audit it.
# --- real model call (needs creds / API key) ---
# import anthropic
# client = anthropic.Anthropic()
# ROLES = {"planner": "Break into 2-4 sub-queries, one per line.",
# "researcher": "Answer with facts; quote the source text.",
# "writer": "Write a short cited report from the findings."}
# def real_call(role, task):
# m = client.messages.create(model="claude-opus-4-8", max_tokens=600,
# system=ROLES[role], messages=[{"role":"user","content":str(task)}])
# return m.content[0].text
# run_crew(question, call=real_call) # after export ANTHROPIC_API_KEY=...
# --- offline: conflict reconciliation (runnable) ---
TIER = {"filing": 3, "blog": 1} # trusted -> untrusted
def reconcile(findings):
# findings: list of dict(topic, value, source_tier)
best, conflicts = {}, []
for f in findings:
cur = best.get(f["topic"])
if cur and cur["value"] != f["value"]:
conflicts.append((f["topic"], cur["value"], f["value"]))
if f["source_tier"] <= cur["source_tier"]:
continue # keep the more-trusted incumbent
best[f["topic"]] = f
return {"kept": {t: v["value"] for t, v in best.items()},
"conflicts": conflicts}
res = reconcile([
{"topic": "revenue", "value": "$10M", "source_tier": TIER["filing"]},
{"topic": "revenue", "value": "$30M", "source_tier": TIER["blog"]}])
print(res["kept"]) # {'revenue': '$10M'} -- filing beats blog
print(res["conflicts"]) # [('revenue', '$10M', '$30M')] -- flagged
✓ You are done when…
python crew.pyprints a report with verified findings and 0 dropped.python -m pytest tests/ -vshows 6 passed, including the planted-lie test.- You can explain why the writer only ever sees verified findings.
- (Optional) A real
model_callplugs into the same orchestrator.
research-crew/
├─ .venv/
├─ requirements.txt
├─ agents.py (roles + Finding + mock_call)
├─ crew.py (critic + sequential orchestrator + cap)
└─ tests/
└─ test_crew.py (6 offline tests incl. planted lie)
| Dimension | Meets the bar | Above the bar |
|---|---|---|
| Task decomposition is crisp | Roles have single, non-overlapping responsibilities; the planner splits the question cleanly. | Decomposition is validated on real tasks; no duplicated work; shared state shows each agent what's done. |
| Handoffs stay on-task | The writer uses only verified findings; each claim is traceable to a source through the chain. | Handoff correctness is tested — a bad finding upstream cannot silently become confident output downstream. |
| The loop is bounded | Hard step/turn caps per agent, a total token budget, and clear termination conditions exist. | Convergence rate (% of runs terminating cleanly within caps) is measured; circular chatter is caught. |
| The critic actually works | A dedicated critic/verifier checks claims and citations against sources before synthesis. | The critic is tested with planted false claims and demonstrably catches them — not a rubber stamp. |
| Cost is justified | Total tokens/calls per task are tracked; the multi-agent cost is known. | Uplift vs a single agent (Project 6) is proven on hard tasks; if it only ties, the crew is dropped. |
| Final report is faithful | Report accuracy, coverage, and citation-completeness are evaluated (LLM-judge + spot human). | A held-out eval set with graded reports; hallucinated citations are near-zero and measured, not hoped. |
Score each row 0 (missing) / 1 (meets) / 2 (above). 0–4: a prototype — keep building. 5–8: a solid build you could take to review. 9–12: staff-level — production-defensible. Any dimension at 0 blocks shipping regardless of the total.
Knowledge check check yourself
Before building a multi-agent crew, the design says to justify it against a single agent. Why, and what specifically must earn its keep?
Show answer
How does the crew guarantee that a lying researcher's fabricated claim never reaches the final report?