A Practical Guide to the OpenAI Batch API with Python and openbatch

TL;DR The OpenAI Batch API offers a 50% cost reduction and significantly higher throughput for large-scale tasks, but its file-based workflow is cumbersome, especially for structured outputs. The openbatch Python library eliminates this friction by providing a convenient, drop-in replacement for the standard OpenAI client. It simplifies batch file creation, streamlines structured data handling using Pydantic, and offers powerful templating, making the cost and speed benefits of batch processing easily accessible without sacrificing developer convenience.

For researchers and developers working with large datasets, the OpenAI Batch API offers significant advantages in cost and speed. However, its asynchronous, file-based workflow can feel cumbersome compared to the simplicity of direct API calls. This guide explores the trade-offs and introduces openbatch, a Python package designed to make the Batch API as convenient to use as the standard sequential API.

Batch Processing is Cheaper and often Faster

The primary motivation for using the Batch API is efficiency. It provides two key benefits:

1. 50% Cost Reduction: Batch API calls are priced at half the cost of the standard API. For large-scale data analysis, classification, or generation tasks, this immediately doubles your budget’s effectiveness.
2. High Throughput: While batch jobs have a 24-hour completion window, they often finish much faster than an equivalent number of sequential calls. In one experiment, a task that would have taken over 10 hours of sequential API calls was completed in under 1 hour using a single batch job.

The trade-off for this (cost-)efficiency has traditionally been convenience. Instead of a simple request-response cycle, the batch workflow involves manually preparing a JSONL file, uploading it, starting the job, and then retrieving the results from a separate file. This is especially cumbersome and challenging when you need to generate prompts dynamically or make use of structured outputs.

Convenient Batch Processing with openbatch

openbatch is a lightweight Python library that simplifies the creation of the batch input file by providing a developer experience that mirrors the official openai Python client. It’s designed to be a near drop-in replacement that can integrate into existing workflows, allowing you to switch between sequential and batch processing with minimal code changes.

The library’s core features directly address the common pain points of the batch workflow:

• Familiar API: The BatchCollector class mimics the structure of openai.OpenAI, so you can write collector.responses.create(...) instead of client.responses.create(...).
• Structured Outputs with Pydantic: Reliably getting JSON output from LLMs can be tricky. openbatch allows you to pass a Pydantic model directly to a parse() method (e.g., collector.responses.parse(text_format=MyModel)), which automatically handles the complex JSON schema generation needed to enforce the output structure.
• Powerful Templating: With the BatchJobManager, you can define a PromptTemplate with placeholders and programmatically generate thousands or millions of requests from a list of data, which is ideal for large-scale, repetitive tasks.
• Full API Coverage: It supports all endpoints available in the Batch API: /v1/responses, /v1/chat/completions, and /v1/embeddings.

Have a look at the full documentation at https://daniel-gomm.github.io/openbatch for more details and how-to guides.

Installation

You can install openbatch via pip:

pip install openbatch

A Practical Workflow: Sentiment Analysis at Scale

To understand the value of openbatch, let’s walk through a common, real-world task: performing sentiment analysis on a large dataset of customer reviews. Imagine you have a file, customer_reviews.csv, containing thousands of reviews that you need to classify.

The goal is to get a structured output for each review, classifying it as ‘Positive’, ‘Neutral’, or ‘Negative’ with a confidence score. We’ll start with a standard sequential approach and then see how openbatch dramatically simplifies the process for batching.

First, let’s define the Pydantic model that will enforce our desired output structure:

from pydantic import BaseModel, Field
from typing import Literal

class SentimentAnalysisModel(BaseModel):
    sentiment: Literal["Positive", "Neutral", "Negative"]
    confidence: float = Field(
        ge=0.0, 
        le=1.0, 
        description="Confidence score for the sentiment classification."
    )

Now, let’s assume our customer_reviews.csv looks like this:

Approach 1: The Standard Sequential Loop

A typical approach would be to loop through the dataset and make an API call for each review. This is easy to write but highly inefficient for large datasets.

import csv
from openai import OpenAI

client = OpenAI()
results = []
instructions = "You are an expert annotator. Judge the sentiment of the user-provided comment. Use the categories 'Positive', 'Neutral', and 'Negative'."

# This loop is slow and expensive at scale
print("Starting sequential processing...")
with open("customer_reviews.csv", "r") as f:
    reader = csv.DictReader(f)
    for row in reader:
        response = client.responses.parse(
            model="gpt-4o-mini",
            instructions=instructions,
            input=[{"role": "user", "content": row["review_text"]}],
            text_format=SentimentAnalysisModel
        )
        results.append({"review_id": row["review_id"], "analysis": response})
        print(f"Processed review {row['review_id']}...")

print("✅ Sequential processing complete.")

This method is simple but has major drawbacks: it’s slow, processing one review at a time, and expensive, as it uses standard API pricing.

Approach 2: The BatchCollector Drop-in Replacement

Here’s where openbatch comes in. You can switch to batch processing with minimal changes. The BatchCollector API is designed to be a drop-in replacement for the OpenAI client within your existing loop.

import csv
from openbatch import BatchCollector

batch_file_path = "sentiment_batch.jsonl"
collector = BatchCollector(batch_file_path=batch_file_path) # Change 1: Instantiate collector
instructions = "You are an expert annotator. Judge the sentiment of the user-provided comment. Use the categories 'Positive', 'Neutral', and 'Negative'."

# The same loop, now preparing a batch file instead of making live calls
print("Preparing batch file with BatchCollector...")
with open("customer_reviews.csv", "r") as f:
    reader = csv.DictReader(f)
    for row in reader:
        # Change 2: Call the collector instead of the client
        collector.responses.parse(
            custom_id=row["review_id"], # Add a custom ID for tracking
            model="gpt-4o-mini",
            instructions=instructions,
            input=[{"role": "user", "content": row["review_text"]}],
            text_format=SentimentAnalysisModel
        )

print(f"✅ Batch file '{batch_file_path}' prepared.")

With just two lines changed, your code now prepares a cost-effective batch job instead of making slow, expensive API calls. You’ve decoupled file preparation from execution.

Approach 3: The Simplified BatchJobManager

For templated, large-scale tasks like this, the BatchJobManager is even more efficient. It eliminates the need for an explicit Python loop entirely, letting you define the task and the data separately.

import csv
from openbatch import BatchJobManager, PromptTemplate, Message, ResponsesRequest, PromptTemplateInputInstance

# 1. Define the prompt template
template = PromptTemplate(messages=[
    Message(role="system", content="You are an expert annotator. Judge the sentiment of the user-provided comment. Use the categories 'Positive', 'Neutral', and 'Negative'."),
    Message(role="user", content="{review_text}")
])

# 2. Define the common request configuration
common_config = ResponsesRequest(model="gpt-4o-mini")
# Enforce the structured output for all requests in this batch
common_config.set_output_structure(SentimentAnalysisModel)

# 3. Load the data and create input instances (no loop needed for API calls)
with open("customer_reviews.csv", "r") as f:
    reader = csv.DictReader(f)
    input_instances = [
        PromptTemplateInputInstance(
            id=row["review_id"],
            prompt_value_mapping={"review_text": row["review_text"]}
        ) for row in reader
    ]

# 4. Generate the entire batch file in one go
batch_file_path = "sentiment_batch_optimized.jsonl"
manager = BatchJobManager()
manager.add_templated_instances(
    prompt=template,
    common_request=common_config,
    input_instances=input_instances,
    save_file_path=batch_file_path
)

print(f"Batch file '{batch_file_path}' generated with BatchJobManager.")

This approach is the cleanest and most declarative, perfectly suited for large-scale, programmatic batch job creation.

Creating the Batch Job

Once the batch-job-file is created, the batch-job can be started. This is possible through the Batches Web UI or programatically via the standard openai client:

from openai import OpenAI

client = OpenAI()

# Upload the file to OpenAI
print(f"Uploading '{batch_file_path}'...")
batch_input_file = client.files.create(
    file=open(batch_file_path, "rb"),
    purpose="batch"
)
print(f"File uploaded. File ID: {batch_input_file.id}")

# Create the batch job, specifying the correct endpoint
print("Creating batch job...")
batch_job = client.batches.create(
    input_file_id=batch_input_file.id,
    endpoint="/v1/responses",  # Must match the API used in your requests
    completion_window="24h"
)
print(f"Batch job created. Job ID: {batch_job.id}, Status: {batch_job.status}")

Monitor and Download the Results

After some time, you can check the job’s status. Once it’s complete, you can download the output file containing the responses. This is again possible through the Batches Web UI or programatically:

# In a real application, you could poll this endpoint periodically
completed_job = client.batches.retrieve(batch_job.id)
print(f"Current job status: {completed_job.status}")

# Once the status is 'completed'
if completed_job.status == 'completed':
    output_file_id = completed_job.output_file_id
    print(f"Job finished! Output file ID: {output_file_id}")
    
    # Download the results file
    results_content = client.files.content(file_id=output_file_id)
    with open("batch_results.jsonl", "wb") as f:
        f.write(results_content.read())
    print("✅ Results saved to 'batch_results.jsonl'.")
elif completed_job.status in ['failed', 'expired', 'cancelled']:
    print(f"❌ Job did not complete. Status: {completed_job.status}")

The downloaded batch_results.jsonl file will contain the output for each of the input requests, which can then be parsed for the application. Thanks to the structured output feature, the JSON responses for the requests will be clean, valid, and ready to be loaded directly into the Pydantic models.

Conclusion

The OpenAI Batch API is a powerful tool for processing large datasets efficiently and cost-effectively. While its asynchronous nature can introduce complexity, libraries like openbatch abstract away the tedious work of file preparation. By providing a familiar, Pydantic-powered interface, it makes the benefits of batch processing accessible without sacrificing developer convenience, allowing researchers and engineers to focus on their results, not their boilerplate.

For more detail on openbatch, including installation instructions and comprehensive documentation, visit the official documentation site.