Process Batch Outputs

Turning Results into Data

Lesson 136 saved every result exactly as the API produced it, and indexed them by custom_id. The saved file is raw material. Your application needs something else: the label for each review, the summary for each ticket, the extracted fields for each invoice, stored in the form your database and reports expect. Processing is the step that converts one into the other.

Processing has four responsibilities. It must interpret each outcome type correctly, extract the content of successful messages, validate that content against the shape you asked for, and account for cost. This lesson covers each one, using a running example: classifying product reviews with structured output, so every answer must match a schema before it is accepted.

Interpreting a Succeeded Result

A succeeded result contains a complete message, which means it has all the properties a synchronous response has. Four of those properties decide what the result means:

A succeeded result with stop_reason: "end_turn" is a complete answer. A succeeded result with max_tokens is an answer that was cut off, and it is labeled succeeded only because the request itself completed. Treat that outcome as incomplete, and the same applies to refusal. The result type alone never tells you the answer is usable. Lesson 43 covered the stop reasons in detail, and they matter just as much in batches.

Here is the extraction step, applied to a single record from the saved file:

def text_of(message: dict) -> str:
    return "".join(block.get("text", "") for block in message["content"] if block.get("type") == "text").strip()

Reading blocks by type is important. A message can contain several blocks, and thinking blocks, when enabled, come before the text. Joining the text blocks, and nothing else, gives the answer the user should see.

Interpreting Errored, Canceled, and Expired Results

The three non-success types each mean something different, and processing should keep them apart:

The documentation's example shows this distinction in code. An errored result's error carries an error.error.type field, and invalid_request_error marks problems you must fix, while other types can be retried directly. Lesson 138 builds the retry logic on this split.

None of the three non-success types is billed, so retrying them adds cost only for requests that actually run.

Extracting Structured Answers

When a batch request asks for structured output (Lesson 101), its successful text is JSON. The processing step parses the JSON and validates it against the same model used to build the request. Validation rejects answers that are well-formed JSON but wrong for the schema, such as a label outside the allowed set, and those answers should be marked for review rather than accepted:

import json
from typing import Literal

from pydantic import BaseModel, ValidationError


class ReviewLabel(BaseModel):
    label: Literal["positive", "negative", "mixed"]
    confidence: float


def parse_label(text: str) -> tuple[ReviewLabel | None, str | None]:
    try:
        return ReviewLabel.model_validate_json(text), None
    except (ValidationError, ValueError) as error:
        return None, str(error).splitlines()[0]

The function returns a pair: the validated label, or the reason it failed. Callers never receive a half-valid object. The ValueError branch covers text that is not JSON at all, which can happen when a request was truncated and the output stopped mid-object.

Processing Each Record

Each record needs a single function that takes one line of the saved file and returns a uniform outcome. Uniform outcomes make the rest of the pipeline simple: storage, reporting, and retries all handle the same shape, regardless of what happened:

from dataclasses import dataclass


@dataclass
class Outcome:
    custom_id: str
    status: str            # "ok", "needs_review", "retry", "fix_request", "not_run"
    value: dict | None
    note: str
    input_tokens: int = 0
    output_tokens: int = 0


def process_record(record: dict) -> Outcome:
    custom_id = record["custom_id"]
    result = record["result"]
    if result["type"] == "succeeded":
        message = result["message"]
        usage = message.get("usage", {})
        tokens = dict(input_tokens=usage.get("input_tokens", 0), output_tokens=usage.get("output_tokens", 0))
        if message["stop_reason"] != "end_turn":
            return Outcome(custom_id, "needs_review", None, f"stop_reason={message['stop_reason']}", **tokens)
        label, problem = parse_label(text_of(message))
        if label is None:
            return Outcome(custom_id, "needs_review", None, problem, **tokens)
        return Outcome(custom_id, "ok", label.model_dump(), "", **tokens)
    if result["type"] == "errored":
        kind = result.get("error", {}).get("error", {}).get("type")
        status = "fix_request" if kind == "invalid_request_error" else "retry"
        return Outcome(custom_id, status, None, kind or "unknown error")
    return Outcome(custom_id, "not_run", None, result["type"])

Every path returns an Outcome. A truncated answer is needs_review, not an error, because the request did run and a person may want to see its partial output. An invalid request is fix_request, and a temporary failure is retry. Canceled and expired requests are not_run, because they never reached the model.

Accounting for Cost

Batch pricing is half the standard price (Lesson 131), so the cost of each result is computed from its usage with the batch rates. The rates are per million tokens, and the table below uses the current batch prices for Sonnet 5.5:

BATCH_PRICES = {"claude-sonnet-5-5": {"input": 1.00, "output": 5.00}}


def cost_of(model: str, input_tokens: int, output_tokens: int) -> float:
    prices = BATCH_PRICES[model]
    return (input_tokens * prices["input"] + output_tokens * prices["output"]) / 1_000_000

Summing cost_of() over every ok and needs_review outcome gives the true cost of the batch, because only succeeded requests carry tokens. The totals belong in the batch's record, so reports can show cost per job rather than cost per request.

Practical Example: Processing a Saved Batch

The example below reads a results file written in the format from Lesson 136, processes every record, and prints a report. The demonstration file contains four records: a clean label, a truncated answer, a request with an invalid parameter, and a canceled request. Each one exercises a different branch.

"""process_batch.py: turn saved batch results into validated labels, with a cost report."""

import json
from collections import Counter
from dataclasses import dataclass
from typing import Literal

from pydantic import BaseModel, ValidationError

BATCH_PRICES = {"claude-sonnet-5-5": {"input": 1.00, "output": 5.00}}


class ReviewLabel(BaseModel):
    label: Literal["positive", "negative", "mixed"]
    confidence: float


@dataclass
class Outcome:
    custom_id: str
    status: str
    value: dict | None
    note: str
    input_tokens: int = 0
    output_tokens: int = 0


def text_of(message: dict) -> str:
    return "".join(b.get("text", "") for b in message["content"] if b.get("type") == "text").strip()


def parse_label(text: str):
    try:
        return ReviewLabel.model_validate_json(text), None
    except (ValidationError, ValueError) as error:
        return None, str(error).splitlines()[0]


def process_record(record: dict) -> Outcome:
    custom_id, result = record["custom_id"], record["result"]
    if result["type"] == "succeeded":
        message = result["message"]
        usage = message.get("usage", {})
        tokens = dict(input_tokens=usage.get("input_tokens", 0), output_tokens=usage.get("output_tokens", 0))
        if message["stop_reason"] != "end_turn":
            return Outcome(custom_id, "needs_review", None, f"stop_reason={message['stop_reason']}", **tokens)
        label, problem = parse_label(text_of(message))
        if label is None:
            return Outcome(custom_id, "needs_review", None, problem, **tokens)
        return Outcome(custom_id, "ok", label.model_dump(), "", **tokens)
    if result["type"] == "errored":
        kind = result.get("error", {}).get("error", {}).get("type")
        status = "fix_request" if kind == "invalid_request_error" else "retry"
        return Outcome(custom_id, status, None, kind or "unknown error")
    return Outcome(custom_id, "not_run", None, result["type"])


def cost_of(model: str, input_tokens: int, output_tokens: int) -> float:
    prices = BATCH_PRICES[model]
    return (input_tokens * prices["input"] + output_tokens * prices["output"]) / 1_000_000


DEMO_FILE = [
    {"custom_id": "review-8841", "result": {"type": "succeeded", "message": {
        "model": "claude-sonnet-5-5", "stop_reason": "end_turn",
        "content": [{"type": "text", "text": '{"label": "positive", "confidence": 0.94}'}],
        "usage": {"input_tokens": 41, "output_tokens": 12}}}},
    {"custom_id": "review-8842", "result": {"type": "succeeded", "message": {
        "model": "claude-sonnet-5-5", "stop_reason": "max_tokens",
        "content": [{"type": "text", "text": '{"label": "neg'}],
        "usage": {"input_tokens": 52, "output_tokens": 200}}}},
    {"custom_id": "review-8843", "result": {"type": "errored", "error": {"type": "error",
        "error": {"type": "invalid_request_error", "message": "max_tokens is too large"}}}},
    {"custom_id": "review-8844", "result": {"type": "canceled"}},
]

if __name__ == "__main__":
    outcomes = [process_record(record) for record in DEMO_FILE]
    for outcome in outcomes:
        print(f"{outcome.custom_id}: {outcome.status:<12} {outcome.value or outcome.note}")

    print("counts:", dict(Counter(o.status for o in outcomes)))
    total = sum(cost_of("claude-sonnet-5-5", o.input_tokens, o.output_tokens) for o in outcomes)
    print(f"batch cost: ${total:.6f}")

Output:

review-8841: ok           {'label': 'positive', 'confidence': 0.94}
review-8842: needs_review stop_reason=max_tokens
review-8843: fix_request  invalid_request_error
review-8844: not_run      canceled
counts: {'ok': 1, 'needs_review': 1, 'fix_request': 1, 'not_run': 1}
batch cost: $0.001153

How It Works

1. Each record goes through process_record() once, and the branch taken depends on the result type first and the stop reason second.
2. The truncated answer is caught by its stop reason, even though its result type is succeeded. Its partial JSON would fail validation anyway, but checking the stop reason first gives a clearer note.
3. The invalid request is classified by error type, so the pipeline knows to fix the request, not retry it unchanged.
4. The canceled request is not_run, which keeps it out of cost totals, because it has no usage.
5. The cost is the sum over outcomes with tokens. The first record costs (41 × $1 + 12 × $5) per million, which is $0.000101. The truncated record costs (52 × $1 + 200 × $5) per million, which is $0.001052. Together they make $0.001153. The errored and canceled records have no usage and add nothing.

Note: the truncated answer is billed for its 200 output tokens, even though its result is marked for review. Batch cost reports should include every succeeded result, not only the accepted ones.

Storing the Outcomes

The outcome objects are the input to storage. A single table with one row per request, keyed by custom_id, holds the status, the validated value, the note, and the tokens. Re-processing the same file then becomes an upsert rather than a duplicate insert, which is the same idempotency principle used throughout the data pipelines in Lesson 109. The status column lets a later job select exactly the rows that need attention, such as every fix_request for correction, or every needs_review for a person.