Capstone C, built end to end

Chapter 55 argued the document-intake swarm as a design; this chapter is the runbook. The stack is Bedrock, Pydantic AI for typed validated extraction, and an event-driven SQS/Lambda fleet with idempotency, a dead-letter queue, and a calibrated judge. The star is throughput under an honest at-least-once delivery contract. AWS steps are follow-along; the Pydantic AI extraction runs locally against Bedrock if you have access.

Goal, gates, patterns. Documents land in bulk; extract structured fields, judge each, auto-approve the confident majority, route the tail to humans. Dominant gates: 5 (failure semantics), 8 (evals via the judge), 9 (observability). Patterns: pipeline, evaluator-optimizer, router, map-reduce, guardrail.

Step 0a: rehearse locally first

This capstone's rung 1 is Lab 0.1 almost exactly, the same worker shape, so start there:

# rung 1: idempotency, the race, and the happy path against fakes
python3 -m pytest local_harness.py -q          # 3 passed

# rung 1: your typed extractor against recorded documents (no Bedrock)
python3 -m pytest tests/test_extract.py -q

# rung 2: the real S3 + DynamoDB + SQS shapes, still local
pip install 'moto[all]'
python3 -m pytest tests/test_worker_moto.py -q

Do not skip the redelivery test. At-least-once delivery means the duplicate will happen in production, and it is free to prove survivable now and expensive to discover later as double-filed invoices.

Architecture

  doc -> S3 event -> EventBridge -> SQS (main)  --(3 fails)--> SQS (DLQ)
                                      |
                     Lambda worker (idempotent, keyed by doc id)
                        |            |            |
                    [guardrail]  [extract]     [judge]
                     PII check   (Pydantic AI) (calibrated 0..1)
                        |
             confidence >= threshold ? --yes--> auto-approve
                                       \--no--> arbitration queue -> human

Step 0: prerequisites and project

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

Step 1: the typed extractor (extract.py)

The output type is the schema; Pydantic AI validates the model's output against it and retries on a validation failure.

# extract.py
from pydantic import BaseModel, Field
from pydantic_ai import Agent

class LineItem(BaseModel):
    sku: str
    quantity: int = Field(ge=0)
    unit_price_cents: int = Field(ge=0)

class Invoice(BaseModel):
    vendor: str
    invoice_number: str
    total_cents: int = Field(ge=0)
    line_items: list[LineItem]

extractor = Agent(
    "bedrock:anthropic.claude-haiku-4-5",          # small model, high volume
    output_type=Invoice,
    system_prompt=("Extract the invoice fields from the document text. "
                   "If a field is illegible, raise rather than guess."))

def extract(document_text: str) -> Invoice:
    return extractor.run_sync(document_text).output   # a validated Invoice

Step 2: the calibrated judge (judge.py)

# judge.py
from pydantic import BaseModel, Field
from pydantic_ai import Agent
from extract import Invoice

class Verdict(BaseModel):
    confidence: float = Field(ge=0, le=1)
    issues: list[str]

judge = Agent(
    "bedrock:anthropic.claude-haiku-4-5",
    output_type=Verdict,
    system_prompt=("Score 0..1 how confident you are the extraction matches "
                   "the document, and list mismatches. Be calibrated: 0.9 "
                   "means right 9 times in 10."))

AUTO_APPROVE = 0.92          # set by step 3, not by guess

def review(doc: str, extracted: Invoice) -> tuple[bool, Verdict]:
    v = judge.run_sync(f"Document:\n{doc}\n\nExtracted:\n{extracted}").output
    return v.confidence >= AUTO_APPROVE, v

Step 3: calibrate the threshold (calibrate.py)

Do not pick AUTO_APPROVE by feel. Run the judge over human-labeled documents and choose the threshold where the false-auto-approve rate is acceptable.

# calibrate.py
import json, sys
from extract import extract
from judge import judge

def calibrate(labeled_path: str):
    rows = [json.loads(l) for l in open(labeled_path)]   # {doc, correct: bool}
    scored = []
    for r in rows:
        v = judge.run_sync(f"Document:\n{r['doc']}\n\nExtracted:\n"
                           f"{extract(r['doc'])}").output
        scored.append((v.confidence, r["correct"]))
    for thr in (0.80, 0.85, 0.90, 0.92, 0.95):
        auto = [c for conf, c in scored if conf >= thr]
        bad = sum(1 for c in auto if not c)
        print(f"threshold {thr}: auto-approve {len(auto)}/{len(scored)}, "
              f"wrong auto-approvals {bad}")

if __name__ == "__main__":
    calibrate(sys.argv[1])          # e.g. python calibrate.py labeled.jsonl

threshold 0.80: auto-approve 88/100, wrong auto-approvals 6
threshold 0.90: auto-approve 71/100, wrong auto-approvals 1
threshold 0.92: auto-approve 64/100, wrong auto-approvals 0     <- pick this

Set AUTO_APPROVE in judge.py to the threshold whose wrong-rate you can live with.

Step 4: the idempotent worker (worker.py)

At-least-once delivery means a document can arrive twice, so the effect is keyed by doc id with a conditional write, making a redelivery a no-op.

# worker.py
import boto3, json, os
from extract import extract
from judge import review

ddb = boto3.client("dynamodb")
sqs = boto3.client("sqs")
s3 = boto3.client("dynamodb") and boto3.client("s3")
TABLE = "hive-intake-results"
HUMAN_Q = os.environ["HUMAN_QUEUE_URL"]

def already_done(doc_id: str) -> bool:
    return "Item" in ddb.get_item(TableName=TABLE,
        Key={"pk": {"S": doc_id}, "sk": {"S": "result"}})

def put_once(doc_id: str, item: dict):
    ddb.put_item(TableName=TABLE, Item=item,
        ConditionExpression="attribute_not_exists(pk)")   # exactly-once effect

def handler(event, _ctx=None):
    for record in event["Records"]:                # SQS batch
        msg = json.loads(record["body"])
        doc_id, key = msg["doc_id"], msg["s3_key"]
        if already_done(doc_id):
            continue                               # duplicate delivery -> no-op
        text = s3.get_object(Bucket=msg["bucket"], Key=key)["Body"].read().decode()
        # guardrail at the boundary (PII) omitted for brevity; see Chapter 59
        extracted = extract(text)
        auto, verdict = review(text, extracted)
        try:
            put_once(doc_id, {
                "pk": {"S": doc_id}, "sk": {"S": "result"},
                "extracted": {"S": extracted.model_dump_json()},
                "confidence": {"N": str(verdict.confidence)},
                "auto": {"BOOL": auto}})
        except ddb.exceptions.ConditionalCheckFailedException:
            continue                               # raced a duplicate; fine
        if not auto:
            sqs.send_message(QueueUrl=HUMAN_Q, MessageBody=json.dumps(
                {"doc_id": doc_id, "issues": verdict.issues}))

Step 5: create the queues, table, and event wiring

# results table
aws dynamodb create-table --table-name hive-intake-results \
  --attribute-definitions AttributeName=pk,AttributeType=S AttributeName=sk,AttributeType=S \
  --key-schema AttributeName=pk,KeyType=HASH AttributeName=sk,KeyType=RANGE \
  --billing-mode PAY_PER_REQUEST --region $AWS_REGION

# dead-letter queue, then main queue with redrive after 3 receives
DLQ=$(aws sqs create-queue --queue-name hive-intake-dlq --query QueueUrl --output text)
DLQ_ARN=$(aws sqs get-queue-attributes --queue-url $DLQ \
  --attribute-names QueueArn --query Attributes.QueueArn --output text)
MAIN=$(aws sqs create-queue --queue-name hive-intake \
  --attributes "{\"VisibilityTimeout\":\"300\",\
\"RedrivePolicy\":\"{\\\"deadLetterTargetArn\\\":\\\"$DLQ_ARN\\\",\\\"maxReceiveCount\\\":\\\"3\\\"}\"}" \
  --query QueueUrl --output text)
HUMAN=$(aws sqs create-queue --queue-name hive-intake-human --query QueueUrl --output text)

The VisibilityTimeout of 300s is above the worker's real per-document time, so a healthy worker is not redelivered mid-run; maxReceiveCount of 3 sends poison documents to the DLQ.

Wire S3 -> EventBridge -> SQS (bucket assumed created and EventBridge notifications enabled):

aws events put-rule --name intake-docs \
  --event-pattern '{"source":["aws.s3"],"detail-type":["Object Created"],
    "detail":{"bucket":{"name":["hive-intake-'$ACCOUNT'"]}}}'
aws events put-targets --rule intake-docs \
  --targets "Id=1,Arn=$(aws sqs get-queue-attributes --queue-url $MAIN \
    --attribute-names QueueArn --query Attributes.QueueArn --output text)"

Step 6: deploy the worker Lambda and attach the queue

pip install -r requirements.txt -t build/ && cp *.py build/
(cd build && zip -qr ../fn.zip .)
aws lambda create-function --function-name hive-intake-worker \
  --runtime python3.12 --handler worker.handler --zip-file fileb://fn.zip \
  --role arn:aws:iam::$ACCOUNT:role/hive-intake-worker \
  --timeout 300 --memory-size 1024 \
  --environment "Variables={HUMAN_QUEUE_URL=$HUMAN}" --region $AWS_REGION

# SQS -> Lambda event source mapping
aws lambda create-event-source-mapping --function-name hive-intake-worker \
  --event-source-arn $(aws sqs get-queue-attributes --queue-url $MAIN \
    --attribute-names QueueArn --query Attributes.QueueArn --output text) \
  --batch-size 5 --region $AWS_REGION

(The hive-intake-worker IAM role grants bedrock:InvokeModel on the Haiku model, s3:GetObject on the intake bucket, DynamoDB read/write on the results table, and SQS receive/send on the two queues, and nothing else, per Chapter 36.)

Step 7: run it

Drop a batch of documents into the intake bucket:

aws s3 cp ./sample-invoices/ s3://hive-intake-$ACCOUNT/inbox/ --recursive

Events fan into SQS; the Lambda fleet drains the queue at its sustainable pace (backpressure, free). Watch the queue depth fall and the results table fill:

aws sqs get-queue-attributes --queue-url $MAIN \
  --attribute-names ApproximateNumberOfMessages --query Attributes
aws dynamodb scan --table-name hive-intake-results --select COUNT

Step 8: validate the results

1. Confirm exactly-once effects. Re-send one message deliberately and confirm the table shows one record, not two:

aws sqs send-message --queue-url $MAIN --message-body \
  '{"doc_id":"inv-42","s3_key":"inbox/inv-42.txt","bucket":"hive-intake-'$ACCOUNT'"}'
# ...send the identical body again...
aws dynamodb get-item --table-name hive-intake-results \
  --key '{"pk":{"S":"inv-42"},"sk":{"S":"result"}}' --query 'Item.confidence'
# one item; the second delivery was a no-op

2. Re-check calibration on a fresh sample:

python calibrate.py fresh-labeled.jsonl     # the 0.92 should still hold

3. Read the auto-approve rate as an economic dial:

aws dynamodb scan --table-name hive-intake-results \
  --filter-expression "auto = :t" \
  --expression-attribute-values '{":t":{"BOOL":true}}' --select COUNT

A sudden jump means the judge got miscalibrated or a document type changed.

4. Audit the human tail (the hive-intake-human queue): the items routed to humans should be genuinely hard. If humans keep overturning auto-approvals, the threshold is too low; if they rubber-stamp the tail, too high.

Troubleshoot

Don't be confused: extraction accuracy vs judge calibration. They are different numbers and both matter. Extraction accuracy is how often the fields are right; judge calibration is how well the confidence predicts that. A well-calibrated judge over a mediocre extractor is safe (it routes the doubtful ones to humans); a miscalibrated judge over a great extractor is a liability, because it auto-approves the occasional confident error and no human ever looks. Calibrate the judge (step 3) before you tune the extractor.

👉 Next: Capstone D, built end to end, the research service on Bedrock and CrewAI, as a runbook with the per-request budget and the verification mesh wired step by step.