Strategies API Reference

CallableStrategy

Adapter for an existing asynchronous SDK, gateway client, agent, or application service. See Use Your Existing Async Client for callback and retry-state semantics.

async_batch_llm.CallableStrategy

CallableStrategy(invoke: InvokeCallback[TOutput], *, identity: ArtifactIdentity | None = None, error_classifier: ErrorClassifier | None = None, prepare: LifecycleCallback | None = None, cleanup: LifecycleCallback | None = None, on_error: ErrorCallback | None = None, dry_run: DryRunCallback[TOutput] | None = None, max_concurrency: int | None = None, concurrency_scope: object | None = None, quota_scope: object | None = None, token_estimator: TokenEstimator | None = None, request_concurrency: RequestConcurrencyCallback | None = None)

Bases: LLMCallStrategy[TOutput]

Adapt an existing async client or application operation to ABL.

The invocation callback has this canonical shape::

async def invoke(
    prompt: str,
    *,
    attempt: int,
    timeout: float,
    state: RetryState | None,
) -> CallOutcome[TOutput]: ...

attempt is ABL's logical attempt number, timeout is the effective executor-owned attempt budget, and state belongs exclusively to the current work item. This class adapts callbacks to LLMCallStrategy; the existing ItemExecutor remains responsible for retries, timeouts, cooldowns, accounting, and cancellation.

Source code in src/async_batch_llm/callable_strategy.py

def __init__(
    self,
    invoke: InvokeCallback[TOutput],
    *,
    identity: ArtifactIdentity | None = None,
    error_classifier: ErrorClassifier | None = None,
    prepare: LifecycleCallback | None = None,
    cleanup: LifecycleCallback | None = None,
    on_error: ErrorCallback | None = None,
    dry_run: DryRunCallback[TOutput] | None = None,
    max_concurrency: int | None = None,
    concurrency_scope: object | None = None,
    quota_scope: object | None = None,
    token_estimator: TokenEstimator | None = None,
    request_concurrency: RequestConcurrencyCallback | None = None,
) -> None:
    if not callable(invoke):
        raise TypeError("CallableStrategy invoke must be callable")
    if not _is_async_callable(invoke):
        raise TypeError(
            "CallableStrategy invoke must be declared with async def and return CallOutcome"
        )
    if max_concurrency is not None and (
        isinstance(max_concurrency, bool)
        or not isinstance(max_concurrency, int)
        or max_concurrency <= 0
    ):
        raise ValueError("max_concurrency must be a positive integer or None")
    if token_estimator is not None and not callable(token_estimator):
        raise TypeError("token_estimator must be callable or None")

    self._invoke = invoke
    self._artifact_identity = identity
    self._error_classifier = error_classifier
    self._prepare_callback = prepare
    self._cleanup_callback = cleanup
    self._on_error_callback = on_error
    self._dry_run_callback = dry_run
    self._max_concurrency = max_concurrency
    self._concurrency_scope = concurrency_scope
    self._quota_scope = quota_scope
    self._token_estimator = token_estimator
    self._request_concurrency_callback = request_concurrency

artifact_identity property

artifact_identity: ArtifactIdentity | None

Stable artifact identity, or None when the caller omitted one.

quota_scope property

quota_scope: object

Identity whose RPM and coordinated cooldown this call consumes.

Defaults to :attr:concurrency_scope; pass quota_scope= when transport capacity and provider/account quota have different owners.

CallOutcome

async_batch_llm.CallOutcome dataclass

CallOutcome(output: TOutput, token_usage: Mapping[str, int] = dict(), metadata: Mapping[str, Any] | None = None)

Bases: Generic[TOutput]

The output, reported token usage, and metadata from an async operation.

Token usage and metadata are copied and validated when the outcome crosses the strategy boundary. An empty usage mapping means the upstream operation did not report usage; ABL does not estimate missing values.

LLMCallStrategy

async_batch_llm.LLMCallStrategy

Bases: ABC, Generic[TOutput]

Abstract base class for LLM call strategies.

A strategy encapsulates how LLM calls are made, including: - Resource initialization (caches, clients) - Call execution with retries - Resource cleanup

The framework calls: 1. prepare() once per unique strategy instance before its first execution 2. execute() for each attempt (including retries) 3. cleanup() once per prepared strategy when the processor exits or shuts down

concurrency_scope property

concurrency_scope: object

Identity whose capacity is shared by concurrent calls.

max_concurrency property

max_concurrency: int | None

Maximum safe concurrent calls advertised by this strategy.

None means the capacity is unknown. Model-backed strategies forward capacity metadata from their model; custom strategies can override this property when they own a bounded client or transport.

quota_scope property

quota_scope: object

Identity whose RPM, TPM, and coordinated cooldown are shared.

The default follows :attr:concurrency_scope for backward-compatible ownership. Override it when one provider/account quota spans multiple clients or when one shared client serves independent quota budgets. Object identity, not equality, defines sharing.

cleanup async

cleanup() -> None

Clean up resources when the processor exits or shuts down.

Called once per prepared strategy instance, not once per work item.

Use this for: - Closing connections/sessions - Releasing locks - Logging final metrics - Deleting temporary files

Do NOT use this for: - Deleting caches intended for reuse across runs - Destructive cleanup that prevents resource reuse

Note on Caches (v0.2.0): For reusable resources like Gemini caches with TTLs, consider letting them expire naturally to enable cost savings across multiple pipeline runs. See GeminiCachedModel for an example.

Default: no-op

Source code in src/async_batch_llm/llm_strategies.py

async def cleanup(self) -> None:
    """
    Clean up resources when the processor exits or shuts down.

    Called once per prepared strategy instance, not once per work item.

    **Use this for:**
    - Closing connections/sessions
    - Releasing locks
    - Logging final metrics
    - Deleting temporary files

    **Do NOT use this for:**
    - Deleting caches intended for reuse across runs
    - Destructive cleanup that prevents resource reuse

    **Note on Caches (v0.2.0):**
    For reusable resources like Gemini caches with TTLs, consider letting
    them expire naturally to enable cost savings across multiple pipeline
    runs. See `GeminiCachedModel` for an example.

    Default: no-op
    """
    pass

dry_run async

dry_run(prompt: str) -> tuple[TOutput, TokenUsage]

Return mock output for dry-run mode (testing without API calls).

Override this method to provide realistic mock data for testing. Default implementation returns placeholder values that may not match your output type.

Parameters:

Name	Type	Description	Default
prompt	str	The prompt that would have been sent to the LLM	required

Returns:

Type	Description
tuple[TOutput, TokenUsage]	Tuple of (mock_output, mock_token_usage)

Default behavior: - Returns string "[DRY-RUN] Mock output" as output - Returns mock token usage: 100 input, 50 output, 150 total

Source code in src/async_batch_llm/llm_strategies.py

async def dry_run(self, prompt: str) -> tuple[TOutput, TokenUsage]:
    """
    Return mock output for dry-run mode (testing without API calls).

    Override this method to provide realistic mock data for testing.
    Default implementation returns placeholder values that may not match
    your output type.

    Args:
        prompt: The prompt that would have been sent to the LLM

    Returns:
        Tuple of (mock_output, mock_token_usage)

    Default behavior:
    - Returns string "[DRY-RUN] Mock output" as output
    - Returns mock token usage: 100 input, 50 output, 150 total
    """
    mock_output: TOutput = f"[DRY-RUN] Mock output for prompt: {prompt[:50]}..."  # ty:ignore[invalid-assignment]
    mock_tokens: TokenUsage = {
        "input_tokens": 100,
        "output_tokens": 50,
        "total_tokens": 150,
    }
    return mock_output, mock_tokens

estimate_tokens

estimate_tokens(prompt: str, attempt: int, state: RetryState | None) -> TokenEstimate | Awaitable[TokenEstimate] | None

Return a local token estimate for TPM admission, if supported.

The default supplies no estimate. Configure ProcessorConfig with a run-level estimator or override this hook. No provider call should be made solely to estimate tokens.

Source code in src/async_batch_llm/llm_strategies.py

def estimate_tokens(
    self,
    prompt: str,
    attempt: int,
    state: RetryState | None,
) -> TokenEstimate | Awaitable[TokenEstimate] | None:
    """Return a local token estimate for TPM admission, if supported.

    The default supplies no estimate. Configure ``ProcessorConfig`` with a
    run-level estimator or override this hook. No provider call should be
    made solely to estimate tokens.
    """
    del prompt, attempt, state
    return None

execute abstractmethod async

execute(prompt: str, attempt: int, timeout: float, state: RetryState | None = None) -> tuple[TOutput, TokenUsage] | tuple[TOutput, TokenUsage, dict[str, Any] | None]

Execute an LLM call for the given attempt.

Parameters:

Name	Type	Description	Default
prompt	str	The prompt to send to the LLM	required
attempt	int	Which logical retry attempt this is (1, 2, 3, ...). Guarantee: rate-limit errors do NOT advance this number. If a call is throttled (429 / coordinated cooldown), the framework retries the same attempt after the cooldown rather than burning an attempt. So a strategy that escalates models on higher attempts (e.g. attempt >= 2 → smarter model) escalates because the output was bad over max_attempts tries, never merely because the endpoint was busy. Rate-limit retries are bounded separately by RetryConfig.max_rate_limit_retries.	required
timeout	float	Maximum time to wait for response (seconds)	required
state	RetryState | None	Optional retry state that persists across attempts (v0.3.0)	None

Returns:

Type	Description
tuple[TOutput, TokenUsage] | tuple[TOutput, TokenUsage, dict[str, Any] | None]	Either a 2-tuple (output, token_usage) or a 3-tuple
tuple[TOutput, TokenUsage] | tuple[TOutput, TokenUsage, dict[str, Any] | None]	(output, token_usage, metadata). token_usage is a
tuple[TOutput, TokenUsage] | tuple[TOutput, TokenUsage, dict[str, Any] | None]	TokenUsage dict with optional keys input_tokens,
tuple[TOutput, TokenUsage] | tuple[TOutput, TokenUsage, dict[str, Any] | None]	output_tokens, total_tokens, cached_input_tokens.
tuple[TOutput, TokenUsage] | tuple[TOutput, TokenUsage, dict[str, Any] | None]	metadata (v0.10.0) is a provider-specific dict forwarded into
tuple[TOutput, TokenUsage] | tuple[TOutput, TokenUsage, dict[str, Any] | None]	WorkItemResult.metadata — typically finish_reason,
tuple[TOutput, TokenUsage] | tuple[TOutput, TokenUsage, dict[str, Any] | None]	model, provider (OpenRouter), safety_ratings (Gemini);
tuple[TOutput, TokenUsage] | tuple[TOutput, TokenUsage, dict[str, Any] | None]	pass None if you have nothing to surface. The 2-tuple shape is
tuple[TOutput, TokenUsage] | tuple[TOutput, TokenUsage, dict[str, Any] | None]	supported for backward compatibility but will be removed in a
tuple[TOutput, TokenUsage] | tuple[TOutput, TokenUsage, dict[str, Any] | None]	future release; built-in strategies all return the 3-tuple shape.

Raises:

Type	Description
Exception	Any exception propagated to trigger a retry (if retryable) or a permanent failure.

Note (v0.3.0): The state parameter allows strategies to maintain state across retry attempts for multi-stage retry patterns. See RetryState documentation for examples.

Source code in src/async_batch_llm/llm_strategies.py

@abstractmethod
async def execute(
    self, prompt: str, attempt: int, timeout: float, state: "RetryState | None" = None
) -> tuple[TOutput, TokenUsage] | tuple[TOutput, TokenUsage, dict[str, Any] | None]:
    """
    Execute an LLM call for the given attempt.

    Args:
        prompt: The prompt to send to the LLM
        attempt: Which *logical* retry attempt this is (1, 2, 3, ...).
            **Guarantee:** rate-limit errors do NOT advance this number. If a
            call is throttled (429 / coordinated cooldown), the framework
            retries the *same* ``attempt`` after the cooldown rather than
            burning an attempt. So a strategy that escalates models on higher
            attempts (e.g. attempt >= 2 → smarter model) escalates because the
            *output* was bad over ``max_attempts`` tries, never merely because
            the endpoint was busy. Rate-limit retries are bounded separately
            by ``RetryConfig.max_rate_limit_retries``.
        timeout: Maximum time to wait for response (seconds)
        state: Optional retry state that persists across attempts (v0.3.0)

    Returns:
        Either a 2-tuple ``(output, token_usage)`` or a 3-tuple
        ``(output, token_usage, metadata)``. ``token_usage`` is a
        TokenUsage dict with optional keys ``input_tokens``,
        ``output_tokens``, ``total_tokens``, ``cached_input_tokens``.
        ``metadata`` (v0.10.0) is a provider-specific dict forwarded into
        ``WorkItemResult.metadata`` — typically ``finish_reason``,
        ``model``, ``provider`` (OpenRouter), ``safety_ratings`` (Gemini);
        pass ``None`` if you have nothing to surface. The 2-tuple shape is
        supported for backward compatibility but will be removed in a
        future release; built-in strategies all return the 3-tuple shape.

    Raises:
        Exception: Any exception propagated to trigger a retry (if
            retryable) or a permanent failure.

    Note (v0.3.0):
        The state parameter allows strategies to maintain state across retry
        attempts for multi-stage retry patterns. See RetryState documentation
        for examples.
    """
    pass

on_error async

on_error(exception: Exception, attempt: int, state: RetryState | None = None) -> None

Handle errors that occur during execute().

Called by the framework when execute() raises an exception, before deciding whether to retry. This allows strategies to: - Inspect the error type to adjust retry behavior - Store error information for use in next attempt - Modify prompts based on validation errors - Track error patterns across attempts

Parameters:

Name	Type	Description	Default
exception	Exception	The exception that was raised during execute()	required
attempt	int	Which attempt number failed (1, 2, 3, ...)	required
state	RetryState | None	Optional retry state that persists across attempts (v0.3.0)	None

Default: no-op

Example

async def on_error( self, exception: Exception, attempt: int, state: RetryState | None = None ) -> None: if state is not None: # Track validation errors separately from other errors if isinstance(exception, ValidationError): count = state.get('validation_failures', 0) + 1 state.set('validation_failures', count) # Save partial results for recovery if hasattr(exception, 'partial_data'): state.set('partial_data', exception.partial_data)

Source code in src/async_batch_llm/llm_strategies.py

async def on_error(
    self, exception: Exception, attempt: int, state: "RetryState | None" = None
) -> None:
    """
    Handle errors that occur during execute().

    Called by the framework when execute() raises an exception, before
    deciding whether to retry. This allows strategies to:
    - Inspect the error type to adjust retry behavior
    - Store error information for use in next attempt
    - Modify prompts based on validation errors
    - Track error patterns across attempts

    Args:
        exception: The exception that was raised during execute()
        attempt: Which attempt number failed (1, 2, 3, ...)
        state: Optional retry state that persists across attempts (v0.3.0)

    Default: no-op

    Example:
        async def on_error(
            self, exception: Exception, attempt: int, state: RetryState | None = None
        ) -> None:
            if state is not None:
                # Track validation errors separately from other errors
                if isinstance(exception, ValidationError):
                    count = state.get('validation_failures', 0) + 1
                    state.set('validation_failures', count)
                    # Save partial results for recovery
                    if hasattr(exception, 'partial_data'):
                        state.set('partial_data', exception.partial_data)
    """
    pass

prepare async

prepare() -> None

Initialize resources before making any LLM calls.

Called once per unique strategy instance before the first work item using that instance executes. Use this to set up shared caches, clients, etc. Per-item retry state belongs in execute()/on_error() via RetryState.

Default: no-op

Source code in src/async_batch_llm/llm_strategies.py

async def prepare(self) -> None:
    """
    Initialize resources before making any LLM calls.

    Called once per unique strategy instance before the first work item using
    that instance executes. Use this to set up shared caches, clients, etc.
    Per-item retry state belongs in execute()/on_error() via RetryState.

    Default: no-op
    """
    pass

recommended_error_classifier

recommended_error_classifier() -> ErrorClassifier | None

Return the error classifier best suited to this strategy's provider.

The execution host calls this once per strategy identity when the caller didn't pass error_classifier explicitly.

Returns None by default ("no preference"), which lets the framework fall back to :class:DefaultErrorClassifier. Provider strategies (GeminiStrategy, OpenAIStrategy, …) override this to return their matching classifier. An explicit error_classifier= on the processor always wins over this recommendation.

Source code in src/async_batch_llm/llm_strategies.py

def recommended_error_classifier(self) -> "ErrorClassifier | None":
    """Return the error classifier best suited to this strategy's provider.

    The execution host calls this once per strategy identity when the caller
    didn't pass ``error_classifier`` explicitly.

    Returns ``None`` by default ("no preference"), which lets the framework
    fall back to :class:`DefaultErrorClassifier`. Provider strategies
    (``GeminiStrategy``, ``OpenAIStrategy``, …) override this to return their
    matching classifier. An explicit ``error_classifier=`` on the processor
    always wins over this recommendation.
    """
    return None

ModelStrategy

Shared base for the provider-named strategies below; delegates to an LLMModel. Use directly for a custom model you don't want a dedicated subclass for.

async_batch_llm.ModelStrategy

ModelStrategy(model: LLMModel, response_parser: None = None, *, temperature: float | None = 0.0, generation_config: dict[str, Any] | None = None, quota_scope: object | None = None, token_estimator: TokenEstimator | None = None)

ModelStrategy(model: LLMModel, response_parser: Callable[[LLMResponse], TOutput], *, temperature: float | None = 0.0, generation_config: dict[str, Any] | None = None, quota_scope: object | None = None, token_estimator: TokenEstimator | None = None)

ModelStrategy(model: LLMModel, response_parser: Callable[[LLMResponse], TOutput] | None = None, *, temperature: float | None = 0.0, generation_config: dict[str, Any] | None = None, quota_scope: object | None = None, token_estimator: TokenEstimator | None = None)

Bases: LLMCallStrategy[TOutput]

Base strategy for any provider exposed as an :class:LLMModel.

Holds the machinery shared by all model-backed strategies: the model reference, an optional response parser, lifecycle delegation to :class:ManagedLLMModel, and an execute() that calls model.generate(), parses the response, and forwards LLMResponse.metadata as the third tuple element.

The provider-named subclasses (:class:GeminiStrategy, :class:OpenAIStrategy, :class:OpenRouterStrategy) are thin shells over this base — they exist so users can pick the strategy named after the provider they're using. Use this base directly for a custom :class:LLMModel you don't want to name a dedicated subclass for.

Added in v0.10.0 (extracted from the formerly-duplicated provider strategy classes).

Initialize strategy.

Parameters:

Name	Type	Description	Default
model	LLMModel	An LLMModel instance (e.g., GeminiModel, OpenAIModel).	required
response_parser	Callable[[LLMResponse], TOutput] | None	Function to parse LLMResponse into TOutput. Defaults to returning response.text — only valid when TOutput is str. When TOutput is any other type, pass a response_parser (enforced by the @overload signatures).	None
temperature	float | None	Default sampling temperature. Pass None to omit the parameter and use the provider default (e.g. for OpenAI reasoning models that reject an explicit temperature).	0.0
generation_config	dict[str, Any] | None	Provider-specific config forwarded to model.generate(config=...) on every call — e.g. Gemini's response_schema / response_mime_type / tools, or an OpenAI-compatible response_format / max_tokens (merged into the model's extra_body). Lets a built-in strategy carry native structured output or grounding without subclassing execute(). A subclass that overrides execute() and needs a per-attempt config can read self.generation_config and merge.	None
quota_scope	object | None	Optional identity shared by strategies consuming the same provider/account quota. Defaults to the model, matching the existing concurrency scope.	None
token_estimator	TokenEstimator | None	Optional strategy-owned local estimator used when TPM admission is enabled and the processor has no override.	None

Source code in src/async_batch_llm/llm_strategies.py

def __init__(
    self,
    model: "LLMModel",
    response_parser: Callable[[LLMResponse], TOutput] | None = None,
    *,
    temperature: float | None = 0.0,
    generation_config: dict[str, Any] | None = None,
    quota_scope: object | None = None,
    token_estimator: TokenEstimator | None = None,
) -> None:
    """
    Initialize strategy.

    Args:
        model: An LLMModel instance (e.g., GeminiModel, OpenAIModel).
        response_parser: Function to parse LLMResponse into TOutput. Defaults to
            returning ``response.text`` — only valid when TOutput is ``str``.
            When TOutput is any other type, pass a response_parser (enforced
            by the @overload signatures).
        temperature: Default sampling temperature. Pass ``None`` to omit the
            parameter and use the provider default (e.g. for OpenAI
            reasoning models that reject an explicit temperature).
        generation_config: Provider-specific config forwarded to
            ``model.generate(config=...)`` on every call — e.g. Gemini's
            ``response_schema`` / ``response_mime_type`` / ``tools``, or an
            OpenAI-compatible ``response_format`` / ``max_tokens`` (merged
            into the model's ``extra_body``). Lets a built-in strategy carry
            native structured output or grounding without subclassing
            ``execute()``. A subclass that overrides ``execute()`` and needs a
            *per-attempt* config can read ``self.generation_config`` and merge.
        quota_scope: Optional identity shared by strategies consuming the
            same provider/account quota. Defaults to the model, matching
            the existing concurrency scope.
        token_estimator: Optional strategy-owned local estimator used when
            TPM admission is enabled and the processor has no override.
    """
    if token_estimator is not None and not callable(token_estimator):
        raise TypeError("token_estimator must be callable or None")
    self.model = model
    # The overloads restrict the None-parser path to TOutput=str, so the cast
    # below is sound at static-analysis time.
    self.response_parser = response_parser or (lambda response: cast(TOutput, response.text))
    self.temperature = temperature
    self.generation_config = generation_config
    self._quota_scope = quota_scope
    self._token_estimator = token_estimator

concurrency_scope property

concurrency_scope: object

Strategies wrapping the same model share one admission limit.

max_concurrency property

max_concurrency: int | None

Forward optional client/transport capacity advertised by the model.

quota_scope property

quota_scope: object

Strategies wrapping the same model share quota by default.

cleanup async

cleanup() -> None

Delegate to model.cleanup() if the model has a managed lifecycle.

Source code in src/async_batch_llm/llm_strategies.py

async def cleanup(self) -> None:
    """Delegate to model.cleanup() if the model has a managed lifecycle."""
    if isinstance(self.model, ManagedLLMModel):
        await self.model.cleanup()

execute async

execute(prompt: str, attempt: int, timeout: float, state: RetryState | None = None) -> tuple[TOutput, TokenUsage, dict[str, Any] | None]

Execute the LLM call via the model and parse the response.

Parameters:

Name	Type	Description	Default
prompt	str	The prompt to send to the LLM.	required
attempt	int	Which retry attempt this is (1, 2, 3, ...).	required
timeout	float	Maximum time for response (enforced by the framework).	required
state	RetryState | None	Optional retry state for cross-attempt persistence.	None

Returns:

Type	Description
TOutput	3-tuple (parsed_output, token_usage, metadata) where metadata
TokenUsage	is forwarded from LLMResponse.metadata (provider, finish_reason,
dict[str, Any] | None	model, safety_ratings, etc.). Added the metadata slot in v0.10.0; the
tuple[TOutput, TokenUsage, dict[str, Any] | None]	framework still accepts the legacy 2-tuple shape from custom
tuple[TOutput, TokenUsage, dict[str, Any] | None]	strategies via a compat shim.

Source code in src/async_batch_llm/llm_strategies.py

async def execute(
    self, prompt: str, attempt: int, timeout: float, state: RetryState | None = None
) -> tuple[TOutput, TokenUsage, dict[str, Any] | None]:
    """Execute the LLM call via the model and parse the response.

    Args:
        prompt: The prompt to send to the LLM.
        attempt: Which retry attempt this is (1, 2, 3, ...).
        timeout: Maximum time for response (enforced by the framework).
        state: Optional retry state for cross-attempt persistence.

    Returns:
        3-tuple ``(parsed_output, token_usage, metadata)`` where ``metadata``
        is forwarded from ``LLMResponse.metadata`` (provider, finish_reason,
        model, safety_ratings, etc.). Added the metadata slot in v0.10.0; the
        framework still accepts the legacy 2-tuple shape from custom
        strategies via a compat shim.
    """
    # Only pass `config` when set, so a custom LLMModel whose generate()
    # doesn't accept the kwarg keeps working — the default path is identical
    # to before generation_config existed.
    gen_kwargs: dict[str, Any] = {"temperature": self.temperature}
    if self.generation_config is not None:
        gen_kwargs["config"] = self.generation_config
    provider_started = time.perf_counter()
    try:
        llm_response = await self.model.generate(prompt, **gen_kwargs)
    finally:
        if state is not None:
            state.set(
                "_abl_last_provider_seconds",
                max(0.0, time.perf_counter() - provider_started),
            )

    try:
        output = self.response_parser(llm_response)
    except Exception as e:
        # The API already billed for this call even though parsing failed —
        # attach the token usage so the framework can account for it.
        _attach_token_usage(e, llm_response.token_usage)

    return output, llm_response.token_usage, llm_response.metadata

prepare async

prepare() -> None

Delegate to model.prepare() if the model has a managed lifecycle.

Source code in src/async_batch_llm/llm_strategies.py

async def prepare(self) -> None:
    """Delegate to model.prepare() if the model has a managed lifecycle."""
    if isinstance(self.model, ManagedLLMModel):
        await self.model.prepare()

request_concurrency async

request_concurrency(concurrency: int) -> bool

Forward a concurrency request to the model, when supported.

Execution surfaces call this when ProcessorConfig.concurrency is set (v0.20.0) so built-in models can right-size their connection pools. Models without the hook (custom LLMModel implementations, Gemini) simply return False and are left untouched.

Source code in src/async_batch_llm/llm_strategies.py

async def request_concurrency(self, concurrency: int) -> bool:
    """Forward a concurrency request to the model, when supported.

    Execution surfaces call this when ``ProcessorConfig.concurrency`` is
    set (v0.20.0) so built-in models can right-size their connection
    pools. Models without the hook (custom ``LLMModel`` implementations,
    Gemini) simply return False and are left untouched.
    """
    hook = getattr(self.model, "request_concurrency", None)
    if hook is None:
        return False
    result = await hook(concurrency)
    return bool(result)

PydanticAIStrategy

async_batch_llm.PydanticAIStrategy

PydanticAIStrategy(agent: Agent[None, TOutput])

Bases: LLMCallStrategy[TOutput]

Strategy for using PydanticAI agents.

This strategy wraps a PydanticAI agent, providing a clean interface for batch processing. The agent handles all model interaction, validation, and parsing.

Best for: Structured output with Pydantic models, using PydanticAI's features.

Initialize PydanticAI strategy.

Parameters:

Name	Type	Description	Default
agent	Agent[None, TOutput]	Configured PydanticAI agent	required

Source code in src/async_batch_llm/llm_strategies.py

def __init__(self, agent: "Agent[None, TOutput]"):
    """
    Initialize PydanticAI strategy.

    Args:
        agent: Configured PydanticAI agent
    """
    if Agent is Any:
        raise ImportError(
            "pydantic-ai is required for PydanticAIStrategy. "
            "Install with: pip install 'async-batch-llm[pydantic-ai]'"
        )

    self.agent = agent

dry_run async

dry_run(prompt: str) -> tuple[TOutput, TokenUsage]

Return mock output based on agent's result_type for dry-run mode.

Source code in src/async_batch_llm/llm_strategies.py

async def dry_run(self, prompt: str) -> tuple[TOutput, TokenUsage]:
    """Return mock output based on agent's result_type for dry-run mode."""
    # Try to create a mock instance of the expected output type
    try:
        from pydantic import BaseModel

        # pydantic-ai 1.x renamed Agent.result_type -> Agent.output_type
        # (the old attribute no longer exists). Prefer the new name, fall
        # back to the legacy one for 0.x agents.
        result_type = getattr(self.agent, "output_type", None)
        if result_type is None:
            result_type = getattr(self.agent, "result_type", None)

        # If output type is a Pydantic model, try to create an instance
        if isinstance(result_type, type) and issubclass(result_type, BaseModel):
            # Use model_construct to create instance without validation
            # This allows creating instances even with required fields
            mock_output: TOutput = result_type.model_construct()
        else:
            # For non-Pydantic types, use base class default
            return await super().dry_run(prompt)

    except Exception:
        # If anything fails, fall back to base class default
        return await super().dry_run(prompt)

    # Return mock output with realistic token usage
    mock_tokens: TokenUsage = {
        "input_tokens": len(prompt.split()),  # Rough estimate
        "output_tokens": 50,
        "total_tokens": len(prompt.split()) + 50,
    }

    return mock_output, mock_tokens

execute async

execute(prompt: str, attempt: int, timeout: float, state: RetryState | None = None) -> tuple[TOutput, TokenUsage, dict[str, Any] | None]

Execute PydanticAI agent call.

Note: timeout parameter is provided for information but timeout enforcement is handled by the framework wrapping this call in asyncio.wait_for().

Parameters:

Name	Type	Description	Default
prompt	str	The prompt to send to the LLM	required
attempt	int	Which retry attempt this is (1, 2, 3, ...)	required
timeout	float	Maximum time to wait for response (seconds)	required
state	RetryState | None	Optional retry state (v0.3.0, unused by this strategy)	None

Returns:

Type	Description
TOutput	3-tuple (output, token_usage, metadata). PydanticAI's result
TokenUsage	object doesn't expose provider-side metadata uniformly, so
dict[str, Any] | None	metadata is currently always None here. (v0.10.0)

Source code in src/async_batch_llm/llm_strategies.py

async def execute(
    self, prompt: str, attempt: int, timeout: float, state: RetryState | None = None
) -> tuple[TOutput, TokenUsage, dict[str, Any] | None]:
    """Execute PydanticAI agent call.

    Note: timeout parameter is provided for information but timeout enforcement
    is handled by the framework wrapping this call in asyncio.wait_for().

    Args:
        prompt: The prompt to send to the LLM
        attempt: Which retry attempt this is (1, 2, 3, ...)
        timeout: Maximum time to wait for response (seconds)
        state: Optional retry state (v0.3.0, unused by this strategy)

    Returns:
        3-tuple ``(output, token_usage, metadata)``. PydanticAI's result
        object doesn't expose provider-side metadata uniformly, so
        ``metadata`` is currently always ``None`` here. (v0.10.0)
    """
    provider_started = time.perf_counter()
    try:
        result = await self.agent.run(prompt)
    finally:
        if state is not None:
            state.set(
                "_abl_last_provider_seconds",
                max(0.0, time.perf_counter() - provider_started),
            )

    # Extract token usage FIRST (before accessing result.output which may fail validation).
    # pydantic-ai 1.x renamed request_tokens/response_tokens -> input_tokens/output_tokens
    # (the old names still exist but emit DeprecationWarning). Prefer the new
    # names, falling back to the legacy ones so both 0.x and 1.x work cleanly.
    usage = _run_result_usage(result)
    tokens: TokenUsage = {
        "input_tokens": _usage_field(usage, "input_tokens", "request_tokens"),
        "output_tokens": _usage_field(usage, "output_tokens", "response_tokens"),
        "total_tokens": _usage_field(usage, "total_tokens"),
    }

    # Access result.output (may raise validation errors)
    try:
        output = result.output
    except Exception as e:
        # Attach token usage to exception so framework can track it
        _attach_token_usage(e, tokens)

    return output, tokens, None

Structured JSON Parsing

async_batch_llm.pydantic_json_parser

pydantic_json_parser(model_cls: type[TModel], *, recover_trailing_markdown: bool = False) -> Callable[[LLMResponse], TModel]

Build a response_parser that fence-strips then validates with Pydantic.

Returns a function suitable for the response_parser argument of any :class:~async_batch_llm.ModelStrategy subclass (OpenAIStrategy, DeepSeekStrategy, GeminiStrategy, etc.). It runs :func:strip_code_fences over LLMResponse.text before calling model_cls.model_validate_json, so markdown-fenced JSON validates cleanly instead of raising. Set recover_trailing_markdown=True to opt into a conservative fallback for one complete top-level JSON object/array followed only by a recognized closing-fence artifact. It never repairs malformed JSON or discards arbitrary prose/multiple values.

Example

from pydantic import BaseModel from async_batch_llm import DeepSeekModel, DeepSeekStrategy from async_batch_llm.parsing import pydantic_json_parser

class Classification(BaseModel): ... label: str ... confidence: float

model = DeepSeekModel.from_api_key("deepseek-chat", json_mode=True) parser = pydantic_json_parser(Classification, recover_trailing_markdown=True) strategy = DeepSeekStrategy(model, parser)

Validation failures raise pydantic.ValidationError, which the built-in error classifiers treat as retryable (the model may produce valid output on a retry).

Source code in src/async_batch_llm/parsing.py

def pydantic_json_parser(
    model_cls: type[TModel], *, recover_trailing_markdown: bool = False
) -> Callable[[LLMResponse], TModel]:
    """Build a ``response_parser`` that fence-strips then validates with Pydantic.

    Returns a function suitable for the ``response_parser`` argument of any
    :class:`~async_batch_llm.ModelStrategy` subclass (``OpenAIStrategy``,
    ``DeepSeekStrategy``, ``GeminiStrategy``, etc.). It runs
    :func:`strip_code_fences` over ``LLMResponse.text`` before calling
    ``model_cls.model_validate_json``, so markdown-fenced JSON validates
    cleanly instead of raising. Set ``recover_trailing_markdown=True`` to opt
    into a conservative fallback for one complete top-level JSON object/array
    followed only by a recognized closing-fence artifact. It never repairs
    malformed JSON or discards arbitrary prose/multiple values.

    Example:
        >>> from pydantic import BaseModel
        >>> from async_batch_llm import DeepSeekModel, DeepSeekStrategy
        >>> from async_batch_llm.parsing import pydantic_json_parser
        >>>
        >>> class Classification(BaseModel):
        ...     label: str
        ...     confidence: float
        >>>
        >>> model = DeepSeekModel.from_api_key("deepseek-chat", json_mode=True)
        >>> parser = pydantic_json_parser(Classification, recover_trailing_markdown=True)
        >>> strategy = DeepSeekStrategy(model, parser)

    Validation failures raise ``pydantic.ValidationError``, which the built-in
    error classifiers treat as retryable (the model may produce valid output on
    a retry).
    """

    def parser(response: LLMResponse) -> TModel:
        candidate = strip_code_fences(response.text)
        try:
            return model_cls.model_validate_json(candidate)
        except ValidationError:
            if not recover_trailing_markdown:
                raise
            recovered = _recover_trailing_markdown_json(candidate)
            if recovered is None:
                raise

        value, reason = recovered
        output = model_cls.model_validate(value)
        metadata = dict(response.metadata or {})
        metadata.update(
            {
                "structured_output_recovered": True,
                "structured_output_recovery_reason": reason,
                "structured_output_retries_avoided": 1,
            }
        )
        response.metadata = metadata
        logger.debug("Recovered structured output from %s", reason)
        return output

    return parser

async_batch_llm.strip_code_fences

strip_code_fences(text: str) -> str

Strip a leading/trailing markdown code fence from text, if present.

Handles the common shapes models emit around JSON:

• ```json\n{...}\n```
• ```\n{...}\n```
• bare text with no fences (returned stripped, unchanged otherwise)

Only an outer fence is removed; fences in the interior of the payload are left alone. Returns the stripped inner content.

Source code in src/async_batch_llm/parsing.py

def strip_code_fences(text: str) -> str:
    """Strip a leading/trailing markdown code fence from ``text``, if present.

    Handles the common shapes models emit around JSON:

    - ```` ```json\\n{...}\\n``` ````
    - ```` ```\\n{...}\\n``` ````
    - bare text with no fences (returned stripped, unchanged otherwise)

    Only an *outer* fence is removed; fences in the interior of the payload are
    left alone. Returns the stripped inner content.
    """
    stripped = text.strip()
    if not stripped.startswith("```"):
        return stripped

    # Drop the opening fence line (``` or ```json, ```JSON, etc.).
    first_newline = stripped.find("\n")
    if first_newline == -1:
        # Pathological single line like "```json" with no body — nothing to do.
        return stripped
    body = stripped[first_newline + 1 :]

    # Drop the trailing closing fence if there is one.
    rstripped = body.rstrip()
    if rstripped.endswith("```"):
        body = rstripped[: rstripped.rfind("```")]

    return body.strip()

GeminiStrategy

async_batch_llm.GeminiStrategy

GeminiStrategy(model: LLMModel, response_parser: None = None, *, temperature: float | None = 0.0, generation_config: dict[str, Any] | None = None, quota_scope: object | None = None, token_estimator: TokenEstimator | None = None)

GeminiStrategy(model: LLMModel, response_parser: Callable[[LLMResponse], TOutput], *, temperature: float | None = 0.0, generation_config: dict[str, Any] | None = None, quota_scope: object | None = None, token_estimator: TokenEstimator | None = None)

GeminiStrategy(model: LLMModel, response_parser: Callable[[LLMResponse], TOutput] | None = None, *, temperature: float | None = 0.0, generation_config: dict[str, Any] | None = None, quota_scope: object | None = None, token_estimator: TokenEstimator | None = None)

Bases: ModelStrategy[TOutput]

Strategy for calling a Gemini model and parsing the response.

Accepts an LLMModel (e.g., GeminiModel or GeminiCachedModel) and a response parser. The model handles the API call and token extraction; the strategy handles response parsing and lifecycle delegation.

For caching, use GeminiStrategy(model=GeminiCachedModel(...)).

v0.6.0: Accepts LLMModel instead of raw client + model string.

Example

model = GeminiModel("gemini-2.5-flash", client) strategy = GeminiStrategy(model, response_parser=lambda r: r.text)

With caching:

cached_model = GeminiCachedModel("gemini-2.5-flash", client, cached_content=[...]) strategy = GeminiStrategy(cached_model, response_parser=lambda r: r.text)

Source code in src/async_batch_llm/llm_strategies.py

def __init__(
    self,
    model: "LLMModel",
    response_parser: Callable[[LLMResponse], TOutput] | None = None,
    *,
    temperature: float | None = 0.0,
    generation_config: dict[str, Any] | None = None,
    quota_scope: object | None = None,
    token_estimator: TokenEstimator | None = None,
) -> None:
    """
    Initialize strategy.

    Args:
        model: An LLMModel instance (e.g., GeminiModel, OpenAIModel).
        response_parser: Function to parse LLMResponse into TOutput. Defaults to
            returning ``response.text`` — only valid when TOutput is ``str``.
            When TOutput is any other type, pass a response_parser (enforced
            by the @overload signatures).
        temperature: Default sampling temperature. Pass ``None`` to omit the
            parameter and use the provider default (e.g. for OpenAI
            reasoning models that reject an explicit temperature).
        generation_config: Provider-specific config forwarded to
            ``model.generate(config=...)`` on every call — e.g. Gemini's
            ``response_schema`` / ``response_mime_type`` / ``tools``, or an
            OpenAI-compatible ``response_format`` / ``max_tokens`` (merged
            into the model's ``extra_body``). Lets a built-in strategy carry
            native structured output or grounding without subclassing
            ``execute()``. A subclass that overrides ``execute()`` and needs a
            *per-attempt* config can read ``self.generation_config`` and merge.
        quota_scope: Optional identity shared by strategies consuming the
            same provider/account quota. Defaults to the model, matching
            the existing concurrency scope.
        token_estimator: Optional strategy-owned local estimator used when
            TPM admission is enabled and the processor has no override.
    """
    if token_estimator is not None and not callable(token_estimator):
        raise TypeError("token_estimator must be callable or None")
    self.model = model
    # The overloads restrict the None-parser path to TOutput=str, so the cast
    # below is sound at static-analysis time.
    self.response_parser = response_parser or (lambda response: cast(TOutput, response.text))
    self.temperature = temperature
    self.generation_config = generation_config
    self._quota_scope = quota_scope
    self._token_estimator = token_estimator

OpenAIStrategy

async_batch_llm.OpenAIStrategy

OpenAIStrategy(model: LLMModel, response_parser: None = None, *, temperature: float | None = 0.0, generation_config: dict[str, Any] | None = None, quota_scope: object | None = None, token_estimator: TokenEstimator | None = None)

OpenAIStrategy(model: LLMModel, response_parser: Callable[[LLMResponse], TOutput], *, temperature: float | None = 0.0, generation_config: dict[str, Any] | None = None, quota_scope: object | None = None, token_estimator: TokenEstimator | None = None)

OpenAIStrategy(model: LLMModel, response_parser: Callable[[LLMResponse], TOutput] | None = None, *, temperature: float | None = 0.0, generation_config: dict[str, Any] | None = None, quota_scope: object | None = None, token_estimator: TokenEstimator | None = None)

Bases: ModelStrategy[TOutput]

Strategy for calling an OpenAI-compatible model and parsing the response.

Accepts an LLMModel (typically OpenAIModel) and an optional response parser. The model handles the API call and token extraction; the strategy handles response parsing and lifecycle delegation.

Added in v0.9.0.

Example

model = OpenAIModel.from_api_key("gpt-4o-mini", api_key="sk-...") strategy = OpenAIStrategy(model)

Structured output via response_parser:

strategy = OpenAIStrategy( ... model, ... response_parser=lambda r: MyModel.model_validate_json(r.text), ... )

Source code in src/async_batch_llm/llm_strategies.py

def __init__(
    self,
    model: "LLMModel",
    response_parser: Callable[[LLMResponse], TOutput] | None = None,
    *,
    temperature: float | None = 0.0,
    generation_config: dict[str, Any] | None = None,
    quota_scope: object | None = None,
    token_estimator: TokenEstimator | None = None,
) -> None:
    """
    Initialize strategy.

    Args:
        model: An LLMModel instance (e.g., GeminiModel, OpenAIModel).
        response_parser: Function to parse LLMResponse into TOutput. Defaults to
            returning ``response.text`` — only valid when TOutput is ``str``.
            When TOutput is any other type, pass a response_parser (enforced
            by the @overload signatures).
        temperature: Default sampling temperature. Pass ``None`` to omit the
            parameter and use the provider default (e.g. for OpenAI
            reasoning models that reject an explicit temperature).
        generation_config: Provider-specific config forwarded to
            ``model.generate(config=...)`` on every call — e.g. Gemini's
            ``response_schema`` / ``response_mime_type`` / ``tools``, or an
            OpenAI-compatible ``response_format`` / ``max_tokens`` (merged
            into the model's ``extra_body``). Lets a built-in strategy carry
            native structured output or grounding without subclassing
            ``execute()``. A subclass that overrides ``execute()`` and needs a
            *per-attempt* config can read ``self.generation_config`` and merge.
        quota_scope: Optional identity shared by strategies consuming the
            same provider/account quota. Defaults to the model, matching
            the existing concurrency scope.
        token_estimator: Optional strategy-owned local estimator used when
            TPM admission is enabled and the processor has no override.
    """
    if token_estimator is not None and not callable(token_estimator):
        raise TypeError("token_estimator must be callable or None")
    self.model = model
    # The overloads restrict the None-parser path to TOutput=str, so the cast
    # below is sound at static-analysis time.
    self.response_parser = response_parser or (lambda response: cast(TOutput, response.text))
    self.temperature = temperature
    self.generation_config = generation_config
    self._quota_scope = quota_scope
    self._token_estimator = token_estimator

OpenRouterStrategy

async_batch_llm.OpenRouterStrategy

OpenRouterStrategy(model: LLMModel, response_parser: None = None, *, temperature: float | None = 0.0, generation_config: dict[str, Any] | None = None, quota_scope: object | None = None, token_estimator: TokenEstimator | None = None)

OpenRouterStrategy(model: LLMModel, response_parser: Callable[[LLMResponse], TOutput], *, temperature: float | None = 0.0, generation_config: dict[str, Any] | None = None, quota_scope: object | None = None, token_estimator: TokenEstimator | None = None)

OpenRouterStrategy(model: LLMModel, response_parser: Callable[[LLMResponse], TOutput] | None = None, *, temperature: float | None = 0.0, generation_config: dict[str, Any] | None = None, quota_scope: object | None = None, token_estimator: TokenEstimator | None = None)

Bases: ModelStrategy[TOutput]

Strategy for calling an OpenRouter-backed model and parsing the response.

Functionally identical to :class:OpenAIStrategy (both delegate to an LLMModel via :class:ModelStrategy); the separate class exists for provider-named symmetry so users can pick the strategy named after the provider they're using. For OpenRouter, LLMResponse.metadata typically includes provider (the upstream that served the request), model (the actually-routed model), and finish_reason.

Added in v0.9.0.

Example

model = OpenRouterModel.from_api_key( ... "anthropic/claude-haiku-4-5", api_key="sk-or-...", ... ) strategy = OpenRouterStrategy(model)

Source code in src/async_batch_llm/llm_strategies.py

def __init__(
    self,
    model: "LLMModel",
    response_parser: Callable[[LLMResponse], TOutput] | None = None,
    *,
    temperature: float | None = 0.0,
    generation_config: dict[str, Any] | None = None,
    quota_scope: object | None = None,
    token_estimator: TokenEstimator | None = None,
) -> None:
    """
    Initialize strategy.

    Args:
        model: An LLMModel instance (e.g., GeminiModel, OpenAIModel).
        response_parser: Function to parse LLMResponse into TOutput. Defaults to
            returning ``response.text`` — only valid when TOutput is ``str``.
            When TOutput is any other type, pass a response_parser (enforced
            by the @overload signatures).
        temperature: Default sampling temperature. Pass ``None`` to omit the
            parameter and use the provider default (e.g. for OpenAI
            reasoning models that reject an explicit temperature).
        generation_config: Provider-specific config forwarded to
            ``model.generate(config=...)`` on every call — e.g. Gemini's
            ``response_schema`` / ``response_mime_type`` / ``tools``, or an
            OpenAI-compatible ``response_format`` / ``max_tokens`` (merged
            into the model's ``extra_body``). Lets a built-in strategy carry
            native structured output or grounding without subclassing
            ``execute()``. A subclass that overrides ``execute()`` and needs a
            *per-attempt* config can read ``self.generation_config`` and merge.
        quota_scope: Optional identity shared by strategies consuming the
            same provider/account quota. Defaults to the model, matching
            the existing concurrency scope.
        token_estimator: Optional strategy-owned local estimator used when
            TPM admission is enabled and the processor has no override.
    """
    if token_estimator is not None and not callable(token_estimator):
        raise TypeError("token_estimator must be callable or None")
    self.model = model
    # The overloads restrict the None-parser path to TOutput=str, so the cast
    # below is sound at static-analysis time.
    self.response_parser = response_parser or (lambda response: cast(TOutput, response.text))
    self.temperature = temperature
    self.generation_config = generation_config
    self._quota_scope = quota_scope
    self._token_estimator = token_estimator

DeepSeekStrategy

async_batch_llm.DeepSeekStrategy

DeepSeekStrategy(model: LLMModel, response_parser: Callable[[LLMResponse], TOutput] | None = None, *, temperature: float | None = 0.0, generation_config: dict[str, Any] | None = None, quota_scope: object | None = None, token_estimator: TokenEstimator | None = None)

Bases: ModelStrategy[TOutput]

Strategy for calling a DeepSeek model and parsing the response.

Functionally identical to :class:OpenAIStrategy (both delegate to an LLMModel via :class:ModelStrategy); the separate class exists for provider-named symmetry. Pair it with :class:DeepSeekModel, which surfaces DeepSeek's native cache-hit token counts.

Added in v0.10.0.

Example

model = DeepSeekModel.from_api_key("deepseek-chat", api_key="sk-...") strategy = DeepSeekStrategy(model)

Use the model's strict-schema parser when one is configured.

Source code in src/async_batch_llm/llm_strategies.py

def __init__(
    self,
    model: "LLMModel",
    response_parser: Callable[[LLMResponse], TOutput] | None = None,
    *,
    temperature: float | None = 0.0,
    generation_config: dict[str, Any] | None = None,
    quota_scope: object | None = None,
    token_estimator: TokenEstimator | None = None,
) -> None:
    """Use the model's strict-schema parser when one is configured."""
    if response_parser is None and getattr(model, "response_schema", None) is not None:
        parser = getattr(model, "parse_structured_response", None)
        if callable(parser):
            response_parser = parser
    if response_parser is None:

        def default_response_parser(response: LLMResponse) -> TOutput:
            return cast(TOutput, response.text)

        response_parser = default_response_parser
    super().__init__(
        model,
        response_parser,
        temperature=temperature,
        generation_config=generation_config,
        quota_scope=quota_scope,
        token_estimator=token_estimator,
    )

artifact_identity property

artifact_identity: Any

Include DeepSeek API surface/schema in automatic replay identity.

Models

GeminiModel

async_batch_llm.GeminiModel

GeminiModel(model: str, client: Client, *, safety_settings: list[dict[str, Any]] | None = None, system_instruction: str | None = None, metadata_extractors: list[MetadataExtractor] | None = None)

LLM model backed by the Google Gemini API.

Wraps a genai.Client and model name, handling API calls, token extraction, and response normalization. Implements the LLMModel protocol.

Example

client = genai.Client(api_key="...") model = GeminiModel("gemini-2.5-flash", client) response = await model.generate("Hello!") print(response.text, response.input_tokens)

Added in v0.6.0.

Parameters:

Name	Type	Description	Default
model	str	Model name (e.g., "gemini-3.1-flash-lite-preview").	required
client	Client	Initialized genai.Client.	required
safety_settings	list[dict[str, Any]] | None	Default safety settings for all calls.	None
system_instruction	str | None	Default system instruction (overridable per-call).	None
metadata_extractors	list[MetadataExtractor] | None	Optional hooks that contribute extra keys to LLMResponse.metadata (e.g. grounding_metadata_extractor). Merged on top of the built-in safety_ratings/finish_reason; user keys win. Added in v0.15.0.	None

Source code in src/async_batch_llm/models.py

def __init__(
    self,
    model: str,
    client: "genai.Client",
    *,
    safety_settings: list[dict[str, Any]] | None = None,
    system_instruction: str | None = None,
    metadata_extractors: list[MetadataExtractor] | None = None,
):
    """
    Args:
        model: Model name (e.g., "gemini-3.1-flash-lite-preview").
        client: Initialized genai.Client.
        safety_settings: Default safety settings for all calls.
        system_instruction: Default system instruction (overridable per-call).
        metadata_extractors: Optional hooks that contribute extra keys to
            ``LLMResponse.metadata`` (e.g. ``grounding_metadata_extractor``).
            Merged on top of the built-in ``safety_ratings``/``finish_reason``;
            user keys win. Added in v0.15.0.
    """
    if genai is None:
        raise ImportError(
            "google-genai is required for GeminiModel. "
            "Install with: pip install 'async-batch-llm[gemini]'"
        )

    self._model = model
    self._client = client
    self._safety_settings = safety_settings
    self._default_system_instruction = system_instruction
    self._metadata_extractors = metadata_extractors

generate async

generate(prompt: str | list[Any], *, temperature: float | None = 0.0, system_instruction: str | None = None, config: dict[str, Any] | None = None) -> LLMResponse

Generate a response from Gemini.

Parameters:

Name	Type	Description	Default
prompt	str | list[Any]	Text prompt or list of content parts (multimodal).	required
temperature	float | None	Sampling temperature. Pass None to omit it and use the model default.	0.0
system_instruction	str | None	Override default system instruction.	None
config	dict[str, Any] | None	Additional provider-specific config entries.	None

Returns:

Type	Description
LLMResponse	Normalized LLMResponse.

Source code in src/async_batch_llm/models.py

async def generate(
    self,
    prompt: str | list[Any],
    *,
    temperature: float | None = 0.0,
    system_instruction: str | None = None,
    config: dict[str, Any] | None = None,
) -> LLMResponse:
    """Generate a response from Gemini.

    Args:
        prompt: Text prompt or list of content parts (multimodal).
        temperature: Sampling temperature. Pass ``None`` to omit it and
            use the model default.
        system_instruction: Override default system instruction.
        config: Additional provider-specific config entries.

    Returns:
        Normalized LLMResponse.
    """
    # Build config dict
    call_config: dict[str, Any] = {}
    if temperature is not None:
        call_config["temperature"] = temperature

    si = system_instruction or self._default_system_instruction
    if si is not None:
        call_config["system_instruction"] = si

    if self._safety_settings:
        call_config["safety_settings"] = self._safety_settings

    if config:
        call_config.update(config)

    # Make the API call (config is built as a dict; the SDK accepts this at runtime
    # even though the type stubs say GenerateContentConfig)
    response = await self._client.aio.models.generate_content(
        model=self._model,
        contents=prompt,
        config=call_config,  # type: ignore[arg-type]  # ty:ignore[invalid-argument-type]
    )

    # Extract tokens
    tokens = _extract_tokens(response)
    input_tokens, output_tokens, total_tokens, cached_tokens = tokens

    # Extract text (may be None if safety-blocked)
    text = response.text
    if text is None:
        metadata = _extract_metadata(response)
        safety_info = ""
        if metadata and "safety_ratings" in metadata:
            safety_info = f" Safety ratings: {metadata['safety_ratings']}"
        _raise_empty_response(
            f"Empty response from model (likely blocked by safety filter).{safety_info}",
            tokens,
        )

    return LLMResponse(
        text=text,
        input_tokens=input_tokens,
        output_tokens=output_tokens,
        total_tokens=total_tokens,
        cached_input_tokens=cached_tokens,
        metadata=self._build_metadata(response),
        raw=response,
    )

GeminiCachedModel

async_batch_llm.GeminiCachedModel

GeminiCachedModel(model: str, client: Client, cached_content: list[Content], *, cache_ttl_seconds: int = 3600, cache_renewal_buffer_seconds: int = 300, auto_renew: bool = True, cache_tags: dict[str, str] | None = None, safety_settings: list[dict[str, Any]] | None = None, metadata_extractors: list[MetadataExtractor] | None = None)

LLM model backed by Google Gemini with context caching.

Wraps a genai.Client with cache lifecycle management. Implements the ManagedLLMModel protocol: call prepare() before first use, cleanup() when done.

IMPORTANT — share one instance across work items. Create ONE GeminiCachedModel and reuse it across every LLMWorkItem that should share the cached context. Constructing a new instance per item defeats caching entirely and can cost 10× more. The framework calls prepare() exactly once per unique instance, so sharing is the intended lifecycle. See examples/example_llm_strategies.py for the pattern.

This provides 70-90% cost savings when shared correctly.

Example

model = GeminiCachedModel( ... "gemini-2.5-flash", client, ... cached_content=[system_instruction, context_docs], ... ) await model.prepare() # finds or creates cache response = await model.generate("Process this") await model.cleanup() # preserves cache for reuse

Added in v0.6.0.

Parameters:

Name	Type	Description	Default
model	str	Model name (e.g., "gemini-2.5-flash").	required
client	Client	Initialized genai.Client.	required
cached_content	list[Content]	Content to cache (system instructions, documents).	required
cache_ttl_seconds	int	Cache TTL in seconds (default: 3600 = 1 hour).	3600
cache_renewal_buffer_seconds	int	Renew this many seconds before expiry (default: 300 = 5 minutes).	300
auto_renew	bool	Auto-renew expired caches in generate() (default: True).	True
cache_tags	dict[str, str] | None	Tags for precise cache matching. Encoded into the cache's display_name at creation (google-genai CreateCachedContentConfig has no metadata field) and decoded on lookup. Keep tag values short — Gemini's display_name has a 128-character limit.	None
safety_settings	list[dict[str, Any]] | None	Default safety settings for all calls.	None
metadata_extractors	list[MetadataExtractor] | None	Optional hooks that contribute extra keys to LLMResponse.metadata (e.g. grounding_metadata_extractor). Merged on top of the built-in metadata; user keys win. Added in v0.15.0.	None

Source code in src/async_batch_llm/models.py

def __init__(
    self,
    model: str,
    client: "genai.Client",
    cached_content: list["Content"],
    *,
    cache_ttl_seconds: int = 3600,
    cache_renewal_buffer_seconds: int = 300,
    auto_renew: bool = True,
    cache_tags: dict[str, str] | None = None,
    safety_settings: list[dict[str, Any]] | None = None,
    metadata_extractors: list[MetadataExtractor] | None = None,
):
    """
    Args:
        model: Model name (e.g., "gemini-2.5-flash").
        client: Initialized genai.Client.
        cached_content: Content to cache (system instructions, documents).
        cache_ttl_seconds: Cache TTL in seconds (default: 3600 = 1 hour).
        cache_renewal_buffer_seconds: Renew this many seconds before expiry
            (default: 300 = 5 minutes).
        auto_renew: Auto-renew expired caches in generate() (default: True).
        cache_tags: Tags for precise cache matching. Encoded into the cache's
            ``display_name`` at creation (google-genai ``CreateCachedContentConfig``
            has no ``metadata`` field) and decoded on lookup. Keep tag values
            short — Gemini's ``display_name`` has a 128-character limit.
        safety_settings: Default safety settings for all calls.
        metadata_extractors: Optional hooks that contribute extra keys to
            ``LLMResponse.metadata`` (e.g. ``grounding_metadata_extractor``).
            Merged on top of the built-in metadata; user keys win.
            Added in v0.15.0.
    """
    if genai is None:
        raise ImportError(
            "google-genai is required for GeminiCachedModel. "
            "Install with: pip install 'async-batch-llm[gemini]'"
        )

    if cache_renewal_buffer_seconds >= cache_ttl_seconds:
        raise ValueError(
            f"cache_renewal_buffer_seconds ({cache_renewal_buffer_seconds}) "
            f"must be less than cache_ttl_seconds ({cache_ttl_seconds})."
        )

    if 10 <= cache_ttl_seconds < 60:
        import warnings

        warnings.warn(
            f"cache_ttl_seconds ({cache_ttl_seconds}) is less than 60 seconds. "
            f"Very short TTLs defeat the purpose of caching. "
            f"Recommended minimum: 300 seconds (5 minutes).",
            UserWarning,
            stacklevel=2,
        )

    if cache_renewal_buffer_seconds < 60:
        import warnings

        warnings.warn(
            f"cache_renewal_buffer_seconds ({cache_renewal_buffer_seconds}) is less than "
            f"60 seconds. Small buffers risk renewing on every call if generation takes "
            f"longer than the buffer. Recommended minimum: 60 seconds.",
            UserWarning,
            stacklevel=2,
        )

    self._model = model
    self._client = client
    self._cached_content = cached_content
    self._cache_ttl_seconds = cache_ttl_seconds
    self._cache_renewal_buffer_seconds = cache_renewal_buffer_seconds
    self._auto_renew = auto_renew
    self._cache_tags = cache_tags or {}
    self._safety_settings = safety_settings
    self._metadata_extractors = metadata_extractors

    self._cache: Any = None
    self._cache_created_at: float | None = None
    self._cache_lock: Any = None
    self._prepared = False

cache_name property

cache_name: str | None

The name of the active cache, or None.

cleanup async

cleanup() -> None

Preserve cache for reuse (does not delete). Idempotent.

Source code in src/async_batch_llm/models.py

async def cleanup(self) -> None:
    """Preserve cache for reuse (does not delete). Idempotent."""
    if self._cache:
        logger.info(
            f"Leaving cache active for reuse: {self._cache.name} "
            f"(TTL: {self._cache_ttl_seconds}s, will expire naturally)"
        )

delete_cache async

delete_cache() -> None

Explicitly delete the cache.

Safe to call concurrently: the cache lock serializes delete attempts so the provider API fires at most once, and late callers that arrive after the cache is cleared return silently.

Source code in src/async_batch_llm/models.py

async def delete_cache(self) -> None:
    """Explicitly delete the cache.

    Safe to call concurrently: the cache lock serializes delete attempts
    so the provider API fires at most once, and late callers that arrive
    after the cache is cleared return silently.
    """
    if self._cache is None:
        return

    import asyncio as _asyncio

    if getattr(self, "_cache_lock", None) is None:
        self._cache_lock = _asyncio.Lock()

    async with self._cache_lock:
        cache = self._cache
        if cache is None:
            # A concurrent caller already finished the delete.
            return

        # Capture the name up front so log messages don't depend on
        # self._cache still existing after concurrent callers clear it.
        cache_name = cache.name
        # Clear state BEFORE the API call so concurrent tasks that
        # re-enter see an empty cache and no-op.
        self._cache = None
        self._cache_created_at = None
        self._prepared = False

        try:
            await self._client.aio.caches.delete(name=cache_name)
            logger.info(f"Deleted Gemini cache: {cache_name}")
        except Exception as e:
            # Keep Exception (not BaseException) so KeyboardInterrupt still propagates;
            # cache-delete failures are best-effort — caches expire on their own.
            logger.warning(
                f"Failed to delete Gemini cache '{cache_name}': {e}. "
                "Cache may have already expired or been deleted.",
                exc_info=True,
            )

generate async

generate(prompt: str | list[Any], *, temperature: float | None = 0.0, system_instruction: str | None = None, config: dict[str, Any] | None = None) -> LLMResponse

Generate a response using the cached context.

Parameters:

Name	Type	Description	Default
prompt	str | list[Any]	Text prompt or multimodal content parts.	required
temperature	float | None	Sampling temperature. Pass None to omit it and use the model default.	0.0
system_instruction	str | None	Not supported with caching — raises ValueError.	None
config	dict[str, Any] | None	Additional provider-specific config entries.	None

Returns:

Type	Description
LLMResponse	Normalized LLMResponse.

Source code in src/async_batch_llm/models.py

async def generate(
    self,
    prompt: str | list[Any],
    *,
    temperature: float | None = 0.0,
    system_instruction: str | None = None,
    config: dict[str, Any] | None = None,
) -> LLMResponse:
    """Generate a response using the cached context.

    Args:
        prompt: Text prompt or multimodal content parts.
        temperature: Sampling temperature. Pass ``None`` to omit it and
            use the model default.
        system_instruction: Not supported with caching — raises ValueError.
        config: Additional provider-specific config entries.

    Returns:
        Normalized LLMResponse.
    """
    if system_instruction is not None:
        raise ValueError(
            "system_instruction cannot be overridden per-call with cached models. "
            "The system instruction is baked into the cache at creation time."
        )

    # Auto-renew if expired
    if self._auto_renew and self._is_cache_expired():
        import asyncio

        if self._cache_lock is None:
            self._cache_lock = asyncio.Lock()

        async with self._cache_lock:
            if self._is_cache_expired():
                age_str = (
                    f"{time.time() - self._cache_created_at:.0f}s"
                    if self._cache_created_at is not None
                    else "unknown (cache not yet initialized)"
                )
                logger.info(
                    "Cache expired or about to expire, renewing before API call "
                    f"(age: {age_str}, "
                    f"renewal buffer: {self._cache_renewal_buffer_seconds}s)"
                )
                await self._renew_cache()

    if self._cache is None:
        raise RuntimeError("Cache not initialized — call prepare() first")

    # Build config with cache reference
    call_config: dict[str, Any] = {
        "cached_content": self._cache.name,
    }
    if temperature is not None:
        call_config["temperature"] = temperature

    if self._safety_settings:
        call_config["safety_settings"] = self._safety_settings

    if config:
        call_config.update(config)

    response = await self._client.aio.models.generate_content(
        model=self._model,
        contents=prompt,
        config=call_config,  # type: ignore[arg-type]  # ty:ignore[invalid-argument-type]
    )

    tokens = _extract_tokens(response)
    input_tokens, output_tokens, total_tokens, cached_tokens = tokens

    text = response.text
    if text is None:
        metadata = _extract_metadata(response)
        safety_info = ""
        if metadata and "safety_ratings" in metadata:
            safety_info = f" Safety ratings: {metadata['safety_ratings']}"
        _raise_empty_response(
            f"Empty response from model (likely blocked by safety filter).{safety_info}",
            tokens,
        )

    return LLMResponse(
        text=text,
        input_tokens=input_tokens,
        output_tokens=output_tokens,
        total_tokens=total_tokens,
        cached_input_tokens=cached_tokens,
        metadata=self._build_metadata(response),
        raw=response,
    )

prepare async

prepare() -> None

Find or create the Gemini cache. Idempotent.

Source code in src/async_batch_llm/models.py

async def prepare(self) -> None:
    """Find or create the Gemini cache. Idempotent."""
    if self._prepared:
        return

    import asyncio

    if self._cache_lock is None:
        self._cache_lock = asyncio.Lock()

    async with self._cache_lock:
        if self._prepared:
            return
        await self._find_or_create_cache()
        self._prepared = True

OpenAICompatibleModel

async_batch_llm.OpenAICompatibleModel

OpenAICompatibleModel(model: str, client: AsyncOpenAI, *, system_instruction: str | None = None, extra_headers: dict[str, str] | None = None, extra_body: dict[str, Any] | None = None, metadata_extractors: list[MetadataExtractor] | None = None)

Base class for OpenAI chat-completions-compatible providers.

Wraps an AsyncOpenAI client pointed at any chat-completions endpoint (OpenAI itself, OpenRouter, DeepSeek, HuggingFace Inference Providers, Together, Fireworks, local vLLM, etc.). Subclasses customize the default base URL, the install-extras hint, the env var read by :meth:from_api_key, and optionally the token/metadata extractors.

Implements the ManagedLLMModel protocol — :meth:cleanup closes the underlying AsyncOpenAI client when this model owns it (i.e. it was constructed via :meth:from_api_key). User-provided clients are left alone.

Models built with :meth:from_api_key and max_connections=N expose max_concurrency=N so execution surfaces can diagnose worker/pool mismatches. It remains None for caller-supplied clients because their effective transport capacity cannot be inspected reliably.

Added in v0.9.0.

Parameters:

Name	Type	Description	Default
model	str	Provider model id (e.g. "gpt-4o-mini" or "anthropic/claude-haiku-4-5").	required
client	AsyncOpenAI	Initialized AsyncOpenAI (point base_url at the desired endpoint). The model does NOT take ownership of the client — use :meth:from_api_key if you want the model to manage the client's lifecycle.	required
system_instruction	str | None	Default system message prepended to each call. Per-call system_instruction argument takes precedence.	None
extra_headers	dict[str, str] | None	Default headers forwarded on every call (e.g. OpenRouter's HTTP-Referer/X-Title).	None
extra_body	dict[str, Any] | None	Default extra body fields forwarded on every call (e.g. OpenRouter provider routing config).	None
metadata_extractors	list[MetadataExtractor] | None	Optional hooks that contribute extra keys to LLMResponse.metadata (e.g. reasoning_content or logprobs). Merged on top of the built-in finish_reason/model metadata; user keys win. Added in v0.15.0.	None

Source code in src/async_batch_llm/models.py

def __init__(
    self,
    model: str,
    client: "AsyncOpenAI",
    *,
    system_instruction: str | None = None,
    extra_headers: dict[str, str] | None = None,
    extra_body: dict[str, Any] | None = None,
    metadata_extractors: list[MetadataExtractor] | None = None,
):
    """
    Args:
        model: Provider model id (e.g. "gpt-4o-mini" or
            "anthropic/claude-haiku-4-5").
        client: Initialized AsyncOpenAI (point ``base_url`` at the desired
            endpoint). The model does NOT take ownership of the client —
            use :meth:`from_api_key` if you want the model to manage the
            client's lifecycle.
        system_instruction: Default system message prepended to each call.
            Per-call ``system_instruction`` argument takes precedence.
        extra_headers: Default headers forwarded on every call (e.g.
            OpenRouter's ``HTTP-Referer``/``X-Title``).
        extra_body: Default extra body fields forwarded on every call
            (e.g. OpenRouter ``provider`` routing config).
        metadata_extractors: Optional hooks that contribute extra keys to
            ``LLMResponse.metadata`` (e.g. ``reasoning_content`` or
            ``logprobs``). Merged on top of the built-in
            ``finish_reason``/``model`` metadata; user keys win.
            Added in v0.15.0.
    """
    if AsyncOpenAI is None:
        raise ImportError(
            f"openai is required for {type(self).__name__}. "
            f"Install with: pip install 'async-batch-llm[{self._install_extras}]'"
        )

    self._model = model
    self._client = client
    self._default_system_instruction = system_instruction
    self._default_extra_headers = extra_headers
    self._default_extra_body = extra_body
    self._metadata_extractors = metadata_extractors
    # Known only when this class constructs and sizes the underlying httpx
    # pool via from_api_key(max_connections=...). User-supplied clients are
    # intentionally not introspected because their transport may be wrapped.
    self.max_concurrency: int | None = None
    # Set to True only by from_api_key(); cleanup() uses this to decide
    # whether to close the underlying httpx connections.
    self._owns_client: bool = False
    # SDK constructor kwargs stashed by from_api_key() when it built the
    # client with the *default* httpx pool — lets request_concurrency()
    # rebuild the client with a right-sized pool before the first request.
    # None when the caller supplied the client or an explicit pool.
    self._rebuild_client_kwargs: dict[str, Any] | None = None

cleanup async

cleanup() -> None

Close the underlying AsyncOpenAI client if this model owns it.

Models constructed directly with OpenAIModel(model, client=...) do NOT own the client — the caller is expected to close it. Models constructed via :meth:from_api_key do own the client and close it here so repeated processor runs don't leak httpx connections.

Source code in src/async_batch_llm/models.py

async def cleanup(self) -> None:
    """Close the underlying AsyncOpenAI client if this model owns it.

    Models constructed directly with ``OpenAIModel(model, client=...)``
    do NOT own the client — the caller is expected to close it. Models
    constructed via :meth:`from_api_key` do own the client and close
    it here so repeated processor runs don't leak httpx connections.
    """
    if not self._owns_client or self._client is None:
        return
    close = getattr(self._client, "close", None)
    if close is None:
        return
    try:
        result = close()
        if hasattr(result, "__await__"):
            await result
    except Exception as e:
        logger.warning(
            f"Failed to close {type(self).__name__} client: {e}",
            exc_info=True,
        )

from_api_key classmethod

from_api_key(model: str, api_key: str | None = None, *, base_url: str | None = None, system_instruction: str | None = None, extra_headers: dict[str, str] | None = None, extra_body: dict[str, Any] | None = None, json_mode: bool = False, max_connections: int | None = None, metadata_extractors: list[MetadataExtractor] | None = None, _instance_kwargs: dict[str, Any] | None = None, **client_kwargs: Any) -> TM

Build the model with a freshly-constructed AsyncOpenAI client.

The returned model owns the client — its connections are released when the framework calls :meth:cleanup (typically when the ParallelBatchProcessor exits).

Uses base_url (if provided) or the class's _default_base_url. Pass client_kwargs to forward additional kwargs (timeout, max_retries, http_client, etc.) to the SDK constructor.

Parameters:

Name	Type	Description	Default
model	str	Provider model id.	required
api_key	str | None	API key. If None: For OpenAIModel the OpenAI SDK auto-reads OPENAI_API_KEY. For OpenRouterModel (and other subclasses with _api_key_env_var set) we read the env var ourselves and forward it to the SDK explicitly. If neither path resolves, raises ValueError.	None
json_mode	bool	When True, request JSON output by adding response_format={"type": "json_object"} to extra_body (forwarded on every call). A convenience over hand-passing it yourself; an explicit response_format in extra_body takes precedence. Most providers still require the word "JSON" somewhere in your prompt/system instruction for this to take effect. Pair with :func:async_batch_llm.pydantic_json_parser on the strategy, since some providers (DeepSeek) still wrap the JSON in markdown fences even in JSON mode (issue #26).	False
max_connections	int | None	Size of the underlying httpx connection pool (both max_connections and max_keepalive_connections). Set this to at least ProcessorConfig.max_workers — the openai SDK otherwise uses httpx's default pool (~100), so raising max_workers above that gives no extra throughput; the excess workers just block waiting for a connection (see issue #25). High-concurrency providers like DeepSeek (which allow thousands of concurrent connections) hit this ceiling first. Mutually exclusive with passing your own http_client; raises ValueError if you pass both. The value is also exposed as model.max_concurrency and forwarded by ModelStrategy for processor/gateway validation.	None

Source code in src/async_batch_llm/models.py

@classmethod
def from_api_key(
    cls: type[TM],
    model: str,
    api_key: str | None = None,
    *,
    base_url: str | None = None,
    system_instruction: str | None = None,
    extra_headers: dict[str, str] | None = None,
    extra_body: dict[str, Any] | None = None,
    json_mode: bool = False,
    max_connections: int | None = None,
    metadata_extractors: list[MetadataExtractor] | None = None,
    _instance_kwargs: dict[str, Any] | None = None,
    **client_kwargs: Any,
) -> TM:
    """Build the model with a freshly-constructed AsyncOpenAI client.

    The returned model owns the client — its connections are released
    when the framework calls :meth:`cleanup` (typically when the
    ``ParallelBatchProcessor`` exits).

    Uses ``base_url`` (if provided) or the class's ``_default_base_url``.
    Pass ``client_kwargs`` to forward additional kwargs (timeout,
    max_retries, http_client, etc.) to the SDK constructor.

    Args:
        model: Provider model id.
        api_key: API key. If ``None``:

            - For ``OpenAIModel`` the OpenAI SDK auto-reads
              ``OPENAI_API_KEY``.
            - For ``OpenRouterModel`` (and other subclasses with
              ``_api_key_env_var`` set) we read the env var ourselves
              and forward it to the SDK explicitly.
            - If neither path resolves, raises ``ValueError``.
        json_mode: When ``True``, request JSON output by adding
            ``response_format={"type": "json_object"}`` to ``extra_body``
            (forwarded on every call). A convenience over hand-passing it
            yourself; an explicit ``response_format`` in ``extra_body``
            takes precedence. Most providers still require the word "JSON"
            somewhere in your prompt/system instruction for this to take
            effect. Pair with :func:`async_batch_llm.pydantic_json_parser`
            on the strategy, since some providers (DeepSeek) still wrap the
            JSON in markdown fences even in JSON mode (issue #26).
        max_connections: Size of the underlying httpx connection pool
            (both ``max_connections`` and ``max_keepalive_connections``).
            **Set this to at least ``ProcessorConfig.max_workers``** — the
            openai SDK otherwise uses httpx's default pool (~100), so
            raising ``max_workers`` above that gives no extra throughput;
            the excess workers just block waiting for a connection (see
            issue #25). High-concurrency providers like DeepSeek (which
            allow thousands of concurrent connections) hit this ceiling
            first. Mutually exclusive with passing your own
            ``http_client``; raises ``ValueError`` if you pass both.
            The value is also exposed as ``model.max_concurrency`` and
            forwarded by ``ModelStrategy`` for processor/gateway validation.
    """
    if AsyncOpenAI is None:
        raise ImportError(
            f"openai is required for {cls.__name__}. "
            f"Install with: pip install 'async-batch-llm[{cls._install_extras}]'"
        )
    if max_connections is not None:
        if "http_client" in client_kwargs:
            raise ValueError(
                "Pass either max_connections or http_client, not both. "
                "max_connections is a convenience for sizing the default "
                "httpx pool; if you build your own http_client, set its "
                "limits there instead."
            )
        if max_connections < 1:
            raise ValueError(f"max_connections must be >= 1; got {max_connections}.")
        client_kwargs["http_client"] = _build_openai_http_client(max_connections)
    effective_base_url = base_url or cls._default_base_url
    if effective_base_url is not None:
        client_kwargs.setdefault("base_url", effective_base_url)

    resolved_key = api_key
    if resolved_key is None and cls._api_key_env_var is not None:
        import os as _os

        resolved_key = _os.environ.get(cls._api_key_env_var)
        if not resolved_key:
            raise ValueError(
                f"No API key for {cls.__name__}: pass api_key= or set the "
                f"{cls._api_key_env_var} environment variable."
            )

    # When resolved_key is None and the SDK can self-resolve (OpenAIModel),
    # don't pass api_key at all — letting the SDK raise its own clear
    # error if neither path produces one.
    if resolved_key is not None:
        client_kwargs["api_key"] = resolved_key
    client = AsyncOpenAI(**client_kwargs)

    if json_mode:
        # Inject a JSON response_format, letting any explicit caller-supplied
        # response_format in extra_body win.
        effective_extra_body: dict[str, Any] = {"response_format": {"type": "json_object"}}
        if extra_body:
            effective_extra_body.update(extra_body)
        extra_body = effective_extra_body

    instance = cls(
        model,
        client,
        system_instruction=system_instruction,
        extra_headers=extra_headers,
        extra_body=extra_body,
        metadata_extractors=metadata_extractors,
        **(_instance_kwargs or {}),
    )
    instance.max_concurrency = max_connections
    instance._owns_client = True
    if max_connections is None and "http_client" not in client_kwargs:
        # Default-pool client: keep the constructor kwargs so
        # request_concurrency() can rebuild it with a right-sized pool.
        instance._rebuild_client_kwargs = dict(client_kwargs)
    return instance

generate async

generate(prompt: str | list[Any], *, temperature: float | None = 0.0, system_instruction: str | None = None, config: dict[str, Any] | None = None) -> LLMResponse

Call client.chat.completions.create and normalize the response.

Parameters:

Name	Type	Description	Default
prompt	str | list[Any]	A string (becomes a single user message) or a list of OpenAI-shaped message dicts (passed through unchanged — used for multimodal content and Anthropic-via-OpenRouter cache_control markers).	required
temperature	float | None	Sampling temperature. Pass None to omit the parameter so the provider uses its own default — required for OpenAI reasoning models (o1/o3/etc.) that reject an explicit temperature.	0.0
system_instruction	str | None	Per-call override for the system message.	None
config	dict[str, Any] | None	Per-call extra kwargs forwarded to the SDK call (merged over the instance's extra_body). Use this to pass max_tokens, response_format, etc.	None

Returns:

Type	Description
LLMResponse	Normalized LLMResponse.

Source code in src/async_batch_llm/models.py

async def generate(
    self,
    prompt: str | list[Any],
    *,
    temperature: float | None = 0.0,
    system_instruction: str | None = None,
    config: dict[str, Any] | None = None,
) -> LLMResponse:
    """Call ``client.chat.completions.create`` and normalize the response.

    Args:
        prompt: A string (becomes a single user message) or a list of
            OpenAI-shaped message dicts (passed through unchanged — used
            for multimodal content and Anthropic-via-OpenRouter
            ``cache_control`` markers).
        temperature: Sampling temperature. Pass ``None`` to omit the
            parameter so the provider uses its own default — required for
            OpenAI reasoning models (o1/o3/etc.) that reject an explicit
            ``temperature``.
        system_instruction: Per-call override for the system message.
        config: Per-call extra kwargs forwarded to the SDK call (merged
            over the instance's ``extra_body``). Use this to pass
            ``max_tokens``, ``response_format``, etc.

    Returns:
        Normalized LLMResponse.
    """
    messages = _coerce_to_messages(prompt)
    si = system_instruction or self._default_system_instruction
    if si is not None and not _has_system_message(messages):
        messages = [{"role": "system", "content": si}, *messages]

    # Merge extra_body: instance defaults + per-call config overrides.
    extra_body: dict[str, Any] = {}
    if self._default_extra_body:
        extra_body.update(self._default_extra_body)
    if config:
        extra_body.update(config)

    call_kwargs: dict[str, Any] = {
        "model": self._model,
        "messages": messages,
    }
    # Omit temperature entirely when None so providers that reject an
    # explicit value (OpenAI reasoning models) use their own default.
    if temperature is not None:
        call_kwargs["temperature"] = temperature
    if extra_body:
        call_kwargs["extra_body"] = extra_body
    if self._default_extra_headers:
        call_kwargs["extra_headers"] = self._default_extra_headers

    response = await self._client.chat.completions.create(**call_kwargs)

    # Some gateways report upstream failures inside an HTTP-200 body;
    # surface those before interpreting the (empty) choices list.
    self._raise_on_response_error(response)

    # Extract tokens up front: even a no-content response was billed for
    # the prompt (and, for finish_reason="length", the full output).
    tokens = self._extract_tokens(response)
    input_tokens, output_tokens, total_tokens, cached_tokens = tokens

    # Validate content present (None typically means a tool call or a
    # finish-reason like "length"/"content_filter").
    if not response.choices:
        _raise_empty_response(f"No choices returned from {type(self).__name__}.", tokens)
    message = response.choices[0].message
    text = getattr(message, "content", None)
    if text is None:
        finish_reason = getattr(response.choices[0], "finish_reason", "unknown")
        _raise_empty_response(
            f"Empty response content from model "
            f"(finish_reason={finish_reason!r}). "
            "This typically indicates a tool call, content filter, "
            "or token limit was reached.",
            tokens,
        )

    return LLMResponse(
        text=text,
        input_tokens=input_tokens,
        output_tokens=output_tokens,
        total_tokens=total_tokens,
        cached_input_tokens=cached_tokens,
        metadata=self._build_metadata(response),
        raw=response,
    )

prepare async

prepare() -> None

No-op; OpenAI-compatible models have nothing to initialize.

Source code in src/async_batch_llm/models.py

async def prepare(self) -> None:
    """No-op; OpenAI-compatible models have nothing to initialize."""
    return

request_concurrency async

request_concurrency(concurrency: int) -> bool

Ask the model to support concurrency parallel requests.

Called by execution surfaces when ProcessorConfig.concurrency is set (v0.20.0, issue #97), before the first request. When this model owns its client (built via :meth:from_api_key) and no explicit max_connections was given, the AsyncOpenAI client is rebuilt with an httpx pool sized to concurrency (the SDK's default pool of ~100 connections would otherwise silently cap throughput) and max_concurrency starts advertising the new size.

Returns True when the pool was resized. Returns False — leaving the model untouched — for caller-supplied clients and for models built with an explicit max_connections (an explicit value always wins; a genuine contradiction is surfaced by the existing capacity warning instead).

Source code in src/async_batch_llm/models.py

async def request_concurrency(self, concurrency: int) -> bool:
    """Ask the model to support ``concurrency`` parallel requests.

    Called by execution surfaces when ``ProcessorConfig.concurrency`` is
    set (v0.20.0, issue #97), before the first request. When this model
    owns its client (built via :meth:`from_api_key`) and no explicit
    ``max_connections`` was given, the ``AsyncOpenAI`` client is rebuilt
    with an httpx pool sized to ``concurrency`` (the SDK's default pool of
    ~100 connections would otherwise silently cap throughput) and
    ``max_concurrency`` starts advertising the new size.

    Returns True when the pool was resized. Returns False — leaving the
    model untouched — for caller-supplied clients and for models built
    with an explicit ``max_connections`` (an explicit value always wins;
    a genuine contradiction is surfaced by the existing capacity warning
    instead).
    """
    if concurrency < 1:
        raise ValueError(f"concurrency must be >= 1; got {concurrency}.")
    if (
        not self._owns_client
        or self.max_concurrency is not None
        or self._rebuild_client_kwargs is None
    ):
        return False
    new_client = AsyncOpenAI(
        http_client=_build_openai_http_client(concurrency),
        **self._rebuild_client_kwargs,
    )
    old_client = self._client
    self._client = new_client
    self.max_concurrency = concurrency
    # The old client made no requests yet (this runs before batch start);
    # close it so its default httpx transport doesn't linger.
    close = getattr(old_client, "close", None)
    if close is not None:
        try:
            result = close()
            if hasattr(result, "__await__"):
                await result
        except Exception as e:
            logger.warning(
                f"Failed to close replaced {type(self).__name__} client: {e}",
                exc_info=True,
            )
    return True

OpenAIModel

async_batch_llm.OpenAIModel

OpenAIModel(model: str, client: AsyncOpenAI, *, system_instruction: str | None = None, extra_headers: dict[str, str] | None = None, extra_body: dict[str, Any] | None = None, metadata_extractors: list[MetadataExtractor] | None = None)

Bases: OpenAICompatibleModel

LLM model backed by OpenAI's chat completions API.

Uses the OpenAI SDK's default base URL (https://api.openai.com/v1). OpenAI's automatic prompt cache surfaces in cached_input_tokens for prompts longer than ~1024 tokens.

Example

model = OpenAIModel.from_api_key("gpt-4o-mini", api_key="sk-...") response = await model.generate("Hello!") print(response.text, response.cached_input_tokens)

Added in v0.9.0.

Source code in src/async_batch_llm/models.py

def __init__(
    self,
    model: str,
    client: "AsyncOpenAI",
    *,
    system_instruction: str | None = None,
    extra_headers: dict[str, str] | None = None,
    extra_body: dict[str, Any] | None = None,
    metadata_extractors: list[MetadataExtractor] | None = None,
):
    """
    Args:
        model: Provider model id (e.g. "gpt-4o-mini" or
            "anthropic/claude-haiku-4-5").
        client: Initialized AsyncOpenAI (point ``base_url`` at the desired
            endpoint). The model does NOT take ownership of the client —
            use :meth:`from_api_key` if you want the model to manage the
            client's lifecycle.
        system_instruction: Default system message prepended to each call.
            Per-call ``system_instruction`` argument takes precedence.
        extra_headers: Default headers forwarded on every call (e.g.
            OpenRouter's ``HTTP-Referer``/``X-Title``).
        extra_body: Default extra body fields forwarded on every call
            (e.g. OpenRouter ``provider`` routing config).
        metadata_extractors: Optional hooks that contribute extra keys to
            ``LLMResponse.metadata`` (e.g. ``reasoning_content`` or
            ``logprobs``). Merged on top of the built-in
            ``finish_reason``/``model`` metadata; user keys win.
            Added in v0.15.0.
    """
    if AsyncOpenAI is None:
        raise ImportError(
            f"openai is required for {type(self).__name__}. "
            f"Install with: pip install 'async-batch-llm[{self._install_extras}]'"
        )

    self._model = model
    self._client = client
    self._default_system_instruction = system_instruction
    self._default_extra_headers = extra_headers
    self._default_extra_body = extra_body
    self._metadata_extractors = metadata_extractors
    # Known only when this class constructs and sizes the underlying httpx
    # pool via from_api_key(max_connections=...). User-supplied clients are
    # intentionally not introspected because their transport may be wrapped.
    self.max_concurrency: int | None = None
    # Set to True only by from_api_key(); cleanup() uses this to decide
    # whether to close the underlying httpx connections.
    self._owns_client: bool = False
    # SDK constructor kwargs stashed by from_api_key() when it built the
    # client with the *default* httpx pool — lets request_concurrency()
    # rebuild the client with a right-sized pool before the first request.
    # None when the caller supplied the client or an explicit pool.
    self._rebuild_client_kwargs: dict[str, Any] | None = None

OpenRouterModel

async_batch_llm.OpenRouterModel

OpenRouterModel(model: str, client: AsyncOpenAI, *, system_instruction: str | None = None, extra_headers: dict[str, str] | None = None, extra_body: dict[str, Any] | None = None, metadata_extractors: list[MetadataExtractor] | None = None)

Bases: OpenAICompatibleModel

LLM model backed by OpenRouter (https://openrouter.ai).

OpenRouter exposes a unified OpenAI-compatible API for many upstream providers (Anthropic, OpenAI, Google, Mistral, DeepSeek, etc.). Model ids are prefixed with the provider, e.g. "anthropic/claude-haiku-4-5".

Caching is provider-dependent:

• OpenAI / Gemini (implicit) / DeepSeek — automatic; cached_input_tokens is populated when the upstream cache hits.
• Anthropic — opt-in. Pass prompt as a list of message dicts with cache_control: {"type": "ephemeral"} markers on the blocks you want cached.

Example

model = OpenRouterModel.from_api_key( ... "anthropic/claude-haiku-4-5", ... api_key="sk-or-...", ... referer="https://my-app.example.com", ... title="My App", ... ) response = await model.generate("Hello!")

Added in v0.9.0.

Source code in src/async_batch_llm/models.py

def __init__(
    self,
    model: str,
    client: "AsyncOpenAI",
    *,
    system_instruction: str | None = None,
    extra_headers: dict[str, str] | None = None,
    extra_body: dict[str, Any] | None = None,
    metadata_extractors: list[MetadataExtractor] | None = None,
):
    """
    Args:
        model: Provider model id (e.g. "gpt-4o-mini" or
            "anthropic/claude-haiku-4-5").
        client: Initialized AsyncOpenAI (point ``base_url`` at the desired
            endpoint). The model does NOT take ownership of the client —
            use :meth:`from_api_key` if you want the model to manage the
            client's lifecycle.
        system_instruction: Default system message prepended to each call.
            Per-call ``system_instruction`` argument takes precedence.
        extra_headers: Default headers forwarded on every call (e.g.
            OpenRouter's ``HTTP-Referer``/``X-Title``).
        extra_body: Default extra body fields forwarded on every call
            (e.g. OpenRouter ``provider`` routing config).
        metadata_extractors: Optional hooks that contribute extra keys to
            ``LLMResponse.metadata`` (e.g. ``reasoning_content`` or
            ``logprobs``). Merged on top of the built-in
            ``finish_reason``/``model`` metadata; user keys win.
            Added in v0.15.0.
    """
    if AsyncOpenAI is None:
        raise ImportError(
            f"openai is required for {type(self).__name__}. "
            f"Install with: pip install 'async-batch-llm[{self._install_extras}]'"
        )

    self._model = model
    self._client = client
    self._default_system_instruction = system_instruction
    self._default_extra_headers = extra_headers
    self._default_extra_body = extra_body
    self._metadata_extractors = metadata_extractors
    # Known only when this class constructs and sizes the underlying httpx
    # pool via from_api_key(max_connections=...). User-supplied clients are
    # intentionally not introspected because their transport may be wrapped.
    self.max_concurrency: int | None = None
    # Set to True only by from_api_key(); cleanup() uses this to decide
    # whether to close the underlying httpx connections.
    self._owns_client: bool = False
    # SDK constructor kwargs stashed by from_api_key() when it built the
    # client with the *default* httpx pool — lets request_concurrency()
    # rebuild the client with a right-sized pool before the first request.
    # None when the caller supplied the client or an explicit pool.
    self._rebuild_client_kwargs: dict[str, Any] | None = None

from_api_key classmethod

from_api_key(model: str, api_key: str | None = None, *, base_url: str | None = None, system_instruction: str | None = None, extra_headers: dict[str, str] | None = None, extra_body: dict[str, Any] | None = None, json_mode: bool = False, max_connections: int | None = None, referer: str | None = None, title: str | None = None, metadata_extractors: list[MetadataExtractor] | None = None, **client_kwargs: Any) -> OpenRouterModel

Build an OpenRouterModel.

If api_key is None, reads OPENROUTER_API_KEY from the environment and raises ValueError if neither is set. (The OpenAI SDK doesn't know about OPENROUTER_API_KEY, so we have to read it ourselves rather than relying on the SDK's default.)

referer and title map to OpenRouter's optional HTTP-Referer and X-Title headers (used for app attribution on openrouter.ai's leaderboard).

Source code in src/async_batch_llm/models.py

@classmethod
def from_api_key(  # type: ignore[override]
    cls,
    model: str,
    api_key: str | None = None,
    *,
    base_url: str | None = None,
    system_instruction: str | None = None,
    extra_headers: dict[str, str] | None = None,
    extra_body: dict[str, Any] | None = None,
    json_mode: bool = False,
    max_connections: int | None = None,
    referer: str | None = None,
    title: str | None = None,
    metadata_extractors: list[MetadataExtractor] | None = None,
    **client_kwargs: Any,
) -> "OpenRouterModel":
    """Build an OpenRouterModel.

    If ``api_key`` is None, reads ``OPENROUTER_API_KEY`` from the
    environment and raises ``ValueError`` if neither is set. (The
    OpenAI SDK doesn't know about ``OPENROUTER_API_KEY``, so we have
    to read it ourselves rather than relying on the SDK's default.)

    ``referer`` and ``title`` map to OpenRouter's optional
    ``HTTP-Referer`` and ``X-Title`` headers (used for app attribution
    on openrouter.ai's leaderboard).
    """
    merged_headers = dict(extra_headers) if extra_headers else {}
    if referer is not None:
        merged_headers.setdefault("HTTP-Referer", referer)
    if title is not None:
        merged_headers.setdefault("X-Title", title)

    # super().from_api_key is generic over cls (returns the calling
    # subclass type), so this returns OpenRouterModel directly — no cast
    # needed. See issue #10.
    return super().from_api_key(
        model,
        api_key,
        base_url=base_url,
        system_instruction=system_instruction,
        extra_headers=merged_headers or None,
        extra_body=extra_body,
        json_mode=json_mode,
        max_connections=max_connections,
        metadata_extractors=metadata_extractors,
        **client_kwargs,
    )

DeepSeekModel

async_batch_llm.DeepSeekModel

DeepSeekModel(model: str, client: AsyncOpenAI, *, system_instruction: str | None = None, extra_headers: dict[str, str] | None = None, extra_body: dict[str, Any] | None = None, thinking: bool | None = None, api_surface: str = 'chat_completions', response_schema: type[BaseModel] | Mapping[str, Any] | None = None, schema_name: str | None = None, json_mode: bool = False, metadata_extractors: list[MetadataExtractor] | None = None)

Bases: OpenAICompatibleModel

LLM model backed by DeepSeek's OpenAI-compatible API.

Points at https://api.deepseek.com and reads DEEPSEEK_API_KEY in :meth:from_api_key. Model ids are bare DeepSeek names, e.g. "deepseek-chat" or "deepseek-reasoner".

DeepSeek's automatic context cache reports hits at the top level of the usage object (prompt_cache_hit_tokens / prompt_cache_miss_tokens) rather than under OpenAI's nested prompt_tokens_details.cached_tokens — so this subclass overrides :meth:_extract_tokens to surface them in cached_input_tokens. Use :attr:CachedTokenRates.DEEPSEEK (~2%) when computing billable tokens.

(Calling DeepSeek through OpenRouter uses :class:OpenRouterModel instead; the native cache fields aren't reliably forwarded there, which is why direct access via this class gives better cache telemetry.)

Thinking mode. DeepSeek's V4 models (deepseek-v4-flash / deepseek-v4-pro) default to thinking, which for a batch classification job is a surprising, expensive default — thinking can emit several times the output tokens (and cost, and latency) of non-thinking. Pass thinking=False to force non-thinking mode explicitly rather than relying on the deepseek-chat (non-thinking) / deepseek-reasoner (thinking) aliases, which DeepSeek is deprecating. Under the hood this sends extra_body={"thinking": {"type": "disabled"}} for Chat Completions, or reasoning={"effort": "none"} on the Responses API.

Strict structured output. Set api_surface="responses" and pass a Pydantic model class or JSON Schema mapping as response_schema. The schema is sent through text.format.type="json_schema" and :class:DeepSeekStrategy parses successful output automatically. DeepSeek currently exposes Responses only for deepseek-v4-flash; unsupported models fail locally instead of silently falling back to weaker JSON mode.

Example

model = DeepSeekModel.from_api_key( ... "deepseek-v4-flash", api_key="sk-...", thinking=False ... ) response = await model.generate("Hello!") print(response.text, response.cached_input_tokens)

Added in v0.10.0.

See :class:OpenAICompatibleModel; adds the DeepSeek thinking toggle and opt-in Responses API structured output.

Source code in src/async_batch_llm/models.py

def __init__(
    self,
    model: str,
    client: "AsyncOpenAI",
    *,
    system_instruction: str | None = None,
    extra_headers: dict[str, str] | None = None,
    extra_body: dict[str, Any] | None = None,
    thinking: bool | None = None,
    api_surface: str = "chat_completions",
    response_schema: type[BaseModel] | Mapping[str, Any] | None = None,
    schema_name: str | None = None,
    json_mode: bool = False,
    metadata_extractors: list[MetadataExtractor] | None = None,
):
    """See :class:`OpenAICompatibleModel`; adds the DeepSeek ``thinking``
    toggle and opt-in Responses API structured output."""
    if api_surface not in {"chat_completions", "responses"}:
        raise ValueError(
            f"api_surface must be 'chat_completions' or 'responses' (got {api_surface!r})."
        )
    if response_schema is not None and api_surface != "responses":
        raise ValueError(
            "response_schema requires api_surface='responses'; Chat Completions only "
            "supports json_mode=True without schema enforcement."
        )
    if response_schema is not None and json_mode:
        raise ValueError("Pass response_schema or json_mode=True, not both.")
    if api_surface == "responses" and model not in self._responses_models:
        raise ValueError(
            "DeepSeek Responses API currently supports only 'deepseek-v4-flash' "
            f"(got {model!r}). Use Chat Completions without response_schema as the "
            "explicit fallback."
        )

    self.api_surface = api_surface
    self.response_schema: dict[str, Any] | None = None
    self.response_schema_name: str | None = None
    self.response_schema_identity: str | None = None
    self.response_schema_hash: str | None = None
    self._response_model: type[BaseModel] | None = None
    if response_schema is not None:
        (
            self.response_schema,
            self.response_schema_name,
            self.response_schema_identity,
            self.response_schema_hash,
        ) = _prepare_response_schema(response_schema, schema_name)
        if isinstance(response_schema, type) and issubclass(response_schema, BaseModel):
            self._response_model = response_schema

    if api_surface == "responses":
        configured_body = _responses_thinking(extra_body, thinking)
        configured_body = dict(configured_body) if configured_body else {}
        if self.response_schema is not None:
            requested_format = {
                "type": "json_schema",
                "name": self.response_schema_name,
                "schema": self.response_schema,
            }
            text_config = configured_body.get("text")
            if text_config is not None and not isinstance(text_config, Mapping):
                raise TypeError("extra_body['text'] must be a mapping")
            existing_format = text_config.get("format") if text_config else None
            if existing_format is not None and existing_format != requested_format:
                raise ValueError("extra_body['text']['format'] conflicts with response_schema")
            configured_body["text"] = {**dict(text_config or {}), "format": requested_format}
        elif json_mode:
            text_config = configured_body.get("text")
            if text_config is not None and not isinstance(text_config, Mapping):
                raise TypeError("extra_body['text'] must be a mapping")
            configured_body["text"] = {
                "format": {"type": "json_object"},
                **dict(text_config or {}),
            }
        effective_extra_body = configured_body or None
    else:
        effective_extra_body = _merge_thinking(extra_body, thinking)
        if json_mode:
            configured_body = {
                "response_format": {"type": "json_object"},
                **dict(effective_extra_body or {}),
            }
            effective_extra_body = configured_body

    super().__init__(
        model,
        client,
        system_instruction=system_instruction,
        extra_headers=extra_headers,
        extra_body=effective_extra_body,
        metadata_extractors=metadata_extractors,
    )

artifact_identity_extra property

artifact_identity_extra: dict[str, Any]

Replay identity fields that distinguish transport/output contracts.

from_api_key classmethod

from_api_key(model: str, api_key: str | None = None, *, base_url: str | None = None, system_instruction: str | None = None, extra_headers: dict[str, str] | None = None, extra_body: dict[str, Any] | None = None, json_mode: bool = False, max_connections: int | None = None, thinking: bool | None = None, api_surface: str = 'chat_completions', response_schema: type[BaseModel] | Mapping[str, Any] | None = None, schema_name: str | None = None, metadata_extractors: list[MetadataExtractor] | None = None, **client_kwargs: Any) -> DeepSeekModel

Build a DeepSeekModel; reads DEEPSEEK_API_KEY when api_key is None. Adds the thinking toggle and Responses API structured-output options (see the class docstring) on top of the shared constructor.

Source code in src/async_batch_llm/models.py

@classmethod
def from_api_key(  # type: ignore[override]
    cls,
    model: str,
    api_key: str | None = None,
    *,
    base_url: str | None = None,
    system_instruction: str | None = None,
    extra_headers: dict[str, str] | None = None,
    extra_body: dict[str, Any] | None = None,
    json_mode: bool = False,
    max_connections: int | None = None,
    thinking: bool | None = None,
    api_surface: str = "chat_completions",
    response_schema: type[BaseModel] | Mapping[str, Any] | None = None,
    schema_name: str | None = None,
    metadata_extractors: list[MetadataExtractor] | None = None,
    **client_kwargs: Any,
) -> "DeepSeekModel":
    """Build a DeepSeekModel; reads ``DEEPSEEK_API_KEY`` when ``api_key`` is
    None. Adds the ``thinking`` toggle and Responses API structured-output
    options (see the class docstring) on top of the shared constructor."""
    return super().from_api_key(
        model,
        api_key,
        base_url=base_url,
        system_instruction=system_instruction,
        extra_headers=extra_headers,
        extra_body=extra_body,
        # DeepSeekModel applies JSON mode according to the selected API
        # surface; suppress the base class's Chat-only injection.
        json_mode=False,
        max_connections=max_connections,
        metadata_extractors=metadata_extractors,
        _instance_kwargs={
            "thinking": thinking,
            "api_surface": api_surface,
            "response_schema": response_schema,
            "schema_name": schema_name,
            "json_mode": json_mode,
        },
        **client_kwargs,
    )

generate async

generate(prompt: str | list[Any], *, temperature: float | None = 0.0, system_instruction: str | None = None, config: dict[str, Any] | None = None) -> LLMResponse

Call the selected DeepSeek API surface and normalize its response.

Source code in src/async_batch_llm/models.py

async def generate(
    self,
    prompt: str | list[Any],
    *,
    temperature: float | None = 0.0,
    system_instruction: str | None = None,
    config: dict[str, Any] | None = None,
) -> LLMResponse:
    """Call the selected DeepSeek API surface and normalize its response."""
    if self.api_surface == "chat_completions":
        result = await super().generate(
            prompt,
            temperature=temperature,
            system_instruction=system_instruction,
            config=config,
        )
        metadata = dict(result.metadata or {})
        metadata["api_surface"] = self.api_surface
        request_id = getattr(result.raw, "id", None)
        if isinstance(request_id, str) and request_id:
            metadata["provider_request_id"] = request_id
        if self._uses_json_mode:
            metadata["output_format"] = "json_object"
        result.metadata = metadata
        return result
    return await self._generate_responses(
        prompt,
        temperature=temperature,
        system_instruction=system_instruction,
        config=config,
    )

parse_structured_response

parse_structured_response(response: LLMResponse) -> Any

Parse provider-enforced output into its Pydantic type or JSON value.

Source code in src/async_batch_llm/models.py

def parse_structured_response(self, response: LLMResponse) -> Any:
    """Parse provider-enforced output into its Pydantic type or JSON value."""
    try:
        if self._response_model is not None:
            return self._response_model.model_validate_json(response.text)
        return json.loads(response.text)
    except (ValidationError, json.JSONDecodeError) as exc:
        raise StructuredOutputValidationError(
            f"DeepSeek returned output that did not validate against "
            f"{self.response_schema_name!r}: {exc}"
        ) from exc

Protocols

LLMModel

async_batch_llm.LLMModel

Bases: Protocol

Protocol for LLM model instances that can generate responses.

Implementations wrap a specific provider's client and model configuration, handling API calls and response normalization. Strategies call generate() without needing to know about provider-specific details.

Added in v0.6.0.

generate async

generate(prompt: str | list[Any], *, temperature: float | None = 0.0, system_instruction: str | None = None, config: dict[str, Any] | None = None) -> LLMResponse

Generate a response from the LLM.

Parameters:

Name	Type	Description	Default
prompt	str | list[Any]	Text prompt, or list of content parts for multimodal input.	required
temperature	float | None	Sampling temperature (0.0 = deterministic). Pass None to omit the parameter entirely and use the provider default — required for models that reject an explicit temperature (e.g. OpenAI reasoning models like o1/o3).	0.0
system_instruction	str | None	System instruction override (None = use default).	None
config	dict[str, Any] | None	Provider-specific configuration (e.g., response_mime_type).	None

Returns:

Type	Description
LLMResponse	Normalized LLMResponse with text, token counts, and metadata.

Source code in src/async_batch_llm/core/protocols.py

async def generate(
    self,
    prompt: str | list[Any],
    *,
    temperature: float | None = 0.0,
    system_instruction: str | None = None,
    config: dict[str, Any] | None = None,
) -> LLMResponse:
    """
    Generate a response from the LLM.

    Args:
        prompt: Text prompt, or list of content parts for multimodal input.
        temperature: Sampling temperature (0.0 = deterministic). Pass ``None``
            to omit the parameter entirely and use the provider default —
            required for models that reject an explicit temperature (e.g.
            OpenAI reasoning models like o1/o3).
        system_instruction: System instruction override (None = use default).
        config: Provider-specific configuration (e.g., response_mime_type).

    Returns:
        Normalized LLMResponse with text, token counts, and metadata.
    """
    ...

ManagedLLMModel

async_batch_llm.ManagedLLMModel

Bases: LLMModel, Protocol

LLMModel with lifecycle management (e.g., caching).

Models that need one-time setup (creating a cache) or cleanup implement this protocol. The strategy delegates prepare/cleanup calls to the model.

Added in v0.6.0.

cleanup async

cleanup() -> None

Release resources. Must be idempotent.

Source code in src/async_batch_llm/core/protocols.py

async def cleanup(self) -> None:
    """Release resources. Must be idempotent."""
    ...

prepare async

prepare() -> None

Initialize resources (e.g., find or create a cache). Must be idempotent.

Source code in src/async_batch_llm/core/protocols.py

async def prepare(self) -> None:
    """Initialize resources (e.g., find or create a cache). Must be idempotent."""
    ...

LLMResponse

async_batch_llm.LLMResponse dataclass

LLMResponse(text: str, input_tokens: int, output_tokens: int, total_tokens: int, cached_input_tokens: int = 0, metadata: dict[str, Any] | None = None, raw: Any = None)

Bases: ProviderOutputViews

Normalized response from any LLM provider.

Returned by LLMModel.generate(). Provides a provider-agnostic interface so strategies don't need to know about Gemini, OpenAI, etc. response formats.

Attributes:

Name	Type	Description
text	str	The response text content.
input_tokens	int	Number of input/prompt tokens.
output_tokens	int	Number of output/completion tokens.
total_tokens	int	Total tokens used.
cached_input_tokens	int	Input tokens served from cache (0 if no caching).
metadata	dict[str, Any] | None	Provider-specific metadata (safety ratings, finish reason, etc.). The keys 'grounding', 'reasoning', 'tool_calls', and 'logprobs' are reserved, with documented dict shapes readable through the typed views .grounding/.reasoning/ .tool_calls/.logprobs (see provider_output.py).
raw	Any	The raw provider response object, for edge cases.

Added in v0.6.0.

token_usage property

token_usage: TokenUsage

Return token counts as a TokenUsage dict.