Metadata-Version: 2.3
Name: llm-batch-inference
Version: 0.2.0
Summary: LLM Batch Inference
Author: Daniel Di Giovanni
Author-email: Daniel Di Giovanni <dannyjdigio@gmail.com>
Requires-Dist: anthropic
Requires-Dist: google-genai
Requires-Dist: litellm
Requires-Dist: mistralai>=2.7.0
Requires-Dist: openai
Requires-Dist: python-dotenv>=1.2.2
Requires-Python: >=3.13
Description-Content-Type: text/markdown

# LBI: LLM Batch Inference

Many LLM providers offer discounts (often 50%) for requests submitted through their batch APIs instead of the regular synchronous endpoints. Each provider's batch API looks different, though: different upload formats, different status fields, different ways of matching results back to requests. This project wraps OpenAI, Anthropic, Mistral, and Gemini's batch APIs in one normalized interface, so you can submit a batch, poll for completion, and read back results the same way regardless of provider.

## Features

- **One interface, four providers** - `BaseBatchProvider` gives OpenAI, Anthropic, Mistral, and Gemini the same `create_batch` / `get_batch` / `get_results` / `cancel_batch` / `list_batches` shape.
- **Normalized request/response models** - build a `BatchRequest` once and run it against any provider; results always come back as `BatchResult`, matched to requests by `custom_id`.
- **Multi-target fan-out** - send the same batch to several provider/model combinations at once and collect results independently, with per-target overrides and no single failure blocking the rest (`run_batches_and_wait`).
- **Async-first, with a one-call convenience** - `submit_and_wait` handles submission, polling, and result retrieval in a single await; the individual steps are also available when you need to submit and collect in separate processes.

## Supported providers

| Provider  | Class                   | Backing API                                              |
| --------- | ----------------------- | -------------------------------------------------------- |
| OpenAI    | `OpenAIBatchProvider`   | Batch API (file upload), or Chat Completions when inline |
| Anthropic | `AnthropicBatchProvider`| Message Batches API                                      |
| Mistral   | `MistralBatchProvider`  | Batch API (file upload)                                  |
| Gemini    | `GeminiBatchProvider`   | Batch API (inline requests)                              |

## Installation

```bash
pip install llm-batch-inference
```

or with uv:

```bash
uv add llm-batch-inference
```

The package is imported as `lbi`.

## Configuration

Every provider accepts an explicit `api_key`, and falls back to its standard environment variable if one isn't passed:

| Provider  | Environment variable                   |
| --------- | -------------------------------------- |
| OpenAI    | `OPENAI_API_KEY`                       |
| Anthropic | `ANTHROPIC_API_KEY`                    |
| Mistral   | `MISTRAL_API_KEY`                      |
| Gemini    | `GEMINI_API_KEY` (or `GOOGLE_API_KEY`) |

See `.env.example` for a template.

## Quickstart

`lbi.v1` is the library's current public API.

```python
import asyncio

import lbi.v1 as lbi


async def main() -> None:
    provider = lbi.create_provider('openai', api_key='sk-...')

    requests = [
        lbi.BatchRequest(
            custom_id='1',
            messages=[
                lbi.Message(role=lbi.Role.USER, content='Say hi in French.'),
            ],
        ),
        lbi.BatchRequest(
            custom_id='2',
            messages=[
                lbi.Message(role=lbi.Role.USER, content='Say hi in German.'),
            ],
        ),
    ]

    results = await provider.submit_and_wait(
        requests,
        model='gpt-4o-mini',
        batch_filename='greetings.jsonl',
    )

    for result in results:
        print(result.custom_id, result.status, result.content)


asyncio.run(main())
```

`submit_and_wait` is a convenience wrapper around three steps you can also call individually. If you want to submit a batch in one process and collect it in another later on:

```python
info = await provider.create_batch(
    requests, model='gpt-4o-mini', batch_filename='greetings.jsonl'
)
info = await provider.wait_for_completion(info.batch_id)  # polls until terminal
results = await provider.get_results(info.batch_id)
```

`get_batch`, `cancel_batch`, and `list_batches` are also available for checking on or managing batches.

## Running one batch across multiple providers or models

```python
import lbi.v1 as lbi

targets = [
    lbi.BatchTarget(
        provider=lbi.create_provider('openai', api_key=OPENAI_KEY),
        model='gpt-4o-mini',
    ),
    lbi.BatchTarget(
        provider=lbi.create_provider('anthropic', api_key=ANTHROPIC_KEY),
        model='claude-haiku-4-5-20251001',
    ),
]

results = await lbi.run_batches_and_wait(requests, targets)

for r in results:
    print(r.label, r.status, r.error)
```

A failure on one target (submission, polling, or downloading results) is recorded on that target's `MultiBatchResult` and never stops the others. Call `lbi.pre_validate(requests, targets)` first to catch per-target constraint violations (e.g. a model that rejects a non-default `temperature`) before submitting anything.

Per-target overrides let you tune requests for one target without touching the shared request list:

```python
lbi.BatchTarget(
    provider=lbi.create_provider('anthropic', api_key=ANTHROPIC_KEY),
    model='claude-sonnet-5',
    overrides=lbi.BatchOverrides(temperature=None),  # this model rejects a set temperature
)
```

## Error handling

All exceptions raised by LBI subclass `lbi.BatchLLMError`:

| Exception                | Raised when                                                      |
| ------------------------ | ---------------------------------------------------------------- |
| `BatchCreationError`     | The provider rejected the batch at submission time               |
| `BatchNotFoundError`     | The batch ID is unknown to the provider                          |
| `ResultsNotReadyError`   | Results were requested before the batch reached a terminal state |
| `BatchPollTimeoutError`  | `wait_for_completion` / `submit_and_wait` exceeded its timeout   |
| `BatchCancelledError`    | An operation was attempted on a cancelled batch                  |
| `ProviderError`          | The underlying provider SDK returned an unexpected error         |

## Running Tests

To run the unit tests:

```bash
uv run pytest
```

To run the integration tests:

```bash
uv run pytest -m integration
```

Use the verbose flag `-v` to get more detailed output and use the `-n` flag to run multiple tests in parallel (`-n auto` runs one test per CPU core in parallel).

Example:

```bash
uv run pytest -n auto -v -m integration
```

You can also run specific tests and fixtures by keyword:

```bash
uv run pytest -n auto -v -m integration -k 'full_batch_lifecycle and mistral'
```

## Publishing to PyPI

1. Bump the version in `[pyproject.toml](./pyproject.toml)`:

   ```
   version = "0.2.x"
   ```

2. Build the package:

   ```
   uv build
   ```

3. Publish package:

   ```
   uv publish
   ```

4. Add tag in Git:

   ```
   git add pyproject.toml
   git commit -m "Bump version to 0.1.2"
   git tag v0.1.2
   git push origin main --tags
   ```

## Contact

If you have any questions or feedback, feel free to connect with me on LinkedIn at [linkedin.com/in/daniel-di-giovanni/](https://www.linkedin.com/in/daniel-di-giovanni/) or send me an email at [dannyjdigio@gmail.com](mailto:dannyjdigio@gmail.com).
