Batch processing¶

This chapter shows how to use OpenAI's batch API to process large amounts of data at half the price. The idea is simple: put all your prompts in a file, upload the file to OpenAI's server, and wait up to 24 hours for the responses. In my tests the results come back much faster than that, but this might change.

The workflow has more steps than a normal API call, so this chapter is a notebook that walks through them one by one: build the task file, upload it, create the batch job, wait for it to finish, download and read the results.

Setup¶

Install the dependencies and configure your API key.

On Google Colab, store your key in Colab Secrets first: click the key icon in the left sidebar, add a secret named OPENAI_API_KEY, and enable notebook access. Running locally, put the key in a .env file or in the OPENAI_API_KEY environment variable (see handling API keys).

In [1]:

%pip install -q openai python-dotenv

%pip install -q openai python-dotenv

/Users/yangkc/working/agent_space/git_repos/standalone/llm_for_css/.venv/bin/python3: No module named pip

Note: you may need to restart the kernel to use updated packages.

In [2]:

import os

from dotenv import load_dotenv

load_dotenv()  # loads a .env file from the working directory, if there is one

if "OPENAI_API_KEY" not in os.environ:
    try:
        # On Google Colab, read the key from Colab Secrets
        from google.colab import userdata

        os.environ["OPENAI_API_KEY"] = userdata.get("OPENAI_API_KEY")
    except ImportError:
        raise RuntimeError("Set OPENAI_API_KEY in your environment or in a .env file")

import os from dotenv import load_dotenv load_dotenv() # loads a .env file from the working directory, if there is one if "OPENAI_API_KEY" not in os.environ: try: # On Google Colab, read the key from Colab Secrets from google.colab import userdata os.environ["OPENAI_API_KEY"] = userdata.get("OPENAI_API_KEY") except ImportError: raise RuntimeError("Set OPENAI_API_KEY in your environment or in a .env file")

Example: sentiment analysis in a batch¶

We use the same sentiment analysis example as the earlier chapters, with five text messages instead of one. First, import the packages and initialize the client:

In [3]:

import json
import time

from openai import OpenAI

import json import time from openai import OpenAI

In [4]:

client = OpenAI()

client = OpenAI()

Here is the list of text messages we want to process:

In [5]:

text_messages = [
    "The service here is very good!",
    "The service here is good.",
    "The service here is ok.",
    "The service here is not very good.",
    "The service here is terrible!",
]

text_messages = [ "The service here is very good!", "The service here is good.", "The service here is ok.", "The service here is not very good.", "The service here is terrible!", ]

Now let's create the batch file. We reuse the prompts from the earlier chapters.

In [6]:

system_prompt = "You are an expert on sentiment analysis. Your job is to evaluate the sentiment of the given text message."

user_instruction = """
    Given the following text message: '{text_message}', please evaluate its sentiment by giving a score in the range of -1 to 1, where -1 means negative and 1 means positive.
    Also explain why.
    """

system_prompt = "You are an expert on sentiment analysis. Your job is to evaluate the sentiment of the given text message." user_instruction = """ Given the following text message: '{text_message}', please evaluate its sentiment by giving a score in the range of -1 to 1, where -1 means negative and 1 means positive. Also explain why. """

For the batch API, you have to write the JSON schema for the response by hand; there is no parse helper. This is the same schema Pydantic generates from the Sentiment class in the structured output chapter.

In [7]:

sentiment_json_schema = {
    "type": "object",
    "title": "Sentiment",
    "required": ["score", "explanation"],
    "properties": {
        "score": {
            "type": "number",
            "title": "Score",
            "minimum": -1,
            "maximum": 1,
            "description": "Sentiment score in the range of -1 to 1, where -1 means negative and 1 means positive.",
        },
        "explanation": {
            "type": "string",
            "title": "Explanation",
            "description": "Explanation of the sentiment score.",
        },
    },
    "additionalProperties": False,
}

sentiment_json_schema = { "type": "object", "title": "Sentiment", "required": ["score", "explanation"], "properties": { "score": { "type": "number", "title": "Score", "minimum": -1, "maximum": 1, "description": "Sentiment score in the range of -1 to 1, where -1 means negative and 1 means positive.", }, "explanation": { "type": "string", "title": "Explanation", "description": "Explanation of the sentiment score.", }, }, "additionalProperties": False, }

Each task is one API call: a unique ID, the endpoint, and the request body you would normally send to the Responses API.

In [8]:

tasks = []
for index, text_message in enumerate(text_messages):
    task = {
        # The API won't return the input text message, so we need a unique ID for each task
        # This way we can merge the results back with the input text message
        # Instead of generating the ID on the fly, it's recommended to assign a unique ID to each input message at the beginning
        "custom_id": f"text_message_{index}",
        "method": "POST",
        "url": "/v1/responses",
        "body": {
            # This is what you would have in your Chat Completions API call
            "model": "gpt-5.6-luna",
            "instructions": system_prompt,
            "input": user_instruction.format(text_message=text_message),
            "text": {
                "format": {
                    "type": "json_schema",
                    "name": "sentiment",
                    "strict": True,
                    "schema": sentiment_json_schema,
                }
            },
        },
    }

    tasks.append(task)

tasks = [] for index, text_message in enumerate(text_messages): task = { # The API won't return the input text message, so we need a unique ID for each task # This way we can merge the results back with the input text message # Instead of generating the ID on the fly, it's recommended to assign a unique ID to each input message at the beginning "custom_id": f"text_message_{index}", "method": "POST", "url": "/v1/responses", "body": { # This is what you would have in your Chat Completions API call "model": "gpt-5.6-luna", "instructions": system_prompt, "input": user_instruction.format(text_message=text_message), "text": { "format": { "type": "json_schema", "name": "sentiment", "strict": True, "schema": sentiment_json_schema, } }, }, } tasks.append(task)

Write the tasks to a file. Each line is one JSON object (the JSONL format).

In [9]:

task_file_name = "text_message_tasks.jsonl"
with open(task_file_name, "w") as f:
    for task in tasks:
        f.write(json.dumps(task) + "\n")

task_file_name = "text_message_tasks.jsonl" with open(task_file_name, "w") as f: for task in tasks: f.write(json.dumps(task) + "\n")

Now upload the task file to OpenAI.

In [10]:

batch_file = client.files.create(
    file=open(task_file_name, "rb"),
    purpose="batch"
)

batch_file = client.files.create( file=open(task_file_name, "rb"), purpose="batch" )

In [11]:

print(batch_file)

print(batch_file)

FileObject(id='file-CUazj6DJ56f1WXqH2DcHBZ', bytes=4896, created_at=1788320157, filename='text_message_tasks.jsonl', object='file', purpose='batch', status='processed', expires_at=1790912157, status_details=None)

Write the file ID down. It is used to create the batch job and to track it later.

After this step, you can also view the file at https://platform.openai.com/storage. Make sure the organization and project shown there match the ones your API key belongs to.

Now create the batch job:

In [12]:

batch_job = client.batches.create(
    input_file_id=batch_file.id,
    endpoint="/v1/responses",
    completion_window="24h"
)

batch_job = client.batches.create( input_file_id=batch_file.id, endpoint="/v1/responses", completion_window="24h" )

This step can also be done at https://platform.openai.com/batch using the website UI.

In [13]:

print(batch_job)

print(batch_job)

Batch(id='batch_6a97999e631c81908bf4edb00f993db8', completion_window='24h', created_at=1788320158, endpoint='/v1/responses', input_file_id='file-CUazj6DJ56f1WXqH2DcHBZ', object='batch', status='validating', cancelled_at=None, cancelling_at=None, completed_at=None, error_file_id=None, errors=None, expired_at=None, expires_at=1788406558, failed_at=None, finalizing_at=None, in_progress_at=None, metadata=None, model=None, output_file_id=None, request_counts=BatchRequestCounts(completed=0, failed=0, total=0), usage=BatchUsage(input_tokens=0, input_tokens_details=InputTokensDetails(cached_tokens=0), output_tokens=0, output_tokens_details=OutputTokensDetails(reasoning_tokens=0), total_tokens=0))

Waiting for the job¶

The job runs on OpenAI's side. Poll its status until it is no longer in progress. The loop below checks every 30 seconds; a five-message job usually finishes in a few minutes. If you close the notebook, you can come back later and retrieve the job by its ID, or check it on the batch page of the website.

In [14]:

while True:
    batch_object = client.batches.retrieve(batch_job.id)
    print(f"{time.strftime('%H:%M:%S')}  status: {batch_object.status}")
    if batch_object.status in ("completed", "failed", "expired", "cancelled"):
        break
    time.sleep(30)

while True: batch_object = client.batches.retrieve(batch_job.id) print(f"{time.strftime('%H:%M:%S')} status: {batch_object.status}") if batch_object.status in ("completed", "failed", "expired", "cancelled"): break time.sleep(30)

23:35:59  status: validating

23:36:29  status: validating

23:36:59  status: validating

23:37:29  status: in_progress

23:38:00  status: in_progress

23:38:30  status: in_progress

23:39:00  status: in_progress

23:39:30  status: completed

Downloading the results¶

After the job completes, download the output file.

In [15]:

result_file_id = batch_object.output_file_id
result = client.files.content(result_file_id).content

result_file_id = batch_object.output_file_id result = client.files.content(result_file_id).content

In [16]:

result_file_name = "openai_text_message_batch_output.jsonl"

with open(result_file_name, 'wb') as file:
    file.write(result)

result_file_name = "openai_text_message_batch_output.jsonl" with open(result_file_name, 'wb') as file: file.write(result)

Now read the results. Each line is one response, and custom_id tells you which input it belongs to.

In [17]:

results = []
with open("openai_text_message_batch_output.jsonl") as f:
    for line in f:
        results.append(json.loads(line))

results = [] with open("openai_text_message_batch_output.jsonl") as f: for line in f: results.append(json.loads(line))

In [18]:

for r in results:
    output_text = r["response"]["body"]["output"][0]["content"][0]["text"]
    print(r["custom_id"], json.loads(output_text))

for r in results: output_text = r["response"]["body"]["output"][0]["content"][0]["text"] print(r["custom_id"], json.loads(output_text))

text_message_0 {'score': 0.9, 'explanation': 'The message expresses strong positive sentiment, praising the service as “very good.”'}
text_message_1 {'score': 0.8, 'explanation': 'The message expresses a clearly positive opinion, describing the service as “good.”'}
text_message_2 {'score': 0.1, 'explanation': 'The message is mildly positive or neutral. “Ok” indicates acceptable service, but it conveys little enthusiasm or strong approval.'}
text_message_3 {'score': -0.7, 'explanation': 'The message expresses clear dissatisfaction with the quality of the service, making it moderately negative in sentiment.'}
text_message_4 {'score': -1, 'explanation': 'The message expresses very strong dissatisfaction with the service, using the word “terrible,” so its sentiment is extremely negative.'}

Creating the batch file with a script¶

OpenAI's website provides a UI for the whole batch workflow. So you could use a script to generate the task file, then use the website to upload it, create the job, check the status, and download the results. The script batch_processing/create_batch_file.py in the repository does the first step.

Further reading¶

OpenAI has a nice demonstration of the batch API, and you should check it out.