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.

llm_minder.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/llm_minder/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"
        )
    try:
        signature = inspect.signature(invoke)
    except (TypeError, ValueError):
        # Some extension callables do not expose a signature; the execution
        # boundary still validates their awaitable and CallOutcome.
        signature = None
    if signature is not None:
        try:
            signature.bind("prompt", attempt=1, timeout=1.0, state=None)
        except TypeError as exc:
            raise TypeError(
                "CallableStrategy invoke must accept "
                "invoke(prompt, *, attempt, timeout, state). "
                f"Got signature {signature}: {exc}"
            ) from exc
    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

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

llm_minder.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() when the processor exits or shuts down; a later explicit close retries an unsuccessful attempt

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. Explicit None falls back to strategy identity. An ordinary property failure raises QuotaScopeError before admission; cancellation and process-control exceptions propagate.

cleanup async

cleanup() -> None

Clean up resources when the processor exits or shuts down.

Called per prepared strategy instance, not per work item. A successful call is never repeated. If cleanup raises or is interrupted, a later explicit close retries it. Implementations must therefore be idempotent and safe to call after a partially completed attempt.

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

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

    Called per prepared strategy instance, not per work item. A successful
    call is never repeated. If cleanup raises or is interrupted, a later
    explicit close retries it. Implementations must therefore be
    idempotent and safe to call after a partially completed attempt.

    **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/llm_minder/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/llm_minder/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, 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
TOutput	A 3-tuple (output, token_usage, metadata). token_usage is a
TokenUsage	TokenUsage dict with optional keys input_tokens,
dict[str, Any] | None	output_tokens, total_tokens, cached_input_tokens.
tuple[TOutput, TokenUsage, dict[str, Any] | None]	metadata (v0.10.0) is a provider-specific dict forwarded into
tuple[TOutput, TokenUsage, dict[str, Any] | None]	WorkItemResult.metadata — typically finish_reason,
tuple[TOutput, TokenUsage, dict[str, Any] | None]	model, provider (OpenRouter), safety_ratings (Gemini);
tuple[TOutput, TokenUsage, dict[str, Any] | None]	pass None if you have nothing to surface. Since 1.0 a 2-tuple
tuple[TOutput, TokenUsage, dict[str, Any] | None]	(output, token_usage) fails the item with a ValueError
tuple[TOutput, TokenUsage, dict[str, Any] | None]	(category logic_error); its token usage is still recorded.

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

@abstractmethod
async def execute(
    self, prompt: str, attempt: int, timeout: float, state: "RetryState | None" = None
) -> 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:
        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. Since 1.0 a 2-tuple
        ``(output, token_usage)`` fails the item with a ``ValueError``
        (category ``logic_error``); its token usage is still recorded.

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

The hooks below are the ones a custom strategy overrides.

estimate_tokens(prompt, attempt, state)

Optional per-strategy TPM estimator. It runs after middleware and coordinated cooldown but before the atomic quota reservation and provider-capacity wait. It may be synchronous or asynchronous and receives retry state so model escalation can change the expected output. ProcessorConfig.token_estimator takes precedence. The default returns None.

quota_scope identifies strategies sharing RPM, TPM, and cooldown by object identity. It defaults to concurrency_scope, which identifies shared provider capacity. Override them independently when account quota and client capacity have different ownership. Explicit None uses strategy identity; unhashable objects are supported. A raising property fails closed with non-retryable QuotaScopeError, without copying the property exception message. Initial configuration raises this error directly (including call_result and pool construction); a middleware-selected strategy failure becomes a per-item quota_scope_error result. No provider request or quota debit follows it.

async def prepare() -> None

Initialize resources before making LLM calls (e.g., create caches, initialize clients).

Default: No-op

async def execute(prompt, attempt, timeout, state=None)

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

Execute an LLM call.

Parameters:

• prompt (str): The prompt to send to the LLM
• attempt (int): Which retry attempt this is (1, 2, 3, ...)
• timeout (float): Maximum time to wait for response (seconds)
• Note: Timeout enforcement is handled by the framework wrapping this call in asyncio.wait_for()
• state (RetryState | None): Mutable per-work-item state provided by the framework so strategies can track partial progress across retries

Returns: Tuple of (output, token_usage, metadata)

• output (TOutput): The LLM response
• token_usage (TokenUsage): Token usage dict with optional keys: input_tokens, output_tokens, total_tokens, cached_input_tokens, cache_write_input_tokens
• metadata (dict[str, Any] | None): Provider metadata forwarded to WorkItemResult.metadata; None when there is none

Since 1.0, returning the 2-tuple (output, token_usage) fails the item with a ValueError (category logic_error, not retried) that names your strategy. Its token usage is still recorded.

Raises: Any exception to trigger retry (if retryable) or failure

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

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

Called when ProcessorConfig(dry_run=True) is set. Override this method to provide realistic mock data for testing.

Parameters:

• prompt (str): 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 for prompt: {prompt[:50]}..." as output
• Returns mock token usage: 100 input, 50 output, 150 total tokens

Example override:

class MyStrategy(LLMCallStrategy[Output]):
    async def dry_run(self, prompt: str) -> tuple[Output, TokenUsage]:
        # Return realistic mock data
        mock_output = Output(result="Test result")
        mock_tokens: TokenUsage = {
            "input_tokens": len(prompt.split()),
            "output_tokens": 50,
            "total_tokens": len(prompt.split()) + 50,
        }
        return mock_output, mock_tokens

async def 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 the next attempt
• Modify prompts based on validation errors
• Track error patterns across attempts
• Make intelligent decisions (e.g., escalate to smarter model only on validation errors)

Parameters:

• exception (Exception): The exception that was raised during execute()
• attempt (int): Which attempt number failed (1, 2, 3, ...)
• state (RetryState | None): Retry state that persists across attempts (v0.3.0)

Default: No-op

Use Cases:

1. Smart Model Escalation - Only escalate to expensive models on validation errors, not network errors:

class SmartModelEscalationStrategy(LLMCallStrategy[Output]):
    async def on_error(self, exception: Exception, attempt: int, state=None) -> None:
        if state is not None and isinstance(exception, ValidationError):
            state.set("validation_failures", state.get("validation_failures", 0) + 1)

    async def execute(self, prompt: str, attempt: int, timeout: float, state=None):
        # Only escalate model on validation errors
        failures = state.get("validation_failures", 0) if state is not None else 0
        model_index = min(failures, len(MODELS) - 1)
        model = MODELS[model_index]
        # Make call with appropriate model...

1. Smart Retry with Partial Parsing - Build better retry prompts based on what failed:

class SmartRetryStrategy(LLMCallStrategy[Output]):
    async def on_error(self, exception: Exception, attempt: int, state=None) -> None:
        if state is not None and isinstance(exception, ValidationError):
            state.set("last_validation_error", exception)
            # last_response set in execute() before raising

    async def execute(self, prompt: str, attempt: int, timeout: float, state=None):
        if attempt > 1 and state and state.get("last_validation_error"):
            # Build smart retry prompt with partial parsing feedback
            prompt = self._create_retry_prompt_with_partial_data(prompt, state)
        # Make call with improved prompt...

1. Error Type Tracking - Distinguish between different error types:

class ErrorTrackingStrategy(LLMCallStrategy[Output]):
    async def on_error(self, exception: Exception, attempt: int, state=None) -> None:
        if state is None:
            return
        if isinstance(exception, ValidationError):
            key = "validation_errors"
        elif isinstance(exception, ConnectionError):
            key = "network_errors"
        elif "429" in str(exception):
            key = "rate_limit_errors"
        else:
            key = "other_errors"
        state.set(key, state.get(key, 0) + 1)

Important Notes:

• Exceptions in on_error() are caught and logged by the framework - they won't crash processing
• on_error() is only called when execute() raises an exception, not on success
• The error is still propagated to the framework's retry logic after on_error() returns
• Share strategy/client instances across work items, and keep all item-specific mutation in the supplied RetryState. Use concurrency-safe metrics or observers for batch-wide counters.

See Also:

• examples/example_smart_model_escalation.py - Complete smart model escalation example
• examples/example_gemini_smart_retry.py - Smart retry with partial parsing

async def cleanup() -> None

Clean up resources after all attempts complete (e.g., delete caches, close clients).

Default: No-op

Custom Strategy Example:

from llm_minder import LLMCallStrategy, TokenUsage

class MyCustomStrategy(LLMCallStrategy[str]):
    async def execute(
        self, prompt: str, attempt: int, timeout: float, state=None
    ) -> tuple[str, TokenUsage, None]:
        # Your custom LLM API call
        response = await my_llm_api.generate(prompt)

        tokens: TokenUsage = {
            "input_tokens": response.input_tokens,
            "output_tokens": response.output_tokens,
            "total_tokens": response.total_tokens,
        }

        return response.text, tokens, 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.

llm_minder.ModelStrategy

ModelStrategy(model: LLMModel, response_parser: None = None, *, temperature: float | None = None, 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 = None, 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 = None, 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. Defaults to None, which omits the parameter so the provider default applies (v0.27; earlier releases defaulted to 0.0). Always passed to model.generate(), so a custom LLMModel must treat None as "omit".	None
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(). For a config that changes per attempt, override :meth:attempt_generation_config instead of execute().	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/llm_minder/llm_strategies.py

def __init__(
    self,
    model: "LLMModel",
    response_parser: Callable[[LLMResponse], TOutput] | None = None,
    *,
    temperature: float | None = None,
    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. Defaults to ``None``,
            which omits the parameter so the provider default applies
            (v0.27; earlier releases defaulted to ``0.0``). Always passed
            to ``model.generate()``, so a custom ``LLMModel`` must treat
            ``None`` as "omit".
        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()``. For a config that changes per attempt, override
            :meth:`attempt_generation_config` instead of ``execute()``.
        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
    self._model_lease: weakref.ReferenceType[_StrategyLease] | None = None
    self._direct_model_lease: _StrategyLease | None = None
    # 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.

attempt_generation_config

attempt_generation_config(attempt: int, state: RetryState | None) -> dict[str, Any] | None

Return config to merge over generation_config for this attempt.

Override it to change provider config per attempt without overriding execute(), for example to raise an output-token bound after a truncated response. The returned keys replace the same top-level keys of generation_config for this call only (a shallow merge); None, the default, sends generation_config unchanged. Don't mutate self.generation_config: the strategy may be shared by concurrent items.

Parameters:

Name	Type	Description	Default
attempt	int	The logical attempt number, as passed to execute().	required
state	RetryState | None	The item's retry state, as passed to execute(); it holds whatever on_error() recorded about earlier attempts.	required

Source code in src/llm_minder/llm_strategies.py

def attempt_generation_config(
    self, attempt: int, state: RetryState | None
) -> dict[str, Any] | None:
    """Return config to merge over ``generation_config`` for this attempt.

    Override it to change provider config per attempt without overriding
    ``execute()``, for example to raise an output-token bound after a
    truncated response. The returned keys replace the same top-level keys of
    ``generation_config`` for this call only (a shallow merge); ``None``, the
    default, sends ``generation_config`` unchanged. Don't mutate
    ``self.generation_config``: the strategy may be shared by concurrent
    items.

    Args:
        attempt: The logical attempt number, as passed to ``execute()``.
        state: The item's retry state, as passed to ``execute()``; it holds
            whatever ``on_error()`` recorded about earlier attempts.
    """
    return None

cleanup async

cleanup() -> None

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

Source code in src/llm_minder/llm_strategies.py

async def cleanup(self) -> None:
    """Delegate to model.cleanup() if the model has a managed lifecycle."""
    if isinstance(self.model, ManagedLLMModel):
        lease = self._model_lease() if self._model_lease is not None else None
        if lease is not None:
            await lease.release(self.model)
            self._direct_model_lease = None
        elif self._model_lease is None:
            # No lease has ever been acquired (e.g. custom managed models).
            # An expired weak reference instead means ownership was already
            # released; a redundant close must not bypass a peer's lease.
            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.

Source code in src/llm_minder/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.
    """
    # 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}
    config = self.generation_config
    attempt_config = self.attempt_generation_config(attempt, state)
    if attempt_config is not None:
        config = {**(config or {}), **attempt_config}
    if config is not None:
        gen_kwargs["config"] = config
    async with provider_call(state):
        llm_response = await self.model.generate(prompt, **gen_kwargs)

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

async def prepare(self) -> None:
    """Delegate to model.prepare() if the model has a managed lifecycle."""
    from ._internal.strategy_lifecycle import _acquire_lease, _retain_child_lease

    if (
        isinstance(self.model, ManagedLLMModel)
        and getattr(self.model, "_abl_owned_lifecycle", False) is True
    ):
        lease = self._model_lease() if self._model_lease is not None else None
        if lease is None or lease.released or lease.loop() is not asyncio.get_running_loop():
            lease = _acquire_lease(self.model)
            self._model_lease = weakref.ref(lease)
        if not _retain_child_lease(self, lease):
            self._direct_model_lease = lease
        await lease.shared.ensure_ready(self.model)
    elif 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/llm_minder/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)

Per-attempt config

To change provider config from one attempt to the next, override attempt_generation_config() rather than execute(). Its keys replace the same top-level keys of generation_config for that call. Timing, token accounting on parse failures and later fixes to execute() still apply.

from typing import Any

from llm_minder import DeepSeekStrategy, RetryState


class GrowingOutputBound(DeepSeekStrategy):
    """Double the output-token bound on each retry: 1024, 2048, 4096, ..."""

    def attempt_generation_config(
        self, attempt: int, state: RetryState | None
    ) -> dict[str, Any] | None:
        return {"max_tokens": 1024 * 2 ** (attempt - 1)}

A truncated response raises TruncatedResponseError, which the built-in classifiers treat as non-retryable response_truncated, so the item would fail on its first attempt. To retry a truncation with the larger bound, and nothing else, mark it retryable in a classifier (new in 1.0 and provisional):

from llm_minder import (
    DefaultErrorClassifier,
    ErrorInfo,
    ParallelBatchProcessor,
    TruncatedResponseError,
)


class RetryTruncations(DefaultErrorClassifier):
    def classify(self, exception: Exception) -> ErrorInfo:
        if isinstance(exception, TruncatedResponseError):
            return ErrorInfo(
                is_retryable=True,
                is_rate_limit=False,
                is_timeout=False,
                error_category="response_truncated",
            )
        return super().classify(exception)


processor = ParallelBatchProcessor(error_classifier=RetryTruncations())

Subclass the classifier your provider uses (for DeepSeek, OpenAIErrorClassifier). Safety blocks and refusals keep their own reason and stay final.

Helpers for custom execute() overrides

A strategy that makes its own provider call can report what the built-in strategies report. provider_call(state) times the call and records it as the attempt's provider time, even when the call raises. attach_token_usage(exception, tokens) keeps a billed response's tokens in the item's totals when handling the response fails afterwards, and keeps the exception's type, so classification and on_error() see the original error. Both are new in 0.32 and provisional.

import json
from typing import Any

from llm_minder import (
    LLMCallStrategy,
    RetryState,
    TokenUsage,
    attach_token_usage,
    provider_call,
)


class MyStrategy(LLMCallStrategy[dict]):
    def __init__(self, client: Any) -> None:
        self.client = client

    async def execute(
        self, prompt: str, attempt: int, timeout: float, state: RetryState | None = None
    ) -> tuple[dict, TokenUsage, dict[str, Any] | None]:
        async with provider_call(state):
            response = await self.client.generate(prompt)
        tokens: TokenUsage = {"total_tokens": response.total_tokens}
        try:
            output = json.loads(response.text)
        except ValueError as e:
            attach_token_usage(e, tokens)
            raise
        return output, tokens, None

Setting the _failed_token_usage attribute on an exception yourself, which is what attach_token_usage() does, is also supported.

llm_minder.provider_call

provider_call(state: RetryState | None)

Time a provider call and record it as this attempt's provider time.

Use it around the call itself, inside execute(), with the state passed to execute()::

async def execute(self, prompt, attempt, timeout, state=None):
    async with provider_call(state):
        response = await self.model.generate(prompt, config=config)
    ...

The time is recorded even when the call raises. It fills AttemptTiming.provider_seconds and the provider side of the timing statistics. Plain with works too. state=None records nothing. Leave response parsing outside the block, so it isn't counted as provider time. If the block runs more than once in an attempt, the last one wins.

Source code in src/llm_minder/provider_calls.py

def __init__(self, state: RetryState | None) -> None:
    self._state = state
    self._started = 0.0

llm_minder.attach_token_usage

attach_token_usage(exception: BaseException, tokens: TokenUsage | Mapping[str, int]) -> None

Attach billed token usage to an exception before re-raising it.

Use it when a response the provider already billed fails afterwards, for example in parsing, so the failed attempt's tokens still count::

try:
    output = parse(response)
except ValueError as e:
    attach_token_usage(e, response.token_usage)
    raise

The exception keeps its type, so classification and on_error() see the original error. The usage is copied onto the exception's _failed_token_usage attribute, which the framework reads first; setting that attribute yourself is also supported.

Source code in src/llm_minder/provider_calls.py

def attach_token_usage(exception: BaseException, tokens: TokenUsage | Mapping[str, int]) -> None:
    """Attach billed token usage to an exception before re-raising it.

    Use it when a response the provider already billed fails afterwards, for
    example in parsing, so the failed attempt's tokens still count::

        try:
            output = parse(response)
        except ValueError as e:
            attach_token_usage(e, response.token_usage)
            raise

    The exception keeps its type, so classification and ``on_error()`` see the
    original error. The usage is copied onto the exception's
    ``_failed_token_usage`` attribute, which the framework reads first; setting
    that attribute yourself is also supported.
    """
    exception.__dict__["_failed_token_usage"] = dict(tokens)

PydanticAIStrategy

llm_minder.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/llm_minder/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 'llm-minder[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/llm_minder/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/llm_minder/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)
    """
    async with provider_call(state):
        result = await self.agent.run(prompt)

    # 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"),
    }
    # pydantic-ai counts cache reads and writes inside input_tokens, as ABL does.
    for source, key in (
        ("cache_read_tokens", "cached_input_tokens"),
        ("cache_write_tokens", "cache_write_input_tokens"),
    ):
        value = getattr(usage, source, None) if usage else None
        if isinstance(value, int) and not isinstance(value, bool) and value > 0:
            tokens[key] = value  # type: ignore[literal-required]

    # 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

llm_minder.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:~llm_minder.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 llm_minder import DeepSeekModel, DeepSeekStrategy from llm_minder.parsing import pydantic_json_parser

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

model = DeepSeekModel.from_api_key("deepseek-v4-flash", 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/llm_minder/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:`~llm_minder.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 llm_minder import DeepSeekModel, DeepSeekStrategy
        >>> from llm_minder.parsing import pydantic_json_parser
        >>>
        >>> class Classification(BaseModel):
        ...     label: str
        ...     confidence: float
        >>>
        >>> model = DeepSeekModel.from_api_key("deepseek-v4-flash", 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

        candidate_text, reason = recovered
        output = model_cls.model_validate_json(candidate_text)
        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

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

pydantic_json_parser(Model) strips a normal outer Markdown fence and performs strict Pydantic JSON/schema validation. Recovery is disabled by default.

parser = pydantic_json_parser(
    Classification,
    recover_trailing_markdown=True,
)
strategy = OpenAIStrategy(model, parser)

With recovery enabled, strict parsing still runs first. On failure, the parser uses a real JSON decoder to read exactly one complete top-level object or array, accepts only the explicitly supported trailing closing-fence artifacts (three backticks, with or without the observed trailing underscore), and then runs normal Pydantic schema validation. It does not repair malformed JSON or accept scalars, multiple JSON values, arbitrary prose, or schema-invalid data. Those failures continue through the configured retry policy.

A recovered WorkItemResult exposes typed recovery properties backed by metadata. Processor stats and MetricsObserver include structured_output_recoveries, structured_output_retries_avoided, and counts by structured_output_recovery_reasons.

GeminiStrategy

llm_minder.GeminiStrategy

GeminiStrategy(model: LLMModel, response_parser: None = None, *, temperature: float | None = None, 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 = None, 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 = None, 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-3.5-flash", client) strategy = GeminiStrategy(model, response_parser=lambda r: r.text)

With caching:

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

Source code in src/llm_minder/llm_strategies.py

def __init__(
    self,
    model: "LLMModel",
    response_parser: Callable[[LLMResponse], TOutput] | None = None,
    *,
    temperature: float | None = None,
    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. Defaults to ``None``,
            which omits the parameter so the provider default applies
            (v0.27; earlier releases defaulted to ``0.0``). Always passed
            to ``model.generate()``, so a custom ``LLMModel`` must treat
            ``None`` as "omit".
        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()``. For a config that changes per attempt, override
            :meth:`attempt_generation_config` instead of ``execute()``.
        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
    self._model_lease: weakref.ReferenceType[_StrategyLease] | None = None
    self._direct_model_lease: _StrategyLease | None = None
    # 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

llm_minder.OpenAIStrategy

OpenAIStrategy(model: LLMModel, response_parser: None = None, *, temperature: float | None = None, 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 = None, 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 = None, 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-6-luna", 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/llm_minder/llm_strategies.py

def __init__(
    self,
    model: "LLMModel",
    response_parser: Callable[[LLMResponse], TOutput] | None = None,
    *,
    temperature: float | None = None,
    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. Defaults to ``None``,
            which omits the parameter so the provider default applies
            (v0.27; earlier releases defaulted to ``0.0``). Always passed
            to ``model.generate()``, so a custom ``LLMModel`` must treat
            ``None`` as "omit".
        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()``. For a config that changes per attempt, override
            :meth:`attempt_generation_config` instead of ``execute()``.
        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
    self._model_lease: weakref.ReferenceType[_StrategyLease] | None = None
    self._direct_model_lease: _StrategyLease | None = None
    # 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

llm_minder.OpenRouterStrategy

OpenRouterStrategy(model: LLMModel, response_parser: None = None, *, temperature: float | None = None, 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 = None, 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 = None, 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/llm_minder/llm_strategies.py

def __init__(
    self,
    model: "LLMModel",
    response_parser: Callable[[LLMResponse], TOutput] | None = None,
    *,
    temperature: float | None = None,
    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. Defaults to ``None``,
            which omits the parameter so the provider default applies
            (v0.27; earlier releases defaulted to ``0.0``). Always passed
            to ``model.generate()``, so a custom ``LLMModel`` must treat
            ``None`` as "omit".
        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()``. For a config that changes per attempt, override
            :meth:`attempt_generation_config` instead of ``execute()``.
        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
    self._model_lease: weakref.ReferenceType[_StrategyLease] | None = None
    self._direct_model_lease: _StrategyLease | None = None
    # 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

llm_minder.DeepSeekStrategy

DeepSeekStrategy(model: LLMModel, response_parser: Callable[[LLMResponse], TOutput] | None = None, *, temperature: float | None = None, 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-v4-flash", api_key="sk-...") strategy = DeepSeekStrategy(model)

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

Source code in src/llm_minder/llm_strategies.py

def __init__(
    self,
    model: "LLMModel",
    response_parser: Callable[[LLMResponse], TOutput] | None = None,
    *,
    temperature: float | None = None,
    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

llm_minder.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, cache_group: str | 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-3.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. Merged on top of the built-in safety_ratings/finish_reason; user keys win. Added in v0.15.0.	None
cache_group	str | None	Label for this model's calls in the per-group cache statistics, in place of the system-instruction hash. A per-call config={"cache_group": ...} overrides it and is never sent. Provisional.	None

Source code in src/llm_minder/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,
    cache_group: str | 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``.
            Merged on top of the built-in ``safety_ratings``/``finish_reason``;
            user keys win. Added in v0.15.0.
        cache_group: Label for this model's calls in the per-group cache
            statistics, in place of the system-instruction hash. A per-call
            ``config={"cache_group": ...}`` overrides it and is never sent.
            Provisional.
    """
    if genai is None:
        raise ImportError(
            "google-genai is required for GeminiModel. "
            "Install with: pip install 'llm-minder[gemini]'"
        )

    self._abl_owned_lifecycle = True
    self._owns_client = False
    self._reopen_kwargs: dict[str, Any] | None = None
    self._client_closed = False
    self._client_lifecycle_lock: asyncio.Lock | None = None
    self._client_lifecycle_loop: asyncio.AbstractEventLoop | None = None
    self._model = model
    self._client = client
    self._safety_settings = safety_settings
    self._default_system_instruction = system_instruction
    self._cache_group_label = _cache_label(cache_group, "cache_group")
    self._metadata_extractors = metadata_extractors

cleanup async

cleanup() -> None

Close both transports of a factory-owned Gemini client.

Source code in src/llm_minder/models.py

async def cleanup(self) -> None:
    """Close both transports of a factory-owned Gemini client."""
    if not self._owns_client:
        return
    async with self._lifecycle_lock():
        if self._client_closed:
            return
        try:
            await self._client.aio.aclose()
        finally:
            try:
                self._client.close()
            finally:
                self._client_closed = True

generate async

generate(prompt: str | list[Any], *, temperature: float | None = None, 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.	None
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/llm_minder/models.py

async def generate(
    self,
    prompt: str | list[Any],
    *,
    temperature: float | None = None,
    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.
    """
    await self.prepare()

    # 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)
    label = _pop_cache_label(call_config) or self._cache_group_label

    # 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 not text:
        metadata = _extract_metadata(response)
        provider_reason, reason = _gemini_empty_reason(response, metadata)
        if text is None or reason in _EMPTY_TEXT_FAILURES:
            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 (finish_reason="
                f"{(metadata or {}).get('finish_reason', 'unknown')}).{safety_info}",
                tokens,
                provider_reason=provider_reason,
                reason=reason,
            )

    return _with_cache_group(
        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,
        ),
        # From the merged config: a per-call system_instruction overrides the default.
        label or _cache_group(None, call_config.get("system_instruction")),
    )

prepare async

prepare() -> None

Reopen a closed factory-owned client; caller clients stay untouched.

Source code in src/llm_minder/models.py

async def prepare(self) -> None:
    """Reopen a closed factory-owned client; caller clients stay untouched."""
    if self._owns_client and self._client_closed:
        async with self._lifecycle_lock():
            if self._client_closed:
                assert self._reopen_kwargs is not None
                self._client = genai.Client(**self._reopen_kwargs)
                self._client_closed = False

GeminiCachedModel

llm_minder.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, cache_group: str | 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-3.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-3.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. Merged on top of the built-in metadata; user keys win. Added in v0.15.0.	None
cache_group	str | None	Label for this model's calls in the per-group cache statistics, in place of the cached-content hash. A per-call config={"cache_group": ...} overrides it and is never sent. Provisional.	None

Source code in src/llm_minder/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,
    cache_group: str | None = None,
):
    """
    Args:
        model: Model name (e.g., "gemini-3.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``.
            Merged on top of the built-in metadata; user keys win.
            Added in v0.15.0.
        cache_group: Label for this model's calls in the per-group cache
            statistics, in place of the cached-content hash. A per-call
            ``config={"cache_group": ...}`` overrides it and is never sent.
            Provisional.
    """
    if genai is None:
        raise ImportError(
            "google-genai is required for GeminiCachedModel. "
            "Install with: pip install 'llm-minder[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 = copy.deepcopy(cached_content)
    self._cache_ttl_seconds = cache_ttl_seconds
    self._cache_renewal_buffer_seconds = cache_renewal_buffer_seconds
    self._auto_renew = auto_renew
    if cache_tags and "abl-content" in cache_tags:
        raise ValueError("cache_tags key 'abl-content' is reserved for content identity")
    canonical_content = _canonical_cache_content(self._cached_content)
    canonical = json.dumps(
        {
            "model": model.rsplit("/models/", 1)[-1].removeprefix("models/"),
            "contents": canonical_content,
        },
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=False,
        allow_nan=False,
    )
    self._cache_tags = dict(cache_tags or {})
    self._cache_tags["abl-content"] = hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:32]
    display_name = _encode_tags_to_display_name(self._cache_tags)
    if len(display_name) > 128:
        raise ValueError(
            "cache_tags exceed the 128-character display_name budget including "
            "the reserved abl-content fingerprint and JSON escaping"
        )
    self._safety_settings = safety_settings
    self._metadata_extractors = metadata_extractors
    self._cache_group_label = _cache_label(cache_group, "cache_group")

    self._cache: Any = None
    self._cache_created_at: float | None = None
    self._cache_expires_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/llm_minder/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/llm_minder/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._cache_expires_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 = None, 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.	None
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/llm_minder/models.py

async def generate(
    self,
    prompt: str | list[Any],
    *,
    temperature: float | None = None,
    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)
    label = _pop_cache_label(call_config) or self._cache_group_label

    requested_cache = call_config["cached_content"]
    owned_cache_name = self._cache.name
    try:
        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]
        )
    except Exception as exc:
        from google.genai.errors import APIError

        missing_cache = (
            isinstance(exc, APIError)
            and exc.code in (403, 404)
            and "cachedcontent" in str(exc.message).lower().replace(" ", "")
            and requested_cache == owned_cache_name
        )
        if not self._auto_renew or not missing_cache:
            raise
        assert self._cache_lock is not None
        async with self._cache_lock:
            if self._cache is not None and self._cache.name == requested_cache:
                self._cache = None
                self._cache_created_at = None
                self._cache_expires_at = None
                self._prepared = False
                await self._find_or_create_cache(renewing=True, exclude_name=requested_cache)
                self._prepared = True
        assert self._cache is not None
        call_config["cached_content"] = self._cache.name
        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 not text:
        metadata = _extract_metadata(response)
        provider_reason, reason = _gemini_empty_reason(response, metadata)
        if text is None or reason in _EMPTY_TEXT_FAILURES:
            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 (finish_reason="
                f"{(metadata or {}).get('finish_reason', 'unknown')}).{safety_info}",
                tokens,
                provider_reason=provider_reason,
                reason=reason,
            )

    return _with_cache_group(
        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,
        ),
        label or "gemini-cache:" + self._cache_tags["abl-content"][:12],
    )

prepare async

prepare() -> None

Find or create the Gemini cache. Idempotent.

Source code in src/llm_minder/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

llm_minder.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, prompt_cache_key: str | None = None, cache_group: str | 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-6-luna" 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
prompt_cache_key	str | None	Sent as prompt_cache_key (in extra_body) on every call, which routes requests that share it to the same prompt cache. "auto" sends abl-si- plus a hash of the system instruction actually sent on each call, and no key when there is none. A prompt_cache_key in per-call config wins; one in extra_body as well raises ValueError. Provisional.	None
cache_group	str | None	Label for this model's calls in the per-group cache statistics, ahead of the key and the system-instruction hash. A per-call config={"cache_group": ...} overrides it and is never sent. Provisional.	None

Source code in src/llm_minder/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,
    prompt_cache_key: str | None = None,
    cache_group: str | None = None,
):
    """
    Args:
        model: Provider model id (e.g. "gpt-6-luna" 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.
        prompt_cache_key: Sent as ``prompt_cache_key`` (in ``extra_body``) on
            every call, which routes requests that share it to the same prompt
            cache. ``"auto"`` sends ``abl-si-`` plus a hash of the system
            instruction actually sent on each call, and no key when there is
            none. A ``prompt_cache_key`` in per-call ``config`` wins; one in
            ``extra_body`` as well raises ``ValueError``. Provisional.
        cache_group: Label for this model's calls in the per-group cache
            statistics, ahead of the key and the system-instruction hash. A
            per-call ``config={"cache_group": ...}`` overrides it and is never
            sent. Provisional.
    """
    if AsyncOpenAI is None:
        raise ImportError(
            f"openai is required for {type(self).__name__}. "
            f"Install with: pip install 'llm-minder[{self._install_extras}]'"
        )

    self._abl_owned_lifecycle = True
    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
    self._prompt_cache_key = _cache_label(prompt_cache_key, "prompt_cache_key")
    if self._prompt_cache_key is not None and extra_body and "prompt_cache_key" in extra_body:
        raise ValueError("Pass prompt_cache_key or extra_body['prompt_cache_key'], not both")
    self._cache_group_label = _cache_label(cache_group, "cache_group")
    # 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
    self._reopen_kwargs: dict[str, Any] | None = None
    self._reopen_pool_size: int | None = None
    self._caller_http_client = False
    self._client_closed = False
    self._client_used = False
    self._client_lock: asyncio.Lock | None = None
    self._client_lock_loop: asyncio.AbstractEventLoop | 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 own the client and close it here, unless an HTTP transport was supplied by the caller. Closing that SDK client would also close the caller's transport, so both remain open. A later prepare reopens an owned client with its configured pool size.

Source code in src/llm_minder/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` own the client and close it here,
    unless an HTTP transport was supplied by the caller. Closing that SDK
    client would also close the caller's transport, so both remain open.
    A later prepare reopens an owned client with its configured pool size.
    """
    if not self._owns_client or self._client is None or self._caller_http_client:
        return
    async with self._lifecycle_lock():
        if self._client_closed:
            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,
            )
        finally:
            self._client_closed = 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, prompt_cache_key: str | None = None, cache_group: str | 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:llm_minder.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. Without it, the openai SDK's own default applies (1000 connections, 100 kept alive; the same in openai 1.66.2 and 3.x): above 100 concurrent requests, connections beyond the keep-alive limit are closed after each response and reopened, and above 1000 the excess workers block waiting for a connection (see issue #25). ABL also can't see the default pool's capacity, so it can't warn or gate on it. 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/llm_minder/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,
    prompt_cache_key: str | None = None,
    cache_group: str | 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:`llm_minder.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``.**
            Without it, the openai SDK's own default applies (1000
            connections, 100 kept alive; the same in openai 1.66.2 and
            3.x): above 100 concurrent requests, connections beyond
            the keep-alive limit are closed after each response and
            reopened, and above 1000 the excess workers block waiting for
            a connection (see issue #25). ABL also can't see the default
            pool's capacity, so it can't warn or gate on it. 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 'llm-minder[{cls._install_extras}]'"
        )
    caller_http_client = client_kwargs.get("http_client") is not None
    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

        key_env = cls._api_key_env_var
        resolved_key = _os.environ.get(key_env)
        if not resolved_key:
            raise ValueError(
                f"No API key for {cls.__name__}: pass api_key= or set the "
                f"{key_env} 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_kwargs.setdefault("max_retries", 0)
    from openai import OpenAIError

    try:
        client = AsyncOpenAI(**client_kwargs)
    except OpenAIError as exc:
        # Keep all SDK authentication mechanisms and configuration errors.
        # Only translate its known missing-credentials diagnostics.
        if str(exc).startswith(
            (
                "Missing credentials.",
                "The api_key client option must be set",
            )
        ):
            raise ValueError(
                f"No credentials for {cls.__name__}: pass SDK credentials or set "
                "OPENAI_API_KEY (or OPENAI_ADMIN_KEY where supported)."
            ) from exc
        raise

    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,
        # Only when set, so subclasses written for the older constructor
        # signature (and DeepSeekModel, which takes no prompt_cache_key) keep
        # working with this factory.
        **{
            name: value
            for name, value in (
                ("prompt_cache_key", prompt_cache_key),
                ("cache_group", cache_group),
            )
            if value is not None
        },
        **(_instance_kwargs or {}),
    )
    instance.max_concurrency = max_connections
    instance._owns_client = True
    instance._caller_http_client = caller_http_client
    instance._reopen_kwargs = dict(client_kwargs)
    instance._reopen_pool_size = max_connections
    if max_connections is not None:
        instance._reopen_kwargs.pop("http_client", None)
    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 = None, 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 reasoning models that reject an explicit temperature.	None
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/llm_minder/models.py

async def generate(
    self,
    prompt: str | list[Any],
    *,
    temperature: float | None = None,
    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
            reasoning models 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.
    """
    await self.prepare()
    self._client_used = True
    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)
    label = self._apply_cache_options(extra_body, _system_text(messages))

    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
    cache_writes = self._extract_cache_write_tokens(response)

    # 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, cache_writes
        )
    message = response.choices[0].message
    text = getattr(message, "content", None)
    if text is None or text == "":
        finish_reason = getattr(response.choices[0], "finish_reason", "unknown")
        provider_reason = _reason_text(finish_reason)
        refused = _reason_text(getattr(message, "refusal", None)) is not None
        reason = "refusal" if refused else _empty_response_reason(provider_reason)
        if text is None or reason in _EMPTY_TEXT_FAILURES:
            _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,
                cache_writes,
                provider_reason=provider_reason,
                reason=reason,
            )

    return _with_cache_group(
        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,
            cache_write_input_tokens=cache_writes,
        ),
        label or _cache_group(extra_body.get("prompt_cache_key"), _system_text(messages)),
    )

prepare async

prepare() -> None

Reopen an owned client after cleanup, preserving its configured pool.

Source code in src/llm_minder/models.py

async def prepare(self) -> None:
    """Reopen an owned client after cleanup, preserving its configured pool."""
    if not self._client_closed or self._reopen_kwargs is None:
        return
    async with self._lifecycle_lock():
        if not self._client_closed:
            return
        kwargs = dict(self._reopen_kwargs)
        if self._reopen_pool_size is not None:
            kwargs["http_client"] = _build_openai_http_client(self._reopen_pool_size)
        self._client = AsyncOpenAI(**kwargs)
        self._client_closed = False
        self._client_used = False

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 allows 1000 connections but keeps only 100 alive between requests, and ABL cannot report an unknown capacity) 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/llm_minder/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
    allows 1000 connections but keeps only 100 alive between requests, and
    ABL cannot report an unknown capacity) 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._client_used
        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
    self._reopen_pool_size = concurrency
    self._client_closed = False
    # 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

llm_minder.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, api_surface: str = 'responses', prompt_cache_key: str | None = None, cache_group: str | None = None)

Bases: _ResponsesSurface, OpenAICompatibleModel

LLM model backed by OpenAI, on the Responses API by default.

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.

API surface. Since v0.27 requests use the Responses API (api_surface="responses") with store=False unless the request sets store. Chat-style request fields are translated: max_tokens → max_output_tokens, response_format → text.format, reasoning_effort → reasoning.effort, logprobs → include, Chat tools/tool_choice → Responses function tools, and Chat message lists (including assistant tool_calls and tool replies) → Responses input items. Fields with no Responses equivalent (n, stop, seed, penalties, logit_bias, legacy functions) and stream / background raise ValueError before any request. Pass api_surface="chat_completions" for the previous Chat Completions behavior. For other OpenAI-compatible servers (vLLM, proxies, Azure-style endpoints) use :class:OpenAICompatibleModel instead.

A completed Responses call that returns only function calls succeeds with text="", finish_reason="tool_calls" and the calls in metadata["tool_calls"]. An incomplete call with text succeeds with finish_reason="length" (token limit) or "content_filter"; without text it raises :class:EmptyResponseError, as does a refusal.

Example

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

Added in v0.9.0.

See :class:OpenAICompatibleModel; adds api_surface ("responses", the default, or "chat_completions").

Source code in src/llm_minder/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,
    api_surface: str = "responses",
    prompt_cache_key: str | None = None,
    cache_group: str | None = None,
):
    """See :class:`OpenAICompatibleModel`; adds ``api_surface``
    (``"responses"``, the default, or ``"chat_completions"``)."""
    if api_surface not in {"chat_completions", "responses"}:
        raise ValueError(
            f"api_surface must be 'responses' or 'chat_completions' (got {api_surface!r})."
        )
    if api_surface == "responses":
        try:
            create_response = getattr(getattr(client, "responses", None), "create", None)
        except Exception:
            create_response = None
        if not callable(create_response):
            raise ValueError(
                "OpenAIModel uses the Responses API by default, which needs a client with "
                "callable responses.create (openai>=1.66.2). Upgrade the SDK, or pass "
                'api_surface="chat_completions".'
            ) from None
    self.api_surface = api_surface
    super().__init__(
        model,
        client,
        system_instruction=system_instruction,
        extra_headers=extra_headers,
        extra_body=extra_body,
        metadata_extractors=metadata_extractors,
        prompt_cache_key=prompt_cache_key,
        cache_group=cache_group,
    )

artifact_identity_extra property

artifact_identity_extra: dict[str, Any]

Replay identity: empty on Chat Completions, so v0.26 checkpoints match.

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, api_surface: str = 'responses', prompt_cache_key: str | None = None, cache_group: str | None = None, **client_kwargs: Any) -> OpenAIModel

Build an OpenAIModel with an owned client; adds api_surface.

json_mode adds response_format={"type": "json_object"}, which the Responses surface translates to text.format.

Source code in src/llm_minder/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,
    metadata_extractors: list[MetadataExtractor] | None = None,
    api_surface: str = "responses",
    prompt_cache_key: str | None = None,
    cache_group: str | None = None,
    **client_kwargs: Any,
) -> "OpenAIModel":
    """Build an OpenAIModel with an owned client; adds ``api_surface``.

    ``json_mode`` adds ``response_format={"type": "json_object"}``, which the
    Responses surface translates to ``text.format``.
    """
    return super().from_api_key(
        model,
        api_key,
        base_url=base_url,
        system_instruction=system_instruction,
        extra_headers=extra_headers,
        extra_body=extra_body,
        json_mode=json_mode,
        max_connections=max_connections,
        metadata_extractors=metadata_extractors,
        prompt_cache_key=prompt_cache_key,
        cache_group=cache_group,
        _instance_kwargs={"api_surface": api_surface},
        **client_kwargs,
    )

generate async

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

Call the selected OpenAI API surface and normalize the response.

Source code in src/llm_minder/models.py

async def generate(
    self,
    prompt: str | list[Any],
    *,
    temperature: float | None = None,
    system_instruction: str | None = None,
    config: dict[str, Any] | None = None,
) -> LLMResponse:
    """Call the selected OpenAI API surface and normalize the response."""
    if self.api_surface == "chat_completions":
        return await super().generate(
            prompt,
            temperature=temperature,
            system_instruction=system_instruction,
            config=config,
        )
    return await self._generate_responses(
        prompt,
        temperature=temperature,
        system_instruction=system_instruction,
        config=config,
    )

OpenRouterModel

llm_minder.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, prompt_cache_key: str | None = None, cache_group: str | 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/llm_minder/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,
    prompt_cache_key: str | None = None,
    cache_group: str | None = None,
):
    """
    Args:
        model: Provider model id (e.g. "gpt-6-luna" 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.
        prompt_cache_key: Sent as ``prompt_cache_key`` (in ``extra_body``) on
            every call, which routes requests that share it to the same prompt
            cache. ``"auto"`` sends ``abl-si-`` plus a hash of the system
            instruction actually sent on each call, and no key when there is
            none. A ``prompt_cache_key`` in per-call ``config`` wins; one in
            ``extra_body`` as well raises ``ValueError``. Provisional.
        cache_group: Label for this model's calls in the per-group cache
            statistics, ahead of the key and the system-instruction hash. A
            per-call ``config={"cache_group": ...}`` overrides it and is never
            sent. Provisional.
    """
    if AsyncOpenAI is None:
        raise ImportError(
            f"openai is required for {type(self).__name__}. "
            f"Install with: pip install 'llm-minder[{self._install_extras}]'"
        )

    self._abl_owned_lifecycle = True
    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
    self._prompt_cache_key = _cache_label(prompt_cache_key, "prompt_cache_key")
    if self._prompt_cache_key is not None and extra_body and "prompt_cache_key" in extra_body:
        raise ValueError("Pass prompt_cache_key or extra_body['prompt_cache_key'], not both")
    self._cache_group_label = _cache_label(cache_group, "cache_group")
    # 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
    self._reopen_kwargs: dict[str, Any] | None = None
    self._reopen_pool_size: int | None = None
    self._caller_http_client = False
    self._client_closed = False
    self._client_used = False
    self._client_lock: asyncio.Lock | None = None
    self._client_lock_loop: asyncio.AbstractEventLoop | 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, prompt_cache_key: str | None = None, cache_group: str | 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/llm_minder/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,
    prompt_cache_key: str | None = None,
    cache_group: str | 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,
        prompt_cache_key=prompt_cache_key,
        cache_group=cache_group,
        **client_kwargs,
    )

DeepSeekModel

llm_minder.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, cache_group: str | None = None)

Bases: _ResponsesSurface, 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-v4-flash" or "deepseek-v4-pro".

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 validates the requested surface and schema locally; model IDs are passed through for provider validation without an implicit surface fallback.

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. DeepSeek's prefix cache takes no key, so there is no prompt_cache_key; cache_group labels calls for the per-group cache statistics.

Source code in src/llm_minder/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,
    cache_group: str | None = None,
):
    """See :class:`OpenAICompatibleModel`; adds the DeepSeek ``thinking``
    toggle and opt-in Responses API structured output. DeepSeek's prefix
    cache takes no key, so there is no ``prompt_cache_key``; ``cache_group``
    labels calls for the per-group cache statistics."""
    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":
        try:
            create_response = getattr(getattr(client, "responses", None), "create", None)
        except Exception:
            create_response = None
        if not callable(create_response):
            raise ValueError(
                "DeepSeek Responses requires a client with callable responses.create. "
                "Install llm-minder[deepseek] or supply a Responses-capable client."
            ) from None

    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,
        cache_group=cache_group,
    )

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, cache_group: str | 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/llm_minder/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,
    cache_group: str | 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,
        cache_group=cache_group,
        _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 = None, 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/llm_minder/models.py

async def generate(
    self,
    prompt: str | list[Any],
    *,
    temperature: float | None = None,
    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/llm_minder/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

llm_minder.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 = None, 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. The default None omits the parameter so the provider's own default applies; some models or reasoning modes reject an explicit value. Pass a number, e.g. 0.0, for more repeatable output where supported.	None
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/llm_minder/core/protocols.py

async def generate(
    self,
    prompt: str | list[Any],
    *,
    temperature: float | None = None,
    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. The default ``None`` omits the
            parameter so the provider's own default applies; some models
            or reasoning modes reject an explicit value. Pass a number,
            e.g. ``0.0``, for more repeatable output where supported.
        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

llm_minder.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. Preparation must support reuse after cleanup; cleanup must tolerate a partially completed previous cleanup. Overlapping hosts share a strategy lifecycle within one event loop. Built-in OpenAI-compatible models also share ownership across distinct ModelStrategy instances.

Added in v0.6.0.

cleanup async

cleanup() -> None

Release resources. Must be idempotent.

Source code in src/llm_minder/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/llm_minder/core/protocols.py

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

LLMResponse

llm_minder.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, cache_write_input_tokens: int = 0)

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.
cache_write_input_tokens	int	Input tokens written to the prompt cache (0 if none were reported).

Added in v0.6.0.

token_usage property

token_usage: TokenUsage

Return token counts as a TokenUsage dict.

MetadataExtractor

A callable that takes the raw provider response and returns extra keys for LLMResponse.metadata, or None. Pass a list of them as a built-in model's metadata_extractors argument; their keys are merged over the built-in metadata.

llm_minder.MetadataExtractor module-attribute

MetadataExtractor = Callable[[Any], dict[str, Any] | None]

Provider notes

DeepSeek strict JSON Schema output

DeepSeek's Responses API can enforce a JSON Schema before ABL parses the result. Pass a Pydantic model class to get that model back directly:

from pydantic import BaseModel

from llm_minder import DeepSeekModel, DeepSeekStrategy

class Verdict(BaseModel):
    valid: bool
    reason: str

model = DeepSeekModel.from_api_key(
    "deepseek-v4-flash",
    api_surface="responses",
    response_schema=Verdict,
    thinking=False,
    max_connections=64,
)
strategy = DeepSeekStrategy(
    model,
    generation_config={"max_tokens": 256},  # mapped to max_output_tokens
)

A JSON Schema mapping is also accepted; the default strategy output is then the decoded JSON value. schema_name= overrides the Pydantic class name or schema title used in the provider request.

This mode runs through the ordinary LLMCallPool and batch strategy path, so capacity admission, proactive quotas, retries, timing, token/cache accounting, and cost calculation remain active. Result metadata includes api_surface, provider_request_id, and response_schema (name, Python/schema identity, and canonical SHA-256). The same API/schema fields participate in inferred JSONL/SQLite artifact identity, preventing Chat JSON-mode or different-schema results from being replayed as compatible strict outputs.

ABL validates the surface and schema locally and passes model IDs through to DeepSeek for validation. See the provider Responses reference for available models. The deepseek extra requires OpenAI SDK 1.66.2 or newer; caller-supplied clients must expose callable responses.create. This does not add multimodal work-item support or an automatic surface fallback. The explicit fallback is Chat Completions with json_mode=True plus pydantic_json_parser(...); it requests valid JSON but does not enforce the schema and therefore cannot safely repair malformed JSON inside the object. Provider/schema rejection is non-retryable under structured_output_schema_rejected; an accepted response that fails local validation is retryable under structured_output_validation_error.

---

Provider classifier categories

PydanticAIStrategy recommends the exported PydanticAIErrorClassifier. It classifies ModelHTTPError by HTTP status (429 rate limit, 401 authentication, 403 permission denied, other deterministic 4xx client errors, transient 5xx server errors). UsageLimitExceeded is non-retryable with category usage_limit_exceeded. Exact UnexpectedModelBehavior remains a retryable validation_error; its content-filter and incomplete-tool-call subclasses retain ordinary retry backoff.

All built-in classifiers handle framework deadlines, aborts and framework timeouts before provider matching, along with named middleware-contract and quota-scope errors. Terminal rate-limit retry exhaustion uses rate_limit_retries_exceeded with is_rate_limit=False. Empty provider responses use non-retryable empty_response, and truncated ones (TruncatedResponseError) non-retryable response_truncated; both exceptions remain ValueError subclasses. Structured-output schema rejection is non-retryable structured_output_schema_rejected; local output validation is retryable structured_output_validation_error.

Gemini provider timeouts use timeout, including bare TimeoutError(). Default, OpenAI and OpenRouter provider timeouts use api_timeout. Gemini daily quota exhaustion uses non-retryable quota_exhausted only when every reported quota violation has an explicit per-day quota ID; ambiguous and mixed daily/minute violations remain rate limits. Gemini RPC retry-delay hints are parsed alongside HTTP headers.

Category changes affect per-category metrics and abort_on_error_categories. The string factory rejects unknown model kwargs with a list of accepted names, and unresolved OpenAI credentials raise ValueError.