Use Batches for Bulk Tasks
Putting the Pieces Together
Lessons 131–138 each covered one stage of batch work: creating batches, assigning identifiers, watching status, listing batches, downloading results, interpreting outcomes, and retrying failures. A bulk task uses all of them at once. This lesson assembles those stages into one pipeline, runs it end to end on a realistic job, and explains the decisions that shape the design: how big each batch should be, how instructions are shared across requests, how results are checkpointed, and how the cost compares with synchronous calls.
The example job is classifying customer reviews. It is simple enough to follow, and it has the properties of most bulk work: many records, a shared set of instructions, a structured answer, and a need for every record to be accounted for.
The Pipeline Stages
Each stage reads from the previous one's output and writes something durable, which is what makes the pipeline restartable:
records ──► build requests ──► split into batches ──► submit ──► wait ──► download ──► process ──► store
(source) (stable IDs, (size limits, (save IDs) (status) (raw file) (outcomes) (database)
shared prefix) Lesson 132) Lesson 133 Lesson 134 Lesson 136 Lesson 137The batch IDs and the raw results file are the two checkpoints. If the process stops while batches are running, the saved IDs are enough to resume waiting. If it stops after download, the raw file is enough to reprocess without calling the API again.
Building Requests with a Shared Prefix
Every review classification uses the same instructions. Repeating those instructions in every request costs input tokens each time. Prompt caching (Lessons 121–130) can make the shared part cheap, and the documentation says it works in batches too, on a best-effort basis. For batches, the recommended approach is a one-hour cache entry on the shared block, because batches often take longer than the default five minutes.
"""bulk_pipeline.py: classify many reviews through batches, end to end."""
import json
from collections import Counter
from batch_builder import MODEL, validate, split_into_batches
from batch_status import wait_for_batch
from process_batch import cost_of, process_record
INSTRUCTIONS = (
"You classify customer reviews of an online course as positive, negative, or mixed. "
"Reply with JSON only, in the form {\"label\": ..., \"confidence\": ...}, where confidence "
"is a number from 0 to 1. Judge the overall opinion, not individual words. "
) * 20 # Long enough to be worth caching
SYSTEM = [{"type": "text", "text": INSTRUCTIONS,
"cache_control": {"type": "ephemeral", "ttl": "1h"}}]
def build_request(record: dict) -> dict:
return {
"custom_id": f"review-{record['id']:04d}",
"params": {
"model": MODEL,
"max_tokens": 64,
"system": SYSTEM,
"messages": [{"role": "user", "content": f"<review>\n{record['text']}\n</review>"}],
},
}
def chunked(items: list, size: int):
for start in range(0, len(items), size):
yield items[start:start + size]Several choices here are deliberate. The custom_id comes from the record's stable numeric ID, zero-padded, so IDs sort correctly and never change when new records are added. The review text is wrapped in tags, which keeps it separate from the instructions (Lesson 37). The output limit is small, because a label and a confidence fit in a few dozen tokens, and a smaller limit reduces the risk of truncation.
Running the Pipeline
The pipeline separates the work from the service it talks to. The three service operations are passed in as functions: submitting a batch, fetching the results of a finished batch, and reporting progress. That separation lets the same pipeline run against the live API, or against a stand-in for testing:
def run_pipeline(records: list[dict], submit_batch, fetch_results, batch_size: int = 5):
requests = [build_request(record) for record in records]
problems = validate(requests)
if problems:
raise ValueError("; ".join(problems))
batch_ids = []
for chunk in chunked(requests, batch_size):
batch_id = submit_batch(chunk)
batch_ids.append(batch_id)
print(f"submitted {batch_id}: {len(chunk)} requests")
outcomes = []
for batch_id in batch_ids:
for line in fetch_results(batch_id):
outcomes.append(process_record(line))
status_counts = Counter(outcome.status for outcome in outcomes)
labels = Counter(outcome.value["label"] for outcome in outcomes if outcome.status == "ok")
tokens = sum(cost_of("claude-sonnet-5-5", o.input_tokens, o.output_tokens) for o in outcomes)
return outcomes, status_counts, labels, tokensThe function validates every request before submitting anything, so a problem in the data is found before the first batch leaves. It submits in chunks, keeps the batch IDs, and processes the results only after all batches have been fetched. The totals at the end are the figures a team actually reports: how many records were classified, how many need attention, and what the job cost.
The Live Functions
With a real client, the three service operations use the calls from the earlier lessons. Submitting returns the ID. Fetching waits for the batch to end using the function from Lesson 134, then streams the results from Lesson 136:
def live_submit(client, chunk: list[dict]) -> str:
return client.messages.batches.create(requests=chunk).id
def live_fetch(client, batch_id: str):
wait_for_batch(client, batch_id)
for result in client.messages.batches.results(batch_id):
yield result.model_dump()A production job would save each batch ID to the registry at the moment of submission (Lesson 133), so an interruption never loses track of a batch that is already running.
A Dry Run Without the API
To test the pipeline without spending money, the stand-in below produces results in memory. It returns successful classifications with realistic token counts, and it returns one invalid request, so the pipeline's handling of a failure is exercised too:
"""bulk_pipeline_demo.py: run the full pipeline against an in-memory stand-in for the API."""
import json
from bulk_pipeline import run_pipeline
LABELS = ["positive", "negative", "mixed"]
RECORDS = [{"id": n, "text": f"Review number {n}."} for n in range(1, 13)]
stored: dict[str, list[dict]] = {}
def dry_submit(chunk: list[dict]) -> str:
batch_id = f"msgbatch_{len(stored) + 1}"
lines = []
for request in chunk:
custom_id = request["custom_id"]
if custom_id == "review-0007":
outcome = {"type": "errored", "error": {"type": "error", "error": {
"type": "invalid_request_error", "message": "unsupported input"}}}
else:
label = LABELS[int(custom_id.split("-")[1]) % 3]
outcome = {"type": "succeeded", "message": {
"model": "claude-sonnet-5-5", "stop_reason": "end_turn",
"content": [{"type": "text", "text": json.dumps({"label": label, "confidence": 0.9})}],
"usage": {"input_tokens": 120, "output_tokens": 4}}}
lines.append({"custom_id": custom_id, "result": outcome})
stored[batch_id] = lines
return batch_id
def dry_fetch(batch_id: str):
yield from stored[batch_id]
if __name__ == "__main__":
outcomes, status_counts, labels, tokens = run_pipeline(RECORDS, dry_submit, dry_fetch, batch_size=5)
print("status:", dict(status_counts))
print("labels:", dict(labels))
print(f"batch cost: ${tokens:.6f}")
synchronous = sum(
(o.input_tokens * 2.00 + o.output_tokens * 10.00) / 1_000_000 for o in outcomes
)
print(f"same work, synchronous prices: ${synchronous:.6f}")Output:
submitted msgbatch_1: 5 requests
submitted msgbatch_2: 5 requests
submitted msgbatch_3: 2 requests
status: {'ok': 11, 'fix_request': 1}
labels: {'negative': 3, 'mixed': 4, 'positive': 4}
batch cost: $0.001540
same work, synchronous prices: $0.003080How It Works
- The records become requests with stable IDs and a shared, cached system prompt. Twelve records produce twelve requests, which the validator accepts.
- The chunker splits the requests into batches of five, giving batches of 5, 5, and 2. In a real job the size would be set by the limits in Lesson 132, and usually be far larger.
- Each batch is submitted and its ID kept, so the run can be resumed after an interruption.
- The results are processed only after every batch has been fetched, using the same
process_record()from Lesson 137, which turns each line into a uniform outcome. - Record 7 comes back as
fix_request, because its invalid-request error is classified exactly as in the previous lesson. It is not counted in the cost, because it has no usage. - The cost is half the synchronous cost. Eleven records at 120 input and 4 output tokens cost $0.001540 at batch prices, against $0.003080 at standard prices. The difference is the 50% batch discount, applied to the same work.
Choosing the Batch Size
The demonstration uses five requests per batch only to make the splitting visible. A real job should use batches as large as the limits and the operational needs allow. Larger batches reduce the number of submissions and status checks. Smaller batches make each unit of work easier to recover, and they let early results arrive sooner, which matters when the job is monitored by people.
The practical guidance is to keep batches large enough that overhead is negligible, and small enough that one failed batch does not hold up the whole job. Thousands of requests per batch is a common range for short tasks like classification, with the byte limit checked on the actual data.
Monitoring a Long Job
For a job that runs for hours, progress reporting matters more than the final numbers. The status function from Lesson 134 can report after every check, and the listing from Lesson 135 can summarize every batch in the job at once. A simple dashboard needs only those two calls: one to show which batches are still running, and one to show the running totals of succeeded and errored requests.
When the Pipeline Is Not Worth Building
A pipeline like this has real overhead: a registry, retries, validation, and monitoring. For a few hundred records that need an answer in the next minute, a synchronous loop is simpler and fast enough. The pipeline pays for itself when the volume is large, the answers can wait, and the work must be reproducible, auditable, and resumable.