Build a Batch Processor
The Goal
A batch processor takes a large set of records, sends them to the model as a batch job at half the standard price, and returns a labeled result for each record along with a clear account of what did not succeed. It is the production form of the bulk work from Section 8: a sentiment label for every review in a catalog, a category for every support ticket in a month, or a summary for every document in an archive.
The project gathers the batch lessons into one program. Requests are built with stable identifiers (Lesson 133), split into batches within the limits (Lesson 132), submitted, collected and processed (Lessons 136 and 137), and failures are classified so that only the right ones are retried (Lesson 138). The output is a report, not just a list of answers.
What You Will Build
A run_batch_job function that:
- builds one request per record, with a stable identifier from the record's own key,
- validates the requests before anything is submitted,
- splits the requests into batches and submits each one,
- collects the results and sorts them into succeeded, needs-fix, and retryable,
- reports the labels, the failures, and the cost at batch prices.
Project Structure
batch_processor/
├── processor.py # building requests, validation, submission, and processing
└── demo.py # an in-memory stand-in for the batch service, with one invalid recordStep 1: Building and Validating Requests
Each record becomes a request whose identifier comes from the record's own identifier, not from its position in the list. Validation checks the identifier pattern and rejects duplicates before a batch is submitted, since the batch service's own check arrives only when processing ends:
"""processor.py: a batch job from records to labels, with validation and cost accounting."""
import re
from collections import Counter
MODEL = "claude-sonnet-5-5"
CUSTOM_ID = re.compile(r"^[a-zA-Z0-9_-]{1,64}$")
BATCH_PRICE = {"input": 1.00, "output": 5.00} # USD per million tokens, batch rates
BATCH_SIZE = 3
def make_request(record: dict) -> dict:
return {
"custom_id": f"rec-{record['id']}",
"params": {
"model": MODEL,
"max_tokens": 16,
"messages": [{"role": "user",
"content": f"Classify the sentiment of this review as positive, negative, "
f"or mixed. Reply with one word.\n\n{record['text']}"}],
},
}
def validate(requests: list[dict]) -> list[str]:
problems, seen = [], set()
for request in requests:
custom_id = request["custom_id"]
if not CUSTOM_ID.match(custom_id):
problems.append(f"invalid custom_id {custom_id}")
if custom_id in seen:
problems.append(f"duplicate custom_id {custom_id}")
seen.add(custom_id)
return problems
def chunked(items: list, size: int):
for start in range(0, len(items), size):
yield items[start:start + size]Step 2: Submitting and Collecting
The job submits each chunk, keeps the batch identifiers, and then reads the results of every batch. The results arrive in no particular order, so each one is stored under its identifier:
def run_batch_job(client, records: list[dict]) -> dict:
requests = [make_request(record) for record in records]
problems = validate(requests)
if problems:
raise ValueError("; ".join(problems))
batch_ids = [client.messages.batches.create(requests=chunk).id
for chunk in chunked(requests, BATCH_SIZE)]
labels: dict[str, str] = {}
needs_fix: list[str] = []
retryable: list[str] = []
cost = 0.0
for batch_id in batch_ids:
for line in client.messages.batches.results(batch_id):
result = line.result
if result.type == "succeeded":
labels[line.custom_id] = "".join(
block.text for block in result.message.content if block.type == "text"
).strip().lower()
usage = result.message.usage
cost += (usage.input_tokens * BATCH_PRICE["input"]
+ usage.output_tokens * BATCH_PRICE["output"]) / 1_000_000
elif result.type == "errored" and result.error.error.type == "invalid_request_error":
needs_fix.append(line.custom_id)
else:
retryable.append(line.custom_id)
return {"batches": len(batch_ids), "labels": labels, "needs_fix": needs_fix,
"retryable": retryable, "cost": cost}How It Works
- Validation runs before submission, so a malformed identifier or a duplicate stops the job immediately, at no cost.
- The chunks are submitted in order, and the identifiers are kept so the job can be followed and resumed (Lesson 139).
- Each result is sorted by its type. A success adds a label and its cost. An invalid request goes to
needs_fix, because resubmitting it unchanged would fail again. Any other failure goes toretryable, since a later attempt may succeed (Lesson 138). - Costs are summed from the usage of successes only. Errored and canceled results carry no tokens, so they add nothing (Lesson 137).
- The report keeps the three outcomes separate, so a person can see what to correct and what to resubmit.
Step 3: A Demonstration
The demonstration uses an in-memory stand-in for the batch service. It processes six records in two batches, returns a label for each record except one, which has an invalid request, and reports usage on each success:
"""demo.py: a batch job over six reviews, with one invalid record."""
from collections import Counter
from types import SimpleNamespace
from processor import run_batch_job
LABELS = {1: "positive", 2: "negative", 3: "positive", 5: "mixed", 6: "positive"}
def success(text):
return SimpleNamespace(type="succeeded", message=SimpleNamespace(
content=[SimpleNamespace(type="text", text=text)],
usage=SimpleNamespace(input_tokens=30, output_tokens=2)))
class DryBatches:
def __init__(self):
self.stored = {}
self.count = 0
def create(self, requests):
self.count += 1
batch_id = f"msgbatch_{self.count}"
lines = []
for request in requests:
number = int(request["custom_id"].split("-")[1])
if number == 4:
result = SimpleNamespace(type="errored", error=SimpleNamespace(
error=SimpleNamespace(type="invalid_request_error")))
else:
result = success(LABELS[number])
lines.append(SimpleNamespace(custom_id=request["custom_id"], result=result))
self.stored[batch_id] = lines
return SimpleNamespace(id=batch_id)
def results(self, batch_id):
yield from self.stored[batch_id]
if __name__ == "__main__":
client = SimpleNamespace(messages=SimpleNamespace(batches=DryBatches()))
records = [{"id": n, "text": f"review {n}"} for n in range(1, 7)]
report = run_batch_job(client, records)
print("batches submitted:", report["batches"])
print("succeeded:", len(report["labels"]))
print("labels:", dict(Counter(report["labels"].values())))
print("needs fix:", report["needs_fix"])
print(f"batch cost: ${report['cost']:.6f}")Output:
batches submitted: 2
succeeded: 5
labels: {'positive': 3, 'negative': 1, 'mixed': 1}
needs fix: ['rec-4']
batch cost: $0.000200How It Works
- The six records make two batches of three, which the report shows.
- Five records succeed, and their labels are counted. Positive appears three times, negative once, and mixed once.
- Record four is an invalid request, so it lands in
needs fix, not in the retry list, because a second attempt would fail identically. - The cost is five successes at forty micro-dollars each, which is 30 input tokens at one dollar per million, plus 2 output tokens at five dollars per million, for each record. The batch rate is half the standard price, which is why the figure is small.
- The report carries the counts a team needs, so the job is judged by what it produced, not only by whether it finished.
Going Live
The stand-in is replaced by Anthropic(), and the job runs the same way. In production, the batch identifiers are written to a registry as they are created (Lesson 133), and the retryable list is resubmitted as a new batch with an attempt count (Lesson 138), so that a failed job can be finished without repeating the successes.
Operating the Job
A large job runs for hours, and its status should be checked with the polling pattern from Lesson 134 while it runs. Its results should be saved as they arrive, so a crash does not lose completed work, and its cost should be reported against the batch budget (Lesson 186). The processor shown here reads results only after the batches finish, which is simple and correct for demonstration. A production version would process each batch as it completes.