claude-batch

Batch processing library for Claude API. Handles concurrent requests with automatic retry logic, rate limiting and progress tracking.

Installation

pip install -r requirements.txt

Quick Start

from claudebatch import BatchProcessor

processor = BatchProcessor(api_key="your-key")

prompts = [
    "Summarize this article...",
    "Translate to French...",
    "Extract key points..."
]

results = processor.run(prompts)
for result in results:
    print(result.content)

Features

• Concurrent request processing with configurable parallelism
• Exponential backoff retry on rate limits and errors
• Progress tracking with callbacks
• Result aggregation and error handling
• Token usage tracking

API Reference

BatchProcessor

processor = BatchProcessor(
    api_key: str,
    model: str = "claude-3-sonnet-20240229",
    max_concurrent: int = 5,
    max_retries: int = 3,
    timeout: int = 60
)

Methods

• run(prompts: List[str]) -> List[Response] - Process batch of prompts
• run_async(prompts: List[str]) -> List[Response] - Async version
• set_progress_callback(fn: Callable) - Set progress callback

Configuration

# Environment variables
ANTHROPIC_API_KEY=your-key
CLAUDE_BATCH_MAX_CONCURRENT=10
CLAUDE_BATCH_TIMEOUT=120

Examples

Processing with progress tracking

def on_progress(completed, total):
    print(f"Progress: {completed}/{total}")

processor = BatchProcessor(api_key=key)
processor.set_progress_callback(on_progress)
results = processor.run(prompts)

Handling errors

results = processor.run(prompts)
for i, result in enumerate(results):
    if result.error:
        print(f"Prompt {i} failed: {result.error}")
    else:
        print(result.content)

License

MIT