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

Installation

pip install llm-batch-inference

or with uv:

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:

See .env.example for a template.

Quickstart

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

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:

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

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:

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:

Running Tests

To run the unit tests:

uv run pytest

To run the integration tests:

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:

uv run pytest -n auto -v -m integration

You can also run specific tests and fixtures by keyword:

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.2.x"
   git tag v0.2.x
   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/ or send me an email at dannyjdigio@gmail.com.