567-labs--instructor/docs/examples/batch_in_memory.md

---
title: In-Memory Batch Processing for Serverless Applications
description: Learn how to use Instructor's in-memory batch processing feature for serverless deployments without disk I/O.
---

## See Also

- [Batch Processing](./batch_job_oai.md) - File-based batch processing
- [Bulk Classification](./bulk_classification.md) - Process multiple classifications
- [from_provider Guide](../concepts/from_provider.md#async-clients) - Async client setup
- [Cost Optimization](./batch_job_oai.md) - Reduce API costs with batch processing

# In-Memory Batch Processing for Serverless

This guide demonstrates how to use Instructor's in-memory batch processing feature, which is perfect for serverless deployments and applications that need to avoid disk I/O.

## Overview

In-memory batch processing allows you to create and submit batch requests without writing to disk, using BytesIO buffers instead of files. This is ideal for:

- **Serverless environments** (AWS Lambda, Google Cloud Functions, Azure Functions)
- **Containerized applications** with read-only file systems
- **Security-sensitive applications** that avoid temporary files
- **High-performance applications** that minimize I/O overhead

## Quick Start

```python
import time
from pydantic import BaseModel
from instructor.batch.processor import BatchProcessor


class User(BaseModel):
    """User model for extraction."""

    name: str
    age: int
    email: str


def main():
    # Initialize batch processor
    processor = BatchProcessor("openai/gpt-5-nano", User)

    # Sample messages for batch processing
    messages_list = [
        [
            {"role": "system", "content": "Extract user information from the text."},
            {
                "role": "user",
                "content": "John Doe is 25 years old and his email is john@example.com",
            },
        ],
        [
            {"role": "system", "content": "Extract user information from the text."},
            {
                "role": "user",
                "content": "Jane Smith, age 30, can be reached at jane.smith@company.com",
            },
        ],
        [
            {"role": "system", "content": "Extract user information from the text."},
            {
                "role": "user",
                "content": "Bob Wilson (bob.wilson@email.com) is 28 years old",
            },
        ],
    ]

    # Create batch in memory (no file_path specified)
    batch_buffer = processor.create_batch_from_messages(
        messages_list,
        file_path=None,  # This triggers in-memory mode
        max_tokens=150,
        temperature=0.1,
    )

    print(f"Created batch buffer: {type(batch_buffer)}")
    print(f"Buffer size: {len(batch_buffer.getvalue())} bytes")

    # Submit the batch using the in-memory buffer
    batch_id = processor.submit_batch(
        batch_buffer, metadata={"description": "In-memory batch example"}
    )

    print(f"Batch submitted successfully! Batch ID: {batch_id}")

    # Poll for completion
    print("Waiting for batch to complete...")
    max_wait_time = 300  # 5 minutes max
    start_time = time.time()

    while time.time() - start_time < max_wait_time:
        status = processor.get_batch_status(batch_id)
        current_status = status.get("status", "unknown")

        print(f"Current status: {current_status}")

        if current_status in ["completed", "failed", "cancelled", "expired"]:
            break

        time.sleep(10)

    # Retrieve and process results
    if status.get("status") == "completed":
        print("Batch completed! Retrieving results...")

        results = processor.get_results(batch_id)

        successful_results = [r for r in results if hasattr(r, "result")]
        error_results = [r for r in results if hasattr(r, "error_message")]

        print(f"Total results: {len(results)}")
        print(f"Successful: {len(successful_results)}")
        print(f"Errors: {len(error_results)}")

        # Show successful extractions
        if successful_results:
            print("\nExtracted Users:")
            for result in successful_results:
                user = result.result
                print(f"   - {user.name}, {user.age} years old, {user.email}")

        # Show any errors
        if error_results:
            print("\nErrors encountered:")
            for error in error_results:
                print(f"   - {error.custom_id}: {error.error_message}")


if __name__ == "__main__":
    main()
```

## File vs In-Memory Comparison

### Traditional File-Based Approach

```python
# File-based approach
processor = BatchProcessor("openai/gpt-5-nano", User)

# Creates file on disk
file_path = processor.create_batch_from_messages(
    messages_list,
    file_path="temp_batch.jsonl",  # Specify file path
    max_tokens=150,
    temperature=0.1,
)

# Submit using file path
batch_id = processor.submit_batch(file_path)

# Remember to clean up
import os

if os.path.exists(file_path):
    os.remove(file_path)
```

### New In-Memory Approach

```python
# In-memory approach
processor = BatchProcessor("openai/gpt-5-nano", User)

# Creates BytesIO buffer in memory
buffer = processor.create_batch_from_messages(
    messages_list,
    file_path=None,  # No file path = in-memory
    max_tokens=150,
    temperature=0.1,
)

# Submit using buffer
batch_id = processor.submit_batch(buffer)

# No cleanup required - buffer is automatically garbage collected
```

## Benefits of In-Memory Processing

### ✅ Perfect for Serverless

```python
# AWS Lambda example
import json


def lambda_handler(event, context):
    """AWS Lambda function using in-memory batch processing."""

    # Extract data from event
    messages_list = event.get("messages", [])

    # Process in memory - no disk I/O
    processor = BatchProcessor("openai/gpt-5-nano", User)
    buffer = processor.create_batch_from_messages(
        messages_list,
        file_path=None,  # Essential for Lambda
    )

    batch_id = processor.submit_batch(buffer)

    return {
        'statusCode': 200,
        'body': json.dumps(
            {'batch_id': batch_id, 'message': 'Batch submitted successfully'}
        ),
    }
```

### ✅ Memory Efficient

```python
# Check buffer size before submission
buffer = processor.create_batch_from_messages(messages_list, file_path=None)

print(f"Buffer size: {len(buffer.getvalue())} bytes")
print(f"Buffer type: {type(buffer)}")

# Buffer content is accessible
buffer.seek(0)
content_preview = buffer.read(200).decode("utf-8")
print(f"Preview: {content_preview}...")

# Reset for submission
buffer.seek(0)
batch_id = processor.submit_batch(buffer)
```

### ✅ Security Benefits

```python
# No temporary files on disk
# No file permissions to manage
# No cleanup required
# Buffer is automatically garbage collected

processor = BatchProcessor("openai/gpt-5-nano", User)

# This approach leaves no trace on the file system
buffer = processor.create_batch_from_messages(
    sensitive_messages,
    file_path=None,  # Keeps everything in memory
)

batch_id = processor.submit_batch(buffer)
# When buffer goes out of scope, it's automatically cleaned up
```

## Error Handling

```python
try:
    # Create batch buffer
    buffer = processor.create_batch_from_messages(
        messages_list,
        file_path=None,
    )

    # Submit batch
    batch_id = processor.submit_batch(buffer)

    # Process results
    results = processor.get_results(batch_id)

except Exception as e:
    print(f"Error during batch processing: {e}")
    #> Error during batch processing: name 'processor' is not defined
    # No file cleanup needed with in-memory approach
```

## Provider Support

All providers support in-memory batch processing:

### OpenAI
```python
processor = BatchProcessor("openai/gpt-5-nano", User)
buffer = processor.create_batch_from_messages(messages_list, file_path=None)
batch_id = processor.submit_batch(buffer)
```

### Anthropic
```python
processor = BatchProcessor("anthropic/claude-3-5-sonnet-20241022", User)
buffer = processor.create_batch_from_messages(messages_list, file_path=None)
batch_id = processor.submit_batch(buffer)
```

### Google GenAI
```python
processor = BatchProcessor("google/gemini-2.5-flash", User)
buffer = processor.create_batch_from_messages(messages_list, file_path=None)
batch_id = processor.submit_batch(buffer)
```

## Best Practices

1. **Always set `file_path=None`** to enable in-memory mode
2. **Monitor buffer size** for large batches to avoid memory issues
3. **Use appropriate models** that support JSON schema (e.g., gpt-4o-mini)
4. **Handle errors gracefully** - no file cleanup needed
5. **Consider memory limits** in serverless environments

## Limitations

- **Memory usage**: Large batches may consume significant memory
- **No debugging files**: Can't inspect batch files for troubleshooting
- **Temporary storage**: Buffer contents are lost if not submitted immediately

## Troubleshooting

### Buffer Size Issues
```python
# Check buffer size before submission
buffer = processor.create_batch_from_messages(messages_list, file_path=None)
size_mb = len(buffer.getvalue()) / (1024 * 1024)
print(f"Buffer size: {size_mb:.2f} MB")

if size_mb > 100:  # Adjust threshold as needed
    print("Warning: Large buffer size, consider splitting batch")
```

### Memory Monitoring
```python
import psutil
import os

# Check memory usage
process = psutil.Process(os.getpid())
memory_before = process.memory_info().rss / 1024 / 1024  # MB

buffer = processor.create_batch_from_messages(messages_list, file_path=None)

memory_after = process.memory_info().rss / 1024 / 1024  # MB
print(f"Memory increase: {memory_after - memory_before:.2f} MB")
```

This in-memory approach makes Instructor's batch processing perfect for modern serverless and containerized applications while maintaining the same powerful API and provider support.