Batch Processing

Process large volumes of requests with 50% cost reduction using the Batch API.

Overview

The Batch API allows you to submit groups of requests for asynchronous processing with a 24-hour completion window, reducing costs by 50%.

┌─────────────────────────────────────────────────────────────────────┐
│                    BATCH API WORKFLOW                                │
├─────────────────────────────────────────────────────────────────────┤
│                                                                      │
│  1. PREPARE           2. SUBMIT           3. WAIT             4. GET│
│  ┌─────────┐         ┌─────────┐         ┌─────────┐        ┌─────┐│
│  │ Create  │────────►│ Upload  │────────►│ Batch   │───────►│Get  ││
│  │ JSONL   │         │ File +  │         │ Runs    │        │Rslt ││
│  │ File    │         │ Submit  │         │ (≤24h)  │        │     ││
│  └─────────┘         └─────────┘         └─────────┘        └─────┘│
│                                                                      │
│  BENEFITS:                                                          │
│  ├─► 50% cost reduction                                             │
│  ├─► Higher rate limits                                             │
│  ├─► Automatic retries                                              │
│  └─► No infrastructure to manage                                    │
│                                                                      │
│  BEST FOR:                                                          │
│  ├─► Data processing pipelines                                      │
│  ├─► Content generation at scale                                    │
│  ├─► Evaluation datasets                                            │
│  └─► Any non-time-sensitive workload                                │
│                                                                      │
└─────────────────────────────────────────────────────────────────────┘

Basic Usage

Step 1: Prepare Requests File

import json

def create_batch_file(requests: list[dict], filename: str = "batch_input.jsonl"):
    """Create JSONL file for batch processing."""

    with open(filename, "w") as f:
        for i, req in enumerate(requests):
            batch_request = {
                "custom_id": req.get("id", f"request-{i}"),
                "method": "POST",
                "url": "/v1/chat/completions",
                "body": {
                    "model": req.get("model", "gpt-4o-mini"),
                    "messages": req["messages"],
                    "max_tokens": req.get("max_tokens", 1000)
                }
            }
            f.write(json.dumps(batch_request) + "\n")

    return filename

# Example requests
requests = [
    {
        "id": "movie-1",
        "messages": [
            {"role": "system", "content": "Categorize this movie into genres."},
            {"role": "user", "content": "The Matrix - A hacker discovers reality is a simulation."}
        ]
    },
    {
        "id": "movie-2",
        "messages": [
            {"role": "system", "content": "Categorize this movie into genres."},
            {"role": "user", "content": "Titanic - A love story aboard the doomed ship."}
        ]
    },
    # ... more requests
]

batch_file = create_batch_file(requests)

Step 2: Upload and Submit

from openai import OpenAI

client = OpenAI()

def submit_batch(filename: str) -> str:
    """Upload file and submit batch job."""

    # Upload the file
    with open(filename, "rb") as f:
        file = client.files.create(file=f, purpose="batch")

    print(f"Uploaded file: {file.id}")

    # Create the batch
    batch = client.batches.create(
        input_file_id=file.id,
        endpoint="/v1/chat/completions",
        completion_window="24h",
        metadata={"description": "Movie categorization batch"}
    )

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

    return batch.id

batch_id = submit_batch("batch_input.jsonl")

Step 3: Monitor Progress

import time

def wait_for_batch(batch_id: str, poll_interval: int = 60) -> dict:
    """Poll batch status until completion."""

    while True:
        batch = client.batches.retrieve(batch_id)

        print(f"Status: {batch.status}")
        print(f"Progress: {batch.request_counts.completed}/{batch.request_counts.total}")

        if batch.status == "completed":
            return batch
        elif batch.status == "failed":
            raise Exception(f"Batch failed: {batch.errors}")
        elif batch.status in ["expired", "cancelled"]:
            raise Exception(f"Batch {batch.status}")

        time.sleep(poll_interval)

completed_batch = wait_for_batch(batch_id)

Step 4: Retrieve Results

def get_batch_results(batch: dict) -> list[dict]:
    """Download and parse batch results."""

    # Download output file
    output_file_id = batch.output_file_id
    content = client.files.content(output_file_id)

    results = []
    for line in content.text.split("\n"):
        if line.strip():
            result = json.loads(line)
            results.append({
                "custom_id": result["custom_id"],
                "response": result["response"]["body"]["choices"][0]["message"]["content"],
                "usage": result["response"]["body"]["usage"]
            })

    return results

results = get_batch_results(completed_batch)
for r in results:
    print(f"{r['custom_id']}: {r['response'][:100]}...")

Complete Example: Movie Categorization

import json
import time
from openai import OpenAI
from dataclasses import dataclass

client = OpenAI()

@dataclass
class Movie:
    id: str
    title: str
    description: str

def categorize_movies_batch(movies: list[Movie]) -> dict[str, str]:
    """Categorize movies using batch API."""

    # Step 1: Create batch file
    filename = "movies_batch.jsonl"
    with open(filename, "w") as f:
        for movie in movies:
            request = {
                "custom_id": movie.id,
                "method": "POST",
                "url": "/v1/chat/completions",
                "body": {
                    "model": "gpt-4o-mini",
                    "messages": [
                        {
                            "role": "system",
                            "content": "Categorize the movie into genres. Return only the genres as a comma-separated list."
                        },
                        {
                            "role": "user",
                            "content": f"{movie.title}: {movie.description}"
                        }
                    ],
                    "max_tokens": 100
                }
            }
            f.write(json.dumps(request) + "\n")

    # Step 2: Upload and submit
    with open(filename, "rb") as f:
        file = client.files.create(file=f, purpose="batch")

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

    print(f"Batch submitted: {batch.id}")

    # Step 3: Wait for completion
    while batch.status not in ["completed", "failed", "expired", "cancelled"]:
        time.sleep(30)
        batch = client.batches.retrieve(batch.id)
        completed = batch.request_counts.completed
        total = batch.request_counts.total
        print(f"Progress: {completed}/{total} ({100*completed/total:.1f}%)")

    if batch.status != "completed":
        raise Exception(f"Batch {batch.status}")

    # Step 4: Get results
    content = client.files.content(batch.output_file_id)

    results = {}
    for line in content.text.split("\n"):
        if line.strip():
            result = json.loads(line)
            movie_id = result["custom_id"]
            genres = result["response"]["body"]["choices"][0]["message"]["content"]
            results[movie_id] = genres

    return results

# Usage
movies = [
    Movie("m1", "The Matrix", "A hacker discovers reality is a simulation"),
    Movie("m2", "Titanic", "A love story aboard the doomed ship"),
    Movie("m3", "Toy Story", "Toys come to life when humans aren't around"),
]

categories = categorize_movies_batch(movies)
for movie_id, genres in categories.items():
    print(f"{movie_id}: {genres}")

Batch with Vision (Image Captioning)

import base64

def create_image_batch(images: list[dict]) -> str:
    """Create batch for image captioning."""

    filename = "images_batch.jsonl"

    with open(filename, "w") as f:
        for img in images:
            # Read and encode image
            with open(img["path"], "rb") as image_file:
                base64_image = base64.b64encode(image_file.read()).decode()

            request = {
                "custom_id": img["id"],
                "method": "POST",
                "url": "/v1/chat/completions",
                "body": {
                    "model": "gpt-4o-mini",
                    "messages": [
                        {
                            "role": "user",
                            "content": [
                                {
                                    "type": "text",
                                    "text": "Describe this image in one sentence."
                                },
                                {
                                    "type": "image_url",
                                    "image_url": {
                                        "url": f"data:image/jpeg;base64,{base64_image}"
                                    }
                                }
                            ]
                        }
                    ],
                    "max_tokens": 100
                }
            }
            f.write(json.dumps(request) + "\n")

    return filename

Batch with Embeddings

def create_embeddings_batch(texts: list[dict]) -> str:
    """Create batch for generating embeddings."""

    filename = "embeddings_batch.jsonl"

    with open(filename, "w") as f:
        for item in texts:
            request = {
                "custom_id": item["id"],
                "method": "POST",
                "url": "/v1/embeddings",
                "body": {
                    "model": "text-embedding-3-small",
                    "input": item["text"]
                }
            }
            f.write(json.dumps(request) + "\n")

    return filename

# Submit as usual, but use embeddings endpoint
batch = client.batches.create(
    input_file_id=file.id,
    endpoint="/v1/embeddings",
    completion_window="24h"
)

Error Handling

def process_batch_with_errors(batch_id: str) -> tuple[list, list]:
    """Process batch and separate successes from failures."""

    batch = client.batches.retrieve(batch_id)

    successes = []
    failures = []

    # Process output file (successes)
    if batch.output_file_id:
        content = client.files.content(batch.output_file_id)
        for line in content.text.split("\n"):
            if line.strip():
                result = json.loads(line)
                if result.get("error"):
                    failures.append(result)
                else:
                    successes.append(result)

    # Process error file (failures)
    if batch.error_file_id:
        error_content = client.files.content(batch.error_file_id)
        for line in error_content.text.split("\n"):
            if line.strip():
                failures.append(json.loads(line))

    return successes, failures

# Retry failed requests
def retry_failed(failures: list) -> str:
    """Create new batch file for failed requests."""

    if not failures:
        return None

    filename = "retry_batch.jsonl"
    with open(filename, "w") as f:
        for failure in failures:
            # Extract original request and resubmit
            original = failure.get("request", {})
            f.write(json.dumps(original) + "\n")

    return filename

Batch Management

List Batches

def list_recent_batches(limit: int = 10):
    """List recent batch jobs."""

    batches = client.batches.list(limit=limit)

    for batch in batches.data:
        print(f"ID: {batch.id}")
        print(f"  Status: {batch.status}")
        print(f"  Created: {batch.created_at}")
        print(f"  Progress: {batch.request_counts.completed}/{batch.request_counts.total}")
        print()

Cancel Batch

def cancel_batch(batch_id: str):
    """Cancel a running batch."""
    batch = client.batches.cancel(batch_id)
    print(f"Cancelled batch {batch_id}: {batch.status}")

Cost Comparison

Standard API vs Batch API (gpt-4o-mini):

Standard:
- Input:  $0.15 / 1M tokens
- Output: $0.60 / 1M tokens

Batch:
- Input:  $0.075 / 1M tokens (50% off)
- Output: $0.30 / 1M tokens  (50% off)

Example: 100,000 requests × 1,000 tokens each = 100M tokens

Standard: 100M × $0.15/1M = $15.00
Batch:    100M × $0.075/1M = $7.50

SAVINGS: $7.50 (50%)

Best Practices

When to Use Batch API

Scenario	Use Batch?	Reason
>1000 requests	✅ Yes	Significant cost savings
Results needed in <1h	❌ No	24h window too slow
Daily data processing	✅ Yes	Schedule overnight
Evaluation datasets	✅ Yes	Non-time-sensitive
Real-time chat	❌ No	Needs immediate response

Batch Size Guidelines

# Recommended batch sizes
BATCH_GUIDELINES = {
    "minimum": 100,        # Worth the overhead
    "optimal": 10000,      # Good balance
    "maximum": 50000,      # API limit per batch
}

# For very large jobs, split into multiple batches
def split_into_batches(requests: list, batch_size: int = 10000):
    for i in range(0, len(requests), batch_size):
        yield requests[i:i+batch_size]

See Also

• Prompt Caching
• Fine-tuning Techniques
• Parallel Agents