### Install Batchata with uv

Source: https://github.com/agamm/batchata/blob/main/README.md

Install the batchata library using uv, a fast Python package installer and resolver.

```bash
uv add batchata
```

--------------------------------

### Install Development Dependencies

Source: https://github.com/agamm/batchata/blob/main/development.md

Installs the necessary development dependencies for the project using uv.

```bash
uv sync --dev
```

--------------------------------

### Load Environment Variables in Examples

Source: https://github.com/agamm/batchata/blob/main/specs/010-remove-dotenv-add-api-key-validation.md

Loads environment variables from a .env file for examples that require them. Ensure dotenv is installed in your development environment.

```python
from dotenv import load_dotenv
load_dotenv()
```

--------------------------------

### Complete Batchata Example with Pydantic and Citations

Source: https://github.com/agamm/batchata/blob/main/README.md

This comprehensive example shows advanced Batchata usage, including Pydantic models for structured output, API key loading from .env, setting state persistence, defining cost and time limits, enabling citations, and processing results with detailed output and error handling.

```python
from batchata import Batch
from pydantic import BaseModel
from dotenv import load_dotenv

load_dotenv()  # Load API keys from .env

# Define structured output
class InvoiceAnalysis(BaseModel):
    invoice_number: str
    total_amount: float
    vendor: str
    payment_status: str

# Create batch configuration
batch = Batch(
        results_dir="./invoice_results",
        max_parallel_batches=1,
        items_per_batch=3
    )
    .set_state(file="./invoice_state.json", reuse_state=False)
    .set_default_params(model="gpt-5.2-pro-2025-12-11", temperature=0.0)
    .add_cost_limit(usd=5.0)
    .add_time_limit(minutes=10)  # Time limit of 10 minutes
    .set_verbosity("warn") 

# Add jobs with structured output and citations
invoice_files = ["path/to/invoice1.pdf", "path/to/invoice2.pdf", "path/to/invoice3.pdf"]
for invoice_file in invoice_files:
    batch.add_job(
        file=invoice_file,
        prompt="Extract the invoice number, total amount, vendor name, and payment status.",
        response_model=InvoiceAnalysis,
        enable_citations=True
    )

# Execute with rich progress display
print("Starting batch processing...")
run = batch.run(print_status=True)

# Or use custom progress callback
run = batch.run(print_status=True)

# Get results
results = run.results()

# Process successful results
for result in results["completed"]:
    analysis = result.parsed_response
    citations = result.citation_mappings
    print(f"\nInvoice: {analysis.invoice_number} (page: {citations.get("invoice_number").page})")
    print(f"  Vendor: {analysis.vendor} (page: {citations.get("vendor").page})")
    print(f"  Total: ${analysis.total_amount:.2f} (page: {citations.get("total_amount").page})")
    print(f"  Status: {analysis.payment_status} (page: {citations.get("payment_status").page})")
    
    # Save each result to JSON file
    result.save_to_json(f"./invoice_results/{result.job_id}.json")

# Process failed/cancelled results  
for result in results["failed"]:
    print(f"\nJob {result.job_id} failed: {result.error}")

for result in results["cancelled"]:
    print(f"\nJob {result.job_id} was cancelled: {result.error}")
```

--------------------------------

### Basic Batch Processing with Batchata

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Demonstrates basic batch processing setup, including setting a results directory, default model parameters, and a cost limit. Jobs are added in a loop, and the batch is executed.

```python
from batchata import Batch

# Simple batch processing
batch = Batch(results_dir="./output")
    .set_default_params(model="claude-sonnet-4-20250514")
    .add_cost_limit(usd=5.0)

# Add jobs
for file in files:
    batch.add_job(file=file, prompt="Summarize this document")

# Execute
run = batch.run()
results = run.results()
```

--------------------------------

### Set API Keys as Environment Variables

Source: https://github.com/agamm/batchata/blob/main/README.md

Configure API keys for supported providers by exporting them as environment variables. This is a common setup method.

```bash
export ANTHROPIC_API_KEY="your-key"
export OPENAI_API_KEY="your-key"
export GOOGLE_API_KEY="your-key"  # For Gemini models
```

--------------------------------

### Install Batchata with pip

Source: https://github.com/agamm/batchata/blob/main/README.md

Install the batchata library using pip. This is the standard method for adding Python packages to your project.

```bash
pip install batchata
```

--------------------------------

### BatchRun Initialization Details

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Provides an in-depth look at the BatchRun constructor, detailing the initialization of various components like cost trackers, state managers, and result directories. It also handles the creation of temporary state files and the setup of raw file directories if enabled.

```python
import shutil
from pathlib import Path

# ... inside BatchRun.__init__ ...

if not config.reuse_state and self.results_dir.exists():
    shutil.rmtree(self.results_dir)

self.results_dir.mkdir(parents=True, exist_ok=True)

# Raw files directory (if enabled)
self.raw_files_dir = None
if config.raw_files:
    self.raw_files_dir = self.results_dir / "raw_files"
    self.raw_files_dir.mkdir(parents=True, exist_ok=True)
```

--------------------------------

### Initialize Batch Configuration

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Instantiate the Batch class to start configuring a batch job. Specify the directory for results and optionally set limits for parallel batches and items per batch.

```python
batch = Batch("./results", max_parallel_batches=10, items_per_batch=10)
```

--------------------------------

### Load API Keys from .env File

Source: https://github.com/agamm/batchata/blob/main/README.md

Alternatively, load API keys from a .env file in the project root using the python-dotenv library. Ensure python-dotenv is installed.

```python
from dotenv import load_dotenv
load_dotenv()

from batchata import Batch
# Your API keys will now be loaded from .env
```

--------------------------------

### Enable State Persistence and Resumption

Source: https://context7.com/agamm/batchata/llms.txt

Configures a JSON checkpoint file for state persistence. If `reuse_state=True`, the run automatically resumes from the last saved state, skipping completed jobs. Set `reuse_state=False` to start fresh.

```python
from batchata import Batch

batch = (
    Batch(results_dir="./resumable_output")
    .set_default_params(model="claude-3-5-sonnet-20241022")
    .set_state(file="./my_batch_state.json", reuse_state=True)
)
# If the process was killed mid-run, simply re-execute the same script.
# Completed jobs will be loaded from the state file; only remaining jobs run.
```

--------------------------------

### Batch API Call Example

Source: https://github.com/agamm/batchata/blob/main/specs/001-initial-ai-batch-mvp-completed.md

Use this snippet to perform batch processing with the AI API. Ensure the 'SpamResult' model is defined and the 'emails' list is populated.

```python
batch(
    messages=[[{"role": "user", "content": f"You are a spam detection expert. Is this spam? {email}"}] for email in emails], 
    model="claude-3-haiku-20240307", 
    response_model=SpamResult
)
```

--------------------------------

### Get Batch Execution Statistics

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Retrieve a dictionary containing detailed statistics about the current batch run. Optionally, print these statistics to the logger.

```python
def status(self, print_status: bool = False) -> Dict:
    """Get current execution statistics."""
    total_jobs = len(self.jobs)
    completed_count = len(self.completed_results) + len(self.failed_jobs) + len(self.cancelled_jobs)
    remaining_count = total_jobs - completed_count
    
    stats = {
        "total": total_jobs,
        "pending": remaining_count,
        "active": 0,  # Always 0 for synchronous execution
        "completed": len(self.completed_results),
        "failed": len(self.failed_jobs),
        "cancelled": len(self.cancelled_jobs),
        "cost_usd": self.cost_tracker.used_usd,
        "cost_limit_usd": self.cost_tracker.limit_usd,
        "is_complete": self.is_complete,
        "batches_total": self.total_batches,
        "batches_completed": self.completed_batches,
        "batches_pending": self.total_batches - self.completed_batches,
        "current_batch_index": self.current_batch_index,
        "current_batch_size": self.current_batch_size,
        "items_per_batch": self.config.items_per_batch
    }
    
    if print_status:
        logger.info("\nBatch Run Status:")
        logger.info(f"  Total jobs: {stats['total']}")
        logger.info(f"  Pending: {stats['pending']}")
        logger.info(f"  Active: {stats['active']}")
        logger.info(f"  Completed: {stats['completed']}")
        logger.info(f"  Failed: {stats['failed']}")
        logger.info(f"  Cancelled: {stats['cancelled']}")
        logger.info(f"  Cost: ${stats['cost_usd']:.6f}")
        if stats['cost_limit_usd']:
            logger.info(f"  Cost limit: ${stats['cost_limit_usd']:.2f}")
        logger.info(f"  Complete: {stats['is_complete']}")
    
    return stats
```

--------------------------------

### Add Job with File Input

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Add a job to the batch that processes content from a file. A prompt is provided to guide the processing of the file's content.

```python
.add_job(file="./path/to/file.pdf", prompt="Generate summary of file")
```

--------------------------------

### Handle Batchata Errors with Specific Exceptions

Source: https://context7.com/agamm/batchata/llms.txt

This example demonstrates how to use a try-except block to catch various `BatchataError` subclasses, including `ValidationError`, `ProviderNotFoundError`, and `ProviderError`. It also shows how cost limit violations are reported as failed jobs rather than exceptions.

```python
from batchata import Batch
from batchata import (
    BatchataError,
    CostLimitExceededError,
    ProviderError,
    ProviderNotFoundError,
    ValidationError,
)

try:
    batch = (
        Batch(results_dir="./output")
        .set_default_params(model="claude-3-5-sonnet-20241022")
        .add_cost_limit(usd=0.001)   # very tight limit
    )
    batch.add_job(messages=[{"role": "user", "content": "Hello"}])
    run = batch.run()
except ValidationError as e:
    print(f"Bad parameters: {e}")
except ProviderNotFoundError as e:
    print(f"Unknown model: {e}")
except ProviderError as e:
    print(f"Provider API error: {e}")
except BatchataError as e:
    print(f"Batchata error: {e}")

# Cost limit violations appear as failed jobs, not exceptions:
results = run.results()
cost_exceeded = [r for r in results["failed"] if "Cost limit" in (r.error or "")]
```

--------------------------------

### Batch.set_state

Source: https://context7.com/agamm/batchata/llms.txt

Enables state persistence by configuring a JSON checkpoint file. If `reuse_state=True` and the file exists, completed jobs are automatically skipped, allowing the run to resume from where it left off. Setting `reuse_state=False` starts a fresh run, clearing the results directory.

```APIDOC
## Batch.set_state – Enable State Persistence and Resume

Configures a JSON checkpoint file. If `reuse_state=True` and the file already exists, completed jobs are skipped automatically so the run resumes where it left off. Pass `reuse_state=False` to start fresh even if the file exists (also clears the results directory).

```python
from batchata import Batch

batch = (
    Batch(results_dir="./resumable_output")
    .set_default_params(model="claude-3-5-sonnet-20241022")
    .set_state(file="./my_batch_state.json", reuse_state=True)
)
# If the process was killed mid-run, simply re-execute the same script.
# Completed jobs will be loaded from the state file; only remaining jobs run.
```
```

--------------------------------

### Simple Batch Processing with Batchata

Source: https://github.com/agamm/batchata/blob/main/README.md

This snippet demonstrates basic batch processing. Configure a Batch instance, set default parameters and cost limits, add jobs with file inputs and prompts, and then run the batch. Results can be retrieved after completion.

```python
from batchata import Batch

# Simple batch processing
batch = Batch(results_dir="./output")
    .set_default_params(model="gpt-5.2-latest")  # or "claude-sonnet-4-5-20250929" or "gemini-3.0-pro-latest"
    .add_cost_limit(usd=5.0)

for file in files:
    batch.add_job(file=file, prompt="Summarize")

run = batch.run()

results = run.results()  # {"completed": [JobResult], "failed": [JobResult], "cancelled": [JobResult]}

# Or preview costs first with dry run
run = batch.run(dry_run=True)  # Shows cost estimates without executing
```

--------------------------------

### Build and Run a Batch Job

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Demonstrates how to configure a batch job with various settings like state management, default parameters, cost limits, and adding jobs. Use this to set up complex batch processing workflows.

```python
batch = Batch("./results", max_parallel_batches=10, items_per_batch=10)
    .set_state(file="./state.json", reuse_state=True)
    .set_default_params(model="claude-sonnet-4-20250514", temperature=0.7)
    .add_cost_limit(usd=15.0)
    .add_job(messages=[{"role": "user", "content": "Hello"}])
    .add_job(file="./path/to/file.pdf", prompt="Generate summary of file")

run = batch.run()

```

--------------------------------

### Basic Batch Processing with Batchata

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Demonstrates setting up a Batch object, defining default parameters, adding a cost limit, and adding jobs for file processing. Use this for simple batch tasks.

```python
from batchata import Batch  # Simple batch processing
batch = Batch(results_dir="./output")     .set_default_params(model="claude-sonnet-4-20250514")     .add_cost_limit(usd=5.0)  # Add jobs for file in files:
    batch.add_job(file=file, prompt="Summarize this document")  # Execute
run = batch.run() 
results = run.results()
```

--------------------------------

### Execute Synchronous Batch Run

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Executes a synchronous batch run and waits for its completion. It prevents the batch run from being started more than once and records the start time.

```python
def execute(self):
    """Execute synchronous batch run and wait for completion."""
    if self._started:
        raise RuntimeError("Batch run already started")
    
    self._started = True
    self._start_time = datetime.now()
    
    # Register signal handler for graceful shutdown
    def signal_handler(signum, frame):
```

--------------------------------

### Get Batch Results by Status

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Organize and retrieve all job results categorized by their final status (completed, failed, or cancelled).

```python
def results(self) -> Dict[str, List[JobResult]]:
    """Get all results organized by status.

    Returns:
        {
            "completed": [JobResult],
            "failed": [JobResult],
            "cancelled": [JobResult]
        }
    """
    return {
        "completed": list(self.completed_results.values()),
        "failed": self._create_failed_results(),
        "cancelled": self._create_cancelled_results()
    }
```

--------------------------------

### Initialize Batch Builder

Source: https://context7.com/agamm/batchata/llms.txt

Instantiates the main Batch builder. Configure output directory, concurrency, and batch item limits.

```python
from batchata import Batch

batch = Batch(
    results_dir="./output",        # directory for result JSON files
    max_parallel_batches=5,        # up to 5 batches in flight at once
    items_per_batch=10,            # 10 jobs per provider batch
    raw_files=True                 # save raw provider JSONL / responses
)
```

--------------------------------

### Get Failed Jobs (Deprecated)

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Retrieves failed jobs with their error messages. This method is deprecated and users should use results()['failed'] instead.

```python
def get_failed_jobs(self) -> Dict[str, str]:
        """Get failed jobs with error messages.

        Note: This method is deprecated. Use results()['failed'] instead.
        """
        return dict(self.failed_jobs)
```

--------------------------------

### Get Batch Execution Statistics

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Retrieves current execution statistics for a batch, including total jobs, pending, completed, failed, cancelled, and cost.

```python
    def status(self, print_status: bool = False) -> Dict:
        """Get current execution statistics."""
        total_jobs = len(self.jobs)
        completed_count = len(self.completed_results) + len(self.failed_jobs) + len(self.cancelled_jobs)
        remaining_count = total_jobs - completed_count
        
        stats = {
            "total": total_jobs,
            "pending": remaining_count,
            "active": 0,  # Always 0 for synchronous execution
            "completed": len(self.completed_results),
            "failed": len(self.failed_jobs),
            "cancelled": len(self.cancelled_jobs),
            "cost_usd": self.cost_tracker.used_usd,
        }
```

--------------------------------

### Configure API Keys and Initialize Batch

Source: https://context7.com/agamm/batchata/llms.txt

This snippet shows how to set environment variables for API keys for Anthropic, OpenAI, and Google Gemini, or alternatively use `python-dotenv`. It then initializes a `Batch` object with a specified results directory and default model.

```python
import os
os.environ["ANTHROPIC_API_KEY"] = "sk-ant-..."
os.environ["OPENAI_API_KEY"]    = "sk-..."
os.environ["GOOGLE_API_KEY"]    = "AIza..."

# Or use python-dotenv:
from dotenv import load_dotenv
load_dotenv()   # reads .env in project root

from batchata import Batch
batch = Batch(results_dir="./out").set_default_params(model="gemini-2.5-flash")
```

--------------------------------

### Initialize BatchRun with Configuration

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Initializes the BatchRun class with configuration parameters and a list of jobs. It sets up logging, cost tracking, state management, and result directories. If not reusing state, it clears the results directory.

```python
config = BatchParams(...)
run = BatchRun(config, jobs)
run.execute()
results = run.results()
```

--------------------------------

### Batch.__init__

Source: https://context7.com/agamm/batchata/llms.txt

Instantiates the top-level Batch builder. Configures output directory, maximum parallel batches, jobs per batch, and whether to save raw provider responses.

```APIDOC
## Batch.__init__ – Create a Batch Builder

Instantiates the top-level builder. `results_dir` controls where per-job JSON files are written. `max_parallel_batches` limits how many provider-level batch requests run concurrently. `items_per_batch` controls how many jobs are packed into each provider batch submission.

```python
from batchata import Batch

batch = Batch(
    results_dir="./output",        # directory for result JSON files
    max_parallel_batches=5,        # up to 5 batches in flight at once
    items_per_batch=10,            # 10 jobs per provider batch
    raw_files=True                 # save raw provider JSONL / responses
)
```
```

--------------------------------

### Get Batch Run Results by Status

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Retrieves all job results organized by their status (completed, failed, cancelled). Failed and cancelled results are converted into JobResult objects.

```python
def results(self) -> Dict[str, List[JobResult]]:
        """Get all results organized by status.

        Returns:
            {
                "completed": [JobResult],
                "failed": [JobResult],
                "cancelled": [JobResult]
            }
        """
        return {
            "completed": list(self.completed_results.values()),
            "failed": self._create_failed_results(),
            "cancelled": self._create_cancelled_results()
        }
```

--------------------------------

### Update pyproject.toml for Wheel Packaging

Source: https://github.com/agamm/batchata/blob/main/specs/009-fix-packaging-import-issues.md

Modify the `pyproject.toml` file to correctly specify the package directory for wheel builds. This ensures the package is installed without an unnecessary `src/` prefix.

```toml
[tool.hatch.build.targets.wheel]
packages = ["batchata"]
```

--------------------------------

### Initialize Batch Configuration

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Initializes the Batch object with essential configuration parameters. Ensure `results_dir` is provided to specify where output will be stored.

```python
def __init__(self, results_dir: str, max_parallel_batches: int = 10, items_per_batch: int = 10, raw_files: Optional[bool] = None):
        """Initialize batch configuration.
        
        Args:
            results_dir: Directory to store results
            max_parallel_batches: Maximum parallel batch requests
            items_per_batch: Number of jobs per provider batch
            raw_files: Whether to save debug files (raw responses, JSONL files) from providers (default: True if results_dir is set, False otherwise)
        """
        # Auto-determine raw_files based on results_dir if not explicitly set
        if raw_files is None:
            raw_files = bool(results_dir and results_dir.strip())
        
        self.config = BatchParams(
            state_file=None,
            results_dir=results_dir,
            max_parallel_batches=max_parallel_batches,
            items_per_batch=items_per_batch,
            reuse_state=True,
            raw_files=raw_files
        )
        self.jobs: List[Job] = []

```

--------------------------------

### Start Time Limit Watchdog Thread

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Initiates a background thread that monitors the batch execution time. If the time limit is exceeded, it logs a warning, sets a flag, and triggers a shutdown.

```python
def time_limit_watchdog():
    """Check for time limit every second and trigger shutdown if exceeded."""
    while not self._shutdown_event.is_set():
        if self._check_time_limit():
            logger.warning("Batch execution time limit exceeded")
            with self._state_lock:
                self._time_limit_exceeded = True
            self._shutdown_event.set()
            break
        time.sleep(1.0)

# Start watchdog as daemon thread
watchdog_thread = threading.Thread(target=time_limit_watchdog, daemon=True)
watchdog_thread.start()
logger.debug(f"Started time limit watchdog thread (time limit: {self.config.time_limit_seconds}s)")
```

--------------------------------

### Prepare Batches for Execution

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Prepares all job batches for execution, grouping them by provider and estimating costs. It also initializes tracking information for each pending batch.

```python
def _prepare_batches(self) -> List[Tuple[str, object, List[Job]]]:
        """Prepare all batches as simple list of (provider_name, provider, jobs)."""
        batches = []
        jobs_by_provider = self._group_jobs_by_provider()
        
        for provider_name, provider_jobs in jobs_by_provider.items():
            provider = get_provider(provider_jobs[0].model)
            job_batches = self._split_into_batches(provider_jobs)
            
            for batch_jobs in job_batches:
                batches.append((provider_name, provider, batch_jobs))
                
                # Pre-populate batch tracking for pending batches
                batch_id = f"pending_{len(self.batch_tracking)}"
                estimated_cost = provider.estimate_cost(batch_jobs)
                self.batch_tracking[batch_id] = {
                    'start_time': None,
                    'status': 'pending',
                    'total': len(batch_jobs),
                    'completed': 0,
                    'cost': 0.0,
                    'estimated_cost': estimated_cost,
                    'provider': provider_name,
                    'jobs': batch_jobs
                }
        
        return batches
```

--------------------------------

### Generate API Documentation

Source: https://github.com/agamm/batchata/blob/main/development.md

Generates API documentation for the batchata package using pdoc and outputs it to the 'docs/' directory. This command should be run for each version.

```bash
uv run pdoc -o docs/ batchata
```

--------------------------------

### Project Structure for Multi-Provider Support

Source: https://github.com/agamm/batchata/blob/main/specs/002-remove-instructor-custom-providers.md

Illustrates the proposed directory structure for organizing provider-specific code. This structure facilitates the addition of new AI providers by creating dedicated modules.

```text
src/
    __init__.py
    ai_batch.py          # Main batch() function
    └── providers/
        ├── __init__.py
        ├── anthropic.py     # prepare_request() + parse_response()
        └── openai.py        # prepare_request() + parse_response() (future)
```

--------------------------------

### Initialize BatchRun

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Initializes the BatchRun class with configuration and a list of jobs. It sets up logging, cost tracking, state management, and result directories. If not reusing state, it clears the results directory.

```python
def __init__(self, config: BatchParams, jobs: List[Job]):
    """Initialize batch run. """
    self.config = config
    self.jobs = {job.id: job for job in jobs}
    # Set logging level based on config
    set_log_level(level=config.verbosity.upper())
    # Initialize components
    self.cost_tracker = CostTracker(limit_usd=config.cost_limit_usd)
    # Use temp file for state if not provided
    state_file = config.state_file
    if not state_file:
        state_file = create_temp_state_file(config)
        config.reuse_state = False
        logger.info(f"Created temporary state file: {state_file}")
    self.state_manager = StateManager(state_file)
    # State tracking
    self.pending_jobs: List[Job] = []
    self.completed_results: Dict[str, JobResult] = {}  # job_id -> result
    self.failed_jobs: Dict[str, str] = {}  # job_id -> error
    self.cancelled_jobs: Dict[str, str] = {}  # job_id -> reason
    # Batch tracking
    self.total_batches = 0
    self.completed_batches = 0
    self.current_batch_index = 0
    self.current_batch_size = 0
    # Execution control
    self._started = False
    self._start_time: Optional[datetime] = None
    self._time_limit_exceeded = False
    self._progress_callback: Optional[Callable[[Dict, float], None]] = None
    self._progress_interval: float = 1.0  # Default to 1 second
    # Threading primitives
    self._state_lock = threading.Lock()
    self._shutdown_event = threading.Event()
    self._progress_lock = threading.Lock()
    self._last_progress_update = 0.0
    # Batch tracking for progress display
    self.batch_tracking: Dict[str, Dict] = {}  # batch_id -> batch_info
    # Active batch tracking for cancellation
    self._active_batches: Dict[str, object] = {}  # batch_id -> provider
    self._active_batches_lock = threading.Lock()
    # Results directory
    self.results_dir = Path(config.results_dir)
    # If not reusing state, clear the results directory
    if not config.reuse_state and self.results_dir.exists():
        import shutil
        shutil.rmtree(self.results_dir)
```

--------------------------------

### Get Batch Run Status

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Returns a dictionary containing the current status statistics of a batch run, including counts of total, pending, active, completed, failed, and cancelled jobs, along with cost information. Optionally logs the status to the logger if print_status is True.

```python
def stats(self) -> Dict[str, Any]:
        """Get batch run statistics."""
        stats = {
            "total": self.total_batches,
            "pending": self.total_batches - self.completed_batches,
            "active": self.active_jobs,
            "completed": self.completed_batches,
            "failed": len(self.failed_jobs),
            "cancelled": len(self.cancelled_jobs),
            "cost_usd": self.cost_tracker.total_usd,
            "cost_limit_usd": self.cost_tracker.limit_usd,
            "is_complete": self.is_complete,
            "batches_total": self.total_batches,
            "batches_completed": self.completed_batches,
            "batches_pending": self.total_batches - self.completed_batches,
            "current_batch_index": self.current_batch_index,
            "current_batch_size": self.current_batch_size,
            "items_per_batch": self.config.items_per_batch
        }

        if print_status:
            logger.info("\nBatch Run Status:")
            logger.info(f"  Total jobs: {stats['total']}")
            logger.info(f"  Pending: {stats['pending']}")
            logger.info(f"  Active: {stats['active']}")
            logger.info(f"  Completed: {stats['completed']}")
            logger.info(f"  Failed: {stats['failed']}")
            logger.info(f"  Cancelled: {stats['cancelled']}")
            logger.info(f"  Cost: ${stats['cost_usd']:.6f}")
            if stats['cost_limit_usd']:
                logger.info(f"  Cost limit: ${stats['cost_limit_usd']:.2f}")
            logger.info(f"  Complete: {stats['is_complete']}")
        
        return stats
```

--------------------------------

### Create and Track Batch Jobs

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Creates a batch of jobs with a given provider, tracks its status, and manages active batches. Handles potential errors during batch creation and updates tracking state.

```python
logger.info(f"Creating batch with {len(batch_jobs)} jobs...")
            raw_files_path = str(self.raw_files_dir) if self.raw_files_dir else None
            batch_id, job_mapping = provider.create_batch(batch_jobs, raw_files_path)
            
            # Track active batch for cancellation
            with self._active_batches_lock:
                self._active_batches[batch_id] = provider
            
            with self._state_lock:
                # Remove pending entry if it exists
                pending_keys = [k for k in self.batch_tracking.keys() if k.startswith('pending_')]
                for pending_key in pending_keys:
                    if self.batch_tracking[pending_key]['jobs'] == batch_jobs:
                        del self.batch_tracking[pending_key]
                        break
                
                # Add actual batch tracking
                self.batch_tracking[batch_id] = {
                    'start_time': datetime.now(),
                    'status': 'running',
                    'total': len(batch_jobs),
                    'completed': 0,
                    'cost': 0.0,
                    'estimated_cost': estimated_cost,
                    'provider': provider.__class__.__name__,
                    'jobs': batch_jobs
                }
```

--------------------------------

### Set API Keys for Providers

Source: https://github.com/agamm/batchata/blob/main/development.md

Sets environment variables for API keys required by AI providers. Replace 'your-api-key' with actual keys.

```bash
export ANTHROPIC_API_KEY="your-api-key"
export OPENAI_API_KEY="your-api-key"
```

--------------------------------

### Initialize Batch Run

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Initializes a batch run, creating necessary directories for results and raw files. It also attempts to resume from a saved state if available.

```python
self.results_dir.mkdir(parents=True, exist_ok=True)
        
        # Raw files directory (if enabled)
        self.raw_files_dir = None
        if config.raw_files:
            self.raw_files_dir = self.results_dir / "raw_files"
            self.raw_files_dir.mkdir(parents=True, exist_ok=True)
        
        # Try to resume from saved state
        self._resume_from_state()
```

--------------------------------

### Batch.run

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Creates a BatchRun instance and executes the jobs synchronously.

```APIDOC
## Batch.run

### Description
Creates a BatchRun instance and executes the jobs synchronously.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Method
This appears to be a Python method call, not an HTTP endpoint.

### Endpoint
N/A

### Arguments
- **on_progress** (Optional[Callable]) - Optional progress callback function that receives (stats_dict, elapsed_time_seconds, batch_data).
- **progress_interval** (float) - Interval in seconds between progress updates (default: 1.0).
- **print_status** (bool) - Whether to show rich progress display (default: False).
- **dry_run** (bool) - If True, only show cost estimation without executing (default: False).

### Returns
- **BatchRun** - BatchRun instance with completed results.

### Raises
- **ValueError** - If no jobs have been added.
```

--------------------------------

### Batch Initialization

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Initializes the Batch class with configuration parameters for managing batch jobs.

```APIDOC
## Batch(results_dir: str, max_parallel_batches: int = 10, items_per_batch: int = 10, raw_files: Optional[bool] = None)

### Description
Initialize batch configuration.

### Parameters
#### Path Parameters
- **results_dir** (str) - Required - Directory to store results
- **max_parallel_batches** (int) - Optional - Maximum parallel batch requests (default: 10)
- **items_per_batch** (int) - Optional - Number of jobs per provider batch (default: 10)
- **raw_files** (Optional[bool]) - Optional - Whether to save debug files (raw responses, JSONL files) from providers (default: True if results_dir is set, False otherwise)
```

--------------------------------

### Create and Validate Job in Batchata

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

This snippet shows the process of creating a Job object, validating parameters against a provider, and handling file paths. It includes checks for model presence, parameter validation, temporary file path warnings, and citation compatibility.

```python
if "model" not in params:
            raise ValueError("Model must be provided either in defaults or job parameters")
        
        # Validate parameters
        provider = get_provider(params["model"])
        # Extract params without model to avoid duplicate
        param_subset = {k: v for k, v in params.items() if k != "model"}
        provider.validate_params(params["model"], **param_subset)
        
        # Convert file path if string
        if isinstance(file, str):
            file = Path(file)
        
        # Warn about temporary file paths that may not persist
        if file:
            file_str = str(file)
            if "/tmp/" in file_str or "/var/folders/" in file_str or "temp" in file_str.lower():
                logger = logging.getLogger("batchata")
                logger.debug(f"File path appears to be in a temporary directory: {file}")
                logger.debug("This may cause issues when resuming from state if temp files are cleaned up")
        
        # Create job
        job = Job(
            id=job_id,
            messages=messages,
            file=file,
            prompt=prompt,
            response_model=response_model,
            enable_citations=enable_citations,
            **params
        )
        
        # Validate citation compatibility
        if response_model and enable_citations:
            from ..utils.validation import validate_flat_model
            validate_flat_model(response_model)
        
        # Validate job with provider (includes PDF validation for Anthropic)
        provider.validate_job(job)
        
        self.jobs.append(job)
        return self
```

--------------------------------

### Instructor's System Message Generation

Source: https://github.com/agamm/batchata/blob/main/specs/002-remove-instructor-custom-providers.md

Illustrates the system message format generated by the instructor library, which includes the JSON schema. This is the behavior that the new provider system aims to replicate.

```text
As a genius expert, your task is to understand the content and provide
the parsed objects in json that match the following json_schema:
{json_schema}

Make sure to return an instance of the JSON, not the schema itself
```

--------------------------------

### Execute Batch with Batch.run

Source: https://context7.com/agamm/batchata/llms.txt

Submits all queued jobs to provider APIs and waits for completion. Supports dry runs for cost estimation and live progress display.

```python
from batchata import Batch

batch = (
    Batch(results_dir="./output")
    .set_default_params(model="gemini-2.5-flash")
    .add_cost_limit(usd=5.0)
)
for text in ["doc1.txt", "doc2.txt"]:
    batch.add_job(file=text, prompt="Summarise in one sentence.")

# Dry run – prints estimated costs, does not submit
dry = batch.run(dry_run=True)

# Live run with Rich progress bar
run = batch.run(print_status=True)

# Live run with a custom progress callback
def on_progress(stats, elapsed, batch_data):
    print(f"[{elapsed:.1f}s] {stats['completed']}/{stats['total']} jobs done, ${stats['cost_usd']:.4f} spent")

run = batch.run(on_progress=on_progress, progress_interval=2.0)
```

--------------------------------

### Run Batch Job

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Execute the configured batch job. This method initiates the processing of all added jobs according to the specified configuration.

```python
run = batch.run()
```

--------------------------------

### Batch.run

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Executes the batch process. Allows for progress tracking and status updates.

```APIDOC
## Batch.run

### Description
Execute the batch.

### Parameters
#### Arguments
- **on_progress** (Callable) - A callback function to be called on progress updates. It should accept a dictionary of job status, a float representing the progress percentage, and a dictionary of job metadata.
- **progress_interval** (float) - The interval in seconds between progress updates.
- **print_status** (bool) - If True, prints the status of the batch execution.
- **dry_run** (bool) - If True, performs a dry run without executing the jobs.

### Returns
- **BatchRun** - An object representing the completed batch run.
```

--------------------------------

### Execute Batch Jobs

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

This method is intended to execute a batch of jobs using a given provider. It returns a dictionary containing jobs, costs, and errors.

```python
def _execute_batch(self, provider, batch_jobs: List[Job]) -> Dict:
    """Execute one batch, return results dict with jobs/costs/errors."""
    if not batch_jobs:
        
```

--------------------------------

### Structured Output with Pydantic

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

This snippet shows how to structure output for batch runs, including details about failed jobs, cancelled jobs, total cost, and configuration parameters. It utilizes Pydantic for data modeling.

```python
{
"timestamp": datetime.now().isoformat()
} for job_id, error in self.failed_jobs.items()
],
"cancelled_jobs": [
{
"id": job_id,
"reason": reason,
"timestamp": datetime.now().isoformat()
} for job_id, reason in self.cancelled_jobs.items()
],
"total_cost_usd": self.cost_tracker.used_usd,
"config": {
"state_file": self.config.state_file,
"results_dir": self.config.results_dir,
"max_parallel_batches": self.config.max_parallel_batches,
"items_per_batch": self.config.items_per_batch,
"cost_limit_usd": self.config.cost_limit_usd,
"default_params": self.config.default_params,
"raw_files": self.config.raw_files
}
}
```

--------------------------------

### Set Batch State Persistence

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Configure state persistence for the batch. Specify a file path for the state file and whether to reuse existing state.

```python
batch.set_state(file="./state.json", reuse_state=True)
```

--------------------------------

### Batch.add_cost_limit

Source: https://context7.com/agamm/batchata/llms.txt

Registers a USD spend ceiling. Before each batch is submitted, its estimated cost is compared against the remaining budget. If the estimate exceeds the cap, the batch is skipped, and its jobs are marked as failed with a 'Cost limit exceeded' reason.

```APIDOC
## Batch.add_cost_limit – Enforce a USD Spend Cap

Registers a hard spending ceiling. Before each batch is submitted, its estimated cost is checked against remaining budget. If the estimate would exceed the cap the entire batch is skipped and those jobs appear in `results()["failed"]` with "Cost limit exceeded".

```python
from batchata import Batch

batch = (
    Batch(results_dir="./output")
    .set_default_params(model="claude-3-5-haiku-20241022")
    .add_cost_limit(usd=10.0)    # stop accepting new batches once $10 is committed
)
```
```

--------------------------------

### Batch.run

Source: https://context7.com/agamm/batchata/llms.txt

Executes all queued jobs and waits for completion. Supports dry runs for cost estimation and live progress display.

```APIDOC
## `Batch.run` – Execute the Batch

Submits all jobs to the provider batch APIs and blocks until every job has reached a terminal state. Returns a `BatchRun` instance. Pass `print_status=True` for a live Rich terminal progress display. Pass `dry_run=True` to preview cost estimates without actually submitting.

### Parameters

- **dry_run** (bool) - Optional - If True, previews cost estimates without submitting jobs.
- **print_status** (bool) - Optional - If True, displays a live Rich terminal progress bar.
- **on_progress** (callable) - Optional - A callback function to receive progress updates.
- **progress_interval** (float) - Optional - The interval in seconds for progress updates when `on_progress` is used.

### Returns

- **BatchRun** - An instance representing the completed batch run.

### Example

```python
from batchata import Batch

batch = (
    Batch(results_dir="./output")
    .set_default_params(model="gemini-2.5-flash")
    .add_cost_limit(usd=5.0)
)
for text in ["doc1.txt", "doc2.txt"]:
    batch.add_job(file=text, prompt="Summarise in one sentence.")

# Dry run – prints estimated costs, does not submit
dry = batch.run(dry_run=True)

# Live run with Rich progress bar
run = batch.run(print_status=True)

# Live run with a custom progress callback
def on_progress(stats, elapsed, batch_data):
    print(f"[{elapsed:.1f}s] {stats['completed']}/{stats['total']} jobs done, ${stats['cost_usd']:.4f} spent")

run = batch.run(on_progress=on_progress, progress_interval=2.0)
```
```

--------------------------------

### Dry Run Logging and Cost Estimation

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Logs details about cost limits, estimated costs, and remaining budget during a dry run. Also logs execution plan details like total batches, parallel batches, items per batch, and results directory.

```python
logger.info(f"Cost limit: ${self.config.cost_limit_usd:.2f}")
if total_estimated_cost > self.config.cost_limit_usd:
    excess = total_estimated_cost - self.config.cost_limit_usd
    logger.warning(f"⚠️ Estimated cost exceeds limit by ${excess:.4f}")
else:
    remaining = self.config.cost_limit_usd - total_estimated_cost
    logger.info(f"✅ Within cost limit (${remaining:.4f} remaining)")
else:
    logger.info("No cost limit set")

# Show execution plan
logger.info(f"\n=== EXECUTION PLAN ===")
total_batches = sum(
    len(jobs) // self.config.items_per_batch + (1 if len(jobs) % self.config.items_per_batch else 0)
    for jobs in provider_groups.values()
)
logger.info(f"Total batches to process: {total_batches}")
logger.info(f"Max parallel batches: {self.config.max_parallel_batches}")
logger.info(f"Items per batch: {self.config.items_per_batch}")
logger.info(f"Results directory: {self.config.results_dir}")

logger.info("\n=== DRY RUN COMPLETE ===")
logger.info("To execute for real, call run() without dry_run=True")

return self
```

--------------------------------

### Set Default Job Parameters

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Define default parameters such as model and temperature that will be applied to all jobs in the batch unless overridden. Ensure the model is valid before setting defaults.

```python
.set_default_params(model="claude-sonnet-4-20250514", temperature=0.7)
```

--------------------------------

### Estimate and Reserve Cost for Batch Jobs

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Estimates the cost for a batch of jobs and reserves the cost if within the budget. Logs warnings and returns failed jobs if the cost limit is exceeded.

```python
logger.info(f"Estimating cost for batch of {len(batch_jobs)} jobs...")
        estimated_cost = provider.estimate_cost(batch_jobs)
        remaining = self.cost_tracker.remaining()
        remaining_str = f"${remaining:.4f}" if remaining is not None else "unlimited"
        logger.info(f"Total estimated cost: ${estimated_cost:.4f}, remaining budget: {remaining_str}")
        
        if not self.cost_tracker.reserve_cost(estimated_cost):
            logger.warning(f"Cost limit would be exceeded, skipping batch of {len(batch_jobs)} jobs")
            failed = {}
            for job in batch_jobs:
                failed[job.id] = "Cost limit exceeded"
            return {"results": [], "failed": failed, "cost": 0.0, "jobs_to_remove": list(batch_jobs)}
```

--------------------------------

### set_state

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Configures the state file for persistence and determines whether to resume from an existing state.

```APIDOC
## set_state

### Description
Set state file configuration.

### Method
python

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
- **file** (Optional[str]) - Path to state file for persistence (default: None)
- **reuse_state** (bool) - Whether to resume from existing state file (default: True)

### Returns
- **Batch** - Self for chaining

### Example
```python
batch.set_state(file="./state.json", reuse_state=True)
```
```

--------------------------------

### BatchRun.execute

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Executes a synchronous batch run and waits for its completion. This method initiates the batch processing pipeline.

```APIDOC
## BatchRun.execute

### Description
Execute synchronous batch run and wait for completion.

### Method
```python
def execute(self):
```

### Raises
- RuntimeError: If the batch run has already been started.
```

--------------------------------

### Create and Print a Citation Object

Source: https://context7.com/agamm/batchata/llms.txt

This snippet shows how to instantiate a `Citation` object and print its text, page number, and confidence level. This is useful for understanding the structure of extracted citations.

```python
from batchata.types import Citation

# Typical citation received from a completed job result:
c = Citation(
    text="Total amount due: $4,250.00",
    source="invoice_001.pdf",
    page=2,
    metadata={"block_index": 3},
    confidence="high",
    match_reason="numeric value matched total_amount field"
)
print(c.text, c.page, c.confidence)
```

--------------------------------

### Structured Output with Pydantic and Citations

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Shows how to use Pydantic models for structured output validation and enables citation extraction for Anthropic models. It includes defining a Pydantic model, adding a job with the model, running the batch, and accessing parsed responses and citation mappings.

```python
from batchata import Batch
from pydantic import BaseModel

class DocumentAnalysis(BaseModel):
    title: str
    summary: str
    key_points: list[str]

batch = Batch(results_dir="./results")
    .set_default_params(model="claude-sonnet-4-20250514")

batch.add_job(
    file="document.pdf",
    prompt="Analyze this document",
    response_model=DocumentAnalysis,
    enable_citations=True  # Anthropic only
)

run = batch.run()
for result in run.results()["completed"]:
    analysis = result.parsed_response  # DocumentAnalysis object
    citations = result.citation_mappings  # Field -> Citation mapping
```

--------------------------------

### Unified batch() Function Signature

Source: https://github.com/agamm/batchata/blob/main/specs/008-combine-batch-functions-complete.md

Defines the signature for the unified `batch()` function, which accepts either `messages` or `files` as input. The `prompt` parameter is mandatory when `files` are provided. Ensure only one of `messages` or `files` is supplied.

```python
def batch(
    messages: Optional[List[List[dict]]] = None,
    files: Optional[Union[List[str], List[Path], List[bytes]]] = None,
    model: str,
    prompt: Optional[str] = None,  # Required when files is provided
    response_model: Optional[Type[T]] = None,
    enable_citations: bool = False,
    provider: str = "anthropic",
    max_tokens: int = 1024,
    temperature: float = 0.0,
    verbose: bool = False,
    raw_results_dir: Optional[str] = None
) -> BatchJob:
    """
    Process conversations or files using batch API.
    
    Either messages OR files must be provided, not both.
    When using files, prompt is required.
    """
```

--------------------------------

### Set Default Parameters for Batch

Source: https://github.com/agamm/batchata/blob/main/docs/batchata.html

Configure default parameters such as model, temperature, and max tokens for batch operations. Returns self for chaining.

```python
batch.set_default_params(model="claude-3-sonnet", temperature=0.7)
```