LLM Batch Runner

=================

Durable, async batched prompting for LLMs with retries, progress tracking, and resumability via a lightweight SQLite database.

Key features

• Async concurrency with backoff retries
• Durable progress stored in SQLite (resume on rerun)
• Pluggable worker: bring your own function or use a Pydantic AI + OpenRouter worker
• Optional structured outputs via Pydantic response_model
• Choose return shape: unique prompts only or expanded to original input length
• Return results in‑memory or export to JSONL

Requirements

• Python 3.12+

Installation

uv add llm-batch-runner git+https://github.com/m-gaster/llm-batch-runner

Quick start

The simplest way is to rely on environment variables (.env works too):

# .env
MODEL=openai/gpt-4o-mini
OPENROUTER_API_KEY=sk-or-...

Then configure the model-specific rate limit table and run a small script:

import asyncio
from llm_batch_runner.main import prompt_map
from llm_batch_runner.utils import MODEL_RATE_LIMITS

MODEL_RATE_LIMITS["openai/gpt-4o-mini"] = {
    "requests_per_minute": 3_900,
    "requests_per_second": 65,
}

prompts = [
    "Summarize: The quick brown fox jumps over the lazy dog.",
    "Give me 3 bullet points on why the sky appears blue.",
    "Rewrite this in pirate speak: Hello, friend!",
]

async def main():
    results = await prompt_map(prompts, concurrency=16, teardown=True)
    for row in results:
        print(row)

asyncio.run(main())

This will:

• Create (or reuse) a SQLite DB at .llm_batch_cache/runs.db
• Run prompts concurrently with retries
• Print progress and return ordered results
• Remove the progress DB on exit when teardown=True (and optionally the results DB when teardown_results=True)

Other ways to provide a worker

• Direct params (OpenRouter):

results = await prompt_map(
    prompts,
    model_name="openai/gpt-4o-mini",
    openrouter_api_key="sk-or-...",
    rate_limit={
        "requests_per_minute": 3_900,
        "requests_per_second": 65,
    },
)

• Custom async worker:

async def echo_worker(p: str) -> str:
    return p.upper()

results = await prompt_map(
    prompts,
    worker=echo_worker,
    rate_limit={
        "requests_per_minute": 120,
        "requests_per_second": 10,
    },
)

Prefer OpenRouter providers

The built-in worker can prioritize particular OpenRouter providers and control whether OpenRouter may fall back to another provider:

results = await prompt_map(
    prompts,
    model_name="anthropic/claude-sonnet-4",
    openrouter_api_key="sk-or-...",
    providers=["anthropic", "amazon-bedrock"],
    allow_provider_fallbacks=False,
    rate_limit={
        "requests_per_minute": 60,
        "requests_per_second": 1,
    },
)

providers is an ordered preference list of OpenRouter provider slugs. Both options are optional and are only available with the built-in Pydantic AI + OpenRouter worker, not a custom worker. Replace the illustrative rate limit with the limits for your provider account.

Structured outputs

You can ask the built-in Pydantic AI worker to return structured data by passing a Pydantic model class as response_model. The result stored in the DB and returned from prompt_map will be a JSON string matching your schema.

from pydantic import BaseModel

class Bullets(BaseModel):
    points: list[str]

results = await prompt_map(
    prompts,
    model_name="openai/gpt-4o-mini",
    openrouter_api_key="sk-or-...",
    response_model=Bullets,
    rate_limit={
        "requests_per_minute": 3_900,
        "requests_per_second": 65,
    },
)
# each row["result"] is a JSON string for Bullets

Process and validate raw results locally

Use result_processor to validate or normalize each new string result before it is cached as done. The processor runs inside the retry loop and must return the string that should be cached. Cached results are not processed again.

def process_result(raw_result: str) -> str:
    return Bullets.model_validate_json(raw_result).model_dump_json()

results = await prompt_map(
    prompts,
    model_name="openai/gpt-4o-mini",
    result_processor=process_result,
    rate_limit={
        "requests_per_minute": 3_900,
        "requests_per_second": 65,
    },
)

Output shape and exporting

By default, prompt_map deduplicates identical prompts internally. You can control the returned shape via output_shape:

results_orig   = await prompt_map(prompts, output_shape="original")  # default
results_unique = await prompt_map(prompts, output_shape="unique")

The results DB (*-results.db) mirrors the chosen output_shape for that call. With original, duplicate prompts are written as multiple rows (distinguished by their idx).

For lower-level progress DB access, fetch_results(engine, keys) is the legacy API that returns plain dictionaries for done jobs only. Use fetch_job_rows(engine, keys) to get typed rows for all job statuses.

Exporting to JSONL If you prefer a file output, you can export after a run:

from llm_batch_runner.utils import export_jsonl, DB_URL_DEFAULT
import asyncio

asyncio.run(export_jsonl(DB_URL_DEFAULT, out="results.jsonl"))

Tuning

• concurrency: maximum simultaneous jobs (default 128)
• rate_limit: optional per-run override with both requests_per_minute and requests_per_second
• MODEL_RATE_LIMITS: model-to-limit mapping used when rate_limit is omitted
• max_attempts: total attempts per job with exponential backoff (default 3)
• temperature: sampling temperature forwarded to the underlying model/worker (default 0.0 for deterministic responses)
• cache_db_url: override progress DB location, e.g. sqlite+aiosqlite:///my_runs.db
• progress_update_every: print frequency for progress updates (default 200)
• teardown: remove the progress/cache DB on completion (default True)
• teardown_results: also remove the separate results DB on completion (default True)
• output_shape: "original" (default) returns one row per input in original order; "unique" returns one row per unique prompt (ordered by first occurrence). Failed prompts appear with status="failed", result=None, and a populated error in dict/Polars forms.
• return_dtype: one of "list[dict]" (default), "list[str]", "list[tuple[str,str]]", or "polars".

High-throughput example (Gemini Flash 2.5 Lite ~4k rpm)

To approach a provider limit of ~4,000 requests/minute (~66.7 rps):

1. Choose an appropriate concurrency. A good rule of thumb is concurrency ≈ rps_target × p95_latency_seconds. For example, with 1.0s p95 latency, start with concurrency=80–120.

2. Configure the limiter to avoid 429s while saturating throughput:

from llm_batch_runner.utils import MODEL_RATE_LIMITS

MODEL_RATE_LIMITS["google/gemini-flash-2.5-latest"] = {
    "requests_per_minute": 3900,
    "requests_per_second": 65,
}

results = await prompt_map(
    prompts,
    # Ensure your provider/model is set, e.g. via .env:
    # MODEL=google/gemini-flash-2.5-latest
    concurrency=128,           # adjust based on observed latency
    teardown=True,
)

3. Watch the progress logs. The runner prints a rolling RPM estimate alongside status counts. Increase/decrease concurrency to push the rolling RPM close to the target without increasing errors.

Notes:

• Very low latencies require lower concurrency; higher latencies may require 200+.
• If you see many retries due to 429s, reduce the configured per-model limit slightly. If you consistently under-target throughput, increase concurrency or raise the configured limit closer to the provider cap.

Notes

• The library uses SQLAlchemy (async) with a simple jobs table and stores pending|inflight|done|failed states.
• With output_shape="unique", results are ordered by the first occurrence index of each unique prompt. With output_shape="original", results are one-per-input in original order.