DEV Community

syncore
syncore

Posted on

Batch API Deep Dive: How to Process 1,000 Prompts at Half the Cost

#python #ai #claude #tutorial

4 min read · 748 words

If you are running large-scale LLM workloads—like content generation, data enrichment, offline evaluation, or customer feedback analysis—you are probably spending way too much money.

When your application doesn't require instantaneous, sub-second response times, sending thousands of synchronous API requests is a waste of cash and rate limits.

Enter the Batch API. Major providers like OpenAI and Anthropic offer dedicated batch endpoints that cut your API costs by 50% while giving you significantly higher rate limits. The catch? Results are processed asynchronously within a 24-hour window (though they often finish much faster).

In this deep dive, we’ll walk through how the Batch API works and how to write a production-ready Python pipeline to process 1,000 prompts for half the price.

Why Use the Batch API?

  1. 50% Discount: You pay exactly half price per token compared to standard REST calls.
  2. Higher Rate Limits: Batch requests sit in a separate queue, meaning you don't hit standard requests-per-minute (RPM) or tokens-per-minute (TPM) ceilings.
  3. Fewer Retries: The provider handles queueing and transient network errors internally.

Step 1: Preparing Your Batch File

The Batch API expects a .jsonl (JSON Lines) file where each line represents an individual API request payload with a unique custom_id.

Here is how you generate a 1,000-prompt JSONL file and submit it using Python and the official openai SDK:

import json
from openai import OpenAI

client = OpenAI()

# Generate 1,000 sample prompts
dataset = [
    f"Extract the main sentiment and product features from review #{i}: 'The battery life is fantastic, but charging is slow.'"
    for i in range(1, 1001)
]

# Write tasks to a local JSONL file
batch_filename = "tasks.jsonl"
with open(batch_filename, "w") as f:
    for idx, prompt in enumerate(dataset):
        task = {
            "custom_id": f"req-review-{idx}",
            "method": "POST",
            "url": "/v1/chat/completions",
            "body": {
                "model": "gpt-4o-mini",
                "messages": [
                    {"role": "system", "content": "You are a concise data extraction assistant."},
                    {"role": "user", "content": prompt}
                ],
                "max_tokens": 100
            }
        }
        f.write(json.dumps(task) + "\n")

# Upload the batch file to OpenAI
uploaded_file = client.files.create(
    file=open(batch_filename, "rb"),
    purpose="batch"
)

# Launch the batch job
batch_job = client.batches.create(
    input_file_id=uploaded_file.id,
    endpoint="/v1/chat/completions",
    completion_window="24h", # Currently the standard window
    metadata={"job_type": "sentiment_analysis_v1"}
)

print(f"Batch Job Created Successfully!")
print(f"Batch ID: {batch_job.id}")
print(f"Status: {batch_job.status}")
Enter fullscreen mode Exit fullscreen mode

Step 2: Polling for Completion and Downloading Results

Once the batch is submitted, you need a mechanism to monitor status. You can either listen for webhooks (if supported by your stack) or run a simple polling script.

Here is a robust script to poll job status and save the output when finished:

import time
from openai import OpenAI

client = OpenAI()

# Replace with your actual batch ID from Step 1
BATCH_ID = "batch_abc123xyz"

def retrieve_batch_results(batch_id: str):
    while True:
        job = client.batches.retrieve(batch_id)
        counts = job.request_counts
        print(f"Status: {job.status} | Completed: {counts.completed}/{counts.total} | Failed: {counts.failed}")

        if job.status == "completed":
            print("\nBatch processing complete! Downloading output...")

            # Fetch the completed output file content
            output_content = client.files.content(job.output_file_id)

            # Save results locally
            with open("results.jsonl", "wb") as f:
                f.write(output_content.content)

            print("Saved results to results.jsonl")
            break

        elif job.status in ["failed", "canceled", "expired"]:
            print(f"\nBatch failed with status: {job.status}")
            if job.error_file_id:
                errors = client.files.content(job.error_file_id)
                print(errors.text)
            break

        # Poll every 60 seconds
        time.sleep(60)

retrieve_batch_results(BATCH_ID)
Enter fullscreen mode Exit fullscreen mode

Parsing the Output

The output file (results.jsonl) mirrors your input file. Each JSON object contains the original custom_id, allowing you to map responses directly back to your database records:

{
  "id": "batch_req_123",
  "custom_id": "req-review-0",
  "response": {
    "status_code": 200,
    "body": {
      "choices": [
        {
          "message": {
            "content": "Sentiment: Positive. Features: Battery life (good), Charging speed (slow)."
          }
        }
      ]
    }
  },
  "error": null
}
Enter fullscreen mode Exit fullscreen mode

Practical Takeaways & Best Practices

  • Validate your JSONL locally: One syntax error on line 500 won't invalidate the whole file, but bad prompt parameters inside body will cause individual request failures. Always test a 5-item batch first.
  • Use custom_id wisely: Pass primary keys (e.g., user_12345 or doc_9876) into the custom_id field so you don't need complex mapping logic when saving responses.
  • Set realistic expectations: Most batch jobs complete within 1 to 3 hours, but the SLA is 24 hours. Don't use Batch API for user-facing, real-time features.

Conclusion

If your workload isn't constrained by live user interaction, using standard synchronous endpoints is leaving money on the table. Switching to the Batch API takes under an hour to write and slashes your LLM bill by 50% overnight.

Have you integrated batch processing into your LLM pipelines yet? Drop a comment below with your setup or any edge cases you've run into!

Top comments (0)