Use this file to discover all available pages before exploring further.
When you need to run the same operation across thousands or millions of records — classifying customer feedback, generating embeddings for a product catalog, summarizing a document archive — making synchronous API calls one at a time is slow and expensive. The OpenAI Batch API is designed for this pattern: you submit up to 50,000 requests in a single JSONL file, the API processes them asynchronously within a 24-hour window, and you retrieve the results when the batch completes. The cost is 50% less than the equivalent synchronous calls, and the per-request rate limits are much higher since the API can schedule work across off-peak capacity.
Each line in the file is a JSON object describing one request: a custom_id you define, the HTTP method and url, and a body identical to what you’d send to the Chat Completions endpoint. The file can contain up to 50,000 lines and must be under 200 MB.
2
Upload the file
Use client.files.create with purpose="batch" to upload your JSONL file. The API returns a file ID you pass to the next step.
3
Create the batch
Call client.batches.create with the file ID, the target endpoint, and a completion window. Currently the only supported window is "24h".
4
Poll for completion
Check the batch status periodically using client.batches.retrieve. The status moves through validating → in_progress → completed (or failed). In practice, batches often complete well under the 24-hour limit.
5
Retrieve results
When the status is completed, download the output file using client.files.content. Each line is a JSON object containing the custom_id from your input and the full API response for that request.
The example below classifies a list of customer reviews by sentiment. Each review becomes one request in the batch.
from openai import OpenAIimport jsonclient = OpenAI()texts = [ "Absolutely love this product, works perfectly!", "Terrible experience, broke after one day.", "It's okay, nothing special but does the job.", "Best purchase I've made all year.", "Would not recommend to anyone.",]# Build the JSONL inputrequests = [ { "custom_id": f"request-{i}", "method": "POST", "url": "/v1/chat/completions", "body": { "model": "gpt-4o-mini", "messages": [{"role": "user", "content": f"Classify sentiment: {text}"}], "max_tokens": 10 } } for i, text in enumerate(texts)]# Write to JSONL filewith open("batch_input.jsonl", "w") as f: for req in requests: f.write(json.dumps(req) + "\n")# Upload and create batchwith open("batch_input.jsonl", "rb") as f: batch_file = client.files.create(file=f, purpose="batch")batch = client.batches.create( input_file_id=batch_file.id, endpoint="/v1/chat/completions", completion_window="24h")print(f"Batch ID: {batch.id}")print(f"Status: {batch.status}")
Use gpt-4o-mini for high-volume classification and extraction tasks. It is significantly cheaper than gpt-4o and handles most structured extraction and short-form generation tasks accurately.
After submitting a batch, poll the status until it transitions to completed or failed. The interval below starts at 60 seconds and doubles up to a cap — a simple exponential backoff that avoids hammering the API while still detecting completion quickly.
import timedef wait_for_batch(client: OpenAI, batch_id: str, initial_wait: int = 60) -> dict: wait = initial_wait while True: batch = client.batches.retrieve(batch_id) print(f"Status: {batch.status} | " f"Completed: {batch.request_counts.completed} | " f"Failed: {batch.request_counts.failed}") if batch.status in ("completed", "failed", "cancelled", "expired"): return batch time.sleep(min(wait, 600)) # cap at 10 minutes wait = int(wait * 1.5)batch = wait_for_batch(client, batch.id)
The output file is a JSONL file where each line corresponds to one input request, keyed by your custom_id. Errors for individual requests appear in the error field rather than causing the whole batch to fail.
def retrieve_results(client: OpenAI, batch) -> list[dict]: if batch.status != "completed": raise RuntimeError(f"Batch did not complete: {batch.status}") content = client.files.content(batch.output_file_id) results = [] for line in content.text.splitlines(): if line.strip(): results.append(json.loads(line)) return resultsresults = retrieve_results(client, batch)for result in results: custom_id = result["custom_id"] response = result.get("response", {}) error = result.get("error") if error: print(f"{custom_id}: ERROR — {error['message']}") else: body = response["body"] content = body["choices"][0]["message"]["content"] print(f"{custom_id}: {content}")
Results are not guaranteed to arrive in input order — always use custom_id to match outputs back to inputs.
If a batch expires (not completed within 24 hours), the partial results are still available via batch.output_file_id for any requests that did complete before expiry. Check batch.request_counts.completed vs batch.request_counts.total to determine how many succeeded.
Every request in a batch costs half the price of the equivalent synchronous API call. For large workloads, this compounds quickly — a job that would cost 100synchronouslycosts50 via the Batch API.
Higher throughput
Batch jobs are not subject to the same per-minute token limits as synchronous calls. The API schedules them across available capacity, enabling much higher effective throughput for large datasets.
The trade-off is latency. Batches process within 24 hours but have no guaranteed completion time. The Batch API is the right choice when results are not needed in real time.
Classify thousands of customer reviews, support tickets, or social mentions by sentiment or category. Each record maps cleanly to one batch request.
Embedding generation
Generate embeddings for a large document corpus or product catalog to build a search index. The Batch API supports the /v1/embeddings endpoint alongside chat completions.
Document summarization
Summarize a large collection of articles, research papers, or support threads overnight. Retrieve a complete set of summaries the next morning.
Data extraction
Extract structured fields from thousands of invoices, contracts, or forms. Combine with Structured Outputs for schema-validated results at scale.
The snippet below puts all the steps together in a single reusable function:
from openai import OpenAIimport jsonimport timedef run_batch(texts: list[str], system_prompt: str, model: str = "gpt-4o-mini") -> dict[str, str]: """ Submit a batch of chat completion requests and return a mapping of custom_id -> response content. """ client = OpenAI() # Build input JSONL requests = [ { "custom_id": f"req-{i}", "method": "POST", "url": "/v1/chat/completions", "body": { "model": model, "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": text}, ], "max_tokens": 256, }, } for i, text in enumerate(texts) ] jsonl = "\n".join(json.dumps(r) for r in requests).encode() # Upload file batch_file = client.files.create( file=("batch_input.jsonl", jsonl, "application/jsonl"), purpose="batch", ) # Create batch batch = client.batches.create( input_file_id=batch_file.id, endpoint="/v1/chat/completions", completion_window="24h", ) print(f"Submitted batch {batch.id}") # Poll until done while batch.status not in ("completed", "failed", "cancelled", "expired"): time.sleep(60) batch = client.batches.retrieve(batch.id) print(f" {batch.status} — {batch.request_counts.completed}/{batch.request_counts.total}") if batch.status != "completed": raise RuntimeError(f"Batch ended with status: {batch.status}") # Parse results output = client.files.content(batch.output_file_id) results = {} for line in output.text.splitlines(): if not line.strip(): continue item = json.loads(line) cid = item["custom_id"] if item.get("error"): results[cid] = f"ERROR: {item['error']['message']}" else: results[cid] = item["response"]["body"]["choices"][0]["message"]["content"] return results
# Usageresults = run_batch( texts=["I love this!", "Terrible product.", "It's alright."], system_prompt="Classify the sentiment as POSITIVE, NEGATIVE, or NEUTRAL. Reply with one word only.",)for key, value in results.items(): print(f"{key}: {value}")
Store your batch.id persistently (e.g. in a database or file) immediately after creating the batch. If your process restarts, you can resume polling from the ID without re-submitting the work.
