Entry points and configuration

process_prompts

async_batch_llm.process_prompts async

process_prompts(strategy: LLMCallStrategy[TOutput], prompts: PromptSource, *, config: ProcessorConfig | None = None, concurrency: int | None = None, progress: bool | ProgressCallbackFunc = False, preserve_order: bool = False, artifact_store: ArtifactStore | None = None, resume: ResumePolicy = ResumePolicy.NONE, **processor_kwargs: Any) -> BatchResult[TOutput, Any]

Run strategy over prompts and collect every result.

The one-liner entry point — drains :func:process_stream into a :class:BatchResult (whose results are in completion order). For lazy, bounded-input processing of very large inputs, use :func:process_stream directly with a positive config.max_queue_size and config.max_result_queue_size and consume results promptly. This collection API still retains every final result in its returned :class:BatchResult.

concurrency=N is shorthand for ProcessorConfig(concurrency=N) — the single knob that coherently sizes workers, provider admission, and built-in model connection pools (v0.20.0).

progress=True renders a tqdm bar when tqdm is installed (pip install 'async-batch-llm[progress]') and falls back to interval logging otherwise; pass a callable (completed, total, current_item_id) for a custom reporter. Works for both collected and streamed runs (v0.20.0).

Source code in src/async_batch_llm/streaming.py

async def process_prompts(
    strategy: LLMCallStrategy[TOutput],
    prompts: PromptSource,
    *,
    config: ProcessorConfig | None = None,
    concurrency: int | None = None,
    progress: bool | ProgressCallbackFunc = False,
    preserve_order: bool = False,
    artifact_store: ArtifactStore | None = None,
    resume: ResumePolicy = ResumePolicy.NONE,
    **processor_kwargs: Any,
) -> BatchResult[TOutput, Any]:
    """Run ``strategy`` over ``prompts`` and collect every result.

    The one-liner entry point — drains :func:`process_stream` into a
    :class:`BatchResult` (whose ``results`` are in completion order). For
    lazy, bounded-input processing of very large inputs, use
    :func:`process_stream` directly with a positive
    ``config.max_queue_size`` and ``config.max_result_queue_size`` and consume
    results promptly. This collection API still retains every final result in
    its returned :class:`BatchResult`.

    ``concurrency=N`` is shorthand for ``ProcessorConfig(concurrency=N)`` —
    the single knob that coherently sizes workers, provider admission, and
    built-in model connection pools (v0.20.0).

    ``progress=True`` renders a tqdm bar when tqdm is installed
    (``pip install 'async-batch-llm[progress]'``) and falls back to interval
    logging otherwise; pass a callable ``(completed, total, current_item_id)``
    for a custom reporter. Works for both collected and streamed runs
    (v0.20.0).
    """
    termination: list[BatchTermination] = []
    started = time.monotonic()
    results = [
        result
        async for result in _process_stream_impl(
            strategy,
            prompts,
            config=config,
            concurrency=concurrency,
            progress=progress,
            artifact_store=artifact_store,
            resume=resume,
            termination_out=termination,
            **processor_kwargs,
        )
    ]
    seen_ids: set[str] = set()
    for result in results:
        if result.item_id in seen_ids:
            warnings.warn(
                f"Duplicate item ID {result.item_id!r} in process_prompts; "
                "use unique IDs or submission_index to distinguish results.",
                UserWarning,
                stacklevel=2,
            )
            break
        seen_ids.add(result.item_id)
    batch = BatchResult(
        results=results,
        termination=termination[0] if termination else BatchTermination(),
        wall_time_seconds=time.monotonic() - started,
    )
    return batch.in_input_order() if preserve_order else batch

process_stream

async_batch_llm.process_stream async

process_stream(strategy: LLMCallStrategy[TOutput], prompts: PromptSource, *, config: ProcessorConfig | None = None, concurrency: int | None = None, progress: bool | ProgressCallbackFunc = False, artifact_store: ArtifactStore | None = None, resume: ResumePolicy = ResumePolicy.NONE, **processor_kwargs: Any) -> AsyncIterator[WorkItemResult[TOutput, Any]]

Yield terminal results in completion order.

Ordered streaming is intentionally not offered: a slow early item would block later completed results and require a potentially unbounded reorder buffer. Use process_prompts(..., preserve_order=True) when collection is acceptable.

concurrency=N is shorthand for ProcessorConfig(concurrency=N) — the single knob that coherently sizes workers, provider admission, and built-in model connection pools (v0.20.0).

Source code in src/async_batch_llm/streaming.py

async def process_stream(
    strategy: LLMCallStrategy[TOutput],
    prompts: PromptSource,
    *,
    config: ProcessorConfig | None = None,
    concurrency: int | None = None,
    progress: bool | ProgressCallbackFunc = False,
    artifact_store: ArtifactStore | None = None,
    resume: ResumePolicy = ResumePolicy.NONE,
    **processor_kwargs: Any,
) -> AsyncIterator[WorkItemResult[TOutput, Any]]:
    """Yield terminal results in completion order.

    Ordered streaming is intentionally not offered: a slow early item would
    block later completed results and require a potentially unbounded reorder
    buffer. Use ``process_prompts(..., preserve_order=True)`` when collection is
    acceptable.

    ``concurrency=N`` is shorthand for ``ProcessorConfig(concurrency=N)`` —
    the single knob that coherently sizes workers, provider admission, and
    built-in model connection pools (v0.20.0).
    """
    stream = _process_stream_impl(
        strategy,
        prompts,
        config=config,
        concurrency=concurrency,
        progress=progress,
        artifact_store=artifact_store,
        resume=resume,
        **processor_kwargs,
    )
    try:
        async for result in stream:
            yield result
    finally:
        # Explicitly close the nested generator so an early consumer exit
        # completes processor and reporter cleanup before aclose() returns.
        await stream.aclose()

llm

async_batch_llm.llm

llm(spec: str, *, response_parser: None = None, temperature: float | None = None, generation_config: dict[str, Any] | None = None, **model_kwargs: Any) -> ModelStrategy[str]

llm(spec: str, *, response_parser: Callable[[LLMResponse], TOutput], temperature: float | None = None, generation_config: dict[str, Any] | None = None, **model_kwargs: Any) -> ModelStrategy[TOutput]

llm(spec: str, *, response_parser: Callable[[LLMResponse], Any] | None = None, temperature: float | None = None, generation_config: dict[str, Any] | None = None, **model_kwargs: Any) -> ModelStrategy[Any]

Build a ready-to-use strategy from a "provider:model" string.

Example

from async_batch_llm import llm strategy = llm("openai:gpt-6-luna") # reads OPENAI_API_KEY strategy = llm("gemini:gemini-3.5-flash") # reads GOOGLE_API_KEY strategy = llm("deepseek:deepseek-v4-flash", thinking=False, max_connections=150) strategy = llm("openrouter:anthropic/claude-haiku-4-5") strategy = llm( ... "openai-compatible:meta-llama/Llama-3.1-8B-Instruct", ... base_url="http://localhost:8000/v1", ... api_key="EMPTY", # local servers often ignore the key ... )

Parameters:

Name	Type	Description	Default
spec	str	"provider:model" — one of gemini:, openai:, openrouter:, deepseek:, openai-compatible:. The last targets any other OpenAI-compatible server (vLLM, Together, proxies) over Chat Completions and requires base_url=; its api_key falls back to OPENAI_API_KEY. Everything after the first colon is the provider's model id (which may itself contain colons, e.g. "openrouter:qwen/qwen3.8-27b:free").	required
response_parser	Callable[[LLMResponse], Any] | None	Optional function parsing :class:LLMResponse into the strategy's output type; defaults to returning response.text.	None
temperature	float | None	Default sampling temperature, forwarded to the strategy. Pass None to omit the parameter and use the provider default.	None
generation_config	dict[str, Any] | None	Provider-specific config forwarded on every call (see :class:ModelStrategy).	None
**model_kwargs	Any	Forwarded to the model constructor — e.g. api_key, system_instruction, max_connections / json_mode / extra_headers (OpenAI-compatible providers), thinking (DeepSeek), safety_settings (Gemini).	{}

Returns:

Type	Description
ModelStrategy[Any]	The same strategy objects the explicit two-object form builds:
ModelStrategy[Any]	class:GeminiStrategy, :class:OpenAIStrategy,
ModelStrategy[Any]	class:OpenRouterStrategy, or :class:DeepSeekStrategy. For the
ModelStrategy[Any]	OpenAI-compatible providers the model is created via
ModelStrategy[Any]	from_api_key and owns its client, so connections are released by
ModelStrategy[Any]	the framework's normal strategy cleanup.

Raises:

Type	Description
ValueError	The spec has no provider: prefix, the prefix is unknown, no API key can be resolved, or an openai-compatible: spec has no base_url=.
ImportError	The provider's optional dependency is not installed; the message names the exact install extra.

Added in v0.20.0.

Source code in src/async_batch_llm/factory.py

def llm(
    spec: str,
    *,
    response_parser: Callable[[LLMResponse], Any] | None = None,
    temperature: float | None = None,
    generation_config: dict[str, Any] | None = None,
    **model_kwargs: Any,
) -> ModelStrategy[Any]:
    """Build a ready-to-use strategy from a ``"provider:model"`` string.

    Example:
        >>> from async_batch_llm import llm
        >>> strategy = llm("openai:gpt-6-luna")            # reads OPENAI_API_KEY
        >>> strategy = llm("gemini:gemini-3.5-flash")       # reads GOOGLE_API_KEY
        >>> strategy = llm("deepseek:deepseek-v4-flash", thinking=False, max_connections=150)
        >>> strategy = llm("openrouter:anthropic/claude-haiku-4-5")
        >>> strategy = llm(
        ...     "openai-compatible:meta-llama/Llama-3.1-8B-Instruct",
        ...     base_url="http://localhost:8000/v1",
        ...     api_key="EMPTY",  # local servers often ignore the key
        ... )

    Args:
        spec: ``"provider:model"`` — one of ``gemini:``, ``openai:``,
            ``openrouter:``, ``deepseek:``, ``openai-compatible:``. The last
            targets any other OpenAI-compatible server (vLLM, Together,
            proxies) over Chat Completions and requires ``base_url=``; its
            ``api_key`` falls back to ``OPENAI_API_KEY``. Everything after the first colon
            is the provider's model id (which may itself contain colons, e.g.
            ``"openrouter:qwen/qwen3.8-27b:free"``).
        response_parser: Optional function parsing :class:`LLMResponse` into
            the strategy's output type; defaults to returning ``response.text``.
        temperature: Default sampling temperature, forwarded to the strategy.
            Pass ``None`` to omit the parameter and use the provider default.
        generation_config: Provider-specific config forwarded on every call
            (see :class:`ModelStrategy`).
        **model_kwargs: Forwarded to the model constructor — e.g. ``api_key``,
            ``system_instruction``, ``max_connections`` / ``json_mode`` /
            ``extra_headers`` (OpenAI-compatible providers), ``thinking``
            (DeepSeek), ``safety_settings`` (Gemini).

    Returns:
        The same strategy objects the explicit two-object form builds:
        :class:`GeminiStrategy`, :class:`OpenAIStrategy`,
        :class:`OpenRouterStrategy`, or :class:`DeepSeekStrategy`. For the
        OpenAI-compatible providers the model is created via
        ``from_api_key`` and owns its client, so connections are released by
        the framework's normal strategy cleanup.

    Raises:
        ValueError: The spec has no ``provider:`` prefix, the prefix is
            unknown, no API key can be resolved, or an ``openai-compatible:``
            spec has no ``base_url=``.
        ImportError: The provider's optional dependency is not installed;
            the message names the exact install extra.

    Added in v0.20.0.
    """
    provider, sep, model_id = spec.partition(":")
    provider = provider.strip().lower()
    model_id = model_id.strip()
    if not sep or not provider or not model_id:
        raise ValueError(
            f"Invalid model spec {spec!r}: expected 'provider:model', e.g. "
            f"'openai:gpt-6-luna'. Valid provider prefixes: {_valid_prefixes()}."
        )
    if provider not in _PROVIDER_EXTRAS:
        raise ValueError(
            f"Unknown provider prefix {provider!r} in {spec!r}. "
            f"Valid prefixes: {_valid_prefixes()}. For any other provider, "
            "construct a model and strategy explicitly (see the 'custom "
            "strategy' docs)."
        )

    _validate_model_kwargs(provider, model_kwargs)

    strategy_kwargs: dict[str, Any] = {
        "temperature": temperature,
        "generation_config": generation_config,
    }
    strategy: ModelStrategy[Any]
    if provider == "gemini":
        strategy = GeminiStrategy(_build_gemini(model_id, model_kwargs), **strategy_kwargs)
    elif provider == "openai":
        strategy = OpenAIStrategy(
            OpenAIModel.from_api_key(model_id, **model_kwargs), **strategy_kwargs
        )
    elif provider == "openrouter":
        strategy = OpenRouterStrategy(
            OpenRouterModel.from_api_key(model_id, **model_kwargs), **strategy_kwargs
        )
    elif provider == "openai-compatible":
        if not model_kwargs.get("base_url"):
            raise ValueError(
                'llm("openai-compatible:...") requires base_url=, the server\'s '
                "OpenAI-compatible endpoint (e.g. http://localhost:8000/v1)."
            )
        strategy = OpenAIStrategy(
            OpenAICompatibleModel.from_api_key(model_id, **model_kwargs), **strategy_kwargs
        )
    else:  # deepseek — the registry above is exhaustive
        strategy = DeepSeekStrategy(
            DeepSeekModel.from_api_key(model_id, **model_kwargs), **strategy_kwargs
        )

    if response_parser is not None:
        strategy.response_parser = response_parser
    return strategy

call and call_result

One-shot calls are documented with the shared call pool on Single Call & Shared Call Pool.

RetryState

async_batch_llm.RetryState dataclass

RetryState(data: dict[str, Any] = dict())

Mutable state that persists across retry attempts for a single work item.

This allows strategies to maintain state between retry attempts, enabling multi-stage retry patterns where each attempt can build on previous attempts.

Example use cases: - Partial recovery: Save successfully parsed fields from failed attempts - Progressive refinement: Track which parts of output need improvement - Smart retry prompts: Build targeted prompts based on what failed - Model escalation: Track validation errors vs network errors separately

Usage

In strategy's on_error():

if state: state.set('last_error_type', type(exception).name) state.set('validation_failures', state.get('validation_failures', 0) + 1)

In strategy's execute():

if state and state.get('validation_failures', 0) > 1: # Use smarter model or modified prompt pass

Added in v0.3.0.

__contains__

__contains__(key: str) -> bool

Check if key exists in state.

Source code in src/async_batch_llm/base.py

def __contains__(self, key: str) -> bool:
    """Check if key exists in state."""
    return key in self.data

__copy__

__copy__() -> RetryState

Copy application data without sharing framework runtime state.

Source code in src/async_batch_llm/base.py

def __copy__(self) -> "RetryState":
    """Copy application data without sharing framework runtime state."""
    return type(self)(self.data.copy())

__deepcopy__

__deepcopy__(memo: dict[int, Any]) -> RetryState

Deep-copy application data without copying framework runtime state.

Source code in src/async_batch_llm/base.py

def __deepcopy__(self, memo: dict[int, Any]) -> "RetryState":
    """Deep-copy application data without copying framework runtime state."""
    import copy

    duplicate = type(self)()
    memo[id(self)] = duplicate
    duplicate.data = copy.deepcopy(self.data, memo)
    return duplicate

__getstate__

__getstate__() -> dict[str, Any]

Persist only the public application-owned mapping.

Source code in src/async_batch_llm/base.py

def __getstate__(self) -> dict[str, Any]:
    """Persist only the public application-owned mapping."""
    return {"data": self.data}

__repr__

__repr__() -> str

String representation showing all state data.

Source code in src/async_batch_llm/base.py

def __repr__(self) -> str:
    """String representation showing all state data."""
    return f"RetryState({self.data!r})"

__setstate__

__setstate__(state: dict[str, Any]) -> None

Restore the public mapping; framework state is rebuilt lazily.

Source code in src/async_batch_llm/base.py

def __setstate__(self, state: dict[str, Any]) -> None:
    """Restore the public mapping; framework state is rebuilt lazily."""
    self.data = state["data"]

clear

clear() -> None

Clear all state data.

Source code in src/async_batch_llm/base.py

def clear(self) -> None:
    """Clear all state data."""
    self.data.clear()

delete

delete(key: str, raise_if_missing: bool = False) -> None

Delete a value from the state.

Parameters:

Name	Type	Description	Default
key	str	State key to delete	required
raise_if_missing	bool	If True, raise KeyError if key doesn't exist. If False, silently ignore missing keys (default).	False

Raises:

Type	Description
KeyError	If key doesn't exist and raise_if_missing=True

Source code in src/async_batch_llm/base.py

def delete(self, key: str, raise_if_missing: bool = False) -> None:
    """
    Delete a value from the state.

    Args:
        key: State key to delete
        raise_if_missing: If True, raise KeyError if key doesn't exist.
                         If False, silently ignore missing keys (default).

    Raises:
        KeyError: If key doesn't exist and raise_if_missing=True
    """
    if raise_if_missing:
        del self.data[key]
    else:
        self.data.pop(key, None)

get

get(key: str, default: Any = None) -> Any

Get a value from the state.

Parameters:

Name	Type	Description	Default
key	str	State key to retrieve	required
default	Any	Value to return if key doesn't exist	None

Returns:

Type	Description
Any	Value associated with key, or default if not found

Source code in src/async_batch_llm/base.py

def get(self, key: str, default: Any = None) -> Any:
    """
    Get a value from the state.

    Args:
        key: State key to retrieve
        default: Value to return if key doesn't exist

    Returns:
        Value associated with key, or default if not found
    """
    return self.data.get(key, default)

set

set(key: str, value: Any) -> None

Set a value in the state.

Parameters:

Name	Type	Description	Default
key	str	State key to set	required
value	Any	Value to store (can be any JSON-serializable type)	required

Source code in src/async_batch_llm/base.py

def set(self, key: str, value: Any) -> None:
    """
    Set a value in the state.

    Args:
        key: State key to set
        value: Value to store (can be any JSON-serializable type)
    """
    self.data[key] = value

Middleware

async_batch_llm.middleware.Middleware

Bases: ABC, Generic[TInput, TOutput, TContext]

Abstract base class for middleware that intercepts the processing pipeline.

Ordering and failure semantics (onion model):

• before_process runs once per logical item in registration order, inside its total deadline and before artifact lookup or strategy preparation; after_process runs in reverse registration order, so the first-registered middleware wraps outermost.
• on_error runs in registration order and the first non-None result wins (remaining middlewares are not consulted).
• A middleware that raises is logged and skipped — the item continues through the rest of the chain (fail-open). Don't rely on an exception in before_process to block an item; return None instead, which skips it explicitly. Invalid returned items are rejected with MiddlewareContractError; cancellation and process-control exceptions propagate.
• Replacement items may change prompt, context, and strategy, but must preserve the accepted item ID. The framework preserves the submission index and discards stale artifact keys.
• after_process runs once on a newly executed success, before artifact append, and never on replay, filtering, or on_error recovery. Version changes to this hook with ArtifactIdentity.application_version or parser_version when effective inputs do not otherwise change.
• Middleware on_error runs for a terminal execution/preparation failure, not each retry. It is bypassed for filtering, invalid replacement items, controlled deadlines, and batch aborts. Strategy on_error remains an attempt-level hook. Recovery result usage is additive: the framework includes usage consumed by failed provider attempts plus the usage returned by middleware. Do not duplicate failed-attempt usage in the returned result.
• An invalid after_process return value preserves the previous result, without retrying the provider or calling strategy on_error. A warning names the class and returned type once per middleware class per dispatcher.

after_process abstractmethod async

after_process(result: WorkItemResult[TOutput, TContext]) -> WorkItemResult[TOutput, TContext]

Called once after a newly executed success, in reverse registration order (onion-style: the first-registered middleware sees the result last).

Parameters:

Name	Type	Description	Default
result	WorkItemResult[TOutput, TContext]	The result of processing	required

Returns:

Type	Description
WorkItemResult[TOutput, TContext]	Modified result (can add metadata, modify output, etc.)

Source code in src/async_batch_llm/middleware/base.py

@abstractmethod
async def after_process(
    self, result: WorkItemResult[TOutput, TContext]
) -> WorkItemResult[TOutput, TContext]:
    """
    Called once after a newly executed success, in reverse registration order
    (onion-style: the first-registered middleware sees the result last).

    Args:
        result: The result of processing

    Returns:
        Modified result (can add metadata, modify output, etc.)
    """
    pass

before_process abstractmethod async

before_process(work_item: LLMWorkItem[TInput, TOutput, TContext]) -> LLMWorkItem[TInput, TOutput, TContext] | None

Called once per accepted logical item per run, in registration order.

Parameters:

Name	Type	Description	Default
work_item	LLMWorkItem[TInput, TOutput, TContext]	The work item about to be processed	required

Returns:

Type	Description
LLMWorkItem[TInput, TOutput, TContext] | None	Modified work item, or None to skip processing this item
LLMWorkItem[TInput, TOutput, TContext] | None	(the result fails with category middleware_filtered and cannot replay).

Source code in src/async_batch_llm/middleware/base.py

@abstractmethod
async def before_process(
    self, work_item: LLMWorkItem[TInput, TOutput, TContext]
) -> LLMWorkItem[TInput, TOutput, TContext] | None:
    """
    Called once per accepted logical item per run, in registration order.

    Args:
        work_item: The work item about to be processed

    Returns:
        Modified work item, or None to skip processing this item
        (the result fails with category ``middleware_filtered`` and cannot replay).
    """
    pass

on_error abstractmethod async

on_error(work_item: LLMWorkItem[TInput, TOutput, TContext], error: Exception) -> WorkItemResult[TOutput, TContext] | None

Called when an error occurs during processing, in registration order; the first middleware returning non-None handles the error.

Parameters:

Name	Type	Description	Default
work_item	LLMWorkItem[TInput, TOutput, TContext]	The work item that failed	required
error	Exception	The exception that was raised	required

Returns:

Type	Description
WorkItemResult[TOutput, TContext] | None	Custom result to use instead of error, or None to use default error handling

Source code in src/async_batch_llm/middleware/base.py

@abstractmethod
async def on_error(
    self,
    work_item: LLMWorkItem[TInput, TOutput, TContext],
    error: Exception,
) -> WorkItemResult[TOutput, TContext] | None:
    """
    Called when an error occurs during processing, in registration
    order; the first middleware returning non-None handles the error.

    Args:
        work_item: The work item that failed
        error: The exception that was raised

    Returns:
        Custom result to use instead of error, or None to use default error handling
    """
    pass

BaseMiddleware

async_batch_llm.BaseMiddleware

Bases: Middleware[TInput, TOutput, TContext]

Base middleware with no-op implementations.

after_process async

after_process(result: WorkItemResult[TOutput, TContext]) -> WorkItemResult[TOutput, TContext]

Default: pass through unchanged.

Source code in src/async_batch_llm/middleware/base.py

async def after_process(
    self, result: WorkItemResult[TOutput, TContext]
) -> WorkItemResult[TOutput, TContext]:
    """Default: pass through unchanged."""
    return result

before_process async

before_process(work_item: LLMWorkItem[TInput, TOutput, TContext]) -> LLMWorkItem[TInput, TOutput, TContext] | None

Default: pass through unchanged.

Source code in src/async_batch_llm/middleware/base.py

async def before_process(
    self, work_item: LLMWorkItem[TInput, TOutput, TContext]
) -> LLMWorkItem[TInput, TOutput, TContext] | None:
    """Default: pass through unchanged."""
    return work_item

on_error async

on_error(work_item: LLMWorkItem[TInput, TOutput, TContext], error: Exception) -> WorkItemResult[TOutput, TContext] | None

Default: use default error handling.

Source code in src/async_batch_llm/middleware/base.py

async def on_error(
    self,
    work_item: LLMWorkItem[TInput, TOutput, TContext],
    error: Exception,
) -> WorkItemResult[TOutput, TContext] | None:
    """Default: use default error handling."""
    return None

TokenEstimator

async_batch_llm.TokenEstimator

Bases: Protocol

Synchronous or asynchronous token estimate callable.

The explicit processor estimator is invoked after middleware and cooldown, before quota or provider-capacity admission. attempt is ABL's logical attempt number; rate-limit retries therefore reuse it while still receiving a fresh estimate and reservation.

TokenEstimate

async_batch_llm.TokenEstimate dataclass

TokenEstimate(input_tokens: int, output_tokens: int = 0)

Estimated input and expected output tokens for one physical attempt.

total_tokens property

total_tokens: int

Return the total token demand reserved by the quota gate.

CharacterTokenEstimator

async_batch_llm.CharacterTokenEstimator dataclass

CharacterTokenEstimator(characters_per_token: float = 4.0, expected_output_tokens: int = 256, minimum_input_tokens: int = 1)

Explicit, approximate character-count token estimator.

This heuristic is intentionally never enabled automatically. It is useful for conservative local smoothing when a model-specific tokenizer is not available, but it does not claim provider- or model-specific accuracy.

__call__

__call__(prompt: str, *, strategy: LLMCallStrategy, attempt: int, state: RetryState | None) -> TokenEstimate

Estimate tokens without making a provider or network call.

Source code in src/async_batch_llm/token_estimation.py

def __call__(
    self,
    prompt: str,
    *,
    strategy: LLMCallStrategy,
    attempt: int,
    state: RetryState | None,
) -> TokenEstimate:
    """Estimate tokens without making a provider or network call."""
    del strategy, attempt, state
    input_tokens = max(
        self.minimum_input_tokens,
        math.ceil(len(prompt) / float(self.characters_per_token)),
    )
    return TokenEstimate(
        input_tokens=input_tokens,
        output_tokens=self.expected_output_tokens,
    )

ErrorClassifier, the built-in classifiers and the rate-limit strategies are on Errors and Classification.