python openai batch api: Practical Usage and Code Examples

The OpenAI Batch API lets you send large numbers of requests asynchronously, which is useful when you need to process thousands of prompts without hitting synchronous rate limits. This article shows you how to use the python openai batch api workflow to upload a JSONL file, create a batch job, poll for completion, and download the results.

What the OpenAI Batch API Provides

The OpenAI Batch API is designed for workloads where you can tolerate a delay between submitting a request and receiving a result. Instead of sending each prompt synchronously and waiting for a response, you upload a file containing many requests, submit a batch job, and then poll for completion. For Python developers, this means you can process large datasets, run evaluations, or generate completions in bulk without managing hundreds of concurrent connections.

The primary benefit is that batch jobs run at a lower priority and are typically charged at a reduced rate compared to synchronous API calls. They also have higher rate limits, so you can submit a much larger volume of work without hitting per-minute or per-token caps. This makes the batch API a good fit for offline processing, nightly jobs, and any scenario where results are not needed immediately.

Prerequisites and Setup

To use the OpenAI Batch API from Python, you need the openai Python package installed. The package is available on PyPI and can be installed with:

pip install openai

You also need an OpenAI API key. Set it as an environment variable or pass it directly to the client. The recommended approach is to use an environment variable to avoid hard-coding credentials in source files.

import os
from openai import OpenAI

client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

The OpenAI client object exposes a batches resource that handles batch job creation, retrieval, and cancellation. The exact method names and parameters may vary slightly across SDK versions, so check the SDK documentation for the version you are using.

Preparing the Input File

The batch API expects an input file in JSONL format. Each line in the file represents one request, and each request must contain the same fields you would use in a synchronous API call. For a chat completion, a line looks like this:

{"custom_id": "request-1", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "gpt-4o-mini", "messages": [{"role": "user", "content": "Summarize this article."}], "max_tokens": 100}}

The custom_id is your own identifier for the request. It is echoed back in the output file, which lets you match results to the original input. The method and url fields specify the API endpoint, and body contains the parameters for that endpoint.

You can generate this file in Python by iterating over your input data and writing each request as a JSON object followed by a newline.

import json

requests = [
    {"custom_id": f"req-{i}", "method": "POST", "url": "/v1/chat/completions",
     "body": {"model": "gpt-4o-mini", "messages": [{"role": "user", "content": prompt}], "max_tokens": 50}}
    for i, prompt in enumerate(prompts)
]

with open("batch_input.jsonl", "w") as f:
    for req in requests:
        f.write(json.dumps(req) + "\n")

The file must be uploaded to OpenAI before you can create a batch job. The client.files.create method uploads the file and returns a file ID.

Creating and Submitting a Batch Job

Once the input file is uploaded, you create a batch job by calling client.batches.create. You pass the file ID, an endpoint specification, and a completion window. The completion window is the maximum time you are willing to wait for the batch to finish; typical values are 24 hours or 7 days.

file_response = client.files.create(
    file=open("batch_input.jsonl", "rb"),
    purpose="batch"
)

batch = client.batches.create(
    input_file_id=file_response.id,
    endpoint="/v1/chat/completions",
    completion_window="24h"
)

The endpoint must match the URL used in the JSONL file. For chat completions, it is /v1/chat/completions. For embeddings, it would be /v1/embeddings. The completion_window is a string like "24h" or "7d".

The create call returns a batch object with an id field. You will use this ID to poll for status and retrieve results.

Polling for Completion and Retrieving Results

Batch jobs run asynchronously. You need to poll the status until it reaches a terminal state. The status can be validating, in_progress, completed, failed, or cancelled. A common pattern is to poll at regular intervals and break when the status is completed or failed.

import time

batch_id = batch.id
while True:
    current = client.batches.retrieve(batch_id)
    if current.status in ("completed", "failed", "cancelled"):
        break
    time.sleep(30)  # wait before checking again

When the batch is completed, the output_file_id field contains the ID of a file with the results. Download that file using client.files.content and parse the JSONL lines.

if current.status == "completed":
    result = client.files.content(current.output_file_id)
    for line in result.text.strip().split("\n"):
        data = json.loads(line)
        # data["custom_id"] matches your input, data["response"] contains the API response

Each line in the output file includes the custom_id you supplied, the response body, and other metadata. This lets you correlate results back to your original requests.

Handling Errors and Partial Failures

A batch job can fail for several reasons. The entire job may fail during validation if the input file is malformed, or individual requests may fail while the rest succeed. The API returns a per-request error field in the output for failed requests. You should inspect the output file for lines that contain an error object and handle them appropriately.

for line in result.text.strip().split("\n"):
    data = json.loads(line)
    if "error" in data:
        # log the error and the custom_id
        print(f"Request {data['custom_id']} failed: {data['error']}")
    else:
        # process the successful response
        process(data["response"])

Common errors include invalid model names, content that violates safety policies, or malformed request bodies. Because the batch API processes requests independently, a single bad request does not cancel the entire batch. You can identify and retry failed requests by constructing a new input file with only those custom IDs.

Rate Limits and Cost Considerations

The batch API is designed to handle large volumes of work without exhausting synchronous rate limits. It uses a separate queue and runs at a lower priority, which is why it can accept more requests and charge less. For Python applications that need to process thousands of prompts, this reduces the need for complex concurrency control or retry logic.

However, batch jobs are not real-time. You must wait for the completion window, and the actual duration depends on the current load on OpenAI's servers. If your application requires immediate responses, the synchronous API is more appropriate. If you can wait minutes or hours, the batch API is often the better choice.

The cost per token is typically lower for batch requests than for synchronous requests. The exact pricing is set by OpenAI and can change, so check the official pricing page for current rates. The higher rate limits also mean you can submit a large file in one go rather than splitting it into many small requests.

When to Use Batch vs. Synchronous Requests

Choosing between the batch API and the synchronous API depends on your latency requirements and workload size. Use the batch API when:

You have a large volume of requests (hundreds or thousands) that do not need immediate answers.
You want to reduce API costs and stay within rate limits without complex throttling.
You are running offline jobs, data pipelines, or evaluations that can run in the background.

Use the synchronous API when:

You need a response in seconds or minutes.
You are handling interactive requests from users.
You have a small number of requests where the overhead of file upload and polling is not worth it.

In many applications, a hybrid approach works well: use synchronous calls for user-facing features and batch calls for background processing. The Python SDK makes both paths straightforward, so you can switch based on the context of each request.