Anthropic API in Practice · Capstone

Batch document pipeline

The Anthropic API in Practice track, assembled. Point the pipeline at a folder of documents and it does the real thing at scale: caches one shared instruction prefix, submits every document as a single Message Batch for ~50% savings, extracts structured data from each, tracks token usage and cost, then retrieves and reconciles results by custom_id. Every SDK call is real and labelled; two helpers run offline with no key so you can execute the shapes today.

⏱️ ~2.5 hours🧪 6 steps🎯 Capstone

This is the capstone for the Anthropic API in Practice track. You met the pieces one lesson at a time — the Message Batches API (AP1), prompt caching hands-on (AP2), the Files API with vision & PDF (AP3), and token counting & usage metrics (AP4). Here you wire them together into a bulk document-processing pipeline: one cached instruction prefix, one batch of thousands of documents, structured records out, and a cost report you can hand to finance.

⚙️ Real SDK code — read now, run when you have a keyThe pipeline blocks are the real anthropic Python SDK — accurate calls, no invented methods — and each is labelled ▶ needs API key + network because it makes real batched API calls that cost money. Two helpers (Step 1 building the request list, Step 5 the cost-report aggregator) are stdlib-only and run offline — labelled (runs offline) — so you can execute the request shapes and the cost math today with plain python file.py.

Learning objectives

Compose the AP track into one pipeline: cache a shared prefix, batch every document, reconcile by custom_id.
Build one Batch request per document, each carrying a stable custom_id — the real SDK request shape.
Put a cache_control breakpoint on the shared instruction prefix so every request reads it cheaply.
Submit with client.messages.batches.create, poll status, and stream results with .results().
Reconcile succeeded/errored/expired results by custom_id into structured records — never by position.
Aggregate usage across the batch into a token & cost report, then harden for production.

Architecture — the whole pipeline on one line essential

Read the pipeline as one sentence: a folder of documents becomes one request per document (each with a stable custom_id and a shared, cached instruction prefix), the requests are submitted as a single Message Batch, we poll until it ends, retrieve the results, reconcile them back to their documents by custom_id, and emit structured records plus a cost report. The rest of the page builds exactly this, one step per stage — climbing from a plain request list up to production hardening.

🗺️ How to read this diagram

This is the whole capstone on one line. A folder of documents enters at the left and flows right through seven stages; the last box is what you hand to finance. Follow the arrows.

Folder of docs is the input — N files on disk (invoices, contracts, scans). The pipeline scans it and processes every file in one run.
Build requests turns each file into one Messages request tagged with a stable custom_id, and attaches the shared, cached instruction prefix — the same rules for every document.
Submit Batch sends the whole list to the Message Batches API in one call — that's where the ~50% discount comes from. Poll status then checks back until the batch has ended (asynchronous — could be minutes to hours).
Retrieve streams the finished results with .results(), and Reconcile maps each result back to its document by custom_id — because results come back in any order.
Records + cost is the payoff: structured data per document plus a summed token/cost report. The eval of a batch pipeline is "did every doc get reconciled, and what did it cost?"

In short: a folder becomes cached requests → one batch → poll → retrieve → reconcile by custom_id → structured records and a cost report. Each step below builds one stage, climbing from a plain request list up to production hardening.

Step 1 · Build the request list — one per document essential

Start with the shape and nothing else. Batching is not magic: it is a list of ordinary Messages requests, each tagged with a custom_id you choose so you can find its answer later. Before touching the network we build that list from a fake folder listing and print the shapes — so you can see the request objects and their ids with no key.

Example code for learning — review, test, and adapt it before running against real or production systems. Commands can create, change, or delete resources. See the Terms & Disclaimer.

Python · build the request list (runs offline)

step1_requests.py# (runs offline) stdlib only — no key, no network. Builds the Batch request
# list from a folder listing and prints each request's shape + custom_id.
import json

# pretend this came from os.listdir("docs/") — the real pipeline scans a folder
DOCS = ["invoice_0001.pdf", "invoice_0002.pdf", "contract_A.pdf"]

SHARED_INSTRUCTION = (
    "You are a precise document-extraction engine. From the document, extract "
    "vendor, invoice_number, total_amount, and due_date as JSON. If a field is "
    "missing, use null. Answer with JSON only."
)

def doc_id(filename):
    """Stable custom_id derived from the filename (no extension, safe chars)."""
    stem = filename.rsplit(".", 1)[0]
    return "doc-" + "".join(c if c.isalnum() else "-" for c in stem).lower()

def build_requests(filenames):
    """One request per document. custom_id keys the result back to the doc.
    (This mirrors the real SDK request shape built in Step 2 — here we just
    print dicts so it runs offline.)"""
    requests = []
    for name in filenames:
        requests.append({
            "custom_id": doc_id(name),
            "params": {
                "model": "claude-opus-4-8",
                "max_tokens": 1024,
                "system": SHARED_INSTRUCTION,   # cached prefix added for real in Step 2
                "messages": [
                    {"role": "user", "content": f"Extract fields from: {name}"}
                ],
            },
        })
    return requests

reqs = build_requests(DOCS)
print("built", len(reqs), "requests")
for r in reqs:
    print(r["custom_id"], "->", r["params"]["model"], "| max_tokens",
          r["params"]["max_tokens"])
# ids must be unique — the reconcile step depends on it
print("unique ids:", len(reqs) == len({r["custom_id"] for r in reqs}))

built 3 requests
doc-invoice-0001 -> claude-opus-4-8 | max_tokens 1024
doc-invoice-0002 -> claude-opus-4-8 | max_tokens 1024
doc-contract-a -> claude-opus-4-8 | max_tokens 1024
unique ids: True

▶ How this works

Batching starts as a plain list of requests — no network yet. This offline helper builds that list from a fake folder listing and prints each request's shape and custom_id, so you can see the objects before any key is involved.

DOCS stands in for os.listdir("docs/") — the real pipeline scans a folder; here we hard-code three filenames so it runs anywhere.
doc_id(filename) derives a stable, unique custom_id from the filename (drop the extension, lowercase, replace unsafe chars). Same file always yields the same id — which is what makes re-runs safe later.
build_requests emits one dict per document: a custom_id plus the params (model, max_tokens, the shared system instruction, and the per-document user message). This is the exact shape the real SDK Request takes in Step 2.
The final line asserts the ids are unique — the reconcile step in Step 4 keys on them, so a collision would silently overwrite a document's result.

What the output means: It prints built 3 requests, one line per document showing its custom_id and model, then unique ids: True — proof the request list is well-formed with no key.

Try this: Add "invoice_0001.txt" to DOCS — it collides with invoice_0001.pdf on custom_id, so unique ids flips to False. That's the check catching a real bug before it reaches the batch.

custom_id is the whole gameBatch results come back in any order, so the custom_id is how you map each answer to the document that produced it. Derive it deterministically from the filename (or a DB row id) and keep it unique — the reconcile step in Step 4 depends on it.

Step 2 · Cache the shared instruction prefix intermediate

Every document in the batch is processed with the same long instruction — extraction rules, the field schema, maybe few-shot examples. That prefix is identical across thousands of requests, so pay to process it once and read it cheaply after. This is AP2 applied at scale: a cache_control breakpoint on the last block of the shared system prefix. Batch + caching stack — you get the ~50% batch discount and the ~90% cache-read discount on the cached prefix.

Python · cached shared prefix + real Request objects (needs API key + network)

pipeline/requests.py# needs: pip install anthropic + ANTHROPIC_API_KEY; makes real batched API calls
# ▶ needs API key + network
import anthropic
from anthropic.types.message_create_params import MessageCreateParamsNonStreaming
from anthropic.types.messages.batch_create_params import Request

client = anthropic.Anthropic()

# The shared prefix is a list of system blocks; cache_control on the LAST block
# caches everything before it. Keep it byte-identical across every request so the
# cache actually hits — no timestamps, no per-doc text in here.
SHARED_SYSTEM = [
    {
        "type": "text",
        "text": (
            "You are a precise document-extraction engine. Extract vendor, "
            "invoice_number, total_amount, and due_date. Missing field -> null. "
            "... (long, stable extraction rules + few-shot examples) ..."
        ),
        "cache_control": {"type": "ephemeral"},   # breakpoint: cache the prefix
    },
]

def build_request(custom_id: str, document_text: str) -> Request:
    """One real Batch request. The shared SHARED_SYSTEM is identical for every
    doc (so it caches); only the per-document user content varies."""
    return Request(
        custom_id=custom_id,
        params=MessageCreateParamsNonStreaming(
            model="claude-opus-4-8",
            max_tokens=1024,
            system=SHARED_SYSTEM,                  # cached prefix, reused verbatim
            messages=[{"role": "user", "content": document_text}],
        ),
    )

# one request per document, same cached prefix on each
requests = [
    build_request("doc-invoice-0001", "Invoice text for 0001 ..."),
    build_request("doc-invoice-0002", "Invoice text for 0002 ..."),
]

▶ How this works

Every document is processed with the same long instruction, so we pay to process that prefix once and read it cheaply after. This is the real SDK request, with a caching breakpoint on the shared prefix. ▶ needs API key + network.

SHARED_SYSTEM is a list of system blocks holding the stable extraction rules and few-shot examples. The cache_control: {"type": "ephemeral"} on the last block is the breakpoint: everything up to it is cached.
build_request returns a real Request object (from the SDK's batch types). The same SHARED_SYSTEM goes on every request unchanged — that byte-for-byte sameness is what lets the cache hit.
Only the per-document text varies, and it lives in the messages user turn — after the cached prefix, so it never invalidates the cache.
Batch and caching stack: you get the ~50% batch discount on the whole request and the ~90% cache-read discount on the shared prefix from the second document onward.

What the output means: Nothing prints — this defines the cached-prefix request builder and a two-request list. The payoff shows up as cache_read_input_tokens in the results (Step 4) and in the cost report (Step 5).

Try this: Imagine slipping datetime.now() into SHARED_SYSTEM. Every request's prefix now differs, the cache never hits, and every document re-pays full price for the instruction — the single most common batch-cost mistake.

Any byte change in the prefix breaks the cacheCaching is a prefix match: put the stable rules and few-shot examples in SHARED_SYSTEM and keep it byte-identical on every request. The per-document text goes in the messages user turn, after the breakpoint. A stray timestamp or per-doc id in the system prefix silently drops the cache — verify with usage.cache_read_input_tokens once results come back.

Step 3 · Submit the batch and poll until it ends advanced

Now hand the whole request list to the Batches API in one call (AP1). The API returns immediately with a batch id and a processing_status; the work happens asynchronously (most batches finish within an hour, max 24h). You poll batches.retrieve until the status is "ended", sleeping between checks — don't hammer it.

Python · create + poll the batch (needs API key + network)

pipeline/submit.py# needs: pip install anthropic + ANTHROPIC_API_KEY; makes real batched API calls
# ▶ needs API key + network
import time
import anthropic

client = anthropic.Anthropic()

# submit ALL requests as one batch (up to 100k requests / 256 MB per batch)
batch = client.messages.batches.create(requests=requests)   # requests from Step 2
print("submitted batch:", batch.id, "status:", batch.processing_status)

# poll until the batch has ended; request_counts shows live progress
while True:
    batch = client.messages.batches.retrieve(batch.id)
    if batch.processing_status == "ended":
        break
    c = batch.request_counts
    print(f"status={batch.processing_status} "
          f"processing={c.processing} succeeded={c.succeeded} errored={c.errored}")
    time.sleep(30)                       # be gentle; batches are not latency-critical

c = batch.request_counts
print(f"ended: succeeded={c.succeeded} errored={c.errored} "
      f"canceled={c.canceled} expired={c.expired}")

▶ How this works

Hand the whole request list to the Batches API in one call, then poll until it finishes. The work happens asynchronously on Anthropic's side — you're not holding the model on the line. ▶ needs API key + network.

client.messages.batches.create(requests=requests) submits everything at once (up to 100k requests / 256 MB) and returns immediately with a batch.id and a processing_status.
The while loop calls batches.retrieve(batch.id) and breaks when processing_status == "ended". In between it prints live request_counts (processing / succeeded / errored) and sleeps so it doesn't hammer the API.
Because the batch runs server-side, you could store batch.id, exit, and poll later from a cron job — a demo polls in a loop; production checks on a schedule.

What the output means: You'd see submitted batch: msgbatch_… status: in_progress, then progress lines as counts move, then a final ended: line with the succeeded/errored/expired tallies.

Try this: Change the sleep(30) to a smaller value only if you're impatient — batches aren't latency-critical, and polling too aggressively just wastes requests. Results stay retrievable for 29 days, so there's no rush.

Polling, not blockingThe batch runs on Anthropic's side whether your process is alive or not, so store batch.id durably and poll on an interval (or from a scheduled job) rather than holding one long-lived process. Results stay retrievable for 29 days after creation.

Step 4 · Retrieve results and reconcile by custom_id professional

The batch ended; now stream the results with batches.results() and map each one back to its document. Results arrive in any order, so we key strictly by custom_id — never by position. Each result carries a result.type of succeeded/errored/canceled/expired; we handle each per-item so one bad document can't sink the run.

Python · stream results + reconcile by custom_id (needs API key + network)

pipeline/reconcile.py# needs: pip install anthropic + ANTHROPIC_API_KEY; makes real batched API calls
# ▶ needs API key + network
import json
import anthropic

client = anthropic.Anthropic()

records = {}     # custom_id -> extracted data or an error marker
usages = []      # collect per-item usage for the Step 5 cost report

for result in client.messages.batches.results(batch.id):   # streams, any order
    cid = result.custom_id
    kind = result.result.type
    if kind == "succeeded":
        msg = result.result.message
        text = next((b.text for b in msg.content if b.type == "text"), "")
        try:
            records[cid] = {"status": "ok", "data": json.loads(text)}
        except json.JSONDecodeError:
            records[cid] = {"status": "bad_json", "raw": text}
        # msg.usage feeds the cost report; keep the model too for pricing
        usages.append({"model": msg.model, "usage": msg.usage})
    elif kind == "errored":
        # invalid_request -> fix and resubmit; api_error -> safe to retry
        records[cid] = {"status": "errored",
                        "error": result.result.error.type}
    elif kind == "expired":
        records[cid] = {"status": "expired"}   # resubmit these
    else:  # canceled
        records[cid] = {"status": kind}

ok = sum(1 for r in records.values() if r["status"] == "ok")
print(f"reconciled {len(records)} results ({ok} ok) keyed by custom_id")

▶ How this works

The batch ended; now stream the results and map each one back to its document. The rule that makes this correct: results arrive in any order, so we key by custom_id, never by position. ▶ needs API key + network.

for result in client.messages.batches.results(batch.id): streams every result. Each carries its custom_id and a result.type.
On succeeded we pull the text block, json.loads it into a record, and stash msg.usage + msg.model for the Step-5 cost report. Bad JSON is recorded as bad_json rather than crashing the run.
errored and expired are handled per item: an invalid request is recorded with its error type; an expired one is marked for resubmission. One bad document can't sink the whole batch.
Everything lands in records[custom_id] — a dict keyed by document id, which is why the ids had to be unique back in Step 1.

What the output means: It prints e.g. reconciled 2 results (2 ok) keyed by custom_id — every result mapped back to the document that produced it, regardless of the order they streamed in.

Try this: Picture the results arriving as doc-2 then doc-1. Because we write records[result.custom_id] and never records[i], both land in the right slot. Swap in position-based indexing and you'd silently attach doc-2's data to doc-1.

Never reconcile by positionThe batch does not preserve request order in its results stream — position i in the output is not document i. The custom_id is the only correct key. Handle errored and expired per-item so a single malformed document is recorded and skipped, not fatal.

Step 5 · Aggregate usage into a cost report professional

Every succeeded result carried a usage object (AP4): input_tokens, output_tokens, and the cache fields cache_creation_input_tokens / cache_read_input_tokens. Sum them across the batch, price each bucket, and apply the 50% batch discount to get the real spend. This helper runs offline over a fake usage list so you can execute the exact cost math with no key.

Python · cost-report aggregator (runs offline)

step5_costreport.py# (runs offline) stdlib only — sums per-item usage into a batch cost report.
# Prices are $ per 1M tokens (Opus 4.8); cache reads ~0.1x, cache writes ~1.25x,
# and the Batch API applies a 50% discount on top.

PRICES = {  # (input_per_mtok, output_per_mtok)
    "claude-opus-4-8": (5.00, 25.00),
}
BATCH_DISCOUNT = 0.50   # Message Batches bill at 50% of standard

# what Step 4 collected: one dict per succeeded result (usage as plain numbers here)
USAGES = [
    {"model": "claude-opus-4-8",
     "usage": {"input_tokens": 400, "output_tokens": 120,
               "cache_creation_input_tokens": 1200, "cache_read_input_tokens": 0}},
    {"model": "claude-opus-4-8",
     "usage": {"input_tokens": 380, "output_tokens": 110,
               "cache_creation_input_tokens": 0, "cache_read_input_tokens": 1200}},
    {"model": "claude-opus-4-8",
     "usage": {"input_tokens": 420, "output_tokens": 130,
               "cache_creation_input_tokens": 0, "cache_read_input_tokens": 1200}},
]

def cost_report(usages, discount=BATCH_DISCOUNT):
    totals = {"input": 0, "output": 0, "cache_write": 0, "cache_read": 0}
    dollars = 0.0
    for row in usages:
        u = row["usage"]
        in_price, out_price = PRICES[row["model"]]
        totals["input"]       += u["input_tokens"]
        totals["output"]      += u["output_tokens"]
        totals["cache_write"] += u["cache_creation_input_tokens"]
        totals["cache_read"]  += u["cache_read_input_tokens"]
        dollars += (
            u["input_tokens"]                * in_price
            + u["output_tokens"]             * out_price
            + u["cache_creation_input_tokens"] * in_price * 1.25   # write premium
            + u["cache_read_input_tokens"]     * in_price * 0.10   # read discount
        ) / 1_000_000
    return totals, dollars * (1 - discount)

totals, spend = cost_report(USAGES)
print(f"docs priced      : {len(USAGES)}")
print(f"input tokens     : {totals['input']}")
print(f"output tokens    : {totals['output']}")
print(f"cache write toks : {totals['cache_write']}")
print(f"cache read toks  : {totals['cache_read']}")
print(f"batch cost (USD) : ${spend:.5f}")

docs priced      : 3
input tokens     : 1200
output tokens    : 360
cache write toks : 1200
cache read toks  : 2400
batch cost (USD) : $0.01185

▶ How this works

Every succeeded result carried a usage object; this offline helper sums those across the batch and prices each token bucket, applying the 50% batch discount. It runs over a fake usage list so you can execute the exact cost math with no key.

PRICES maps the model to its (input, output) dollars per million tokens; BATCH_DISCOUNT = 0.50 is the Message Batches 50%-off rate.
USAGES is what Step 4 collected — one row per succeeded document. Note the first row has cache_creation tokens (it wrote the prefix), and the rest have cache_read tokens (they read it).
cost_report tallies each bucket and prices them: input/output at list price, cache writes at a 1.25x premium, cache reads at 0.1x. It divides by 1,000,000 (prices are per-MTok) and multiplies the total by (1 - discount) for the batch price.
The four token totals plus one dollar figure are exactly the report you'd hand to finance after a nightly run.

What the output means: It prints the four token totals and batch cost (USD) : $0.01185 — the summed spend after the 50% batch discount.

Try this: Zero out every cache_read and re-run: the cost jumps because each document now re-pays full input price for the shared prefix. That gap is the value of caching, in dollars.

The cache pays for itself after doc oneNotice the trace: the first document writes the prefix to cache (the 1.25x premium), and every document after it reads the prefix at ~0.1x. Over thousands of documents the shared instruction is billed once at write price and near-free thereafter — the whole reason to cache the prefix rather than inline it per request.

Step 6 · Production hardening tech-lead

A pipeline that runs once on three files is a demo; a pipeline finance trusts on a nightly folder of 50,000 documents needs the tech-lead concerns. The four that matter most: idempotency (never double-process a document), chunking (a batch caps at 100k requests / 256 MB, so split larger runs), monitoring (watch counts + cache-hit rate + spend), and cleanup (persist records, resubmit expired items).

Concern	Why it bites at scale	The control
Idempotency	A retried or re-run job double-charges you and duplicates records.	Derive `custom_id` deterministically from the doc; skip ids already persisted before submitting.
Chunking > N	One batch caps at 100k requests / 256 MB; a huge folder overflows a single submit.	Split the request list into chunks (e.g. 10k) and submit one batch per chunk, tracking each `batch.id`.
Monitoring	A silent cache miss or a spike in `errored` quietly burns money.	Log `request_counts`, the cache-read ratio, and per-batch spend; alert on regressions.
Cleanup / resubmit	`expired` and transient `api_error` items are lost if ignored.	Persist reconciled records; collect `expired` + retryable `errored` ids and resubmit as a new batch.

Python · idempotency + chunking + resubmit (needs API key + network)

pipeline/orchestrate.py# needs: pip install anthropic + ANTHROPIC_API_KEY; makes real batched API calls
# ▶ needs API key + network
import anthropic

client = anthropic.Anthropic()

MAX_PER_BATCH = 10_000   # stay well under the 100k / 256 MB batch ceiling

def already_done(custom_id: str) -> bool:
    """Idempotency gate: True if this doc's record is already persisted.
    (Back this with your DB / object store.)"""
    return custom_id in PERSISTED_IDS   # your durable set of finished ids

def chunk(seq, size):
    for i in range(0, len(seq), size):
        yield seq[i:i + size]

def submit_all(requests):
    """Skip finished docs (idempotency), then submit in chunks (chunking)."""
    fresh = [r for r in requests if not already_done(r["custom_id"])]
    batch_ids = []
    for group in chunk(fresh, MAX_PER_BATCH):
        batch = client.messages.batches.create(requests=group)
        batch_ids.append(batch.id)            # persist these for polling/cleanup
    return batch_ids

def resubmit_expired(records, all_requests):
    """Cleanup: gather expired + retryable ids and resubmit as a new batch."""
    retry_ids = {cid for cid, r in records.items()
                 if r["status"] in ("expired", "errored")}
    retry = [r for r in all_requests if r["custom_id"] in retry_ids]
    return submit_all(retry) if retry else []

▶ How this works

Running once on three files is a demo; a nightly folder of 50,000 documents needs the tech-lead concerns. This orchestrator adds idempotency, chunking, and resubmission around the pieces you built. ▶ needs API key + network.

already_done(custom_id) is the idempotency gate: before submitting, skip any document whose record is already persisted, so a re-run never double-processes or double-charges.
chunk + MAX_PER_BATCH handle chunking: a single batch caps at 100k requests / 256 MB, so submit_all splits the fresh requests into groups and submits one batch per group, collecting each batch.id to poll and clean up later.
resubmit_expired is cleanup: it gathers expired and retryable errored ids from the reconciled records and submits them as a fresh batch, reusing the same idempotent path.
Monitoring is the fourth concern (see the table): log request_counts, the cache-read ratio, and spend, and alert when any regress.

What the output means: Nothing prints on its own — these are the orchestration functions. In production submit_all returns the list of batch.ids you persist and poll, and resubmit_expired returns the ids for a retry batch.

Try this: Call submit_all twice on the same requests. Because already_done filters out anything persisted, the second call submits an empty (or much smaller) batch — that's idempotency saving you from paying twice for the same folder.

The one number to watch is the cache-read ratioIn production, cache_read_input_tokens / (cache_read + cache_creation) should approach 1.0 across a batch after the first document. If it stays near zero, a silent invalidator crept into SHARED_SYSTEM — every document is re-paying full price for the prefix and your batch costs balloon. Alert on it.

Extend it — take the pipeline further

Context: A capstone is a starting point, not a finish line. Each extension pulls in a real API-in-practice feature — vision, structured outputs, or pre-flight token counting — and the discipline is proving the cost math still balances afterward.

Your task: Pick one extension and wire it end-to-end into the pipeline, then re-run the Step-5 cost report to prove the numbers still balance.

Requirements:

Choose exactly one: Files API + vision, structured outputs, or a token-count pre-flight
Files API + vision: upload real PDFs and reference each by file id in its request
Structured outputs: replace the JSON-only instruction with a schema so every record is validated, not hand-parsed
Token-count pre-flight: estimate the batch bill on a sample document before submitting
After wiring it in, re-run the cost report and show the math still balances

💡 Hint: Change one thing at a time and re-derive the cost report against it — the extension is only done when the numbers reconcile.

🪜 Practice ladder beginner → industry

Six graded exercises, easy to real-world. Try each before opening its solution.

Exercise 1 · Milestone 1 — derive a stable custom_id per documentBeginner

Context: Batch results come back in any order, so every request needs a stable, unique custom_id to reconcile each result to its source document. Get this wrong and answers silently attach to the wrong file.

Your task: Derive a deterministic custom_id from each filename and build the batch request list from a folder listing — runnable offline.

Requirements:

Derive the id deterministically from the filename (strip extension, lowercase, sanitise unsafe chars)
The same filename always yields the same id
Build the request list keyed by these ids from a folder listing
Assert the ids are unique — no collisions across the folder
Runs fully offline with no API call

💡 Hint: A pure string transform (drop extension, lowercase, replace non-alphanumerics with hyphens) keeps ids stable; a set check proves uniqueness.

Show solution

Stable custom_id + request list (pure stdlib, runnable):

def doc_id(filename):
    stem = filename.rsplit(".", 1)[0]
    return "doc-" + "".join(c if c.isalnum() else "-" for c in stem).lower()

def build_requests(filenames, system_prompt):
    requests = []
    for name in filenames:
        requests.append({
            "custom_id": doc_id(name),
            "params": {
                "model": "claude-opus-4-8",
                "max_tokens": 1024,
                "system": system_prompt,
                "messages": [{"role": "user", "content": f"Extract fields from: {name}"}],
            },
        })
    return requests

reqs = build_requests(["Invoice_0001.pdf", "invoice 0002.PDF"], "extract fields")
print([r["custom_id"] for r in reqs])   # ['doc-invoice-0001', 'doc-invoice-0002']

The custom_id is the join key: batch results stream back in any order, so a stable, filename-derived id is what lets you reconcile each result to its source document. Deriving it deterministically means a re-run produces the same ids — the basis for idempotency later.

Exercise 2 · Milestone 2 — cache the shared instruction prefix (needs API key)Intermediate

Context: Every request repeats the same long extraction instructions. Marking that prefix with a cache breakpoint lets later requests re-read it cheaply — the caching win that stacks on top of the batch discount.

Your task: Show the correct Request shape that carries the shared instruction prefix with a cache_control breakpoint, reused byte-identical across every request. Mark this rung as needing the SDK.

Requirements:

A shared system prefix defined once, with a cache_control ephemeral breakpoint on its last block
The prefix is reused byte-identical on every request so the cache hits
Per-document text goes in the user turn, after the cached prefix
Build a real Request object per document via a builder function
Label the rung as requiring the SDK

💡 Hint: The prefix must be identical every time for the cache to hit — build it once and reuse the same object; the breakpoint goes on the final block of the shared prefix.

Show solution

The cached shared prefix — needs pip install anthropic + ANTHROPIC_API_KEY (documented types):

from anthropic.types.message_create_params import MessageCreateParamsNonStreaming
from anthropic.types.messages.batch_create_params import Request

SHARED_SYSTEM = [
    {
        "type": "text",
        "text": ("You are a precise document-extraction engine. Extract vendor, "
                 "invoice_number, total_amount, due_date. Missing field -> null. "
                 "... long, stable rules + few-shot examples ..."),
        "cache_control": {"type": "ephemeral"},      # CACHE BREAKPOINT
    },
]

def build_request(custom_id, document_text):
    return Request(
        custom_id=custom_id,
        params=MessageCreateParamsNonStreaming(
            model="claude-opus-4-8",
            max_tokens=1024,
            system=SHARED_SYSTEM,                    # reused byte-identical -> cache hit
            messages=[{"role": "user", "content": document_text}],
        ),
    )
print(build_request("doc-invoice-0001", "Invoice text ...").custom_id)

The cache_control: ephemeral breakpoint marks the stable prefix; the first request writes the cache and every later request re-reads it at ~10% of the input price. The prefix must be byte-identical across requests to hit — so the shared system block is built once and reused verbatim.

Exercise 3 · Milestone 3 — submit the batch and poll to completion (needs API key)Advanced

Context: A Message Batch is asynchronous server-side work. You submit the whole request list at once, then poll the batch's status until it ends — watching the per-request counts as they progress.

Your task: Submit all requests in one batch and poll until it ends, reading the processing status and the per-request counts each loop. Mark this rung as needing the SDK.

Requirements:

Submit the entire request list in a single batch-create call
Poll by retrieving the batch and checking its processing status
Print the live request counts (processing / succeeded / errored) each iteration
Sleep between polls rather than busy-waiting
Exit the loop when the status reaches ended
Label the rung as requiring the SDK

💡 Hint: One create call, then a while loop of retrieve + status check with a sleep — the server does the work, you just watch the counts move.

Show solution

Submit + poll — needs anthropic + ANTHROPIC_API_KEY (documented batch API):

import time, anthropic
client = anthropic.Anthropic()

batch = client.messages.batches.create(requests=requests)   # submit ALL at once
print("submitted:", batch.id, batch.processing_status)

while True:
    batch = client.messages.batches.retrieve(batch.id)
    if batch.processing_status == "ended":
        break
    c = batch.request_counts
    print(f"status={batch.processing_status} processing={c.processing} "
          f"succeeded={c.succeeded} errored={c.errored}")
    time.sleep(30)

c = batch.request_counts
print(f"ended: succeeded={c.succeeded} errored={c.errored} "
      f"canceled={c.canceled} expired={c.expired}")

Batches are asynchronous: you submit the whole set, then poll processing_status until it is "ended", reading request_counts for progress. This is why batch is ~50% cheaper than real-time — you trade latency for a large throughput discount, ideal for a nightly document run.

Exercise 4 · Milestone 4 — reconcile results by custom_id (needs API key)Expert

Context: Results stream back in any order and each has its own status, so you must key every result by its custom_id — never by position — and handle succeeded, errored, and expired distinctly.

Your task: Stream the batch results, key each by custom_id, and branch on succeeded / errored / expired, decoding the structured payload on success. Mark this rung as needing the SDK.

Requirements:

Iterate the streamed results (which arrive in arbitrary order)
Store each record under its custom_id, never a positional index
On success, extract the content and JSON-decode it, handling bad JSON
On errored, record the error type; on expired, mark it for resubmission
Collect per-result usage for the later cost report
Label the rung as requiring the SDK

💡 Hint: The load-bearing detail is records[custom_id], not records[i] — positional indexing silently misaligns out-of-order results.

Show solution

Reconcile by custom_id — needs anthropic + ANTHROPIC_API_KEY (documented result shape):

import json, anthropic
client = anthropic.Anthropic()

records, usages = {}, []
for result in client.messages.batches.results(batch.id):   # streams, any order
    cid, kind = result.custom_id, result.result.type
    if kind == "succeeded":
        msg = result.result.message
        text = next((b.text for b in msg.content if b.type == "text"), "")
        try:
            records[cid] = {"status": "ok", "data": json.loads(text)}
        except json.JSONDecodeError:
            records[cid] = {"status": "bad_json", "raw": text}
        usages.append({"model": msg.model, "usage": msg.usage})
    elif kind == "errored":
        records[cid] = {"status": "errored", "error": result.result.error.type}
    elif kind == "expired":
        records[cid] = {"status": "expired"}

ok = sum(1 for r in records.values() if r["status"] == "ok")
print(f"reconciled {len(records)} results ({ok} ok) keyed by custom_id")

Because results arrive unordered, the custom_id is the only reliable way to map each back to its document. Handling succeeded/errored/expired distinctly — and guarding JSON parsing — is what turns a raw batch stream into a clean, per-document record set you can trust.

Exercise 5 · Milestone 5 — aggregate the cost reportProfessional

Context: The pipeline's value proposition is cost, so it has to be measured. Roll captured usage into a report that prices each token bucket correctly and applies the batch discount — offline, on captured numbers.

Your task: Aggregate token usage into a cost report: sum input, output, and cache token buckets, price them (cache-write at 1.25× input, cache-read at 0.10× input), and apply the 50% batch discount.

Requirements:

A price table mapping the model to input/output per-million rates
Sum input, output, cache-creation, and cache-read tokens across all usages
Price cache-write at 1.25× and cache-read at 0.10× the input rate
Apply the 50% batch discount to the total
Return the four token totals and the final dollar spend
Runs offline on captured usage numbers

💡 Hint: Price each bucket separately then discount the sum; the first doc pays the cache-write premium and later docs collect the cache-read discount.

Show solution

The cost aggregator — batch discount + cache pricing (pure arithmetic, runnable):

PRICES = {"claude-opus-4-8": (5.00, 25.00)}   # (input, output) per 1M tokens
BATCH_DISCOUNT = 0.50

USAGES = [
    {"model": "claude-opus-4-8",
     "usage": {"input_tokens": 400, "output_tokens": 120,
               "cache_creation_input_tokens": 1200, "cache_read_input_tokens": 0}},
    {"model": "claude-opus-4-8",
     "usage": {"input_tokens": 380, "output_tokens": 110,
               "cache_creation_input_tokens": 0, "cache_read_input_tokens": 1200}},
]

def cost_report(usages, discount=BATCH_DISCOUNT):
    dollars = 0.0
    for row in usages:
        u = row["usage"]; in_p, out_p = PRICES[row["model"]]
        dollars += (u["input_tokens"] * in_p
                    + u["output_tokens"] * out_p
                    + u["cache_creation_input_tokens"] * in_p * 1.25   # cache write
                    + u["cache_read_input_tokens"]     * in_p * 0.10   # cache read
                   ) / 1_000_000
    return round(dollars * (1 - discount), 6)

print(f"batch cost (USD): ${cost_report(USAGES)}")

The report separates the four token types because they price differently: cache writes cost 1.25x input, cache reads only 0.10x, and the whole batch gets a 50% discount. Aggregating real usage is how you verify caching actually paid off — the second request re-reads the prefix cheaply instead of re-billing it.

Exercise 6 · Milestone 6 — harden for 50k docs/month as pipeline ownerIndustry scenario

Context: At 50k docs a month the pipeline must survive crashes and stay within batch limits. As owner you add idempotency, chunking under the batch-size cap, and resubmission of expired or errored items.

Your task: Make the pipeline production-safe: skip already-processed ids (idempotency), chunk requests under the batch-size limit, and resubmit expired/errored items after a crash — runnable offline.

Requirements:

Track processed ids in a durable set and skip ids already done
A chunker that splits requests into groups under the max-per-batch limit
A submission planner that filters done ids then chunks the remainder
A resubmit step that collects only expired/errored ids into a retry batch
Runs offline, modelling the persisted-state and crash-recovery behaviour

💡 Hint: Idempotency is a membership check against a persisted id set; recovery is just re-planning over the items whose status was expired or errored.

Show solution

Idempotency + chunking + resubmit (pure stdlib, runnable):

MAX_PER_BATCH = 10_000
PERSISTED_IDS = {"doc-invoice-0001"}          # results already stored

def already_done(custom_id):
    return custom_id in PERSISTED_IDS

def chunk(seq, size):
    for i in range(0, len(seq), size):
        yield seq[i:i + size]

def plan_submission(requests):
    fresh = [r for r in requests if not already_done(r["custom_id"])]
    return [len(g) for g in chunk(fresh, MAX_PER_BATCH)]   # batch group sizes

def resubmit_expired(records, all_requests):
    retry_ids = {cid for cid, r in records.items()
                 if r["status"] in ("expired", "errored")}
    return [r for r in all_requests if r["custom_id"] in retry_ids]

reqs = [{"custom_id": f"doc-{i:05d}"} for i in range(3)]
reqs[0]["custom_id"] = "doc-invoice-0001"     # this one is already done
print("groups:", plan_submission(reqs))       # skips the persisted id
print("resubmit:", resubmit_expired({"doc-00002": {"status": "expired"}}, reqs))

Industry scenario: a finance team runs 50,000 invoices/month; a Lambda crash mid-run must not double-process or drop documents. Idempotency (skip persisted ids) makes re-runs safe, chunking respects the batch-size limit, and resubmitting only expired/errored items recovers cleanly. The stable custom_id from milestone 1 is what makes all three possible.

✓ Checkpoint — you can move on when you can…

Explain the pipeline on one line: folder → cached requests → batch → poll → retrieve → reconcile by custom_id → records + cost.
Build a request list where every document has a unique, deterministic custom_id.
Place a cache_control breakpoint on the shared instruction prefix and keep it byte-identical.
Submit with batches.create, poll batches.retrieve until "ended", and stream batches.results().
Reconcile succeeded/errored/expired results by custom_id and never by position.
Aggregate usage into a cost report and name the four production concerns (idempotency, chunking, monitoring, cleanup).

📋 📋 Grade your batch pipeline

Dimension	Meets the bar	Above the bar (staff)
Batch correctness	One request per doc with a unique custom_id; results reconciled back correctly	Handles partial/failed items per-custom_id without losing the batch
Caching	Shared instruction prefix carries a cache_control breakpoint	Verified cache hits in usage; measurable cost/latency win
Cost model	Reports real $ from usage; shows the batch vs realtime saving	Attributes cost per doc and flags outliers; budget-aware
Error handling	Per-item errors are captured, not fatal to the run	Idempotent re-runs; retries only the failed items
Scale	Chunks jobs beyond the batch size limit	Monitors long-running batches; backpressure/queueing under load
Structured output	Extracted data is schema-valid and stored cleanly	Validation-gated; malformed items quarantined, not silently dropped

Score each 0/1/2. 6–8 = solid; 9–12 = staff-level. Blocking: results not reconciled by custom_id, or no cost reporting.

Knowledge check check yourself

✓ Knowledge check

Why must results be reconciled back to their documents by custom_id and never by position in the results list?

Show answer

The Message Batches API returns results in any order and asynchronously, so positional matching would misalign answers with documents. A stable per-document custom_id is the only reliable key to map each succeeded/errored/expired result back to its source document.

✓ Knowledge check

Why put a single cache_control breakpoint on the shared instruction prefix, and how does it combine with batching for savings?

Show answer

The same long instruction prefix is repeated across every document's request; a cache breakpoint lets every request read that prefix cheaply instead of re-billing it each time. Combined with the batch submission's ~50% discount, caching the prefix plus batching stacks the cost savings.

Educational content, provided as-is and without warranty. Code samples are examples — review, test, and adapt them before using in production. See the Terms of Use & Disclaimer. Use at your own risk.

Learning objectives

Architecture — the whole pipeline on one line essential

Step 1 · Build the request list — one per document essential

Step 2 · Cache the shared instruction prefix intermediate

Step 3 · Submit the batch and poll until it ends advanced

Step 4 · Retrieve results and reconcile by custom_id professional

Step 5 · Aggregate usage into a cost report professional

Step 6 · Production hardening tech-lead

Extend it — take the pipeline further

🪜 Practice ladder beginner → industry

✓ Checkpoint — you can move on when you can…

Knowledge check check yourself

More in Projects