Single Call & Shared Call Pool

Convenience surfaces for running individual calls through the full resilience pipeline — error-type-aware retries, the coordinated rate-limit cooldown, and token accounting — without constructing a ParallelBatchProcessor.

• call / call_result — one prompt, one result. No queue, workers, or result stream are created.
• LLMCallPool — a long-lived, shared entry point for the request path: many concurrent callers, one coordinated cooldown, concurrency bounded by a semaphore.

For large bulk jobs, keep using ParallelBatchProcessor / process_prompts; these surfaces are for single calls, not batches.

Example

from async_batch_llm import LLMCallPool, OpenAIModel, OpenAIStrategy, ProcessorConfig, call, call_result

strategy = OpenAIStrategy(OpenAIModel.from_api_key("gpt-4o-mini"))

# One prompt through the full pipeline — no queue, workers, or result stream.
summary = await call(strategy, "Summarize: ...")        # output, or raises

# A long-lived, shared entry point for a web service's request path.
async with LLMCallPool(
    strategy,
    config=ProcessorConfig(max_workers=5),
    max_pending=100,     # admission cap: reject instantly when saturated
    submit_timeout=30,   # per-caller latency budget (seconds)
) as pool:
    reply = await pool.submit("Answer this one request")

Failure semantics

• call() / LLMCallPool.submit() re-raise the provider's own exception on failure, preserving its type. LLMCallError is raised only when no provider exception was preserved — the pool's max_pending / submit_timeout rejections.
• call_result() / LLMCallPool.submit_result() never raise for a request failure; they return the full WorkItemResult — inspect success, error, token_usage, metadata, and the originating exception.

result = await call_result(strategy, "Summarize: ...")
if not result.success:
    print(result.error, result.exception)   # exception preserves the provider type

The pool drains already-admitted requests on aclose() (the async with exit) before cleaning up the shared strategy, so in-flight calls aren't cut off by shutdown. Set submit_timeout to bound how long shutdown waits for that drain — otherwise it waits as long as the admitted work takes.

Timeout and capacity semantics

The pool semaphore is acquired before item execution, so semaphore wait does not consume attempt_timeout. submit_timeout is the end-to-end caller budget: it includes semaphore wait, provider-capacity admission, coordinated cooldown, retry backoff, and every attempt. If a model built with from_api_key(max_connections=N) advertises less capacity than config.max_workers, pool construction emits a UserWarning and the shared executor gates attempts at N. Use ProcessorConfig(max_provider_concurrency=N) for a custom client whose capacity cannot be inspected.

Avoid wrapping an unbounded input in asyncio.gather(gw.submit(...)): the pool bounds active provider calls, but every gathered coroutine is still a pending task. Bound the outer producer or use process_stream with a bounded max_queue_size for large batch workloads. The bounded-work guide includes both the recommended batch path and a bounded shared-call task window.

LLMCallPool is an in-process, queue-less object. Each caller directly uses the shared ItemExecutor under a semaphore; there is no background dispatcher, network service, or provider-routing layer. The strategy beneath it may use a direct SDK, CallableStrategy, or a real third-party gateway client.

LLMGateway remains an exact, warning-free compatibility alias in v0.20:

from async_batch_llm import LLMCallPool, LLMGateway

assert LLMCallPool is LLMGateway

call

async_batch_llm.call async

call(strategy: LLMCallStrategy[TOutput], prompt: str, *, config: ProcessorConfig | None = None, error_classifier: Any = None) -> TOutput

Run one prompt and return its output, raising on failure.

Set a timeout or retry policy via config (e.g. ProcessorConfig(attempt_timeout=20)).

Source code in src/async_batch_llm/single.py

async def call(
    strategy: LLMCallStrategy[TOutput],
    prompt: str,
    *,
    config: ProcessorConfig | None = None,
    error_classifier: Any = None,
) -> TOutput:
    """Run one prompt and return its output, raising on failure.

    Set a timeout or retry policy via ``config`` (e.g.
    ``ProcessorConfig(attempt_timeout=20)``).
    """
    result = await call_result(strategy, prompt, config=config, error_classifier=error_classifier)
    return unwrap_result(result)

call_result

async_batch_llm.call_result async

call_result(strategy: LLMCallStrategy[TOutput], prompt: str, *, config: ProcessorConfig | None = None, error_classifier: Any = None) -> WorkItemResult[TOutput, Any]

Run one prompt through the full pipeline and return the WorkItemResult.

Inspect .success / .output / .error / .token_usage. Use this when you want token accounting or to branch on failure without exceptions; for the happy-path one-liner use :func:call.

Source code in src/async_batch_llm/single.py

async def call_result(
    strategy: LLMCallStrategy[TOutput],
    prompt: str,
    *,
    config: ProcessorConfig | None = None,
    error_classifier: Any = None,
) -> WorkItemResult[TOutput, Any]:
    """Run one prompt through the full pipeline and return the WorkItemResult.

    Inspect ``.success`` / ``.output`` / ``.error`` / ``.token_usage``. Use this
    when you want token accounting or to branch on failure without exceptions;
    for the happy-path one-liner use :func:`call`.
    """
    host: ExecutorHost[Any, TOutput, Any] = ExecutorHost(
        config or ProcessorConfig(max_workers=1),
        strategy=strategy,
        error_classifier=error_classifier,
    )
    try:
        work_item: LLMWorkItem[Any, TOutput, Any] = LLMWorkItem(
            item_id=_SINGLE_CALL_ITEM_ID, strategy=strategy, prompt=prompt
        )
        return await host.executor.execute(work_item)
    finally:
        await host.aclose()

LLMCallPool

async_batch_llm.LLMCallPool module-attribute

LLMCallPool = LLMGateway

LLMGateway Compatibility Alias

async_batch_llm.LLMGateway

LLMGateway(strategy: LLMCallStrategy[TOutput], *, config: ProcessorConfig | None = None, error_classifier: Any = None, max_pending: int | None = None, submit_timeout: float | None = None)

Bases: Generic[TOutput]

An in-process shared call pool for single LLM calls from many callers.

Create one at startup and call :meth:submit from any number of concurrent handlers. config.max_workers is the global concurrency budget. The strategy is shared, so use a stateless strategy or one whose prepare() yields a reusable resource.

A long rate_limit.cooldown_seconds stalls all callers during a 429 — tune it via config to trade upstream protection against latency.

Two opt-in knobs bound the request path under load (both off by default, so the default is pure backpressure — callers beyond max_workers wait):

• max_pending — an admission cap. With it set, at most max_workers + max_pending requests may be in flight (running or waiting on the semaphore); further submits are rejected instantly with a failed result instead of growing an unbounded waiter list. Bounds memory.
• submit_timeout — a per-caller latency bound (seconds). An admitted request that hasn't completed within the budget (e.g. stuck behind a cooldown) is cancelled and returns a failed result. A firing timeout cancels an in-flight request that might have been about to succeed — the right trade for a web latency budget where the client is already gone.

Source code in src/async_batch_llm/gateway.py

def __init__(
    self,
    strategy: LLMCallStrategy[TOutput],
    *,
    config: ProcessorConfig | None = None,
    error_classifier: Any = None,
    max_pending: int | None = None,
    submit_timeout: float | None = None,
) -> None:
    if max_pending is not None and max_pending < 0:
        raise ValueError(f"max_pending must be >= 0 (got {max_pending})")
    if submit_timeout is not None and submit_timeout <= 0:
        raise ValueError(f"submit_timeout must be > 0 (got {submit_timeout})")

    cfg = config or ProcessorConfig(max_workers=5)
    # Always an int after ProcessorConfig.__post_init__ resolution.
    gateway_workers = cast(int, cfg.max_workers)
    warn_if_worker_capacity_exceeded(
        strategy=strategy,
        max_workers=gateway_workers,
        surface="LLMGateway",
        stacklevel=3,
    )
    self._strategy = strategy
    self._host: ExecutorHost[Any, TOutput, Any] = ExecutorHost(
        cfg, strategy=strategy, error_classifier=error_classifier
    )
    self._sem = asyncio.Semaphore(gateway_workers)
    self._seq = 0
    self._closed = False

    # Admission cap: max requests in flight (running + waiting). None = off.
    self._max_inflight = None if max_pending is None else gateway_workers + max_pending
    self._inflight = 0
    self._submit_timeout = submit_timeout
    # Set whenever no request is in flight; aclose() waits on it to drain
    # already-admitted work before cleaning up the shared strategy.
    self._idle = asyncio.Event()
    self._idle.set()
    # The single drain+cleanup task; concurrent aclose() callers all await it
    # so every `await gw.aclose()` returns only once cleanup is complete.
    self._close_task: asyncio.Task[None] | None = None

aclose async

aclose() -> None

Stop accepting work and run the strategy's cleanup(). Idempotent.

Marks the pool closed (new submits raise immediately), then waits for already-admitted requests — running or still waiting on the semaphore — to drain before cleaning up the shared strategy, whose clients/caches may still be in use. Set submit_timeout to bound how long an admitted request can hold up shutdown (otherwise the drain waits indefinitely).

Concurrent callers share one drain+cleanup task, so every await gw.aclose() returns only once cleanup has actually completed — not just because another call set the closed flag first. The shared task is shield-ed, so cancelling one caller's aclose() doesn't abort the drain/cleanup for the others.

Source code in src/async_batch_llm/gateway.py

async def aclose(self) -> None:
    """Stop accepting work and run the strategy's cleanup(). Idempotent.

    Marks the pool closed (new submits raise immediately), then waits for
    already-admitted requests — running *or* still waiting on the semaphore —
    to drain before cleaning up the shared strategy, whose clients/caches may
    still be in use. Set ``submit_timeout`` to bound how long an admitted
    request can hold up shutdown (otherwise the drain waits indefinitely).

    Concurrent callers share one drain+cleanup task, so *every*
    ``await gw.aclose()`` returns only once cleanup has actually completed —
    not just because another call set the closed flag first. The shared task
    is ``shield``-ed, so cancelling one caller's ``aclose()`` doesn't abort
    the drain/cleanup for the others.
    """
    # Set synchronously so new submits are rejected immediately, even before
    # the cleanup task below is scheduled.
    self._closed = True
    if self._close_task is None:
        self._close_task = asyncio.ensure_future(self._drain_and_close())
    # shield so a cancelled waiter doesn't cancel the shared drain/cleanup.
    await asyncio.shield(self._close_task)

submit async

submit(prompt: str, *, timeout: float | None = None) -> TOutput

Submit one prompt and await its output, raising on failure.

Blocks on the semaphore when the pool is saturated (backpressure), unless max_pending is set (then an over-cap submit raises immediately). A rejected/timed-out submit raises :class:LLMCallError. timeout overrides the pool's submit_timeout for this call.

Source code in src/async_batch_llm/gateway.py

async def submit(self, prompt: str, *, timeout: float | None = None) -> TOutput:
    """Submit one prompt and await its output, raising on failure.

    Blocks on the semaphore when the pool is saturated (backpressure), unless
    ``max_pending`` is set (then an over-cap submit raises immediately). A
    rejected/timed-out submit raises :class:`LLMCallError`. ``timeout``
    overrides the pool's ``submit_timeout`` for this call.
    """
    return unwrap_result(await self.submit_result(prompt, timeout=timeout))

submit_result async

submit_result(prompt: str, *, timeout: float | None = None) -> WorkItemResult[TOutput, Any]

Like :meth:submit but returns the WorkItemResult instead of raising.

Gives access to token_usage / metadata and lets callers branch on .success without exception handling. A request rejected by the admission cap or cut off by the timeout comes back as a failed result (success=False) rather than raising.

timeout overrides the pool's submit_timeout for this call.

Source code in src/async_batch_llm/gateway.py

async def submit_result(
    self, prompt: str, *, timeout: float | None = None
) -> WorkItemResult[TOutput, Any]:
    """Like :meth:`submit` but returns the WorkItemResult instead of raising.

    Gives access to ``token_usage`` / ``metadata`` and lets callers branch
    on ``.success`` without exception handling. A request rejected by the
    admission cap or cut off by the timeout comes back as a failed result
    (``success=False``) rather than raising.

    ``timeout`` overrides the pool's ``submit_timeout`` for this call.
    """
    if self._closed:
        raise RuntimeError("Shared call pool is closed")

    self._seq += 1
    item_id = f"req-{self._seq}"

    # Admission cap. The check + increment below run with no `await` between
    # them, so in asyncio this is race-free and correctly counts waiters
    # toward the bound.
    if self._max_inflight is not None and self._inflight >= self._max_inflight:
        return WorkItemResult(
            item_id=item_id, success=False, error="shared call pool saturated", context=None
        )

    effective_timeout = self._submit_timeout if timeout is None else timeout
    work_item: LLMWorkItem[Any, TOutput, Any] = LLMWorkItem(
        item_id=item_id, strategy=self._strategy, prompt=prompt
    )

    self._inflight += 1
    self._idle.clear()
    try:
        run = self._run(work_item)
        if effective_timeout is None:
            return await run
        # Wrap the whole `async with self._sem: execute(...)` coroutine — NOT
        # `wait_for(sem.acquire())`, which leaks the permit. Cancellation here
        # unwinds through __aexit__, releasing the slot.
        try:
            return await asyncio.wait_for(run, effective_timeout)
        except (TimeoutError, asyncio.TimeoutError):
            return WorkItemResult(
                item_id=item_id, success=False, error="submit timed out", context=None
            )
    finally:
        self._inflight -= 1
        if self._inflight == 0:
            self._idle.set()

LLMCallError

async_batch_llm.LLMCallError

LLMCallError(message: str, *, result: WorkItemResult[Any, Any])

Bases: RuntimeError

Raised by :func:call / LLMCallPool.submit when a request fails and no originating provider exception was preserved.

When the failure carried a provider exception — the usual case for an exhausted-retry or permanent error — that exact exception is re-raised instead, preserving its type. LLMCallError is the fallback for failures that produce no exception: the pool's admission-cap (max_pending) and submit_timeout rejections.

Carries the originating :class:WorkItemResult on .result so callers can still reach token usage, metadata, and the error string.

Source code in src/async_batch_llm/single.py

def __init__(self, message: str, *, result: WorkItemResult[Any, Any]) -> None:
    super().__init__(message)
    self.result = result