Metadata-Version: 2.4
Name: async-batch-llm
Version: 0.18.0
Summary: Provider-neutral asyncio toolkit for concurrent LLM calls with retries, coordinated rate limits, bounded streaming, resumable checkpoints, and deadlines.
Project-URL: Homepage, https://github.com/geoff-davis/async-batch-llm
Project-URL: Documentation, https://geoff-davis.github.io/async-batch-llm/
Project-URL: Repository, https://github.com/geoff-davis/async-batch-llm
Project-URL: Issues, https://github.com/geoff-davis/async-batch-llm/issues
Author-email: Geoff Davis <geoff@keksi.ai>
License: MIT
License-File: LICENSE
Keywords: AI,LLM,anthropic,asyncio,batch-processing,checkpointing,concurrency,deadlines,deepseek,fail-fast,gemini,jsonl,langchain,observability,openai,openrouter,orchestration,parallel,prompt-engineering,pydantic,pydantic-ai,rate-limiting,resumable,retry,streaming,structured-output
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Classifier: Topic :: Software Development :: Libraries
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: aiolimiter>=1.2.1
Requires-Dist: pydantic>=2.0.0
Requires-Dist: typing-extensions>=4.4.0
Provides-Extra: all
Requires-Dist: google-genai>=1.49.0; extra == 'all'
Requires-Dist: openai>=1.50.0; extra == 'all'
Requires-Dist: pydantic-ai>=1.0.0; extra == 'all'
Provides-Extra: deepseek
Requires-Dist: openai>=1.50.0; extra == 'deepseek'
Provides-Extra: dev
Requires-Dist: google-genai>=1.49.0; extra == 'dev'
Requires-Dist: mypy>=1.0.0; extra == 'dev'
Requires-Dist: openai>=1.50.0; extra == 'dev'
Requires-Dist: pre-commit>=3.0.0; extra == 'dev'
Requires-Dist: pydantic-ai>=1.0.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.21.0; extra == 'dev'
Requires-Dist: pytest-cov>=4.0.0; extra == 'dev'
Requires-Dist: pytest-timeout>=2.3.0; extra == 'dev'
Requires-Dist: pytest>=7.0.0; extra == 'dev'
Requires-Dist: ruff>=0.1.0; extra == 'dev'
Requires-Dist: ty>=0.0.1a16; extra == 'dev'
Provides-Extra: docs
Requires-Dist: matplotlib>=3.8; extra == 'docs'
Requires-Dist: mkdocs-material>=9.5.0; extra == 'docs'
Requires-Dist: mkdocs>=1.5.0; extra == 'docs'
Requires-Dist: mkdocstrings[python]>=0.24.0; extra == 'docs'
Requires-Dist: pymdown-extensions>=10.0; extra == 'docs'
Provides-Extra: gemini
Requires-Dist: google-genai>=1.49.0; extra == 'gemini'
Provides-Extra: openai
Requires-Dist: openai>=1.50.0; extra == 'openai'
Provides-Extra: openrouter
Requires-Dist: openai>=1.50.0; extra == 'openrouter'
Provides-Extra: pydantic-ai
Requires-Dist: pydantic-ai>=1.0.0; extra == 'pydantic-ai'
Description-Content-Type: text/markdown

# async-batch-llm

**Run independent LLM calls concurrently with production-grade retries,
coordinated rate-limit cooldowns, bounded input buffering, resumable
checkpoints, deadlines, and complete token accounting.**

The execution pipeline is provider-neutral: use the built-in OpenAI-compatible,
Gemini, or PydanticAI strategies, or bring your own strategy. Use it when you
need results during the current workflow; latency-tolerant jobs may be better
suited to a provider's native batch API.

[![PyPI version](https://badge.fury.io/py/async-batch-llm.svg)](https://badge.fury.io/py/async-batch-llm)
[![Python 3.10-3.14](https://img.shields.io/badge/python-3.10--3.14-blue.svg)](https://www.python.org/downloads/)
[![Tests](https://github.com/geoff-davis/async-batch-llm/workflows/Tests/badge.svg)](https://github.com/geoff-davis/async-batch-llm/actions)
[![Coverage](https://raw.githubusercontent.com/geoff-davis/async-batch-llm/python-coverage-comment-action-data/badge.svg)](https://github.com/geoff-davis/async-batch-llm/tree/python-coverage-comment-action-data)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

**[Documentation](https://geoff-davis.github.io/async-batch-llm/)** ·
[Getting started](https://geoff-davis.github.io/async-batch-llm/getting-started/) ·
[Examples](https://github.com/geoff-davis/async-batch-llm/tree/main/examples) ·
[Changelog](https://github.com/geoff-davis/async-batch-llm/blob/main/CHANGELOG.md)

## Quick start

### Install

Install the provider extra you need. For OpenAI:

```bash
pip install 'async-batch-llm[openai]'
```

Other extras are `pydantic-ai`, `gemini`, `openrouter`, `deepseek`, and `all`.
Install only the core package with `pip install async-batch-llm`.

### Run a batch

```python
import asyncio

from async_batch_llm import OpenAIModel, OpenAIStrategy, ProcessorConfig, process_prompts


async def main() -> None:
    strategy = OpenAIStrategy(
        OpenAIModel.from_api_key("gpt-4o-mini")  # reads OPENAI_API_KEY
    )
    batch = await process_prompts(
        strategy,
        ["Summarize document A", "Summarize document B"],
        config=ProcessorConfig(max_workers=10),
    )

    print(f"{batch.succeeded}/{batch.total_items} succeeded")
    for item in batch.results:  # completion order by default
        value = item.output if item.success else item.error
        print(item.item_id, value)


asyncio.run(main())
```

Pass `(item_id, prompt)` pairs to control IDs, or `(item_id, prompt, context)`
triples to carry application data into each result. Collected results remain in
completion order by default. Pass `preserve_order=True`, or call
`batch.in_input_order()`, when stable submission order is required.

For incremental handling, stream results while a bounded work queue applies
backpressure to the producer:

```python
from async_batch_llm import ProcessorConfig, process_stream

config = ProcessorConfig(max_workers=50, max_queue_size=200)

async for item in process_stream(strategy, huge_prompt_source, config=config):
    await save(item)  # completion order
```

`max_queue_size` bounds pending input. The result handoff queue is unbounded, so
consume streamed results promptly. `process_prompts()` retains every result by
design.

### Choose an execution surface

| Need | API |
| --- | --- |
| Collect a finite run | `process_prompts()` |
| Handle results incrementally | `process_stream()` |
| Execute one resilient request | `call()` / `call_result()` |
| Share limits across service requests | `LLMGateway` |
| Customize queueing and lifecycle | `ParallelBatchProcessor` |

Batch, streaming, single-call, and gateway execution share the same retry,
timing, provider-admission, and token-accounting pipeline. See the
[single-call and gateway guide](https://geoff-davis.github.io/async-batch-llm/api/single-gateway/)
and [core API](https://geoff-davis.github.io/async-batch-llm/api/core/) for the
lower-level surfaces.

## What the operational layer adds

| Capability | Behavior |
| --- | --- |
| Error-aware retries | Separate budgets for content/transport failures and rate limits |
| Coordinated cooldowns | One worker's rate limit pauses the shared execution scope |
| Bounded input | Lazy sync or async sources stop producing when the work queue fills |
| Durable resume | Versioned JSONL checkpoints replay only compatible prior results |
| Guardrails | End-to-end item deadlines, batch deadlines, and category-based fail-fast |
| Accounting | Attempt timing and tokens include retries and failed provider calls |
| Observability | Typed lifecycle events, metrics, middleware, and progress callbacks |

### Why not just use `gather`?

A semaphore plus `asyncio.gather()` is enough when all you need is a concurrency
cap. It does not provide coordinated 429 cooldowns, validation-aware retries,
lazy producer backpressure, checkpoint-before-publication durability, or token
accounting for failed attempts. `return_exceptions=True` also leaves application
code to interpret exception objects mixed into the result list.

Use `gather()` for a small script when those operational guarantees do not
matter. Use a provider's native batch API when delayed results are acceptable
and its current pricing or throughput is a better fit.

### A dated benchmark snapshot

In a **June 10, 2026** GSM8K benchmark using a pre-release v0.12-era build:

- Thirty serial calls took 39–65 seconds; bounded worker pools completed them in
  2.1–4.2 seconds on the uncapped providers.
- At equal concurrency over 1,000 prompts, the framework processed 72 items/s
  on DeepSeek and 108 items/s on Gemini 3.1, compared with 58 and 55 items/s for
  the benchmark's semaphore pool.
- The full 1,319-item run exposed retries, model escalations, permanent errors,
  token use, and cost/latency/accuracy tradeoffs in one result model.

These figures are historical evidence, not current provider guarantees. See the
[methodology, model IDs, pricing snapshot, and complete tables](https://geoff-davis.github.io/async-batch-llm/benchmarks/).

## Production checkpoints and guardrails

The complete
[production resume example](https://github.com/geoff-davis/async-batch-llm/blob/main/examples/example_production_resume.py)
is runnable. The core configuration looks like this once your `strategy` and
`prompts` are defined:

```python
from pathlib import Path

from async_batch_llm import (
    AbortMode,
    ArtifactIdentity,
    GuardrailConfig,
    JsonlArtifactStore,
    ProcessorConfig,
    ResumePolicy,
    process_prompts,
)

store = JsonlArtifactStore(
    "runs/invoice-extraction.jsonl",
    identity=ArtifactIdentity(
        provider="openai",
        model="gpt-4o-mini",
        prompt_version="invoice-v4",
        parser_version="invoice-schema-v2",
        application_version="billing-pipeline-v7",
    ),
    fsync=True,
)
config = ProcessorConfig(
    max_workers=20,
    timeout_per_item=30,  # one provider attempt
    guardrails=GuardrailConfig(
        total_timeout_per_item=180,  # admission, waits, calls, and retries
        batch_timeout=3600,
        abort_on_error_categories=frozenset(
            {"authentication", "insufficient_balance"}
        ),
        abort_mode=AbortMode.DRAIN_ACTIVE,
    ),
)

batch = await process_prompts(
    strategy,
    prompts,
    config=config,
    artifact_store=store,
    resume=ResumePolicy.REUSE_SUCCESSES,
    preserve_order=True,
)

if batch.termination.kind != "completed":
    print("controlled stop:", batch.termination)
Path("summary.json").write_text(batch.to_json(), encoding="utf-8")
```

Important operational details:

- Check `batch.termination`. Batch deadlines and configured fail-fast stops
  return completed and collateral terminal results rather than disguising the
  controlled stop as an unexpected exception.
- Each newly executed terminal result is appended and flushed before it is
  returned or streamed. `fsync=True` requests stronger durability; the default
  is flush-only.
- `JsonlArtifactStore` serializes concurrent writes within one process. It does
  not claim cross-process append safety.
- Replay compatibility includes item ID, prompt, participating context, and the
  complete artifact identity—not merely `item_id`.
- `REUSE_SUCCESSES` reruns prior failures. `REUSE_ALL` also replays compatible
  terminal failures.
- Raw prompts and contexts are excluded by default. Outputs and metadata are
  included by default and may contain sensitive application data.
- Historical replay tokens remain on each result for audit, while live
  processor statistics exclude them from current-run consumption.

Read [Results, Artifacts, and Resume](https://geoff-davis.github.io/async-batch-llm/results-and-artifacts/)
and [Deadlines and Fail-Fast Guardrails](https://geoff-davis.github.io/async-batch-llm/guardrails/)
for schema compatibility, privacy controls, abort modes, and deadline details.

## Provider-neutral execution

Built-in strategies cover:

- `OpenAIStrategy`, `OpenRouterStrategy`, and `DeepSeekStrategy` through the
  shared OpenAI-compatible model layer.
- `GeminiStrategy`, including structured response parsing and shared context
  caching.
- `PydanticAIStrategy` for PydanticAI agents and typed output.

Anthropic can be used through PydanticAI or a custom strategy. Other
OpenAI-compatible services can reuse the common model layer; any async provider
can implement `LLMCallStrategy`. Swapping a compatible strategy does not require
changing processor code.

Model identifiers and service limits change independently of this package.
Confirm current provider documentation when choosing a model, connection pool,
or concurrency limit. See the
[custom strategy guide](https://geoff-davis.github.io/async-batch-llm/examples/custom-strategies/)
and [OpenAI-compatible high-throughput guide](https://geoff-davis.github.io/async-batch-llm/openai-high-throughput/).

## Timing, retry, and ordering semantics

- `timeout_per_item` limits one provider execution attempt. It retains its
  original meaning for backward compatibility.
- `GuardrailConfig.total_timeout_per_item` limits the complete logical item,
  including coordinated cooldown, startup ramp, proactive rate limiting,
  provider-capacity admission, calls, retry cooldowns, and backoff.
- `GuardrailConfig.batch_timeout` starts when the processor run starts.
- A fail-fast category triggers only after an item reaches terminal failure;
  retryable intermediate attempts do not abort the batch.
- `AbortMode.DRAIN_ACTIVE` lets an in-progress provider call finish.
  `AbortMode.CANCEL_ACTIVE` cancels unfinished accepted work while preserving
  external caller-cancellation semantics.
- Collected and streamed results remain completion ordered by default.
  `process_prompts(..., preserve_order=True)` orders a collected batch by its
  stable submission index. Streaming intentionally remains completion ordered
  to avoid blocking behind a slow early item and buffering later results.

See the [production checklist](https://geoff-davis.github.io/async-batch-llm/production-checklist/)
and [bounded-work guide](https://geoff-davis.github.io/async-batch-llm/bounded-work/)
for queue, connection-pool, and lifecycle guidance.

## Results, serialization, and accounting

`BatchResult` aggregates input, cached, output, and total tokens across retries,
including usage recovered from failed attempts. Cost remains caller-supplied;
the package does not bundle a provider price table:

```python
cost = batch.estimated_cost(
    input_per_mtok=current_input_rate,
    output_per_mtok=current_output_rate,
    cached_token_rate=current_cache_rate,
)
```

`WorkItemResult` and `BatchResult` support strict, versioned JSON and JSONL
serialization. Unsupported values raise instead of silently falling back to
`repr()`. Dataclasses, Pydantic models, enums, dates, UUIDs, paths, tuples, and
sets serialize to JSON-safe values; use an encoder/decoder pair when typed
reconstruction is required. Exception descriptors never restore arbitrary
classes or tracebacks.

See the [artifact and serialization API](https://geoff-davis.github.io/async-batch-llm/api/artifacts/)
and [core API](https://geoff-davis.github.io/async-batch-llm/api/core/).

## Testing without provider calls

Use the included fake strategies and `MockAgent` to exercise latency, rate
limits, retryable failures, and terminal failures without spending API quota.
The project test suite makes no live provider calls. See the
[testing guide](https://geoff-davis.github.io/async-batch-llm/testing/).

## Examples

Start with these runnable examples:

- [Production checkpoints and guardrails](https://github.com/geoff-davis/async-batch-llm/blob/main/examples/example_production_resume.py)
- [OpenAI batch processing](https://github.com/geoff-davis/async-batch-llm/blob/main/examples/example_openai.py)
- [Single calls and a shared gateway](https://github.com/geoff-davis/async-batch-llm/blob/main/examples/example_gateway.py)
- [Validation-aware model escalation](https://github.com/geoff-davis/async-batch-llm/blob/main/examples/example_smart_model_escalation.py)
- [Custom embedding strategies](https://github.com/geoff-davis/async-batch-llm/blob/main/examples/example_embeddings.py)

Browse the [complete examples directory](https://github.com/geoff-davis/async-batch-llm/tree/main/examples)
for Gemini, DeepSeek, OpenRouter, Anthropic, LangChain, caching, grounding, and
benchmark walkthroughs.

## Documentation

- [Getting Started](https://geoff-davis.github.io/async-batch-llm/getting-started/)
- [Production Checklist](https://geoff-davis.github.io/async-batch-llm/production-checklist/)
- [Results, Artifacts, and Resume](https://geoff-davis.github.io/async-batch-llm/results-and-artifacts/)
- [Deadlines and Fail-Fast Guardrails](https://geoff-davis.github.io/async-batch-llm/guardrails/)
- [Bounded Work and Backpressure](https://geoff-davis.github.io/async-batch-llm/bounded-work/)
- [API Reference](https://geoff-davis.github.io/async-batch-llm/api/core/)

## Contributing

Clone the repository and use its pinned development environment:

```bash
git clone https://github.com/geoff-davis/async-batch-llm.git
cd async-batch-llm
uv sync --all-extras
make ci
```

See the [contributing guide](https://geoff-davis.github.io/async-batch-llm/contributing/)
or open an [issue](https://github.com/geoff-davis/async-batch-llm/issues).

## License

MIT License. See [LICENSE](https://github.com/geoff-davis/async-batch-llm/blob/main/LICENSE).
