Skip to content

Batch Processing

Batch processing lets you send multiple requests in a single operation, saving up to 50% on costs. Instructor supports batch processing across multiple providers.

Supported Providers

Provider Models Cost Savings
OpenAI gpt-4o, gpt-4.1-mini, gpt-5.4-mini 50%
Anthropic claude-3-5-sonnet, claude-3-opus, claude-3-haiku 50%
Google GenAI gemini-2.5-flash, gemini-2.0-flash, gemini-pro 50%

Basic Usage

from instructor.batch import BatchProcessor
from pydantic import BaseModel


class User(BaseModel):
    name: str
    age: int


processor = BatchProcessor("openai/gpt-4.1-mini", User)

messages_list = [
    [
        {"role": "system", "content": "Extract user information from text."},
        {"role": "user", "content": "Hi, I'm Alice and I'm 28 years old."},
    ],
    [
        {"role": "system", "content": "Extract user information from text."},
        {"role": "user", "content": "Hello, I'm Bob, 35 years old."},
    ],
]

# Create batch file
processor.create_batch_from_messages(
    file_path="batch_requests.jsonl",
    messages_list=messages_list,
    max_tokens=200,
    temperature=0.1,
)

# Submit batch job
batch_id = processor.submit_batch("batch_requests.jsonl")
print(f"Batch job submitted: {batch_id}")

# Check status and retrieve results
status = processor.get_batch_status(batch_id)
if status["status"] in ["completed", "ended", "JOB_STATE_SUCCEEDED"]:
    from instructor.batch import extract_results

    all_results = processor.retrieve_results(batch_id)
    for user in extract_results(all_results):
        print(f"Name: {user.name}, Age: {user.age}")

In-Memory Processing

For serverless deployments, use in-memory mode by setting file_path=None:

import time
from instructor.batch import BatchProcessor
from pydantic import BaseModel


class User(BaseModel):
    name: str
    age: int


processor = BatchProcessor("openai/gpt-4.1-mini", User)

messages_list = [
    [{"role": "user", "content": "Extract: John is 25 years old"}],
    [{"role": "user", "content": "Extract: Jane is 30 years old"}],
]

# Create in-memory buffer (no file_path)
buffer = processor.create_batch_from_messages(
    messages_list,
    file_path=None,
    max_tokens=150,
)

# Submit and poll for results
batch_id = processor.submit_batch(buffer)

while True:
    status = processor.get_batch_status(batch_id)
    if status.get("status") in ["completed", "failed", "cancelled"]:
        break
    time.sleep(10)

if status.get("status") == "completed":
    results = processor.get_results(batch_id)
    for r in results:
        if hasattr(r, "result"):
            print(f"{r.result.name}, {r.result.age}")

When to Use Each Approach

Use Case Approach
Serverless (Lambda, Cloud Functions) In-memory
Large batch jobs File-based
Security-sensitive environments In-memory
Debugging/audit requirements File-based

Provider Setup

OpenAI

export OPENAI_API_KEY="your-openai-key"

processor = BatchProcessor("openai/gpt-4.1-mini", User)

Anthropic

export ANTHROPIC_API_KEY="your-anthropic-key"

processor = BatchProcessor("anthropic/claude-3-5-sonnet-20241022", User)

Google GenAI

export GOOGLE_CLOUD_PROJECT="your-project-id"
export GCS_BUCKET="your-gcs-bucket-name"
export GOOGLE_APPLICATION_CREDENTIALS="/path/to/service-account.json"

processor = BatchProcessor("google/gemini-2.5-flash", User)

Required permissions: roles/aiplatform.user and roles/storage.objectUser.

Processing Results

OpenAI batch requests use strict JSON schemas. All named fields are required in the output, even when the Pydantic model has defaults. Nullable fields still allow null. Arbitrary-key dictionaries are not supported by this strict batch path; use named model fields or a list of key/value objects instead. Anthropic batch requests keep their provider-specific tool schema, including mapping value types.

Results use a Maybe/Result pattern for type-safe handling:

For completed OpenAI batches, retrieval reads both the output and error files. For ended Anthropic batches, it reads every result, including errors. A batch where every request failed returns BatchError records rather than raising because no requests succeeded. Errors retain the request's custom_id, provider error details, and original record in raw_data. Jobs that are still running cannot be retrieved.

from instructor.batch import (
    filter_successful,
    filter_errors,
    extract_results,
    get_results_by_custom_id,
)

all_results = processor.retrieve_results(batch_id)

# Filter by type
successful = filter_successful(all_results)  # List[BatchSuccess[T]]
errors = filter_errors(all_results)  # List[BatchError]
objects = extract_results(all_results)  # List[T]

# Access by custom_id
by_id = get_results_by_custom_id(all_results)
if "request-1" in by_id:
    result = by_id["request-1"]
    if result.success:
        print(f"Success: {result.result}")
    else:
        print(f"Error: {result.error_message}")

Custom IDs

By default, requests get the IDs request-0, request-1, and so on. Pass custom_ids to use your own IDs. This makes it easier to match results, including failed ones, back to your data:

processor.create_batch_from_messages(
    messages_list,
    file_path="batch_requests.jsonl",
    custom_ids=["user_1", "user_2", "user_3"],  # one per conversation, same order
)

IDs must be unique and match the length of messages_list. Anthropic also requires IDs to match ^[a-zA-Z0-9_-]{1,64}$. Invalid IDs raise a ValueError before any request is written.

API Reference

Method Description
create_batch_from_messages(messages_list, file_path=None, ..., custom_ids=None) Create batch file or buffer
submit_batch(file_path_or_buffer, metadata=None) Submit batch job, returns job ID
get_batch_status(batch_id) Get job status
retrieve_results(batch_id) Download and parse results
parse_results(content) Parse raw results content

CLI Commands

# List batch jobs
instructor batch list --model "openai/gpt-4.1-mini"

# Create batch from file
instructor batch create-from-file --file-path batch.jsonl --model "openai/gpt-4.1-mini"

# Get batch results
instructor batch results --batch-id "batch_abc123" --output-file results.jsonl

Best Practices

  1. Batch size: Include at least 25,000 requests per job for optimal efficiency
  2. Cost optimization: Use batch processing for non-urgent workloads
  3. Error handling: Always check both successful and error results
  4. Timeouts: Batch jobs have execution limits (24 hours for Google)
  5. Storage: For Google, ensure GCS bucket is in the same region as your batch job

Troubleshooting

Issue Solution
Missing GCS_BUCKET (Google) Set the GCS_BUCKET environment variable
Permission Denied (Google) Add aiplatform.user and storage.objectUser roles
Invalid Model Name Use format provider/model-name
Authentication Error Verify API keys are set correctly