Capstone D, built end to end

Chapter 56 argued the research service as a design; this chapter is the runbook. The stack is Bedrock, CrewAI for the role crew (planner, reader, fact-checker, synthesizer), a per-request budget ledger, the verification mesh, and an API Gateway front door. The stars are budgets-as-a-product-feature and verification-as-the-product. AWS steps are follow-along; the crew runs locally against Bedrock if you have access.

Goal, gates, patterns. Given a question and a budget, research it and return a cited report where every claim was verified, at a bounded cost. Dominant gates: 3 (budgets) and 8 (verification). Patterns: plan-and-execute, orchestrator-workers, debate/judge panel, agentic retrieval, budget-bounded admission, reflection.

Step 0a: rehearse locally first

The budget ledger and the verification panel are both pure logic, so rung 1 covers the two properties that define this service:

# rung 1: the ledger admits until the cap, then lands a clean partial
python3 -m pytest tests/test_ledger.py -q

# rung 1: the panel kills planted decoys (no model needed, replay fakes)
python3 -m pytest tests/test_verify.py -q

# rung 1: the whole flow against a replaying fake model
python3 -c "from service import research; print(research('test question', 120000))"

Budget-as-a-ceiling is a logic property, not a cloud property: if admit() runs after dispatch rather than before, a local test catches it, and no amount of careful cloud testing will.

Architecture

  POST /research {question, budget} -> API Gateway -> Lambda coordinator
      |
   admit against per-request budget (ledger)
      |
   [planner] (frontier) -> sub-questions
      |
   fan out readers (middle tier) over sub-questions, each admitted
      |
   claims -> dedup -> fact-checker panel (3 blind, frontier) --keep--> verified
      |
   [synthesizer] -> cited report (verified claims only)
      |
   exhausted budget -> partial report with remainder enumerated

Step 0: prerequisites and project

export AWS_REGION=us-east-1 ACCOUNT=$(aws sts get-caller-identity --query Account --output text)
mkdir hive-research && cd hive-research
python3 -m venv .venv && source .venv/bin/activate
printf 'crewai\ncrewai-tools\nboto3\n' > requirements.txt
pip install -r requirements.txt

Step 1: the budget ledger (ledger.py)

Admission, not accounting: a reader runs only if the budget covers its projected cost, and the run lands as a partial when it does not.

# ledger.py
class Ledger:
    def __init__(self, cap: int):
        self.cap, self.spent, self.exhausted = cap, 0, False
    def admit(self, cost: int) -> bool:
        if self.spent + cost > self.cap:
            self.exhausted = True
            return False
        self.spent += cost
        return True

Step 2: the crew (crew.py)

CrewAI expresses each specialist as a role/goal agent on a Bedrock model.

# crew.py
from crewai import Agent, LLM
from crewai_tools import SerperDevTool          # or a Bedrock-KB / MCP tool

frontier = LLM(model="bedrock/anthropic.claude-opus-4-8")
middle   = LLM(model="bedrock/anthropic.claude-sonnet-5")
search = SerperDevTool()                          # web search tool

planner = Agent(role="Research planner", llm=frontier,
    goal="Decompose a question into 3-6 independent, searchable sub-questions.",
    backstory="You scope research so sub-questions do not overlap.",
    verbose=False)

reader = Agent(role="Source reader", llm=middle, tools=[search],
    goal="Answer a sub-question with claims, each carrying its source URL.",
    backstory="You never state a claim without the URL that supports it.",
    verbose=False)

fact_checker = Agent(role="Adversarial fact-checker", llm=frontier, tools=[search],
    goal="Try to REFUTE a claim against its cited source; default to refuted "
         "when the source does not clearly support it.",
    backstory="You are the reason the report can be trusted.", verbose=False)

synthesizer = Agent(role="Synthesizer", llm=frontier,
    goal="Write a report using only verified claims, each with its citation.",
    backstory="Every sentence must trace to a verified claim.", verbose=False)

Step 3: the verification panel (verify.py)

# verify.py
import json
from crewai import Task, Crew, Process
from crew import fact_checker

def panel_keeps(claim: dict, votes: int = 3) -> bool:
    kept = 0
    for _ in range(votes):                        # blind, independent
        t = Task(description=f"Claim: {json.dumps(claim)}. Refute it against "
                             f"its source. Answer only JSON {{\"real\":bool}}.",
                 expected_output='{"real": true|false}', agent=fact_checker)
        out = Crew(agents=[fact_checker], tasks=[t], process=Process.sequential
                   ).kickoff()
        try:
            kept += 1 if json.loads(str(out)).get("real") else 0
        except json.JSONDecodeError:
            pass                                   # unparseable -> not a keep
    return kept > votes // 2

Step 4: the coordinator (service.py)

The flow: admit, plan, fan out readers under the budget, dedup, verify, synthesize. This is the Lambda handler behind the API.

# service.py
import json
from crewai import Task, Crew, Process
from crew import planner, reader, synthesizer
from verify import panel_keeps
from ledger import Ledger

READ_COST, PLAN_COST, SYNTH_COST = 40_000, 15_000, 25_000

def run_agent(agent, description, expected):
    t = Task(description=description, expected_output=expected, agent=agent)
    return str(Crew(agents=[agent], tasks=[t], process=Process.sequential).kickoff())

def research(question: str, budget: int) -> dict:
    led = Ledger(budget)
    led.admit(PLAN_COST)
    plan = run_agent(planner, f"Question: {question}. List sub-questions as a "
                     f"JSON array of strings.", '["...","..."]')
    try:
        sub_qs = json.loads(plan)
    except json.JSONDecodeError:
        sub_qs = [question]
    claims = []
    for sq in sub_qs:                              # fan out (map)
        if not led.admit(READ_COST):               # budget-bounded
            break                                  # -> partial result
        out = run_agent(reader, f"Answer: {sq}. Output a JSON array of "
                        f"{{\"text\",\"source\"}} claims.", '[{"text","source"}]')
        try:
            claims += json.loads(out)
        except json.JSONDecodeError:
            pass
    seen, deduped = set(), []
    for c in claims:                               # dedup
        if c["text"] not in seen:
            seen.add(c["text"]); deduped.append(c)
    verified = [c for c in deduped if panel_keeps(c)]   # the mesh
    led.admit(SYNTH_COST)
    report = run_agent(synthesizer, f"Write a cited report for {question!r} "
                       f"using ONLY these verified claims: {json.dumps(verified)}. "
                       f"Cite each with its source.", "a cited report")
    return {"report": report, "verified": len(verified),
            "sub_questions": len(sub_qs), "spent": led.spent,
            "exhausted": led.exhausted}

def handler(event, _ctx=None):                     # API Gateway proxy integration
    body = json.loads(event.get("body") or "{}")
    out = research(body["question"], int(body.get("budget", 500_000)))
    return {"statusCode": 200, "body": json.dumps(out)}

Step 5: deploy the Lambda and the API

pip install -r requirements.txt -t build/ && cp *.py build/
(cd build && zip -qr ../fn.zip .)
aws lambda create-function --function-name hive-research \
  --runtime python3.12 --handler service.handler --zip-file fileb://fn.zip \
  --role arn:aws:iam::$ACCOUNT:role/hive-research \
  --timeout 900 --memory-size 2048 --region $AWS_REGION

# HTTP API in front
API=$(aws apigatewayv2 create-api --name hive-research --protocol-type HTTP \
  --target arn:aws:lambda:$AWS_REGION:$ACCOUNT:function:hive-research \
  --query ApiId --output text)
aws lambda add-permission --function-name hive-research \
  --statement-id apigw --action lambda:InvokeFunction \
  --principal apigateway.amazonaws.com \
  --source-arn "arn:aws:execute-api:$AWS_REGION:$ACCOUNT:$API/*"
echo "endpoint: https://$API.execute-api.$AWS_REGION.amazonaws.com/research"

(The hive-research role grants bedrock:InvokeModel on the two model ARNs and nothing else beyond logs; the search tool's key lives in Identity/Secrets, not the code.)

Step 6: run it

Expose "quick" and "thorough" as budget tiers, so the price ceiling is a product surface, not an internal guardrail:

ENDPOINT=https://$API.execute-api.$AWS_REGION.amazonaws.com/research

# a thorough run:
curl -s -X POST $ENDPOINT -H 'content-type: application/json' \
  -d '{"question":"What caused the 2026 checkout latency incidents industrywide?","budget":500000}' | jq

{
  "report": "Verified findings: ... [cited]",
  "verified": 14,
  "sub_questions": 5,
  "spent": 415000,
  "exhausted": false
}

Step 7: validate the results

1. Confirm the budget is a hard ceiling. Send a question that wants far more than its budget and confirm a graceful partial, not an overrun:

curl -s -X POST $ENDPOINT -H 'content-type: application/json' \
  -d '{"question":"Exhaustively survey all agent frameworks","budget":80000}' \
  | jq '{verified, spent, exhausted}'
# {"verified": 3, "spent": 80000, "exhausted": true}   <- capped, partial

If spent ever exceeds budget, the ledger is accounting after the fact, move the admit() check before each reader dispatch.

2. Confirm every claim is verified and cited. Sample the report; each sentence must trace to a verified claim and a source URL. An unverified claim means the synthesizer invented, tighten it to use only the verified set.

3. Measure precision against planted decoys. In a test, inject two known-false claims into the reader output; the panel must kill both:

python -c "from verify import panel_keeps; \
print(panel_keeps({'text':'the moon is made of cheese','source':'http://x'}))"
# False   <- the panel refuted the decoy

4. Watch the verify-to-find token ratio. Spending several times more on checking than on finding is healthy; a low ratio means the panel is too small and precision will suffer.

Troubleshoot

Don't be confused: coverage vs correctness. They pull in opposite directions and both are the product. Coverage is finding everything relevant (more readers, broader sub-questions); correctness is that what you report is true (the fact-checker panel). Optimize only for coverage and you ship a thorough report full of plausible-but-unchecked claims; only for correctness and you ship a narrow over-verified answer that missed the point. The budget balances them, buy coverage up front, spend the majority on verification, and the verify-to-find ratio is the dial.

The four builds, closed

Part 14 built all four capstones as runbooks, each on a different open-source framework over the same Bedrock model plane and the same Hive platform: the audit fleet on Strands, the incident responder on LangGraph, the document swarm on Pydantic AI, and this research service on CrewAI. The framework changed with the job; the platform did not. Pick your patterns from the catalog and your framework from the landscape, and build them on the planes, budgets, isolation, verification, scheduling, that no framework supplies and this book taught you to own.

👉 The glossary and references close the book.