async-batch-llm 0.18.0

Provider-neutral asyncio toolkit for concurrent LLM calls with retries, coordinated rate limits, bounded streaming, resumable checkpoints, and deadlines.

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.

Documentation · Getting started · Examples · Changelog

Quick start

Install

Install the provider extra you need. For OpenAI:

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

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:

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 and core API 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.

Production checkpoints and guardrails

The complete production resume example is runnable. The core configuration looks like this once your strategy and prompts are defined:

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 and Deadlines and Fail-Fast 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 and OpenAI-compatible high-throughput guide.

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 and bounded-work guide 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:

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 and core API.

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.

Examples

Start with these runnable examples:

• Production checkpoints and guardrails
• OpenAI batch processing
• Single calls and a shared gateway
• Validation-aware model escalation
• Custom embedding strategies

Browse the complete examples directory for Gemini, DeepSeek, OpenRouter, Anthropic, LangChain, caching, grounding, and benchmark walkthroughs.

Documentation

• Getting Started
• Production Checklist
• Results, Artifacts, and Resume
• Deadlines and Fail-Fast Guardrails
• Bounded Work and Backpressure
• API Reference

Contributing

Clone the repository and use its pinned development environment:

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 or open an issue.

License

MIT License. See LICENSE.