Core API Reference

ParallelBatchProcessor

llm_minder.ParallelBatchProcessor

ParallelBatchProcessor(*, post_processor: PostProcessorFunc[TOutput, TContext] | None = None, config: ProcessorConfig | None = None, error_classifier: ErrorClassifier | None = None, rate_limit_strategy: RateLimitStrategy | None = None, middlewares: list[Middleware[TInput, TOutput, TContext]] | None = None, observers: list[ProcessorObserver] | None = None, progress_callback: ProgressCallbackFunc | None = None, artifact_store: ArtifactStore | None = None, resume: ResumePolicy = ResumePolicy.NONE)

Batch processor that executes items in parallel as individual agent calls.

This refactored version uses: - Pluggable error classification (provider-agnostic) - Pluggable rate limit strategies - Middleware pipeline for extensibility - Observer pattern for monitoring - Configuration objects for easier setup

Initialize the parallel batch processor.

Parameters:

Name	Type	Description	Default
`post_processor`	`PostProcessorFunc[TOutput, TContext] \| None`	Optional async function called after each successful item	`None`
`config`	`ProcessorConfig \| None`	Processor configuration; `ProcessorConfig()` defaults when omitted. All arguments are keyword-only since 1.0, which also removed three parameters: use `ProcessorConfig(max_workers=...)`, `ProcessorConfig(attempt_timeout=...)` and `ProcessorConfig(rate_limit=RateLimitConfig(cooldown_seconds=...))` instead of `max_workers`, `timeout_per_item` and `rate_limit_cooldown`.	`None`
`error_classifier`	`ErrorClassifier \| None`	Optional global classifier override. When omitted, each strategy's recommendation is resolved independently.	`None`
`rate_limit_strategy`	`RateLimitStrategy \| None`	Strategy for handling rate limits	`None`
`middlewares`	`list[Middleware[TInput, TOutput, TContext]] \| None`	List of middleware to apply	`None`
`observers`	`list[ProcessorObserver] \| None`	List of observers for events	`None`
`progress_callback`	`ProgressCallbackFunc \| None`	Optional callback(completed, total, current_item_id) for progress updates	`None`

Source code in src/llm_minder/parallel.py

@functools.partial(suggest_keyword_errors, removed=REMOVED_PROCESSOR_KEYWORDS)
def __init__(
    self,
    *,
    post_processor: PostProcessorFunc[TOutput, TContext] | None = None,
    config: ProcessorConfig | None = None,
    error_classifier: ErrorClassifier | None = None,
    rate_limit_strategy: RateLimitStrategy | None = None,
    middlewares: list[Middleware[TInput, TOutput, TContext]] | None = None,
    observers: list[ProcessorObserver] | None = None,
    progress_callback: "ProgressCallbackFunc | None" = None,
    artifact_store: ArtifactStore | None = None,
    resume: ResumePolicy = ResumePolicy.NONE,
):
    """
    Initialize the parallel batch processor.

    Args:
        post_processor: Optional async function called after each successful item
        config: Processor configuration; ``ProcessorConfig()`` defaults when
            omitted. All arguments are keyword-only since 1.0, which also
            removed three parameters: use ``ProcessorConfig(max_workers=...)``,
            ``ProcessorConfig(attempt_timeout=...)`` and
            ``ProcessorConfig(rate_limit=RateLimitConfig(cooldown_seconds=...))``
            instead of ``max_workers``, ``timeout_per_item`` and
            ``rate_limit_cooldown``.
        error_classifier: Optional global classifier override. When omitted,
            each strategy's recommendation is resolved independently.
        rate_limit_strategy: Strategy for handling rate limits
        middlewares: List of middleware to apply
        observers: List of observers for events
        progress_callback: Optional callback(completed, total, current_item_id) for progress updates
    """
    if config is None:
        config = ProcessorConfig()

    # Always an int after ProcessorConfig.__post_init__ resolution.
    resolved_max_workers = cast(int, config.max_workers)

    super().__init__(
        resolved_max_workers,
        post_processor,
        max_queue_size=config.max_queue_size,
        progress_callback=progress_callback,
        progress_callback_timeout=config.progress_callback_timeout,
        max_result_queue_size=config.max_result_queue_size,
    )
    self.config = config
    self.artifact_store = artifact_store
    self._artifact_store_closed = False
    # Steps an owning surface (e.g. the streaming API's progress reporter)
    # appends so they run inside the same ordered close, last.
    self._extra_cleanup_steps: list[CleanupStep] = []
    self.resume = ResumePolicy(resume)
    self._abort_controller: AbortController | None = AbortController(
        config.guardrails.abort_mode
    )
    self._guardrails_started = False
    self._batch_timeout_task: asyncio.Task[None] | None = None
    # One tracker per processor run; a reused config gets a fresh budget.
    self._budget_tracker: BudgetTracker | None = (
        BudgetTracker(
            config.guardrails,
            trip_now=self._trip_abort_now,
            announce=self._announce_abort,
        )
        if config.guardrails.budget_configured
        else None
    )
    # Always created, so per-group cache stats work without a floor.
    self._cache_tracker: CacheHitTracker | None = CacheHitTracker(
        config.guardrails,
        trip_now=self._trip_abort_now,
        announce=self._announce_abort,
    )

    # Diagnostic: high max_workers can outrun the OS open-file limit.
    _warn_if_fd_limit_low(resolved_max_workers)

    # Automatic classifiers are resolved per actual strategy. Keep the old
    # host-wide attribute only as a compatibility/debug alias.
    self._classifier_resolver = StrategyClassifierResolver(error_classifier)
    self.error_classifier: ErrorClassifier = self._classifier_resolver.compatibility_classifier
    self._strategy_configurations: dict[int, _StrategyConfiguration] = {}
    self._capacity_warning_source: tuple[str, int, str] | None = None
    self.rate_limit_strategy = rate_limit_strategy or ExponentialBackoffStrategy(
        initial_cooldown=config.rate_limit.cooldown_seconds,
        max_cooldown=config.rate_limit.max_cooldown_seconds,
        backoff_multiplier=config.rate_limit.backoff_multiplier,
        slow_start_items=config.rate_limit.slow_start_items,
        slow_start_initial_delay=config.rate_limit.slow_start_initial_delay,
        slow_start_final_delay=config.rate_limit.slow_start_final_delay,
    )

    # Set up middleware and observers
    self.middlewares = middlewares or []
    self.observers = list(observers or [])

    # Event + middleware dispatch. Delegates observer emits and the
    # middleware chain (before/after/on_error) to a stateless helper.
    self._events: EventDispatcher[TInput, TOutput, TContext] = EventDispatcher(
        observers=self.observers, middlewares=self.middlewares
    )

    self._admission_registry = AdmissionRegistry(
        rate_limit_strategy=self.rate_limit_strategy,
        events=self._events,
        max_requests_per_minute=config.max_requests_per_minute,
        max_tokens_per_minute=config.max_tokens_per_minute,
        max_cooldown_seconds=config.rate_limit.max_cooldown_seconds,
        quota_burst_seconds=config.quota_burst_seconds,
        max_workers=config.max_workers,
        fallback_cooldown_seconds=config.rate_limit.cooldown_seconds,
    )
    # A private compatibility coordinator serves direct legacy calls made
    # before a strategy exists. Once the first item is admitted, the old
    # `_rate_limit_coord` alias points at that item's real scoped state.
    self._compatibility_rate_limit_coord = RateLimitCoordinator(
        rate_limit_strategy=self.rate_limit_strategy,
        events=self._events,
        max_cooldown_seconds=config.rate_limit.max_cooldown_seconds,
        fallback_cooldown_seconds=config.rate_limit.cooldown_seconds,
    )
    self._rate_limit_coord = self._compatibility_rate_limit_coord
    self._compatibility_scope_bound = False

    # Thread-safety locks (_stats_lock / _results_lock) live on the base
    # class so both batch and streaming modes share them.

    # Strategy lifecycle management (v0.2.0, extracted in v0.7.0).
    # Tracks prepared strategies via a WeakSet so sharing one instance
    # across work items invokes prepare() exactly once.
    self._strategy_lifecycle: StrategyLifecycle[TOutput] = StrategyLifecycle()
    # Back-compat aliases used by existing private methods and tests.
    self._prepared_strategies = self._strategy_lifecycle._prepared
    self._strategy_lock = self._strategy_lifecycle._lock
    self._capacity_limiter = CapacityLimiter(
        config.max_provider_concurrency,
        max_workers=resolved_max_workers,
        startup_ramp=config.startup_ramp,
    )

    # Centralized token-usage extraction across all exception shapes.
    self._token_extractor = TokenExtractor()

    # Per-item execution engine (extracted so the single-call helper and the
    # gateway can share the exact same retry/rate-limit/token pipeline). The
    # processor is one host for the executor; it reads its deps live from
    # `self`, so it must be built here, before start(), and the worker keeps
    # calling the instance methods below (which delegate to it) so tests that
    # monkeypatch them still take effect.
    self._executor: ItemExecutor[TInput, TOutput, TContext] = ItemExecutor(self)

aborted `property`

aborted: bool

Whether this run has entered controlled guardrail termination.

add_work `async`

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

Queue a work item and register its identity-scoped admission state.

Source code in src/llm_minder/parallel.py

async def add_work(self, work_item: LLMWorkItem[TInput, TOutput, TContext]) -> None:
    """Queue a work item and register its identity-scoped admission state."""
    self._check_accepting_work()
    assert self._abort_controller is not None
    if self._abort_controller.aborted:
        raise BatchAdmissionStopped("Batch is no longer accepting work")
    has_preprocessing = bool(self._events.middlewares) or (
        type(self)._run_middlewares_before is not ParallelBatchProcessor._run_middlewares_before
    )
    # Capture before a worker can replace the strategy. Direct configuration
    # retains the caller's stack already; streaming captures once before its
    # producer task is spawned. Neither path creates a warning registry here.
    work_item._capacity_warning_source = self._capacity_warning_source
    if has_preprocessing and work_item._capacity_warning_source is None:
        work_item._capacity_warning_source = capture_capacity_warning_source()
    if not has_preprocessing:
        try:
            await self._configure_item_strategy(work_item)
        except AdmissionGateClosed as exc:
            raise BatchAdmissionClosedError(
                "Batch admission is closed; create a new processor for additional work"
            ) from exc
    await super().add_work(work_item)

get_stats `async`

get_stats() -> dict

Get processor statistics (thread-safe).

Returns:

Type	Description
`dict`	Dictionary containing processing statistics including:
`dict`	processed: Number of items processed
`dict`	succeeded: Number of successful items
`dict`	failed: Number of failed items
`dict`	rate_limit_count: Number of rate limit errors encountered
`dict`	error_counts: Dictionary of error types and their counts
`dict`	total: Total number of items queued
`dict`	start_time: Timestamp when processing started

Source code in src/llm_minder/parallel.py

async def get_stats(self) -> dict:
    """
    Get processor statistics (thread-safe).

    Returns:
        Dictionary containing processing statistics including:
        - processed: Number of items processed
        - succeeded: Number of successful items
        - failed: Number of failed items
        - rate_limit_count: Number of rate limit errors encountered
        - error_counts: Dictionary of error types and their counts
        - total: Total number of items queued
        - start_time: Timestamp when processing started
    """
    async with self._stats_lock:
        stats = self._stats.copy()
    if self._budget_tracker is not None:
        stats.update(self._budget_tracker.stats())
    if self._cache_tracker is not None:
        stats.update(self._cache_tracker.stats())
    return stats

shutdown `async`

shutdown()

Release every owned resource; safe to call repeatedly.

Same ordered close as async with exit: runtime tasks and callbacks, admission resources, prepared strategies, then the artifact store (including after an early stream exit or a failed run). Concurrent calls share one attempt; a step that succeeded is never repeated and a failed one is retried by the next call. Raises the first cleanup failure after every step has been attempted.

Source code in src/llm_minder/parallel.py

async def shutdown(self):
    """Release every owned resource; safe to call repeatedly.

    Same ordered close as ``async with`` exit: runtime tasks and
    callbacks, admission resources, prepared strategies, then the
    artifact store (including after an early stream exit or a failed
    run). Concurrent calls share one attempt; a step that succeeded is
    never repeated and a failed one is retried by the next call. Raises
    the first cleanup failure after every step has been attempted.
    """
    await self.cleanup()

start

start() -> None

Start streaming workers and the batch deadline clock.

Source code in src/llm_minder/parallel.py

def start(self) -> None:
    """Start streaming workers and the batch deadline clock."""
    super().start()
    self._start_guardrail_run()

wait_for_abort `async`

wait_for_abort() -> None

Wait until a configured batch deadline or fail-fast abort trips.

Source code in src/llm_minder/parallel.py

async def wait_for_abort(self) -> None:
    """Wait until a configured batch deadline or fail-fast abort trips."""
    assert self._abort_controller is not None
    await self._abort_controller.event.wait()

LLMWorkItem

llm_minder.LLMWorkItem `dataclass`

LLMWorkItem(item_id: str, strategy: LLMCallStrategy[TOutput], prompt: str = '', context: TContext | None = None, submission_index: int | None = None, _artifact_key: Any = None, _capacity_warning_source: tuple[str, int, str] | None = None)

Bases: Generic[TInput, TOutput, TContext]

Represents a single work item to be processed by an LLM strategy.

Attributes:

Name	Type	Description
`item_id`	`str`	Unique identifier for this work item
`strategy`	`LLMCallStrategy[TOutput]`	LLM call strategy that encapsulates how to make the LLM call
`prompt`	`str`	The prompt/input to pass to the LLM
`context`	`TContext \| None`	Optional context data passed through to results/post-processor

__post_init__

__post_init__()

Validate work item fields.

Source code in src/llm_minder/base.py

def __post_init__(self):
    """Validate work item fields."""
    self._validate_fields()

WorkItemResult

llm_minder.WorkItemResult `dataclass`

WorkItemResult(item_id: str, success: bool, output: TOutput | None = None, error: str | None = None, context: TContext | None = None, token_usage: TokenUsage = (lambda: {'input_tokens': 0, 'output_tokens': 0, 'total_tokens': 0})(), metadata: dict[str, Any] | None = None, exception: Exception | None = None, admission_wait_seconds: float = 0.0, timing: WorkItemTiming = WorkItemTiming(), submission_index: int | None = None, error_category: str | None = None, replayed_from_artifact: bool = False)

Bases: ProviderOutputViews, Generic[TOutput, TContext]

Result of processing a single work item.

Attributes:

Name	Type	Description
`item_id`	`str`	ID of the work item
`success`	`bool`	Whether processing succeeded
`output`	`TOutput \| None`	Agent output if successful, None if failed
`error`	`str \| None`	Error message if failed, None if successful
`context`	`TContext \| None`	Context data from the work item
`token_usage`	`TokenUsage`	Token usage stats (input_tokens, output_tokens, total_tokens)
`metadata`	`dict[str, Any] \| None`	Provider-specific metadata returned alongside the response — e.g. `{"provider": "Anthropic", "finish_reason": "stop", "model": "anthropic/claude-haiku-4-5"}`. Populated when the strategy returns a 3-tuple `(output, tokens, metadata)` from `execute()`; `None` when the strategy returned none. 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`). Added in v0.10.0. Gemini safety ratings are in `metadata['safety_ratings']`.
`exception`	`Exception \| None`	The originating exception for a failed result, when one was raised (all retries exhausted, or a permanent non-retryable error). `None` for successes and for non-error outcomes such as a middleware filter-skip. `call()` / `LLMCallPool.submit()` re-raise this exact exception (preserving the provider's type) rather than a generic `LLMCallError`. Batch execution detaches tracebacks after diagnostic logging to avoid retaining frame locals. `call()`, `call_result()` and `LLMCallPool` retain the original traceback. Excluded from equality so two failed results with distinct exception instances still compare equal.
`admission_wait_seconds`	`float`	Total time this item spent waiting for provider capacity across all attempts. This wait occurs before the per-attempt execution timeout starts.
`timing`	`WorkItemTiming`	Structured end-to-end and per-attempt timing details.

quota_wait_seconds `property`

quota_wait_seconds: float

Combined RPM/TPM wait, separate from provider-capacity admission.

from_dict `classmethod`

from_dict(data: Mapping[str, Any], *, output_decoder: ValueDecoder | None = None, context_decoder: ValueDecoder | None = None) -> WorkItemResult[Any, Any]

Restore a result from :meth:to_dict output.

Without decoders, custom values (including Pydantic models and dataclasses) are restored as JSON-native mappings/lists.

Source code in src/llm_minder/base.py

@classmethod
def from_dict(
    cls,
    data: Mapping[str, Any],
    *,
    output_decoder: "ValueDecoder | None" = None,
    context_decoder: "ValueDecoder | None" = None,
) -> "WorkItemResult[Any, Any]":
    """Restore a result from :meth:`to_dict` output.

    Without decoders, custom values (including Pydantic models and
    dataclasses) are restored as JSON-native mappings/lists.
    """
    from .serialization import work_item_result_from_dict

    return work_item_result_from_dict(
        data,
        output_decoder=output_decoder,
        context_decoder=context_decoder,
    )

to_dict

to_dict(*, encoder: ValueEncoder | None = None) -> dict[str, Any]

Return a versioned, JSON-safe representation of this result.

encoder may convert application-specific output/context values to supported JSON-safe values. Unsupported values raise :class:~llm_minder.ResultSerializationError.

Source code in src/llm_minder/base.py

def to_dict(self, *, encoder: "ValueEncoder | None" = None) -> dict[str, Any]:
    """Return a versioned, JSON-safe representation of this result.

    ``encoder`` may convert application-specific output/context values to
    supported JSON-safe values. Unsupported values raise
    :class:`~llm_minder.ResultSerializationError`.
    """
    from .serialization import work_item_result_to_dict

    return work_item_result_to_dict(self, encoder=encoder)

AttemptTiming

llm_minder.AttemptTiming `dataclass`

AttemptTiming(attempt: int, try_number: int, total_seconds: float = 0.0, admission_wait_seconds: float = 0.0, startup_ramp_wait_seconds: float = 0.0, execution_seconds: float = 0.0, provider_seconds: float | None = None, cooldown_wait_seconds: float = 0.0, retry_backoff_seconds: float = 0.0, success: bool = False, error_type: str | None = None, error_category: str | None = None, timeout_category: str | None = None, quota_wait_seconds: float = 0.0, estimated_input_tokens: int | None = None, estimated_output_tokens: int | None = None, reserved_tokens: int = 0, reported_tokens: int | None = None, reconciliation_delta_tokens: int | None = None, quota_scope_id: int | None = None)

Timing and outcome details for one physical strategy execution try.

WorkItemTiming

llm_minder.WorkItemTiming `dataclass`

WorkItemTiming(total_seconds: float = 0.0, attempts: list[AttemptTiming] = list(), timeout_category: str | None = None)

End-to-end timing for one item, including every retry try.

quota_wait_seconds `property`

quota_wait_seconds: float

Combined RPM/TPM wait across physical attempts.

ProcessorConfig

llm_minder.ProcessorConfig `dataclass`

ProcessorConfig(*, max_workers: int | None = None, post_processor_timeout: float = 90.0, concurrent_post_processing: bool = False, retry: RetryConfig = RetryConfig(), rate_limit: RateLimitConfig = RateLimitConfig(), max_requests_per_minute: float | None = None, progress_interval: int = 10, progress_callback_timeout: float | None = 5.0, max_queue_size: int = 0, dry_run: bool = False, max_provider_concurrency: int | None = None, startup_ramp: StartupRampConfig | None = None, guardrails: GuardrailConfig = GuardrailConfig(), attempt_timeout: float | None = None, concurrency: int | None = None, max_result_queue_size: int = 0, progress_refresh_interval_seconds: float = 0.1, max_tokens_per_minute: int | None = None, token_estimator: TokenEstimator | None = None, quota_burst_seconds: float | None = None)

Complete configuration for batch processor.

__post_init__

__post_init__() -> None

Resolve defaults and derived knobs, then validate.

Source code in src/llm_minder/core/config.py

def __post_init__(self) -> None:
    """Resolve defaults and derived knobs, then validate."""
    if self.attempt_timeout is None:
        self.attempt_timeout = 120.0
    # Derive unset knobs from the single concurrency knob (explicit values
    # win; see the concurrency field comment).
    if self.concurrency is not None:
        if self.max_workers is None:
            self.max_workers = self.concurrency
        if self.max_provider_concurrency is None:
            self.max_provider_concurrency = self.concurrency
    if self.max_workers is None:
        self.max_workers = 5
    self.validate()

validate

validate() -> None

Validate complete configuration.

Source code in src/llm_minder/core/config.py

def validate(self) -> None:
    """Validate complete configuration."""
    for name in (
        "concurrency",
        "max_workers",
        "max_provider_concurrency",
        "progress_interval",
        "max_queue_size",
        "max_result_queue_size",
        "max_tokens_per_minute",
    ):
        value = getattr(self, name)
        if value is not None:
            _validate_number(value, name=name, integer=True)
    for name in (
        "attempt_timeout",
        "post_processor_timeout",
        "progress_callback_timeout",
        "progress_refresh_interval_seconds",
        "quota_burst_seconds",
    ):
        value = getattr(self, name)
        if value is not None:
            _validate_number(value, name=name)
    if self.quota_burst_seconds is not None and self.quota_burst_seconds <= 0:
        raise ValueError("quota_burst_seconds must be finite and > 0 or None")
    if self.concurrency is not None and self.concurrency < 1:
        raise ValueError(
            f"concurrency must be >= 1 or None (got {self.concurrency}). "
            f"Set config.concurrency to a positive integer (typical: 5-50)."
        )
    if self.max_workers is None or self.max_workers < 1:
        raise ValueError(
            f"max_workers must be >= 1 (got {self.max_workers}). "
            f"Set config.max_workers to a positive integer (typical: 5-20)."
        )
    if self.max_provider_concurrency is not None and self.max_provider_concurrency < 1:
        raise ValueError(
            "max_provider_concurrency must be >= 1 or None "
            f"(got {self.max_provider_concurrency})."
        )
    if self.startup_ramp is not None:
        self.startup_ramp.validate()
    self.guardrails.validate()
    if self.attempt_timeout is not None and self.attempt_timeout <= 0:
        raise ValueError(
            f"attempt_timeout must be > 0 (got {self.attempt_timeout}). "
            f"Set config.attempt_timeout to a positive number in seconds (typical: 60-300)."
        )
    if self.post_processor_timeout <= 0:
        raise ValueError(
            f"post_processor_timeout must be > 0 (got {self.post_processor_timeout}). "
            f"Set config.post_processor_timeout to a positive number in seconds (typical: 30-120)."
        )
    if self.progress_interval < 1:
        raise ValueError(
            f"progress_interval must be >= 1 (got {self.progress_interval}). "
            f"Set config.progress_interval to a positive integer."
        )
    if self.progress_callback_timeout is not None and self.progress_callback_timeout <= 0:
        raise ValueError(
            f"progress_callback_timeout must be > 0 (got {self.progress_callback_timeout}). "
            f"Set config.progress_callback_timeout to None to disable or a positive number of seconds."
        )
    if self.max_queue_size < 0:
        raise ValueError(
            f"max_queue_size must be >= 0 (got {self.max_queue_size}). "
            f"Set config.max_queue_size to 0 for unlimited, or a positive number to limit queue size."
        )
    if (
        isinstance(self.max_result_queue_size, bool)
        or not isinstance(self.max_result_queue_size, int)
        or self.max_result_queue_size < 0
    ):
        raise ValueError(
            "max_result_queue_size must be a non-negative integer "
            f"(got {self.max_result_queue_size!r}). Set it to 0 for unlimited, "
            "or a positive integer to bound completed results waiting for a consumer."
        )
    if (
        isinstance(self.progress_refresh_interval_seconds, bool)
        or not isinstance(self.progress_refresh_interval_seconds, (int, float))
        or not math.isfinite(self.progress_refresh_interval_seconds)
        or self.progress_refresh_interval_seconds <= 0
    ):
        raise ValueError(
            "progress_refresh_interval_seconds must be finite and > 0 "
            f"(got {self.progress_refresh_interval_seconds!r})."
        )
    if self.max_requests_per_minute is not None:
        try:
            _validate_number(self.max_requests_per_minute, name="max_requests_per_minute")
            if self.max_requests_per_minute <= 0:
                raise ValueError
        except ValueError:
            raise ValueError(
                "max_requests_per_minute must be > 0 or None (and finite when set) "
                f"(got {self.max_requests_per_minute!r}). Set it to None to disable "
                "proactive rate limiting, or a positive number (including fractional RPM)."
            ) from None
    if self.max_tokens_per_minute is not None and (
        isinstance(self.max_tokens_per_minute, bool)
        or not isinstance(self.max_tokens_per_minute, int)
        or self.max_tokens_per_minute <= 0
    ):
        raise ValueError(
            "max_tokens_per_minute must be a positive integer or None "
            f"(got {self.max_tokens_per_minute!r}). Set it to None to disable "
            "proactive token admission."
        )
    if self.token_estimator is not None and not callable(self.token_estimator):
        raise TypeError("token_estimator must be callable or None")

    # Validate nested configs first
    self.retry.validate()
    self.rate_limit.validate()

    # Cross-field validations
    if self.max_queue_size > 0 and self.max_queue_size < self.max_workers:
        logger.warning(
            f"max_queue_size ({self.max_queue_size}) is less than max_workers ({self.max_workers}). "
            f"This may cause workers to starve waiting for work. "
            f"Consider setting max_queue_size >= max_workers or 0 for unlimited."
        )

RetryConfig

llm_minder.core.RetryConfig `dataclass`

RetryConfig(*, max_attempts: int = 3, initial_wait: float = 1.0, max_wait: float = 60.0, exponential_base: float = 2.0, jitter: bool = True, max_rate_limit_retries: int = 20)

Configuration for retry behavior.

max_attempts bounds retries for content/transport failures (validation errors, timeouts, connection errors, generic 5xx). Rate-limit errors are exempt from this budget: a 429 / quota / coordinated-cooldown signal is a "wait and try again", not a failed attempt, so it does not consume an attempt — the same logical attempt number is retried after the cooldown. Rate-limit retries are instead bounded separately by max_rate_limit_retries (a backstop against an endpoint that throttles forever). This keeps model-escalation strategies (which key behavior off the attempt number) from escalating just because the endpoint was busy.

__post_init__

__post_init__() -> None

Validate configuration on construction.

Source code in src/llm_minder/core/config.py

def __post_init__(self) -> None:
    """Validate configuration on construction."""
    self.validate()

validate

validate() -> None

Validate retry configuration.

Source code in src/llm_minder/core/config.py

def validate(self) -> None:
    """Validate retry configuration."""
    for name in ("max_attempts", "max_rate_limit_retries"):
        _validate_number(getattr(self, name), name=name, integer=True)
    for name in ("initial_wait", "max_wait", "exponential_base"):
        _validate_number(getattr(self, name), name=name)
    if self.max_attempts < 1:
        raise ValueError(
            f"max_attempts must be >= 1 (got {self.max_attempts}). "
            f"Set retry.max_attempts to a positive integer."
        )
    if self.max_rate_limit_retries < 0:
        raise ValueError(
            f"max_rate_limit_retries must be >= 0 (got {self.max_rate_limit_retries}). "
            f"Set retry.max_rate_limit_retries to 0 (rate limits fail immediately) "
            f"or a positive integer."
        )
    if self.initial_wait <= 0:
        raise ValueError(
            f"initial_wait must be > 0 (got {self.initial_wait}). "
            f"Set retry.initial_wait to a positive number in seconds."
        )
    if self.max_wait < self.initial_wait:
        raise ValueError(
            f"max_wait must be >= initial_wait (got max_wait={self.max_wait}, initial_wait={self.initial_wait}). "
            f"Set retry.max_wait to be at least as large as retry.initial_wait."
        )
    if self.exponential_base < 1:
        raise ValueError(
            f"exponential_base must be >= 1 (got {self.exponential_base}). "
            f"Set retry.exponential_base to 1.0 or higher (typical values: 2.0-3.0)."
        )

RateLimitConfig

llm_minder.core.RateLimitConfig `dataclass`

RateLimitConfig(*, cooldown_seconds: float = 300.0, slow_start_items: int = 50, slow_start_initial_delay: float = 2.0, slow_start_final_delay: float = 0.1, backoff_multiplier: float = 1.5, max_cooldown_seconds: float = _DEFAULT_MAX_COOLDOWN_SECONDS)

Configuration for rate limit handling.

__post_init__

__post_init__() -> None

Validate configuration on construction.

Source code in src/llm_minder/core/config.py

def __post_init__(self) -> None:
    """Validate configuration on construction."""
    for name in ("cooldown_seconds", "max_cooldown_seconds"):
        _validate_number(getattr(self, name), name=name)
        if getattr(self, name) < 0:
            raise ValueError(f"{name} must be >= 0")
    # The cap applies to the *escalated* cooldown, so it can never sit
    # below the base cooldown — lift it to match rather than rejecting
    # the config. This must be an unconditional, idempotent lift (not an
    # error, and not gated on the cap equaling its default) because
    # dataclasses.replace() re-runs __post_init__ on the *resolved*
    # values of the source config: a previously lifted cap no longer
    # equals the default and is indistinguishable from an explicit one.
    # Warn when the lift overrides a value the user plausibly set.
    if self.max_cooldown_seconds < self.cooldown_seconds:
        if self.max_cooldown_seconds != _DEFAULT_MAX_COOLDOWN_SECONDS:
            logger.warning(
                f"max_cooldown_seconds ({self.max_cooldown_seconds}) is below "
                f"cooldown_seconds ({self.cooldown_seconds}); lifting the cap to "
                f"match. Set max_cooldown_seconds >= cooldown_seconds to silence."
            )
        self.max_cooldown_seconds = self.cooldown_seconds
    self.validate()

validate

validate() -> None

Validate rate limit configuration.

Source code in src/llm_minder/core/config.py

def validate(self) -> None:
    """Validate rate limit configuration."""
    _validate_number(self.slow_start_items, name="slow_start_items", integer=True)
    for name in (
        "cooldown_seconds",
        "slow_start_initial_delay",
        "slow_start_final_delay",
        "backoff_multiplier",
        "max_cooldown_seconds",
    ):
        _validate_number(getattr(self, name), name=name)
    if self.max_cooldown_seconds < 0:
        raise ValueError("max_cooldown_seconds must be >= 0")
    if self.cooldown_seconds < 0:
        raise ValueError(
            f"cooldown_seconds must be >= 0 (got {self.cooldown_seconds}). "
            f"Set rate_limit.cooldown_seconds to a non-negative number."
        )
    if self.slow_start_items < 0:
        raise ValueError(
            f"slow_start_items must be >= 0 (got {self.slow_start_items}). "
            f"Set rate_limit.slow_start_items to 0 to disable or a positive number."
        )
    if self.slow_start_initial_delay < self.slow_start_final_delay:
        raise ValueError(
            f"slow_start_initial_delay must be >= slow_start_final_delay "
            f"(got initial={self.slow_start_initial_delay}, final={self.slow_start_final_delay}). "
            f"The delay should decrease during slow start, not increase."
        )
    if self.backoff_multiplier < 1.0:
        raise ValueError(
            f"backoff_multiplier must be >= 1.0 (got {self.backoff_multiplier}). "
            f"Set rate_limit.backoff_multiplier to 1.0 (no increase) or higher (typical: 1.5-2.0)."
        )
    # No max_cooldown_seconds >= cooldown_seconds check here: __post_init__
    # lifts the cap unconditionally, so the invariant holds by construction.
    if self.slow_start_final_delay < 0:
        raise ValueError(
            f"slow_start_final_delay must be >= 0 (got {self.slow_start_final_delay})."
        )

GuardrailConfig

llm_minder.GuardrailConfig `dataclass`

GuardrailConfig(*, total_timeout_per_item: float | None = None, batch_timeout: float | None = None, abort_on_error_categories: frozenset[str] = frozenset(), abort_mode: AbortMode = AbortMode.DRAIN_ACTIVE, max_total_tokens: int | None = None, max_total_cost: float | None = None, cost_function: Callable[[AttemptUsage], float] | None = None, min_cache_hit_rate: float | None = None, cache_hit_window: int = 50)

Optional deadlines, fail-fast policy, and a total token/cost budget.

max_total_tokens and max_total_cost are soft caps on provider usage for one processor run. Usage is counted per physical provider attempt, including failed attempts and retries, as each attempt finishes; replayed results are not counted. Reaching a cap stops the run like a batch deadline: no new provider call starts, abort_mode decides whether calls already running drain or are cancelled, and unstarted items finish with batch_budget_exceeded. Calls already running when the cap is reached can overshoot it; see the guardrails documentation for the bound.

max_total_cost requires cost_function, a synchronous callable that prices one :class:~llm_minder.AttemptUsage. A cost_function without max_total_cost only tracks cost. Budgets apply to processor runs; call(), call_result() and LLMCallPool reject them.

min_cache_hit_rate (provisional) stops the run when a cache group's hit rate falls below it. Each provider attempt that returns a result is counted under the cache_group in its metadata; the rate is cached over input tokens across the group's last cache_hit_window attempts that reported positive input, and a group isn't checked until that window is full. A trip ends the run as fail_fast with the termination category cache_hit_rate_below_floor; unstarted items finish batch_aborted. Processor runs only, like budgets. LLMCallPool uses cache_hit_window for the rolling rate in cache_stats(), without a floor.

budget_configured `property`

budget_configured: bool

Whether a token cap, cost cap, or cost tracking is configured.

AbortMode

llm_minder.AbortMode

Bases: str, Enum

How a controlled batch abort treats provider calls already running.

AttemptUsage

llm_minder.AttemptUsage `dataclass`

AttemptUsage(item_id: str, attempt: int, try_number: int, strategy: Any, usage: Mapping[str, int], success: bool)

Provider-observed usage of one physical attempt, passed to a cost function.

GuardrailConfig.cost_function receives one of these for every provider attempt whose usage the provider reported, success or failure. Fields may be added in later releases, so read them by name.

Attributes:

Name	Type	Description
`item_id`	`str`	The work item's ID.
`attempt`	`int`	The logical attempt number (as in `AttemptTiming.attempt`).
`try_number`	`int`	The physical try number, which also counts rate-limit retries (as in `AttemptTiming.try_number`).
`strategy`	`Any`	The strategy that made the call. It is the live object; don't mutate it from a cost function.
`usage`	`Mapping[str, int]`	A read-only copy of the attempt's token usage (`input_tokens`, `output_tokens`, `total_tokens`, `cached_input_tokens` when reported).
`success`	`bool`	Whether the provider call succeeded.

CacheGroupTracker

llm_minder.CacheGroupTracker

CacheGroupTracker(*, window: int = DEFAULT_WINDOW)

Prompt-cache reads and writes per cache group, with a sliding window.

Provisional: new in 0.30. The same accounting backs a processor's get_stats(), LLMCallPool.cache_stats() and the min_cache_hit_rate floor.

Parameters:

Name	Type	Description	Default
`window`	`int`	How many of a group's most recent measured calls `window_hit_rate` and :meth:`below_floor` cover. A group's window rate is `None` until it has that many.	`DEFAULT_WINDOW`

Groups are labels from a call's cache_group metadata. A non-string or empty label counts as ungrouped (None), and a label longer than 128 characters is replaced by "sha256:" plus its hash. At most 64 groups, the ungrouped one included, are tracked; calls for further groups are counted together in cache_groups_overflow, with one warning.

Not thread-safe: use one tracker from one event loop or thread.

Source code in src/llm_minder/cache_groups.py

def __init__(self, *, window: int = DEFAULT_WINDOW) -> None:
    self._window = _validate_window(window)
    self.reset()

below_floor

below_floor(floor: float) -> list[CacheFloorBreach]

Return every breach of floor, the check min_cache_hit_rate applies.

A group breaches when its window is full and its window hit rate is below floor; groups are listed lowest rate first. Overflow is a breach too, listed last, since the floor can't be checked for untracked groups. An empty list means the floor holds.

Source code in src/llm_minder/cache_groups.py

def below_floor(self, floor: float) -> list[CacheFloorBreach]:
    """Return every breach of ``floor``, the check ``min_cache_hit_rate`` applies.

    A group breaches when its window is full and its window hit rate is below
    ``floor``; groups are listed lowest rate first. Overflow is a breach too,
    listed last, since the floor can't be checked for untracked groups. An
    empty list means the floor holds.
    """
    floor = _validate_floor(floor)
    breaches = [
        breach
        for label, group in self._groups.items()
        if (breach := self._breach(label, group, floor)) is not None
    ]
    breaches.sort(key=lambda breach: breach.window_hit_rate or 0.0)
    if self._overflow is not None:
        breaches.append(self._overflow_breach(floor))
    return breaches

from_dict `classmethod`

from_dict(data: Mapping[str, Any]) -> CacheGroupTracker

Restore a tracker saved with :meth:to_dict. Raises ValueError on bad data.

Source code in src/llm_minder/cache_groups.py

@classmethod
def from_dict(cls, data: Mapping[str, Any]) -> CacheGroupTracker:
    """Restore a tracker saved with :meth:`to_dict`. Raises ``ValueError`` on bad data."""
    if not isinstance(data, Mapping):
        raise ValueError("tracker state must be a mapping")
    if data.get("format") != _FORMAT:
        raise ValueError(f"not a CacheGroupTracker state (format {data.get('format')!r})")
    if data.get("version") != _VERSION:
        raise ValueError(f"unsupported CacheGroupTracker state version {data.get('version')!r}")
    tracker = cls(window=_validate_window(data.get("window")))
    groups = data.get("groups")
    if not isinstance(groups, list) or len(groups) > MAX_CACHE_GROUPS:
        raise ValueError(f"groups must be a list of at most {MAX_CACHE_GROUPS} entries")
    for index, entry in enumerate(groups):
        where = f"groups[{index}]"
        if not isinstance(entry, Mapping):
            raise ValueError(f"{where} must be a mapping")
        label = entry.get("group")
        if label is not None and normalize_cache_group(label) != label:
            raise ValueError(f"{where}.group must be None or a string of 1-128 characters")
        if label in tracker._groups:
            raise ValueError(f"{where}.group {label!r} appears more than once")
        group = _Group()
        group.load(entry, where)
        recent = entry.get("recent")
        retained = min(group.measured_calls, tracker._window)
        if not isinstance(recent, list) or len(recent) != retained:
            raise ValueError(
                f"{where}.recent must hold the last {retained} measured calls "
                "(min of measured_calls and window)"
            )
        for pair in recent:
            if (
                not isinstance(pair, list)
                or len(pair) != 2
                or any(isinstance(v, bool) or not isinstance(v, int) or v < 0 for v in pair)
                or pair[0] <= 0
            ):
                raise ValueError(
                    f"{where}.recent entries must be [input_tokens > 0, cached_input_tokens >= 0]"
                )
            group.push(pair[0], pair[1], tracker._window)
        # The retained calls are part of the totals, and are all of them when
        # every measured call is still in the window.
        complete = retained == group.measured_calls
        for name, part, total in (
            ("input_tokens", group.recent_input, group.input_tokens),
            ("cached_input_tokens", group.recent_cached, group.cached_input_tokens),
        ):
            if part > total or (complete and part != total):
                raise ValueError(f"{where}.recent doesn't match {name}")
        # Each measured call that left the window reported at least one input token.
        if group.input_tokens - group.recent_input < group.measured_calls - retained:
            raise ValueError(f"{where}.recent doesn't match input_tokens")
        tracker._groups[label] = group
    overflow = data.get("overflow")
    if overflow is not None:
        if not isinstance(overflow, Mapping):
            raise ValueError("overflow must be a mapping or None")
        if len(tracker._groups) < MAX_CACHE_GROUPS:
            raise ValueError(f"overflow is only possible with {MAX_CACHE_GROUPS} groups")
        tracker._overflow = _GroupCounters()
        tracker._overflow.load(overflow, "overflow")
    return tracker

record

record(group: object, usage: Mapping[str, Any] | None, *, miss_reason: object = None) -> None

Count one call that returned a result.

Parameters:

Name	Type	Description	Default
`group`	`object`	The call's `cache_group` (normalized as described on the class).	required
`usage`	`Mapping[str, Any] \| None`	Its token usage: `input_tokens`, `cached_input_tokens` and `cache_write_input_tokens`. `None`, or no positive `input_tokens`, counts the call as unmeasured.	required
`miss_reason`	`object`	A provider's explanation of a cache miss, kept per group as `last_miss_reason` (non-strings and empty strings are ignored; longer ones are cut to 64 characters).	`None`

Source code in src/llm_minder/cache_groups.py

def record(
    self,
    group: object,
    usage: Mapping[str, Any] | None,
    *,
    miss_reason: object = None,
) -> None:
    """Count one call that returned a result.

    Args:
        group: The call's ``cache_group`` (normalized as described on the class).
        usage: Its token usage: ``input_tokens``, ``cached_input_tokens`` and
            ``cache_write_input_tokens``. ``None``, or no positive
            ``input_tokens``, counts the call as unmeasured.
        miss_reason: A provider's explanation of a cache miss, kept per group
            as ``last_miss_reason`` (non-strings and empty strings are ignored;
            longer ones are cut to 64 characters).
    """
    if usage is not None and not isinstance(usage, Mapping):
        raise TypeError(f"usage must be a mapping or None (got {type(usage).__name__})")
    self._record(normalize_cache_group(group), usage, normalize_miss_reason(miss_reason))

record_result

record_result(result: WorkItemResult[Any, Any]) -> None

Count a finished result, for rebuilding statistics from saved rows.

Uses the result's cache_group and cache_miss_reason metadata and its token_usage. Failed results are skipped. A result's usage adds up every attempt of its item, billed failed attempts included, whereas live tracking counts only the attempt that returned; a tracker rebuilt this way can read lower than the live one did when retries were billed. Save and restore with :meth:to_dict/:meth:from_dict for an exact resume.

Source code in src/llm_minder/cache_groups.py

def record_result(self, result: WorkItemResult[Any, Any]) -> None:
    """Count a finished result, for rebuilding statistics from saved rows.

    Uses the result's ``cache_group`` and ``cache_miss_reason`` metadata and
    its ``token_usage``. Failed results are skipped. A result's usage adds up
    every attempt of its item, billed failed attempts included, whereas live
    tracking counts only the attempt that returned; a tracker rebuilt this way
    can read lower than the live one did when retries were billed. Save and
    restore with :meth:`to_dict`/:meth:`from_dict` for an exact resume.
    """
    if not result.success:
        return
    metadata = result.metadata if isinstance(result.metadata, Mapping) else {}
    self.record(
        metadata.get(CACHE_GROUP_KEY),
        result.token_usage,
        miss_reason=metadata.get(CACHE_MISS_REASON_KEY),
    )

reset

reset() -> None

Forget every group and the overflow counters.

Source code in src/llm_minder/cache_groups.py

def reset(self) -> None:
    """Forget every group and the overflow counters."""
    # Keys are group labels; ``None`` is the ungrouped bucket. Overflow is
    # tracked separately, never as a key, so no caller string can collide.
    self._groups: dict[str | None, _Group] = {}
    self._overflow: _GroupCounters | None = None

stats

stats() -> dict[str, Any]

Return {"cache_groups": [...], "cache_groups_overflow": ...}.

Groups are listed by input tokens, most first. Each has group, measured_calls, unmeasured_calls, input_tokens, cached_input_tokens, cache_write_input_tokens, hit_rate (cumulative), window_hit_rate and last_miss_reason. The overflow entry has the same counters without group and window_hit_rate, and is None unless there was overflow.

Source code in src/llm_minder/cache_groups.py

def stats(self) -> dict[str, Any]:
    """Return ``{"cache_groups": [...], "cache_groups_overflow": ...}``.

    Groups are listed by input tokens, most first. Each has ``group``,
    ``measured_calls``, ``unmeasured_calls``, ``input_tokens``,
    ``cached_input_tokens``, ``cache_write_input_tokens``, ``hit_rate``
    (cumulative), ``window_hit_rate`` and ``last_miss_reason``. The overflow
    entry has the same counters without ``group`` and ``window_hit_rate``,
    and is ``None`` unless there was overflow.
    """
    ordered = sorted(self._groups.items(), key=lambda item: item[1].input_tokens, reverse=True)
    return {
        "cache_groups": [
            {
                "group": label,
                **group.snapshot(),
                "window_hit_rate": group.window_rate(self._window),
            }
            for label, group in ordered
        ],
        "cache_groups_overflow": (
            self._overflow.snapshot() if self._overflow is not None else None
        ),
    }

to_dict

to_dict() -> dict[str, Any]

Return the tracker's full state as JSON-serializable data.

Includes every group's counters and its window of recent calls, so :meth:from_dict restores a tracker that behaves exactly as this one.

Source code in src/llm_minder/cache_groups.py

def to_dict(self) -> dict[str, Any]:
    """Return the tracker's full state as JSON-serializable data.

    Includes every group's counters and its window of recent calls, so
    :meth:`from_dict` restores a tracker that behaves exactly as this one.
    """
    return {
        "format": _FORMAT,
        "version": _VERSION,
        "window": self._window,
        "groups": [
            {
                "group": label,
                **group.state(),
                "recent": [[input_tokens, cached] for input_tokens, cached in group.recent],
            }
            for label, group in self._groups.items()
        ],
        "overflow": self._overflow.state() if self._overflow is not None else None,
    }

CacheFloorBreach

llm_minder.CacheFloorBreach `dataclass`

CacheFloorBreach(group: str | None, window_hit_rate: float | None, floor: float, window: int, last_miss_reason: str | None = None, overflow: bool = False)

One reason :meth:CacheGroupTracker.below_floor reports a floor breach.

Either a group whose full window falls below the floor (overflow is false; group is None for ungrouped calls), or, with overflow true, the tracker having seen more than 64 groups, beyond which it can't check the floor.

message `property`

message: str

The stop reason a processor run reports for this breach.

BatchBudgetExceeded

llm_minder.BatchBudgetExceeded

BatchBudgetExceeded(message: str, *, item_id: str | None = None)

Bases: BatchAbortedError

The run's token or cost budget stopped an accepted item before completion.

Source code in src/llm_minder/strategies/errors.py

def __init__(self, message: str, *, item_id: str | None = None) -> None:
    super().__init__(message)
    self.item_id = item_id

StartupRampConfig

llm_minder.StartupRampConfig `dataclass`

StartupRampConfig(*, initial_concurrency: int = 1, concurrency_step: int = 1, ramp_interval_seconds: float = 1.0, max_concurrency: int | None = None, jitter_seconds: float = 0.0)

Optional concurrency ramp applied when an execution host starts.

BatchResult

llm_minder.BatchResult `dataclass`

BatchResult(results: list[WorkItemResult[TOutput, TContext]], termination: BatchTermination = BatchTermination(), wall_time_seconds: float | None = None)

Bases: Generic[TOutput, TContext]

Result of processing a batch of work items.

Attributes:

Name	Type	Description
`results`	`list[WorkItemResult[TOutput, TContext]]`	Individual work item results, in completion order — the order items finished, which (with parallel workers, retries, and rate-limit cooldowns) is generally NOT the order they were added. Use :meth:`by_id` to look results up by `item_id`.
`total_items`	`int`	Total number of items in the batch
`succeeded`	`int`	Number of successful items
`failed`	`int`	Number of failed items
`total_input_tokens`	`int`	Sum of input tokens across all items
`total_output_tokens`	`int`	Sum of output tokens across all items
`total_cached_tokens`	`int`	Sum of cached input tokens across all items (v0.2.0)
`total_cache_write_tokens`	`int`	Sum of input tokens written to a prompt cache across all items

cache_hit_rate `property`

cache_hit_rate: float

Percentage (0.0–100.0) of input tokens served from cache.

A property since v0.20.0 (issue #93), consistent with the sibling zero-arg scalars (total_cached_tokens, succeeded, ...).

failures `property`

failures: list[WorkItemResult[TOutput, TContext]]

The failed results only, in completion order.

successes `property`

successes: list[WorkItemResult[TOutput, TContext]]

The successful results only, in completion order.

__post_init__

__post_init__()

Calculate summary statistics from results.

Source code in src/llm_minder/base.py

def __post_init__(self):
    """Calculate summary statistics from results."""
    self.total_items = len(self.results)
    self.succeeded = sum(1 for r in self.results if r.success)
    self.failed = sum(1 for r in self.results if not r.success)
    self.total_input_tokens = sum(r.token_usage.get("input_tokens", 0) for r in self.results)
    self.total_output_tokens = sum(r.token_usage.get("output_tokens", 0) for r in self.results)
    # v0.2.0: Aggregate cached tokens
    self.total_cached_tokens = sum(
        r.token_usage.get("cached_input_tokens", 0) for r in self.results
    )
    self.total_cache_write_tokens = sum(
        r.token_usage.get("cache_write_input_tokens", 0) for r in self.results
    )

by_id

by_id() -> dict[str, WorkItemResult[TOutput, TContext]]

Map item_id -> result for direct lookup.

Results are ordered by completion, so use this when you need to align outputs back to specific inputs. If two results somehow share an item_id, the later-completed one wins.

Source code in src/llm_minder/base.py

def by_id(self) -> dict[str, WorkItemResult[TOutput, TContext]]:
    """Map ``item_id`` -> result for direct lookup.

    Results are ordered by completion, so use this when you need to align
    outputs back to specific inputs. If two results somehow share an
    ``item_id``, the later-completed one wins.
    """
    return {r.item_id: r for r in self.results}

effective_input_tokens

effective_input_tokens(cached_token_rate: float, *, cache_write_rate: float | None = None) -> int

Estimate billable input tokens after the cache discount and write premium.

cached_token_rate is the fraction of the normal input-token price you pay for tokens served from cache. For example, Gemini charges 10% of the normal price (rate = 0.10), so 1000 cached tokens cost the same as 100 uncached tokens.

Use the named constants on :class:CachedTokenRates to avoid hardcoding magic numbers:

.. code-block:: python

result.effective_input_tokens(CachedTokenRates.OPENAI)
result.effective_input_tokens(CachedTokenRates.GEMINI)

Parameters:

Name	Type	Description	Default
`cached_token_rate`	`float`	Fraction (0.0–1.0) of the normal input price paid for cached tokens. Required since 1.0: rates differ widely between providers (Gemini 0.10, OpenAI ~0.50).	required
`cache_write_rate`	`float \| None`	Multiple of the normal input price paid for tokens written to a prompt cache (`total_cache_write_tokens`), e.g. `CachedTokenRates.OPENAI_WRITE` (1.25). Omitted, written tokens are priced as normal input, as before.	`None`

Returns:

Type	Description
`int`	Effective input tokens billed. The discount is computed by
`int`	truncating `cached_tokens * (1 - rate)` toward zero with
`int`	`int()`, and the write premium
`int`	`cache_write_tokens * (cache_write_rate - 1)` is rounded up with
`int`	`math.ceil()`, so the returned billable estimate is rounded
`int`	up when either has a fractional part — a deliberately
`int`	conservative choice for cost reporting.

Raises:

Type	Description
`ValueError`	If `cached_token_rate` is not in [0.0, 1.0], or `cache_write_rate` is negative or not finite.

Source code in src/llm_minder/base.py

def effective_input_tokens(
    self,
    cached_token_rate: float,
    *,
    cache_write_rate: float | None = None,
) -> int:
    """
    Estimate billable input tokens after the cache discount and write premium.

    ``cached_token_rate`` is the fraction of the normal input-token price
    you pay for tokens served from cache. For example, Gemini charges
    10% of the normal price (rate = 0.10), so 1000 cached tokens cost
    the same as 100 uncached tokens.

    Use the named constants on :class:`CachedTokenRates` to avoid
    hardcoding magic numbers:

    .. code-block:: python

        result.effective_input_tokens(CachedTokenRates.OPENAI)
        result.effective_input_tokens(CachedTokenRates.GEMINI)

    Args:
        cached_token_rate: Fraction (0.0–1.0) of the normal input price
            paid for cached tokens. Required since 1.0: rates differ
            widely between providers (Gemini 0.10, OpenAI ~0.50).
        cache_write_rate: Multiple of the normal input price paid for
            tokens written to a prompt cache (``total_cache_write_tokens``),
            e.g. ``CachedTokenRates.OPENAI_WRITE`` (1.25). Omitted, written
            tokens are priced as normal input, as before.

    Returns:
        Effective input tokens billed. The discount is computed by
        truncating ``cached_tokens * (1 - rate)`` toward zero with
        ``int()``, and the write premium
        ``cache_write_tokens * (cache_write_rate - 1)`` is rounded up with
        ``math.ceil()``, so the returned billable estimate is rounded
        **up** when either has a fractional part — a deliberately
        conservative choice for cost reporting.

    Raises:
        ValueError: If ``cached_token_rate`` is not in [0.0, 1.0], or
            ``cache_write_rate`` is negative or not finite.
    """
    if not 0.0 <= cached_token_rate <= 1.0:
        raise ValueError(
            f"cached_token_rate must be in [0.0, 1.0]; got {cached_token_rate}. "
            f"This is the fraction of normal price paid for cached tokens "
            f"(0.0 = free, 1.0 = no discount). For named provider rates, "
            f"use CachedTokenRates."
        )
    # cached_token_rate is what you PAY; (1 - rate) is the discount.
    # int() floors the discount toward zero -> conservative (over-)estimate
    # of effective billable tokens. See the Returns docstring.
    discount = int(self.total_cached_tokens * (1.0 - cached_token_rate))
    premium = 0
    if cache_write_rate is not None:
        if not math.isfinite(cache_write_rate) or cache_write_rate < 0.0:
            raise ValueError(
                f"cache_write_rate must be a finite number >= 0; got {cache_write_rate}. "
                f"This is the multiple of the normal input price paid for tokens "
                f"written to a prompt cache (e.g. 1.25). For named provider rates, "
                f"use CachedTokenRates."
            )
        premium = math.ceil(self.total_cache_write_tokens * (cache_write_rate - 1.0))
    return self.total_input_tokens - discount + premium

estimated_cost

estimated_cost(input_per_mtok: float, output_per_mtok: float, cached_token_rate: float, *, cache_write_rate: float | None = None) -> float

Estimate total spend from per-million-token prices.

Applies the cache discount (and, with cache_write_rate, the cache write premium) to input tokens via :meth:effective_input_tokens, so cached tokens are billed at their reduced rate.

Parameters:

Name	Type	Description	Default
`input_per_mtok`	`float`	Price per 1,000,000 input tokens (in your currency).	required
`output_per_mtok`	`float`	Price per 1,000,000 output tokens.	required
`cached_token_rate`	`float`	Fraction of the normal input price paid for cached tokens (see :class:`CachedTokenRates`). Required.	required
`cache_write_rate`	`float \| None`	Multiple of the normal input price paid for tokens written to a prompt cache (see :attr:`CachedTokenRates.OPENAI_WRITE`). Omitted, written tokens are priced as normal input.	`None`

Returns:

Type	Description
`float`	Estimated total cost: ``effective_input / 1e6 * input_per_mtok +
`float`	output / 1e6 * output_per_mtok``.

Source code in src/llm_minder/base.py

def estimated_cost(
    self,
    input_per_mtok: float,
    output_per_mtok: float,
    cached_token_rate: float,
    *,
    cache_write_rate: float | None = None,
) -> float:
    """Estimate total spend from per-million-token prices.

    Applies the cache discount (and, with ``cache_write_rate``, the cache
    write premium) to input tokens via :meth:`effective_input_tokens`, so
    cached tokens are billed at their reduced rate.

    Args:
        input_per_mtok: Price per 1,000,000 input tokens (in your currency).
        output_per_mtok: Price per 1,000,000 output tokens.
        cached_token_rate: Fraction of the normal input price paid for
            cached tokens (see :class:`CachedTokenRates`). Required.
        cache_write_rate: Multiple of the normal input price paid for
            tokens written to a prompt cache (see
            :attr:`CachedTokenRates.OPENAI_WRITE`). Omitted, written tokens
            are priced as normal input.

    Returns:
        Estimated total cost: ``effective_input / 1e6 * input_per_mtok +
        output / 1e6 * output_per_mtok``.
    """
    billable_input = self.effective_input_tokens(
        cached_token_rate, cache_write_rate=cache_write_rate
    )
    input_cost = billable_input / 1_000_000 * input_per_mtok
    output_cost = self.total_output_tokens / 1_000_000 * output_per_mtok
    return input_cost + output_cost

from_dict `classmethod`

from_dict(data: Mapping[str, Any], *, output_decoder: ValueDecoder | None = None, context_decoder: ValueDecoder | None = None) -> BatchResult[Any, Any]

Restore a batch from :meth:to_dict output.

Source code in src/llm_minder/base.py

@classmethod
def from_dict(
    cls,
    data: Mapping[str, Any],
    *,
    output_decoder: "ValueDecoder | None" = None,
    context_decoder: "ValueDecoder | None" = None,
) -> "BatchResult[Any, Any]":
    """Restore a batch from :meth:`to_dict` output."""
    from .serialization import batch_result_from_dict

    return batch_result_from_dict(
        data,
        output_decoder=output_decoder,
        context_decoder=context_decoder,
    )

from_json `classmethod`

from_json(value: str | bytes, *, output_decoder: ValueDecoder | None = None, context_decoder: ValueDecoder | None = None) -> BatchResult[Any, Any]

Restore a batch from a JSON string or UTF-8 bytes.

Source code in src/llm_minder/base.py

@classmethod
def from_json(
    cls,
    value: str | bytes,
    *,
    output_decoder: "ValueDecoder | None" = None,
    context_decoder: "ValueDecoder | None" = None,
) -> "BatchResult[Any, Any]":
    """Restore a batch from a JSON string or UTF-8 bytes."""
    from .serialization import batch_result_from_json

    return batch_result_from_json(
        value,
        output_decoder=output_decoder,
        context_decoder=context_decoder,
    )

from_jsonl `classmethod`

from_jsonl(path: str | Path, *, output_decoder: ValueDecoder | None = None, context_decoder: ValueDecoder | None = None) -> BatchResult[Any, Any]

Restore a batch from :meth:to_jsonl output.

Source code in src/llm_minder/base.py

@classmethod
def from_jsonl(
    cls,
    path: "str | Path",
    *,
    output_decoder: "ValueDecoder | None" = None,
    context_decoder: "ValueDecoder | None" = None,
) -> "BatchResult[Any, Any]":
    """Restore a batch from :meth:`to_jsonl` output."""
    from .serialization import batch_result_from_jsonl

    return batch_result_from_jsonl(
        path,
        output_decoder=output_decoder,
        context_decoder=context_decoder,
    )

in_input_order

in_input_order() -> BatchResult[TOutput, TContext]

Return a new batch whose results are sorted by submission order.

The current batch and its result list are not mutated. Ordering is never inferred from item_id because IDs may be duplicated.

Source code in src/llm_minder/base.py

def in_input_order(self) -> "BatchResult[TOutput, TContext]":
    """Return a new batch whose results are sorted by submission order.

    The current batch and its result list are not mutated. Ordering is
    never inferred from ``item_id`` because IDs may be duplicated.
    """
    missing = [result.item_id for result in self.results if result.submission_index is None]
    if missing:
        preview = ", ".join(repr(item_id) for item_id in missing[:3])
        suffix = "..." if len(missing) > 3 else ""
        raise ValueError(
            "Cannot order results by input: "
            f"{len(missing)} result(s) lack submission_index ({preview}{suffix})."
        )
    ordered = sorted(self.results, key=lambda result: cast(int, result.submission_index))
    return BatchResult(
        results=ordered,
        termination=self.termination,
        wall_time_seconds=self.wall_time_seconds,
    )

outputs

outputs() -> Iterator[TOutput]

outputs(*, with_ids: Literal[True]) -> Iterator[tuple[str, TOutput]]

outputs(*, with_ids: bool = False) -> Iterator[TOutput] | Iterator[tuple[str, TOutput]]

Iterate over the outputs of successful results, in result order.

The happy-path accessor — no if item.success loop needed:

.. code-block:: python

for output in batch.outputs():
    print(output)
for item_id, output in batch.outputs(with_ids=True):
    print(item_id, output)

Parameters:

Name	Type	Description	Default
`with_ids`	`bool`	When True, yield `(item_id, output)` pairs instead of bare outputs.	`False`

Added in v0.20.0. Failed results are skipped; iterate .failures (or check batch.failed) to handle them.

Source code in src/llm_minder/base.py

def outputs(
    self, *, with_ids: bool = False
) -> Iterator[TOutput] | Iterator[tuple[str, TOutput]]:
    """Iterate over the outputs of successful results, in result order.

    The happy-path accessor — no ``if item.success`` loop needed:

    .. code-block:: python

        for output in batch.outputs():
            print(output)
        for item_id, output in batch.outputs(with_ids=True):
            print(item_id, output)

    Args:
        with_ids: When True, yield ``(item_id, output)`` pairs instead of
            bare outputs.

    Added in v0.20.0. Failed results are skipped; iterate ``.failures``
    (or check ``batch.failed``) to handle them.
    """
    if with_ids:
        return ((r.item_id, cast(TOutput, r.output)) for r in self.results if r.success)
    return (cast(TOutput, r.output) for r in self.results if r.success)

summary

summary() -> str

Return a printable plain-text report of the whole run.

print(batch.summary()) is a complete post-run report: item counts, termination, retry totals, token totals (with replayed work accounted separately, consistent with v0.18 replay accounting), admission/execution percentiles, wall time, and failures grouped by error category. Works identically on collected batches and on batches restored via from_dict/from_json/from_jsonl.

Added in v0.20.0.

Source code in src/llm_minder/base.py

def summary(self) -> str:
    """Return a printable plain-text report of the whole run.

    ``print(batch.summary())`` is a complete post-run report: item
    counts, termination, retry totals, token totals (with replayed work
    accounted separately, consistent with v0.18 replay accounting),
    admission/execution percentiles, wall time, and failures grouped by
    error category. Works identically on collected batches and on
    batches restored via ``from_dict``/``from_json``/``from_jsonl``.

    Added in v0.20.0.
    """
    current = [r for r in self.results if not r.replayed_from_artifact]
    replayed = [r for r in self.results if r.replayed_from_artifact]

    def _tokens(results: list[WorkItemResult[TOutput, TContext]]) -> tuple[int, str, int]:
        cached = sum(r.token_usage.get("cached_input_tokens", 0) for r in results)
        written = sum(r.token_usage.get("cache_write_input_tokens", 0) for r in results)
        cache = f"cached {cached:,}" + (f", written {written:,}" if written else "")
        return (
            sum(r.token_usage.get("input_tokens", 0) for r in results),
            cache,
            sum(r.token_usage.get("output_tokens", 0) for r in results),
        )

    def _fmt_seconds(seconds: float) -> str:
        if seconds >= 100:
            return f"{seconds:,.0f}s"
        if seconds >= 10:
            return f"{seconds:.1f}s"
        return f"{seconds:.2f}s"

    def _percentile_line(label: str, values: list[float]) -> str:
        ordered = sorted(values)
        picks = []
        for q in (0.50, 0.95, 0.99):
            # Nearest-rank percentile on the sorted sample.
            index = min(len(ordered) - 1, max(0, math.ceil(q * len(ordered)) - 1))
            picks.append(_fmt_seconds(ordered[index]))
        return f"  {label:<15} p50 {picks[0]}  p95 {picks[1]}  p99 {picks[2]}"

    lines = ["Batch summary", "============="]

    replay_note = f" ({len(replayed)} replayed from artifact)" if replayed else ""
    lines.append(
        f"Items:     {self.total_items} total — {self.succeeded} succeeded, "
        f"{self.failed} failed{replay_note}"
    )

    termination = self.termination.kind
    if self.termination.kind != "completed":
        details = [
            part
            for part in (
                self.termination.reason,
                f"category={self.termination.error_category}"
                if self.termination.error_category
                else None,
                f"item={self.termination.triggering_item_id}"
                if self.termination.triggering_item_id
                else None,
            )
            if part
        ]
        if details:
            termination += f" — {'; '.join(details)}"
    lines.append(f"Stopped:   {termination}")

    extra_tries = sum(max(0, len(r.timing.attempts) - 1) for r in current)
    retried_items = sum(1 for r in current if len(r.timing.attempts) > 1)
    lines.append(f"Retries:   {extra_tries} extra attempt(s) across {retried_items} item(s)")

    input_tokens, cache_info, output_tokens = _tokens(current)
    lines.append(f"Tokens:    in {input_tokens:,} ({cache_info}) · out {output_tokens:,}")
    if replayed:
        r_in, r_cache_info, r_out = _tokens(replayed)
        lines.append(
            f"Replayed:  in {r_in:,} ({r_cache_info}) · out {r_out:,} "
            "(prior run; excluded above)"
        )

    wall = _fmt_seconds(self.wall_time_seconds) if self.wall_time_seconds is not None else "n/a"
    lines.append(f"Wall time: {wall}")

    timed = [r for r in current if r.timing.attempts]
    if timed:
        lines.append(
            _percentile_line("admission wait", [r.admission_wait_seconds for r in timed])
        )
        quota_attempts = [
            attempt
            for result in timed
            for attempt in result.timing.attempts
            if attempt.quota_scope_id is not None
        ]
        if quota_attempts:
            lines.append(_percentile_line("quota wait", [r.quota_wait_seconds for r in timed]))
            reserved = sum(attempt.reserved_tokens for attempt in quota_attempts)
            reported = sum(
                attempt.reported_tokens or 0
                for attempt in quota_attempts
                if attempt.reported_tokens is not None
            )
            refunds = sum(
                max(0, -(attempt.reconciliation_delta_tokens or 0))
                for attempt in quota_attempts
            )
            debt = sum(
                max(0, attempt.reconciliation_delta_tokens or 0) for attempt in quota_attempts
            )
            unknown = sum(
                1
                for attempt in quota_attempts
                if attempt.reserved_tokens
                and attempt.reported_tokens is None
                and attempt.reconciliation_delta_tokens is None
            )
            if reserved or reported or refunds or debt or unknown:
                lines.append(
                    f"  quota tokens    reserved {reserved:,} · reported {reported:,} · "
                    f"refunded {refunds:,} · debt {debt:,} · unknown {unknown:,}"
                )
        lines.append(_percentile_line("execution", [r.timing.execution_seconds for r in timed]))

    if self.failed:
        lines.append("Failures by category:")
        by_category: dict[str, list[str]] = {}
        for r in self.results:
            if not r.success:
                by_category.setdefault(r.error_category or "uncategorized", []).append(
                    r.item_id
                )
        for category in sorted(by_category):
            ids = by_category[category]
            shown = ", ".join(ids[:3]) + ("..." if len(ids) > 3 else "")
            lines.append(f"  {category:<15} {len(ids)} — {shown}")

    repeated: dict[tuple[str | None, str], int] = {}
    for result in self.results:
        if not result.success and result.error is not None:
            key = (result.error_category, result.error)
            repeated[key] = repeated.get(key, 0) + 1
    for (_, error), count in repeated.items():
        if count > 1:
            lines.append(f"{count} items failed with the same error: {error}")

    return "\n".join(lines)

to_dict

to_dict(*, encoder: ValueEncoder | None = None) -> dict[str, Any]

Return a versioned, JSON-safe representation of the batch.

Source code in src/llm_minder/base.py

def to_dict(self, *, encoder: "ValueEncoder | None" = None) -> dict[str, Any]:
    """Return a versioned, JSON-safe representation of the batch."""
    from .serialization import batch_result_to_dict

    return batch_result_to_dict(self, encoder=encoder)

to_json

to_json(*, encoder: ValueEncoder | None = None, indent: int | None = 2) -> str

Serialize this batch to a JSON string.

Source code in src/llm_minder/base.py

def to_json(self, *, encoder: "ValueEncoder | None" = None, indent: int | None = 2) -> str:
    """Serialize this batch to a JSON string."""
    from .serialization import batch_result_to_json

    return batch_result_to_json(self, encoder=encoder, indent=indent)

to_jsonl

to_jsonl(path: str | Path, *, encoder: ValueEncoder | None = None) -> None

Write one versioned result record per UTF-8 JSONL line.

Source code in src/llm_minder/base.py

def to_jsonl(self, path: "str | Path", *, encoder: "ValueEncoder | None" = None) -> None:
    """Write one versioned result record per UTF-8 JSONL line."""
    from .serialization import batch_result_to_jsonl

    batch_result_to_jsonl(self, path, encoder=encoder)

BatchTermination

llm_minder.BatchTermination `dataclass`

BatchTermination(kind: Literal['completed', 'batch_timeout', 'fail_fast', 'budget_exceeded', 'artifact_error'] = 'completed', reason: str | None = None, error_category: str | None = None, triggering_item_id: str | None = None)

Serializable reason a batch stopped accepting or executing work.

Grounding

llm_minder.Grounding `dataclass`

Grounding(sources: list[GroundingSource] = list(), queries: list[str] = list(), supports: list[dict[str, Any]] = list())

Web-grounding data from a grounded call (e.g. Gemini google_search).

Attributes:

Name	Type	Description
`sources`	`list[GroundingSource]`	The web sources the answer was grounded in.
`queries`	`list[str]`	Search queries the model issued (`web_search_queries`).
`supports`	`list[dict[str, Any]]`	Answer-span → source-index links, as plain dicts (`{"text", "start_index", "end_index", "chunk_indices"}`). Kept untyped for now.

from_metadata `classmethod`

from_metadata(data: Any) -> Grounding | None

Parse a metadata['grounding'] dict; lenient, never raises.

Source code in src/llm_minder/provider_output.py

@classmethod
def from_metadata(cls, data: Any) -> Grounding | None:
    """Parse a ``metadata['grounding']`` dict; lenient, never raises."""
    if not isinstance(data, dict) or not data:
        return None
    raw_sources = data.get("sources")
    sources = [
        parsed
        for entry in (raw_sources if isinstance(raw_sources, list) else ())
        if (parsed := GroundingSource.from_metadata(entry)) is not None
    ]
    raw_queries = data.get("queries")
    queries = [
        q for q in (raw_queries if isinstance(raw_queries, list) else ()) if isinstance(q, str)
    ]
    raw_supports = data.get("supports")
    supports = [
        s
        for s in (raw_supports if isinstance(raw_supports, list) else ())
        if isinstance(s, dict)
    ]
    if not sources and not queries and not supports:
        return None
    return cls(sources=sources, queries=queries, supports=supports)

GroundingSource

llm_minder.GroundingSource `dataclass`

GroundingSource(uri: str, title: str | None = None, snippet: str | None = None)

One web source backing a grounded response.

Attributes:

Name	Type	Description
`uri`	`str`	Source URL (always present; entries without one are dropped).
`title`	`str \| None`	Human-readable page title, when the provider supplied one.
`snippet`	`str \| None`	Excerpt from the source, when the provider supplied one.

from_metadata `classmethod`

from_metadata(data: Any) -> GroundingSource | None

Parse one metadata['grounding']['sources'] entry; lenient, never raises.

Source code in src/llm_minder/provider_output.py

@classmethod
def from_metadata(cls, data: Any) -> GroundingSource | None:
    """Parse one ``metadata['grounding']['sources']`` entry; lenient, never raises."""
    if not isinstance(data, dict):
        return None
    uri = data.get("uri")
    if not isinstance(uri, str) or not uri:
        return None
    title = data.get("title")
    snippet = data.get("snippet")
    return cls(
        uri=uri,
        title=title if isinstance(title, str) else None,
        snippet=snippet if isinstance(snippet, str) else None,
    )

ToolCall

llm_minder.ToolCall `dataclass`

ToolCall(id: str | None, name: str, arguments: str)

One tool/function call the model requested. Visibility only — the framework never executes tools; feed these to your own dispatch loop.

Attributes:

Name	Type	Description
`id`	`str \| None`	Provider call id, when supplied.
`name`	`str`	Tool/function name (always present; entries without one are dropped).
`arguments`	`str`	The raw JSON-string arguments, deliberately unparsed — parse with `json.loads` (and validate) yourself.

from_metadata `classmethod`

from_metadata(data: Any) -> ToolCall | None

Parse one metadata['tool_calls'] entry; lenient, never raises.

Source code in src/llm_minder/provider_output.py

@classmethod
def from_metadata(cls, data: Any) -> ToolCall | None:
    """Parse one ``metadata['tool_calls']`` entry; lenient, never raises."""
    if not isinstance(data, dict):
        return None
    name = data.get("name")
    if not isinstance(name, str) or not name:
        return None
    call_id = data.get("id")
    arguments = data.get("arguments")
    return cls(
        id=call_id if isinstance(call_id, str) else None,
        name=name,
        arguments=arguments if isinstance(arguments, str) else "",
    )

ErrorCategory

llm_minder.ErrorCategory

Bases: _StrValueEnum

Values of WorkItemResult.error_category and BatchTermination.error_category produced by the library.

Every member except CACHE_HIT_RATE_BELOW_FLOOR can be an item's category. That one only names why a run stopped; the items it stops finish batch_aborted.

API_ERROR `class-attribute` `instance-attribute`

API_ERROR = 'api_error'

A provider error with no recognized status. Retried conservatively.

API_TIMEOUT `class-attribute` `instance-attribute`

API_TIMEOUT = 'api_timeout'

The SDK or transport timed out waiting for the provider. Retried.

ARTIFACT_PREPARATION_ERROR `class-attribute` `instance-attribute`

ARTIFACT_PREPARATION_ERROR = 'artifact_preparation_error'

The item's input could not be prepared for the artifact store.

ARTIFACT_SERIALIZATION_ERROR `class-attribute` `instance-attribute`

ARTIFACT_SERIALIZATION_ERROR = 'artifact_serialization_error'

The item's result could not be serialized for the artifact store.

AUTHENTICATION `class-attribute` `instance-attribute`

AUTHENTICATION = 'authentication'

The provider rejected the credentials (HTTP 401). Not retried.

BATCH_ABORTED `class-attribute` `instance-attribute`

BATCH_ABORTED = 'batch_aborted'

The run stopped (fail-fast or an explicit abort) before this item finished.

BATCH_BUDGET_EXCEEDED `class-attribute` `instance-attribute`

BATCH_BUDGET_EXCEEDED = 'batch_budget_exceeded'

The run reached its token or cost budget before this item finished.

BATCH_DEADLINE_EXCEEDED `class-attribute` `instance-attribute`

BATCH_DEADLINE_EXCEEDED = 'batch_deadline_exceeded'

The run's batch_timeout expired before this item finished.

CACHE_HIT_RATE_BELOW_FLOOR `class-attribute` `instance-attribute`

CACHE_HIT_RATE_BELOW_FLOOR = 'cache_hit_rate_below_floor'

Termination only (provisional): a cache group's hit rate fell below GuardrailConfig.min_cache_hit_rate. Items the stop ends are batch_aborted.

CLASSIFIER_ERROR `class-attribute` `instance-attribute`

CLASSIFIER_ERROR = 'classifier_error'

The error classifier itself raised. Not retried.

CLIENT_ERROR `class-attribute` `instance-attribute`

CLIENT_ERROR = 'client_error'

Another non-retryable 4xx rejection, such as a bad request or unknown model.

CONNECTION_ERROR `class-attribute` `instance-attribute`

CONNECTION_ERROR = 'connection_error'

A connection failure seen by the default classifier. Retried.

EMPTY_RESPONSE `class-attribute` `instance-attribute`

EMPTY_RESPONSE = 'empty_response'

The provider returned a billed response with no usable text. Not retried.

FRAMEWORK_TIMEOUT `class-attribute` `instance-attribute`

FRAMEWORK_TIMEOUT = 'framework_timeout'

One attempt exceeded attempt_timeout. Retried.

FRAMEWORK_TOTAL_ITEM_TIMEOUT `class-attribute` `instance-attribute`

FRAMEWORK_TOTAL_ITEM_TIMEOUT = 'framework_total_item_timeout'

The item's end-to-end deadline (total_timeout_per_item) expired.

INSUFFICIENT_BALANCE `class-attribute` `instance-attribute`

INSUFFICIENT_BALANCE = 'insufficient_balance'

Billing or credit exhaustion (for example OpenAI insufficient_quota). Not retried.

LOGIC_ERROR `class-attribute` `instance-attribute`

LOGIC_ERROR = 'logic_error'

A programming error in application or strategy code (ValueError, TypeError, KeyError and similar). Not retried.

MIDDLEWARE_CONTRACT_ERROR `class-attribute` `instance-attribute`

MIDDLEWARE_CONTRACT_ERROR = 'middleware_contract_error'

Middleware returned a value that breaks its contract.

MIDDLEWARE_FILTERED `class-attribute` `instance-attribute`

MIDDLEWARE_FILTERED = 'middleware_filtered'

Middleware filtered the item out, so it was not executed. Never replayed.

NETWORK_ERROR `class-attribute` `instance-attribute`

NETWORK_ERROR = 'network_error'

A connection failure reported by the OpenAI-compatible SDK. Retried.

PERMISSION_DENIED `class-attribute` `instance-attribute`

PERMISSION_DENIED = 'permission_denied'

The credentials lack access to the model or resource (HTTP 403). Not retried.

QUOTA_EXHAUSTED `class-attribute` `instance-attribute`

QUOTA_EXHAUSTED = 'quota_exhausted'

A Gemini 429 whose quota violations are all daily limits, so waiting minutes won't help. Not retried.

QUOTA_SCOPE_ERROR `class-attribute` `instance-attribute`

QUOTA_SCOPE_ERROR = 'quota_scope_error'

A strategy's quota_scope could not be resolved.

RATE_LIMIT `class-attribute` `instance-attribute`

RATE_LIMIT = 'rate_limit'

The provider rate-limited the request (HTTP 429 or equivalent). Retried after a coordinated cooldown.

RATE_LIMIT_RETRIES_EXCEEDED `class-attribute` `instance-attribute`

RATE_LIMIT_RETRIES_EXCEEDED = 'rate_limit_retries_exceeded'

The item hit more rate limits than RetryConfig.max_rate_limit_retries allows. Not retried.

RESPONSE_TRUNCATED `class-attribute` `instance-attribute`

RESPONSE_TRUNCATED = 'response_truncated'

The response hit its output-token bound before producing usable text (TruncatedResponseError). Not retried by default.

SERVER_ERROR `class-attribute` `instance-attribute`

SERVER_ERROR = 'server_error'

A retryable 5xx response from the provider.

SERVER_OVERLOAD `class-attribute` `instance-attribute`

SERVER_OVERLOAD = 'server_overload'

The provider reported it is overloaded (Gemini 503). Retried.

SERVER_TIMEOUT `class-attribute` `instance-attribute`

SERVER_TIMEOUT = 'server_timeout'

The provider timed out upstream (Gemini 504). Retried.

STRUCTURED_OUTPUT_SCHEMA_REJECTED `class-attribute` `instance-attribute`

STRUCTURED_OUTPUT_SCHEMA_REJECTED = 'structured_output_schema_rejected'

The provider rejected the requested schema itself. Not retried.

STRUCTURED_OUTPUT_VALIDATION_ERROR `class-attribute` `instance-attribute`

STRUCTURED_OUTPUT_VALIDATION_ERROR = 'structured_output_validation_error'

Structured output did not match the requested schema. Retried.

TIMEOUT `class-attribute` `instance-attribute`

TIMEOUT = 'timeout'

A timeout reported by the Gemini SDK. Retried.

TOKEN_ESTIMATE_EXCEEDS_LIMIT `class-attribute` `instance-attribute`

TOKEN_ESTIMATE_EXCEEDS_LIMIT = 'token_estimate_exceeds_limit'

One attempt's estimate can never fit in the TPM limit.

TOKEN_ESTIMATION_ERROR `class-attribute` `instance-attribute`

TOKEN_ESTIMATION_ERROR = 'token_estimation_error'

The token estimator raised or returned an invalid estimate.

TOKEN_ESTIMATOR_REQUIRED `class-attribute` `instance-attribute`

TOKEN_ESTIMATOR_REQUIRED = 'token_estimator_required'

TPM admission is enabled but no token estimator applies to the item.

UNKNOWN `class-attribute` `instance-attribute`

UNKNOWN = 'unknown'

An exception no classifier recognized. Retried, in case it is transient.

UPSTREAM_ERROR `class-attribute` `instance-attribute`

UPSTREAM_ERROR = 'upstream_error'

OpenRouter could not reach or use an upstream provider. Retried.

USAGE_LIMIT_EXCEEDED `class-attribute` `instance-attribute`

USAGE_LIMIT_EXCEEDED = 'usage_limit_exceeded'

PydanticAI stopped the run at a configured usage limit. Not retried.

VALIDATION_ERROR `class-attribute` `instance-attribute`

VALIDATION_ERROR = 'validation_error'

The response failed validation (for example a Pydantic ValidationError). Retried, since the model may do better on another attempt.

TimeoutCategory

llm_minder.TimeoutCategory

Bases: _StrValueEnum

Values of AttemptTiming.timeout_category and WorkItemTiming.timeout_category.

ADMISSION_TIMEOUT `class-attribute` `instance-attribute`

ADMISSION_TIMEOUT = 'admission_timeout'

An item or batch deadline expired while the item waited for provider capacity.

FRAMEWORK_EXECUTION_TIMEOUT `class-attribute` `instance-attribute`

FRAMEWORK_EXECUTION_TIMEOUT = 'framework_execution_timeout'

One provider attempt exceeded attempt_timeout.

FRAMEWORK_TOTAL_ITEM_TIMEOUT `class-attribute` `instance-attribute`

FRAMEWORK_TOTAL_ITEM_TIMEOUT = 'framework_total_item_timeout'

The item deadline expired during a provider call.

PROVIDER_OR_TRANSPORT_TIMEOUT `class-attribute` `instance-attribute`

PROVIDER_OR_TRANSPORT_TIMEOUT = 'provider_or_transport_timeout'

The provider SDK or transport raised its own timeout.

LLMMinderError

llm_minder.LLMMinderError

Bases: Exception

Base class for the exception types llm-minder defines.

Catch it to handle the library's own errors in one place. They keep their other bases too (for example :class:ItemDeadlineExceeded is still a :class:TimeoutError and :class:ArtifactError a :class:RuntimeError), so existing except clauses still match. Invalid arguments are still reported with the built-in ValueError/TypeError, and provider SDK exceptions and errors raised by your own strategies are not wrapped.

Every exception type the library exports subclasses it, alongside its original built-in base, so except LLMMinderError catches the library's own errors and existing except TimeoutError (or RuntimeError, ValueError) clauses keep working.

Lifecycle exceptions

llm_minder.StreamFinalizationError

StreamFinalizationError(message: str, *, cause: BaseException | None = None)

Bases: RuntimeError, LLMMinderError

The result stream could not finish normally.

Raised by :meth:ParallelBatchProcessor.results when finalization was cancelled (for example by shutdown() before finish() completed) or when a process-control exception interrupted it. Deliberately distinct from :class:asyncio.CancelledError: the consumer itself was not cancelled. The originating exception, when there is one, is __cause__.

Source code in src/llm_minder/base.py

def __init__(self, message: str, *, cause: BaseException | None = None) -> None:
    super().__init__(message)
    if cause is not None:
        self.__cause__ = cause

llm_minder.CleanupInterruptedError

CleanupInterruptedError(step_name: str, *, cause: BaseException | None = None)

Bases: RuntimeError, LLMMinderError

A cleanup step was interrupted without the closing caller being cancelled.

Raised when a step's private task was cancelled by a third party, or when a user cleanup() raised :class:asyncio.CancelledError itself. The step is not checkpointed; a later explicit close retries it. This is an ordinary :class:Exception so it never marks the caller's task cancelled.

Source code in src/llm_minder/_internal/cleanup.py

def __init__(self, step_name: str, *, cause: BaseException | None = None) -> None:
    self.step_name = step_name
    super().__init__(
        f"Cleanup step {step_name!r} was interrupted before it completed; "
        "a later explicit close retries it"
    )
    if cause is not None:
        self.__cause__ = cause

Admission closure

llm_minder.BatchAdmissionClosedError

Bases: RuntimeError, LLMMinderError

The processor no longer accepts work; create a new processor to submit more.

add_work() raises BatchAdmissionClosedError after finish, shutdown, or abort. Rejected submissions receive no submission index. Use a new processor for further work.

Cleanup and shutdown

`async def cleanup() -> None` / `async def shutdown() -> None`

Release every owned resource in dependency order. cleanup(), shutdown(), and async with exit all run the same ordered close:

Runtime tasks and callbacks: the stream finalizer, workers, queued work, progress callbacks, background post-processors, and callback threads.
Admission resources (quota scopes and the rate-limit coordinator).
Prepared strategies (strategy.cleanup()).
The artifact store, including after an early stream exit or a failed run.

await processor.shutdown()

Behavior (see docs/cleanup-lifecycle-contract.md for the full contract):

Every step is attempted. A failing strategy cleanup() does not stop sibling strategies or the artifact store from closing. Every failure is logged with a traceback; with no body exception, the first failure is raised after all steps have run. Inside async with, a body exception stays primary and cleanup failures are only logged.
Repeat calls are safe. A step that succeeded is never repeated. A step that failed or was interrupted is retried by the next explicit close. Concurrent calls share one in-flight attempt. Because a retry re-invokes a strategy's whole cleanup(), user cleanup() implementations must be idempotent and safe after a partially completed earlier attempt.
No deadline. Cleanup waits for the artifact store and the call-pool drain to finish. The two-second worker and progress thresholds only log a warning and keep waiting; any step still running after 30 seconds logs one warning. A dependent resource is never closed while the resource it depends on is still running, including synchronous callback threads.
Cancellation. Cancelling the task that is closing once defers that cancellation until cleanup finishes, then re-raises it (cleanup errors are logged, not raised). Cancelling it a second time force-aborts: the running step is abandoned, later steps are skipped and logged, and the cancellation propagates immediately. This is the operator's escape from a cleanup that never returns and deliberately sacrifices durability.
Interruptions. A strategy cleanup() that raises CancelledError itself, or whose private task is cancelled by something else, surfaces as CleanupInterruptedError (an ordinary Exception); the caller's task is not cancelled and the next close retries the step.
KeyboardInterrupt and SystemExit raised by a cleanup step keep their type and take precedence over a deferred cancellation.
Once a close has started, preparing a new strategy raises RuntimeError.

Provider metadata

Response metadata (`WorkItemResult.metadata`)

Provider metadata (Gemini safety ratings and finish reason, OpenRouter provider/routed model, etc.) flows into WorkItemResult.metadata — a plain dict[str, Any] | None. The parsed output stays in WorkItemResult.output; you no longer wrap it in a separate response object.

Conservative structured-output recovery uses three reserved metadata keys: structured_output_recovered, structured_output_recovery_reason, and structured_output_retries_avoided. Read them through the corresponding typed properties on LLMResponse or WorkItemResult.

Removed: the old GeminiResponse wrapper and the include_metadata opt-in were removed in v0.6.0. Read metadata off result.metadata instead. For Gemini safety ratings specifically, result.metadata["safety_ratings"] carries them (the result.gemini_safety_ratings field was removed in 1.0).

Usage:

result = await processor.process_all()
first = result.results[0]
ratings = (first.metadata or {}).get("safety_ratings")
if ratings and ratings.get("HARM_CATEGORY_HATE_SPEECH") == "HIGH":
    log_flagged_content(first.output)

Two more reserved keys feed the per-group cache statistics (provisional, new in 0.30): cache_group, a string naming the prompt-cache group an attempt belongs to, and cache_miss_reason, a provider's explanation of a cache miss. Built-in models set cache_group; see Cache hit-rate floor.

Typed auxiliary output (grounding, reasoning, tool calls, logprobs)

Stable from 1.0, except logprobs. The grounding, reasoning and tool_calls views and reserved keys are covered by the stability promise. logprobs carries each provider's own structure and stays provisional, so read it defensively.

Provider-specific structured output travels through metadata under four reserved keys with documented plain-dict shapes (JSON-serializable, so persisting metadata as-is works):

Key	Shape	Emitted by
`grounding`	`{"sources": [{"uri", "title"}], "queries": [str], "supports": [dict]}`	Gemini models, when the response carries `google_search` grounding
`reasoning`	`str`: the model's reasoning/thinking trace	Chat Completions: `reasoning_content` (e.g. DeepSeek), falling back to `reasoning` (e.g. OpenRouter). Responses API: reasoning-item text, else the joined reasoning summaries (OpenAI, when requested)
`tool_calls`	`[{"id": str\\|None, "name": str, "arguments": str}]`, with `arguments` as the raw JSON string	OpenAI-compatible models: Chat `tool_calls`, or Responses `function_call` items (`id` is the `call_id`)
`logprobs`	provider-shaped `dict`/`list` (plain JSON)	OpenAI-compatible models, when requested: the Chat logprobs object, or a list of Responses per-token entries

Both LLMResponse and WorkItemResult expose lazy read-only typed views over these keys — parsed from metadata on each access, never cached, never stored twice:

result = await processor.process_all()
item = result.results[0]

if item.grounding:                       # Grounding | None
    for source in item.grounding.sources:  # list[GroundingSource]
        print(source.uri, source.title)
    print(item.grounding.queries)          # list[str]

print(item.reasoning)                    # str | None
for call in item.tool_calls or []:       # list[ToolCall] | None
    print(call.name, call.arguments)     # arguments: raw JSON string
print(item.logprobs)                     # Any | None (provider-shaped)

The view classes (Grounding, GroundingSource, ToolCall) are exported at the top level. Parsing is lenient: malformed metadata yields None (or drops the bad entry) rather than raising — which also means a future shape change degrades to None views rather than errors.

Boundaries:

tool_calls is visibility only — the framework never executes tools. Feed them to your own dispatch loop (or use an agent framework via PydanticAIStrategy). Extracted for OpenAI-compatible models only; Gemini function-call parts are not extracted.
On Chat Completions, a response whose content is null (for example a pure tool-call turn) raises EmptyResponseError before any result exists, so tool_calls surfaces only when the model returned text alongside the calls. On OpenAIModel's Responses surface, a function-call-only reply succeeds with empty text and tool_calls populated.
Auxiliary output does not survive empty/safety-blocked responses — the call fails first.

TokenUsage

llm_minder.TokenUsage

Bases: TypedDict

Token usage statistics from LLM API calls.

All fields are optional to accommodate different provider APIs. Different providers may return different subsets of these fields.

Fields

input_tokens: Number of tokens in the input/prompt output_tokens: Number of tokens in the output/completion total_tokens: Provider-reported total, or input + output when omitted. A valid explicit total is authoritative even when the components differ. cached_input_tokens: Number of input tokens served from cache. Part of input_tokens. cache_write_input_tokens: Number of input tokens written to a prompt cache (OpenAI cache_write_tokens, Anthropic cache_creation_input_tokens). Part of input_tokens; some providers bill them above the normal input price.

CachedTokenRates

llm_minder.CachedTokenRates

Named cached-token price rates for the providers we support.

Each constant is the fraction of the normal input-token price you pay for tokens served from cache (so 0.10 = "10% of normal" = "90% discount"). Pass these to :meth:BatchResult.effective_input_tokens to compute provider-aware billable token estimates:

.. code-block:: python

result.effective_input_tokens(CachedTokenRates.OPENAI)

Rates are accurate as of v0.9.0 (early 2026); confirm with each provider's pricing page before using these for invoicing.

The *_WRITE* constants are the other direction: the multiple of the normal input price paid for tokens written to a cache. Pass them as cache_write_rate to :meth:BatchResult.effective_input_tokens or :meth:BatchResult.estimated_cost:

.. code-block:: python

result.estimated_cost(
    1.25, 10.0,
    cached_token_rate=CachedTokenRates.OPENAI,
    cache_write_rate=CachedTokenRates.OPENAI_WRITE,
)

Notes

Cache writes are reported in cache_write_input_tokens. They are part of input_tokens, so without a cache_write_rate they are priced as normal input.
OpenRouter routes to many upstream providers, each with its own rate. Pick the constant for the upstream that actually served your request (visible in LLMResponse.metadata['provider']).

ANTHROPIC_READ `class-attribute` `instance-attribute`

ANTHROPIC_READ: float = 0.1

Anthropic prompt cache reads: 10% of normal.

ANTHROPIC_WRITE `class-attribute` `instance-attribute`

ANTHROPIC_WRITE: float = 1.25

Anthropic prompt cache writes with the default 5-minute TTL: 1.25× normal.

ANTHROPIC_WRITE_1H `class-attribute` `instance-attribute`

ANTHROPIC_WRITE_1H: float = 2.0

Anthropic prompt cache writes with the 1-hour TTL: 2× normal.

DEEPSEEK `class-attribute` `instance-attribute`

DEEPSEEK: float = 0.02

DeepSeek context cache: cached tokens cost ~2% of normal (V4 Flash bills cache hits at $0.0028/M vs $0.14/M cache-miss input, a price drop that took effect April 2026; earlier the discount was ~10%).

GEMINI `class-attribute` `instance-attribute`

GEMINI: float = 0.1

Gemini context cache: cached tokens cost 10% of normal.

OPENAI `class-attribute` `instance-attribute`

OPENAI: float = 0.5

OpenAI prompt caching: cached tokens cost 50% of normal for gpt-4o-class models. Other model families may differ; check OpenAI's pricing page.

OPENAI_WRITE `class-attribute` `instance-attribute`

OPENAI_WRITE: float = 1.25

OpenAI prompt cache writes (30-minute retention): 1.25× normal input.

Type aliases

SimpleBatchProcessor[T], SimpleWorkItem[T] and SimpleResult[T] shorten the generic types for string prompts with no context. PostProcessorFunc and ProgressCallbackFunc are the callable types accepted by post_processor and progress_callback.

llm_minder.SimpleBatchProcessor `module-attribute`

SimpleBatchProcessor = ParallelBatchProcessor[str, _T, None]

Type alias for ParallelBatchProcessor[str, T, None].

Use when you have string prompts, a typed output, and no context.

Example

async with SimpleBatchProcessorMyOutput as processor: ...

llm_minder.SimpleWorkItem `module-attribute`

SimpleWorkItem = LLMWorkItem[str, _T, None]

Type alias for LLMWorkItem[str, T, None].

Use when creating work items with string prompts, typed output, and no context.

Example

item = SimpleWorkItemMyOutput

llm_minder.SimpleResult `module-attribute`

SimpleResult = WorkItemResult[_T, None]

Type alias for WorkItemResult[T, None].

Use when working with results that have no context.

Example

result: SimpleResult[MyOutput] = results[0]

llm_minder.PostProcessorFunc `module-attribute`

PostProcessorFunc = Callable[[WorkItemResult[TOutput, TContext]], Awaitable[None] | None]

Type of post_processor: called with each WorkItemResult; sync or async.

llm_minder.ProgressCallbackFunc `module-attribute`

ProgressCallbackFunc = Callable[[int, int, str], Awaitable[None] | None]

Type of progress_callback: called as (completed, total, current_item_id); sync or async.

Removed names

BatchProcessor, ProcessingStats, LLMGateway and grounding_metadata_extractor were removed in 1.0; see Removed in 1.0 for replacements.

Core API Reference

ParallelBatchProcessor

llm_minder.ParallelBatchProcessor

aborted property

add_work async

get_stats async

shutdown async

start

wait_for_abort async

LLMWorkItem

llm_minder.LLMWorkItem dataclass

__post_init__

WorkItemResult

llm_minder.WorkItemResult dataclass

quota_wait_seconds property

from_dict classmethod

to_dict

AttemptTiming

llm_minder.AttemptTiming dataclass

WorkItemTiming

llm_minder.WorkItemTiming dataclass

quota_wait_seconds property

ProcessorConfig

llm_minder.ProcessorConfig dataclass

__post_init__

validate

RetryConfig

llm_minder.core.RetryConfig dataclass

__post_init__

validate

RateLimitConfig

llm_minder.core.RateLimitConfig dataclass

__post_init__

validate

GuardrailConfig

llm_minder.GuardrailConfig dataclass

budget_configured property

AbortMode

llm_minder.AbortMode

AttemptUsage

llm_minder.AttemptUsage dataclass

CacheGroupTracker

llm_minder.CacheGroupTracker

below_floor

from_dict classmethod

record

record_result

reset

stats

to_dict

CacheFloorBreach

llm_minder.CacheFloorBreach dataclass

message property

BatchBudgetExceeded

llm_minder.BatchBudgetExceeded

StartupRampConfig

llm_minder.StartupRampConfig dataclass

BatchResult

llm_minder.BatchResult dataclass

cache_hit_rate property

failures property

successes property

__post_init__

by_id

effective_input_tokens

estimated_cost

from_dict classmethod

from_json classmethod

from_jsonl classmethod

in_input_order

outputs

summary

to_dict

to_json

to_jsonl

BatchTermination

llm_minder.BatchTermination dataclass

Grounding

llm_minder.Grounding dataclass

from_metadata classmethod

aborted `property`

add_work `async`

get_stats `async`

shutdown `async`

wait_for_abort `async`

llm_minder.LLMWorkItem `dataclass`

llm_minder.WorkItemResult `dataclass`

quota_wait_seconds `property`

from_dict `classmethod`

llm_minder.AttemptTiming `dataclass`

llm_minder.WorkItemTiming `dataclass`

quota_wait_seconds `property`

llm_minder.ProcessorConfig `dataclass`

llm_minder.core.RetryConfig `dataclass`

llm_minder.core.RateLimitConfig `dataclass`

llm_minder.GuardrailConfig `dataclass`

budget_configured `property`

llm_minder.AttemptUsage `dataclass`

from_dict `classmethod`

llm_minder.CacheFloorBreach `dataclass`

message `property`

llm_minder.StartupRampConfig `dataclass`

llm_minder.BatchResult `dataclass`

cache_hit_rate `property`

failures `property`

successes `property`

from_dict `classmethod`

from_json `classmethod`

from_jsonl `classmethod`

llm_minder.BatchTermination `dataclass`

llm_minder.Grounding `dataclass`

from_metadata `classmethod`

llm_minder.GroundingSource `dataclass`

from_metadata `classmethod`

llm_minder.ToolCall `dataclass`

from_metadata `classmethod`

API_ERROR `class-attribute` `instance-attribute`

API_TIMEOUT `class-attribute` `instance-attribute`

ARTIFACT_PREPARATION_ERROR `class-attribute` `instance-attribute`

ARTIFACT_SERIALIZATION_ERROR `class-attribute` `instance-attribute`

AUTHENTICATION `class-attribute` `instance-attribute`

BATCH_ABORTED `class-attribute` `instance-attribute`

BATCH_BUDGET_EXCEEDED `class-attribute` `instance-attribute`

BATCH_DEADLINE_EXCEEDED `class-attribute` `instance-attribute`

CACHE_HIT_RATE_BELOW_FLOOR `class-attribute` `instance-attribute`

CLASSIFIER_ERROR `class-attribute` `instance-attribute`

CLIENT_ERROR `class-attribute` `instance-attribute`

CONNECTION_ERROR `class-attribute` `instance-attribute`

EMPTY_RESPONSE `class-attribute` `instance-attribute`

FRAMEWORK_TIMEOUT `class-attribute` `instance-attribute`

FRAMEWORK_TOTAL_ITEM_TIMEOUT `class-attribute` `instance-attribute`

INSUFFICIENT_BALANCE `class-attribute` `instance-attribute`

LOGIC_ERROR `class-attribute` `instance-attribute`

MIDDLEWARE_CONTRACT_ERROR `class-attribute` `instance-attribute`

MIDDLEWARE_FILTERED `class-attribute` `instance-attribute`

NETWORK_ERROR `class-attribute` `instance-attribute`

PERMISSION_DENIED `class-attribute` `instance-attribute`

QUOTA_EXHAUSTED `class-attribute` `instance-attribute`

QUOTA_SCOPE_ERROR `class-attribute` `instance-attribute`

RATE_LIMIT `class-attribute` `instance-attribute`

RATE_LIMIT_RETRIES_EXCEEDED `class-attribute` `instance-attribute`

RESPONSE_TRUNCATED `class-attribute` `instance-attribute`

SERVER_ERROR `class-attribute` `instance-attribute`

SERVER_OVERLOAD `class-attribute` `instance-attribute`

SERVER_TIMEOUT `class-attribute` `instance-attribute`

STRUCTURED_OUTPUT_SCHEMA_REJECTED `class-attribute` `instance-attribute`

STRUCTURED_OUTPUT_VALIDATION_ERROR `class-attribute` `instance-attribute`

TIMEOUT `class-attribute` `instance-attribute`

TOKEN_ESTIMATE_EXCEEDS_LIMIT `class-attribute` `instance-attribute`

TOKEN_ESTIMATION_ERROR `class-attribute` `instance-attribute`

TOKEN_ESTIMATOR_REQUIRED `class-attribute` `instance-attribute`

UNKNOWN `class-attribute` `instance-attribute`

UPSTREAM_ERROR `class-attribute` `instance-attribute`

USAGE_LIMIT_EXCEEDED `class-attribute` `instance-attribute`

VALIDATION_ERROR `class-attribute` `instance-attribute`

ADMISSION_TIMEOUT `class-attribute` `instance-attribute`

FRAMEWORK_EXECUTION_TIMEOUT `class-attribute` `instance-attribute`

FRAMEWORK_TOTAL_ITEM_TIMEOUT `class-attribute` `instance-attribute`

PROVIDER_OR_TRANSPORT_TIMEOUT `class-attribute` `instance-attribute`

`async def cleanup() -> None` / `async def shutdown() -> None`