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 llm_minder import LLMCallPool, OpenAIModel, OpenAIStrategy, ProcessorConfig, call, call_result

strategy = OpenAIStrategy(OpenAIModel.from_api_key("gpt-6-luna"))

# 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.

With GuardrailConfig(abort_on_error_categories=...), the first request that finishes failed in one of those categories stops the pool. Later submits return batch_aborted results (submit() raises BatchAbortedError) without calling the provider, and pool.termination (provisional) names the triggering item, category and classifier hint. The pool stays stopped until pool.resume() (provisional), which returns the cleared termination. See Configurable fail-fast.

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.

pool.cache_stats() reports prompt-cache reads and writes per cache group, with a rolling hit rate, and pool.reset_cache_stats() clears them. The pool reports these statistics but has no hit-rate floor. pool.cache_tracker is the CacheGroupTracker behind them, for saving, restoring (cache_tracker=) or checking a floor yourself. See Cache hit-rate floor.

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, the pre-v0.20 name, was removed in 1.0; use LLMCallPool, the same class.

call

llm_minder.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/llm_minder/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

llm_minder.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/llm_minder/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`.
    """
    validate_strategy(strategy)
    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
        )
        result = await host.executor.execute(work_item)
    except BaseException as exc:
        # The body exception (or the caller's cancellation) stays primary;
        # cleanup still runs to completion and logs its own failures.
        await host.close_report(primary_exception=exc)
        raise
    # A completed, already-billed result is preserved when the only cleanup
    # failure came from the user strategy's own cleanup(); runtime and
    # admission failures still raise.
    report = await host.close_report()
    report.raise_first(preserve_completed_result=True)
    return result

LLMCallPool

llm_minder.LLMCallPool

LLMCallPool(strategy: LLMCallStrategy[TOutput], *, config: ProcessorConfig | None = None, error_classifier: Any = None, max_pending: int | None = None, submit_timeout: float | None = None, cache_tracker: CacheGroupTracker | 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.

cache_tracker (provisional) is the :class:~llm_minder.CacheGroupTracker behind :meth:cache_stats. By default the pool creates one with config.guardrails.cache_hit_window; pass one restored with CacheGroupTracker.from_dict to continue a previous pool's statistics. Its own window then applies.

config.guardrails.abort_on_error_categories stops the pool on the first request that finishes failed in one of those categories, as it stops a processor run. From then on every submit, including those waiting for a slot, returns a batch_aborted result without calling the provider; calls already running drain or are cancelled per config.guardrails.abort_mode. The pool stays stopped, with :attr:termination saying why, until :meth:resume is called (for example after re-authenticating); a caller that never resumes it has aborted it.

Source code in src/llm_minder/call_pool.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,
    cache_tracker: CacheGroupTracker | None = None,
) -> None:
    if cache_tracker is not None and not isinstance(cache_tracker, CacheGroupTracker):
        raise TypeError(
            f"cache_tracker must be a CacheGroupTracker (got {type(cache_tracker).__name__})"
        )
    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})")

    validate_strategy(strategy)
    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="LLMCallPool",
        stacklevel=3,
    )
    self._strategy = strategy
    self._cache_tracker = (
        cache_tracker
        if cache_tracker is not None
        else CacheGroupTracker(window=cfg.guardrails.cache_hit_window)
    )
    self._abort_categories = cfg.guardrails.abort_on_error_categories
    self._abort: AbortController | None = (
        AbortController(cfg.guardrails.abort_mode) if self._abort_categories else None
    )
    self._host: ExecutorHost[Any, TOutput, Any] = ExecutorHost(
        cfg,
        strategy=strategy,
        error_classifier=error_classifier,
        cache_tracker=self._cache_tracker,
        abort_controller=self._abort,
    )
    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; the close drains it before
    # cleaning up the shared strategy.
    self._idle = asyncio.Event()
    self._idle.set()
    # Concurrent aclose() callers share one ordered close attempt.
    self._closer = SharedCloser(self._cleanup_steps, name="LLMCallPool", logger=logger)

cache_tracker property

cache_tracker: CacheGroupTracker

The pool's live :class:~llm_minder.CacheGroupTracker (provisional).

Save it with pool.cache_tracker.to_dict() and pass CacheGroupTracker.from_dict(saved) as cache_tracker= to a later pool to carry the statistics across a restart, or call pool.cache_tracker.below_floor(...) to apply a floor yourself.

termination property

termination: BatchTermination | None

Why the pool stopped, or None while it is accepting work (provisional).

Set when a request fails in one of config.guardrails.abort_on_error_categories; kind is "fail_fast" and reason names the item, the category and any classifier hint.

aclose async

aclose() -> None

Stop accepting work, drain in-flight requests, then run cleanup().

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. The drain waits indefinitely; set submit_timeout to bound how long an admitted request can hold up shutdown.

Concurrent callers share one ordered close attempt, so every await gw.aclose() returns only once cleanup has actually completed. A successful step is never repeated; a failed one is retried by the next call. Cancelling one waiter defers that cancellation until the shared attempt finishes; cancelling it a second time force-aborts.

Source code in src/llm_minder/call_pool.py

async def aclose(self) -> None:
    """Stop accepting work, drain in-flight requests, then run cleanup().

    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. The drain waits indefinitely; set ``submit_timeout`` to
    bound how long an admitted request can hold up shutdown.

    Concurrent callers share one ordered close attempt, so every
    ``await gw.aclose()`` returns only once cleanup has actually completed.
    A successful step is never repeated; a failed one is retried by the
    next call. Cancelling one waiter defers that cancellation until the
    shared attempt finishes; cancelling it a second time force-aborts.
    """
    # Set synchronously so new submits are rejected immediately, even before
    # the close attempt is scheduled.
    self._closed = True
    report = await self._closer.close()
    report.raise_first()

cache_stats

cache_stats() -> dict[str, Any]

Prompt-cache reads and writes per cache group since creation or the last reset.

Returns {"cache_groups": [...], "cache_groups_overflow": ...} in the shape ParallelBatchProcessor.get_stats() uses. Each group also reports window_hit_rate over its last cache_tracker.window measured calls (None until that many), since a long-lived pool's cumulative hit_rate changes slowly. Statistics only: the pool has no cache hit-rate floor. Call it from the pool's event loop; it doesn't await.

Source code in src/llm_minder/call_pool.py

def cache_stats(self) -> dict[str, Any]:
    """Prompt-cache reads and writes per cache group since creation or the last reset.

    Returns ``{"cache_groups": [...], "cache_groups_overflow": ...}`` in the
    shape ``ParallelBatchProcessor.get_stats()`` uses. Each group also reports
    ``window_hit_rate`` over its last ``cache_tracker.window`` measured calls
    (``None`` until that many), since a long-lived pool's
    cumulative ``hit_rate`` changes slowly. Statistics only: the pool has no
    cache hit-rate floor. Call it from the pool's event loop; it doesn't await.
    """
    return self._cache_tracker.stats()

reset_cache_stats

reset_cache_stats() -> None

Clear the per-group cache statistics, including the 64-group cap's overflow.

A call still running at the reset is counted when it returns.

Source code in src/llm_minder/call_pool.py

def reset_cache_stats(self) -> None:
    """Clear the per-group cache statistics, including the 64-group cap's overflow.

    A call still running at the reset is counted when it returns.
    """
    self._cache_tracker.reset()

resume

resume() -> BatchTermination | None

Accept work again after a fail-fast stop (provisional).

Returns the :attr:termination it clears, or None when the pool wasn't stopped. Later submits call the provider again, and a later failure in a configured category stops the pool again. Calls that started before the resume finish as they would have, but their failures don't stop the resumed pool. Call it from the pool's event loop; it doesn't await.

Source code in src/llm_minder/call_pool.py

def resume(self) -> BatchTermination | None:
    """Accept work again after a fail-fast stop (provisional).

    Returns the :attr:`termination` it clears, or ``None`` when the pool
    wasn't stopped. Later submits call the provider again, and a later failure
    in a configured category stops the pool again. Calls that started before
    the resume finish as they would have, but their failures don't stop the
    resumed pool. Call it from the pool's event loop; it doesn't await.
    """
    if self._abort is None or not self._abort.aborted:
        return None
    cleared = self._abort.termination()
    self._abort = AbortController(self._abort.mode)
    self._host._abort_controller = self._abort
    logger.warning("LLMCallPool resumed after: %s", cleared.reason)
    return cleared

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/llm_minder/call_pool.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/llm_minder/call_pool.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}"

    if self._abort is not None and self._abort.aborted:
        return self._abort.result_for(
            LLMWorkItem(item_id=item_id, strategy=self._strategy, prompt=prompt)
        )

    # 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:
            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

llm_minder.LLMCallError

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

Bases: RuntimeError, LLMMinderError

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/llm_minder/single.py

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