Batch API: Massive Asynchronous Processing

Updated November 3, 2026 based on the Batch API guide and the Responses API response structure: reading results no longer assumes the first element of output is the message (a reasoning element may precede it), and the error code key, written without quotes, raised a NameError. Source: Batch API.

Trade Latency for Cost

The Batch API operates on a simple market principle: you accept that your requests will be processed asynchronously within a window of up to twenty-four hours, and in exchange, you pay fifty percent less. Anything that doesn’t need to serve a human waiting at their screen falls into this category. At scale, this isn’t a micro-optimization; it cuts your budget line in half.

The habit to build within a team is to ask, for every new processing task, whether someone is waiting for the result. Analyzing a stock of contracts, generating product sheets for a catalog, running a benchmark on thousands of test cases, classifying a six-month backlog of tickets, extracting fields from a batch of invoices: in all these cases, the answer lands in a dashboard or a database, not in a conversational interface. The delay is invisible, but the savings are not.

Build and Launch a Batch

The input format is a JSONL file: one line per request, where each line is a complete, independent object. The custom_id field is what allows you to link each response back to its original request, since nothing guarantees the order of the output file. Choose an ID that is stable and meaningful—a document ID from your database rather than an iteration number—otherwise the reconciliation phase will become a nightmare.

import json

requests = [
    {
        "custom_id": f"doc-{i}",
        "method": "POST",
        "url": "/v1/responses",
        "body": {
            "model": "gpt-5.6-luna",
            "input": f"Summarize this document in 3 key points: {doc}",
            "max_output_tokens": 500,
        },
    }
    for i, doc in enumerate(documents_list)
]

# Write the JSONL file
with open("batch_input.jsonl", "w") as f:
    for request in requests:
        f.write(json.dumps(request) + "\n")

The file is then uploaded and referenced when creating the batch. Metadata serves no technical purpose, but it will save you the day three batches are running in parallel and you need to find out which one corresponds to which prompt version.

import openai

client = openai.OpenAI()

# Upload the file
file_obj = client.files.create(
    file=open("batch_input.jsonl", "rb"),
    purpose="batch",
)

# Create the batch
batch = client.batches.create(
    input_file_id=file_obj.id,
    endpoint="/v1/responses",
    completion_window="24h",
    metadata={"project": "contract-analysis", "version": "2.1"},
)

print(f"Batch created: {batch.id}")
print(f"Status: {batch.status}")

Monitoring involves polling the batch until it leaves the in-progress state, then downloading the output file, which is also in JSONL format. The loop below distinguishes three terminal outcomes—failed, expired, cancelled—and raises an exception for each: an expired batch is one where the twenty-four-hour window elapsed before everything was processed, and treating it as a silent partial success would be the worst option.

import time

def wait_for_batch(batch_id: str, interval: int = 60) -> list:
    """Waits for a batch to finish and returns the results."""
    while True:
        batch = client.batches.retrieve(batch_id)
        print(
            f"Status: {batch.status} | "
            f"Completed: {batch.request_counts.completed}/"
            f"{batch.request_counts.total} | "
            f"Failed: {batch.request_counts.failed}"
        )

        if batch.status == "completed":
            break
        elif batch.status in ("failed", "expired", "cancelled"):
            raise RuntimeError(f"Batch failed: {batch.status}")

        time.sleep(interval)

    # Download results
    content = client.files.content(batch.output_file_id)
    results = []
    for line in content.text.strip().split("\n"):
        results.append(json.loads(line))

    return results


results = wait_for_batch(batch.id)

# Process results
for result in results:
    custom_id = result["custom_id"]
    response = result["response"]
    if response["status_code"] == 200:
        # output may contain other elements before the message
        # (reasoning, for example): look for the message text
        text = next(
            part["text"]
            for item in response["body"]["output"] if item["type"] == "message"
            for part in item["content"] if part["type"] == "output_text"
        )
        print(f"{custom_id}: {text[:100]}...")
    else:
        print(f"{custom_id}: ERROR {response['status_code']}")

Plan for Partial Failures

On a large batch, some requests will fail: a truncated document, a context overflow, or a transient server-side error. A batch is therefore never simply “successful” or “failed”—it is partially successful, and you must know how to isolate the missing part. The pattern to adopt is to immediately separate successes from failures and regenerate a JSONL file ready to be relaunched, rather than digging through logs three days later to reconstruct what is missing.

def handle_batch_errors(results: list[dict]) -> tuple[list, list]:
    """Separates successes from failures for reprocessing."""
    successes = []
    failures = []

    for r in results:
        if r["response"]["status_code"] == 200:
            successes.append(r)
        else:
            failures.append(r)

    if failures:
        print(f"Warning: {len(failures)} failed requests out of {len(results)}")
        # Rewrite a JSONL file with failures for retry
        with open("batch_retry.jsonl", "w") as f:
            for failure in failures:
                original_request = {
                    "custom_id": failure["custom_id"],
                    "method": "POST",
                    "url": "/v1/responses",
                    "body": failure["request"]["body"],
                }
                f.write(json.dumps(original_request) + "\n")

    return successes, failures

Compound Your Savings

On a processing run of ten thousand documents with GPT-5.6 Luna, the synchronous API represents the baseline cost, the Batch API represents half of that, and prompt caching adds on top since cached tokens also benefit from their own reduction. The two mechanisms do not compete: if your ten thousand requests share the same long system prompt, structure it as a stable prefix and you will compound both effects.

Finally, there is the question of splitting. A batch should not be infinite, and it is more comfortable to monitor several manageable batches than a single monolith whose failure costs an entire day. The following function slices a list of requests into batches of bounded size.

def create_optimized_batches(
    requests: list[dict],
    max_size: int = 50_000,
) -> list[list[dict]]:
    """Splits into batches of optimal size."""
    batches = []
    current_batch = []

    for request in requests:
        current_batch.append(request)
        if len(current_batch) >= max_size:
            batches.append(current_batch)
            current_batch = []

    if current_batch:
        batches.append(current_batch)

    return batches

Key Takeaways

• The Batch API offers a 50% reduction for non-real-time processing
• Prepare your requests in a JSONL file with a unique custom_id per request
• Plan a retry mechanism for failed requests
• Combine with prompt caching to maximize savings