Classifying, normalizing, or deduplicating millions of records with an LLM is an engineering problem, not a prompting problem: batch endpoints (50% discount), prompt caching (~90% off repeated prefixes), model-tier routing, and token accounting decide whether the job costs $40 or $4,000.
Tier: Core · Workflow area: LLM Data Pipelines & Cost Engineering
This playbook builds a batch enrichment pipeline with cost controls and sampling QA baked in.
For sub-1k rows, a plain loop with pb-14's extraction function is fine — skip the batch machinery.
polars or Spark for the surrounding dataframe work.Estimate cost before anything runs. Count tokens on 100 representative rows and extrapolate:
n = client.messages.count_tokens(model="claude-haiku-4-5",
system=SYSTEM_PROMPT, messages=[{"role": "user", "content": row_text}])
# total_cost ≈ rows * (avg_in * price_in + avg_out * price_out) * 0.5 (batch discount)
Present the estimate per tier. If the number is unacceptable, fix it here — shorter prompt, cheaper tier, truncated inputs — not after the invoice.
Structure prompts for caching. Put the static instruction block + few-shot examples first with a cache breakpoint; only the row payload varies:
system=[{"type": "text", "text": INSTRUCTIONS_AND_EXAMPLES,
"cache_control": {"type": "ephemeral"}}]
Verify with usage.cache_read_input_tokens > 0 on the second call; a zero means a silent invalidator (timestamp, unsorted JSON, varying prefix). Note minimum cacheable prefix sizes are model-dependent (~1–4k tokens) — pad the shared block with examples rather than shrinking below the threshold.
Route by tier with an escalation rule. Default every row to the cheap tier; escalate to the frontier tier only rows where the cheap model abstains or reports low confidence (add confidence: Literal["high","low"] to the schema). Typical result: 90–95% of rows at ~1/5 the price.
Submit in batch, keyed by stable IDs. Chunk into batches (≤100k requests / 256 MB each on Anthropic), custom_id = your primary key:
from anthropic.types.messages.batch_create_params import Request
from anthropic.types.message_create_params import MessageCreateParamsNonStreaming
batch = client.messages.batches.create(requests=[
Request(custom_id=str(row.id), params=MessageCreateParamsNonStreaming(
model="claude-haiku-4-5", max_tokens=200,
system=cached_system, messages=[{"role": "user", "content": row.text}]))
for row in chunk])
Poll processing_status == "ended", then stream results and join on custom_id — results arrive in any order, never join by position. Handle all four result types (succeeded/errored/canceled/expired); resubmit expired, dead-letter errored.
Sampling QA, then full run. Run the pilot batch, score it against your 200 labeled rows, and only then submit the full job. During the full run, continuously sample 0.5–1% of outputs into a review queue; alert if the "unknown"/abstain rate or per-field distributions drift from the pilot.
Make the pipeline idempotent. Persist raw responses keyed by (custom_id, prompt-version, model); reruns skip already-enriched keys. A prompt or schema change bumps the version and triggers selective re-enrichment, not a full re-spend.
The 50% batch discount, cache reads, and tier routing multiply: cheap-tier + cached prefix + batch is routinely 20–30x cheaper per row than naive frontier-tier synchronous calls — measure your own stack's factor and put it in the report.
client.messages.batches.create(...) as above; caching works inside batches, so the shared system block gets cache pricing across the batch.client.files.create(file=..., purpose="batch"), then client.batches.create(input_file_id=..., endpoint="/v1/chat/completions", completion_window="24h"); results come back as a JSONL output_file_id keyed by custom_id. Same 50% discount.usage fields; effective cost per 1k rows recomputed from actual token counts and within 15% of the pre-run estimate.custom_id.max_tokens and a schema that leaves no room for prose.