Errors, Classification and Rate Limits

Every exception type the library exports subclasses LLMMinderError and keeps its built-in base (TimeoutError, RuntimeError or ValueError), so existing except clauses still match. BatchBudgetExceeded is documented with the budget settings on the core page, LLMCallError with single calls, and the artifact errors on the artifacts page.

For which category each built-in classifier assigns, see Provider classifier categories.

Error classification

ErrorClassifier

llm_minder.strategies.ErrorClassifier

Bases: ABC

Abstract base class for classifying LLM provider errors.

classify abstractmethod

classify(exception: Exception) -> ErrorInfo

Classify an exception and determine handling strategy.

Parameters:

Name	Type	Description	Default
exception	Exception	The exception to classify	required

Returns:

Type	Description
ErrorInfo	ErrorInfo with classification details

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

@abstractmethod
def classify(self, exception: Exception) -> ErrorInfo:
    """
    Classify an exception and determine handling strategy.

    Args:
        exception: The exception to classify

    Returns:
        ErrorInfo with classification details
    """
    pass

ErrorInfo

llm_minder.ErrorInfo dataclass

ErrorInfo(is_retryable: bool, is_rate_limit: bool, is_timeout: bool, error_category: str, suggested_wait: float | None = None, hint: str | None = None)

Structured information about an error.

Attributes:

Name	Type	Description
is_retryable	bool	Whether the framework should retry the call.
is_rate_limit	bool	Whether this is a rate-limit error (triggers a coordinated cooldown rather than per-item backoff).
is_timeout	bool	Whether this is a timeout.
error_category	str	A short label for stats/logging.
suggested_wait	float | None	For rate limits, a server-suggested minimum wait in seconds (e.g. parsed from a Retry-After header). The RateLimitCoordinator honors this as a floor on the cooldown: the backoff strategy may wait longer, subject to the configured maximum. None means "no suggestion; use the strategy's value as-is".
hint	str | None	Optional human-readable remediation hint for the operator (e.g. "top up your prepaid balance" for a 402). Surfaced in the logs when the error is non-retryable so a misconfiguration doesn't look like a generic API bug. None means no extra guidance.

DefaultErrorClassifier

llm_minder.DefaultErrorClassifier

Bases: ErrorClassifier

Default error classifier that handles common error types.

classify

classify(exception: Exception) -> ErrorInfo

Classify common errors with conservative defaults.

Precedence: framework errors, an HTTP 429 carried by the exception, validation and programming-error types, then message heuristics for untyped exceptions.

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

def classify(self, exception: Exception) -> ErrorInfo:
    """Classify common errors with conservative defaults.

    Precedence: framework errors, an HTTP 429 carried by the exception,
    validation and programming-error types, then message heuristics for
    untyped exceptions.
    """
    error_str = str(exception).lower()

    info = self._framework_prelude(exception)
    if info is not None:
        return info

    # PydanticAIStrategy uses this default classifier. Preserve immediate
    # validation retries by the exact optional SDK type, never its name.
    # ContentFilterError and IncompleteToolCall subclasses keep backoff.
    try:
        from pydantic_ai.exceptions import UnexpectedModelBehavior

        if type(exception) is UnexpectedModelBehavior:
            return ErrorInfo(
                is_retryable=True,
                is_rate_limit=False,
                is_timeout=False,
                error_category="validation_error",
            )
    except ImportError:
        pass

    info = self._structured_rate_limit(exception) or self._typed_failure(exception)
    if info is not None:
        return info

    # Message heuristics apply only to exceptions not classified by type or
    # status above (e.g. SDKs without a built-in classifier, test doubles).
    if _insufficient_quota(exception):
        return _insufficient_quota_info()
    if http_status(exception) is None and self._matches_rate_limit(error_str):
        return ErrorInfo(
            is_retryable=True,  # Rate limits are retryable - framework handles cooldown
            is_rate_limit=True,
            is_timeout=False,
            error_category="rate_limit",
        )

    # Check for API timeout (retryable - might be transient)
    if isinstance(exception, (TimeoutError, asyncio.TimeoutError)) or "timeout" in error_str:
        return ErrorInfo(
            is_retryable=True,
            is_rate_limit=False,
            is_timeout=True,
            error_category="api_timeout",
        )

    # Check for connection errors
    if isinstance(exception, ConnectionError) or "connection" in error_str:
        return ErrorInfo(
            is_retryable=True,
            is_rate_limit=False,
            is_timeout=False,
            error_category="connection_error",
        )

    return self._generic_tail(exception)

GeminiErrorClassifier

llm_minder.GeminiErrorClassifier

Bases: ErrorClassifier

Google Gemini-specific error classification.

Dispatches on google.genai.errors.APIError.code (the HTTP status) when the SDK is installed, then falls back to generic message/type-based classification — so mocks and non-Gemini exceptions still classify sensibly even without the [gemini] extra.

classify

classify(exception: Exception) -> ErrorInfo

Classify Gemini-specific errors.

Source code in src/llm_minder/classifiers/gemini.py

def classify(self, exception: Exception) -> ErrorInfo:
    """Classify Gemini-specific errors."""
    info = self._framework_prelude(exception)
    if info is not None:
        return info

    # Dispatch on the genai SDK's status code when available. When the
    # SDK isn't installed (or the exception isn't a genai one), fall
    # through to the generic checks below so rate limits, timeouts, and
    # logic bugs are still recognized.
    info = self._classify_genai_exception(exception)
    if info is not None:
        return info

    # Check for PydanticAI validation errors
    try:
        from pydantic_ai.exceptions import UnexpectedModelBehavior

        if isinstance(exception, UnexpectedModelBehavior):
            return ErrorInfo(
                is_retryable=True,  # Retry validation errors
                is_rate_limit=False,
                is_timeout=False,
                error_category="validation_error",
            )
    except ImportError:
        pass

    # A non-genai exception carrying HTTP 429, then validation and
    # programming-error types, all outrank message text.
    info = self._structured_rate_limit(exception) or self._typed_failure(exception)
    if info is not None:
        return info

    # Fallback for untyped exceptions: billing exhaustion, then rate-limit
    # text (mocks and other providers).
    error_str = str(exception)
    info = self._message_rate_limit(exception, RATE_LIMIT_PATTERNS)
    if info is not None:
        return info

    # Check if it looks like a timeout
    if isinstance(exception, (TimeoutError, asyncio.TimeoutError)) or self._matches_any_pattern(
        error_str, TIMEOUT_PATTERNS
    ):
        return ErrorInfo(
            is_retryable=True,
            is_rate_limit=False,
            is_timeout=True,
            error_category="timeout",
        )

    return self._generic_tail(exception)

OpenAIErrorClassifier

llm_minder.OpenAIErrorClassifier

Bases: ErrorClassifier

Classifier for OpenAI SDK exceptions plus generic fallbacks.

Designed to be subclassed: provider-specific classifiers (e.g. :class:OpenRouterErrorClassifier) override :meth:classify to handle extra cases first and delegate to super().classify() for the rest.

OpenRouterErrorClassifier

llm_minder.OpenRouterErrorClassifier

Bases: OpenAIErrorClassifier

OpenAI-compatible classifier with OpenRouter-specific overrides.

PydanticAIErrorClassifier

llm_minder.PydanticAIErrorClassifier

Bases: OpenAIErrorClassifier

Use HTTP status policy for model failures and stop on usage limits.

Rate-limit strategies

RateLimitConfig controls cooldown and slow-start; a RateLimitStrategy decides how long each cooldown lasts.

RateLimitStrategy

llm_minder.RateLimitStrategy

Bases: ABC

Strategy for handling rate limits.

on_rate_limit abstractmethod async

on_rate_limit(worker_id: int, consecutive_limit_count: int) -> float

Called when rate limit detected.

Parameters:

Name	Type	Description	Default
worker_id	int	ID of the worker that hit the rate limit	required
consecutive_limit_count	int	Number of consecutive rate limits	required

Returns:

Type	Description
float	Cooldown duration in seconds

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

@abstractmethod
async def on_rate_limit(
    self,
    worker_id: int,
    consecutive_limit_count: int,
) -> float:
    """
    Called when rate limit detected.

    Args:
        worker_id: ID of the worker that hit the rate limit
        consecutive_limit_count: Number of consecutive rate limits

    Returns:
        Cooldown duration in seconds
    """
    ...

should_apply_slow_start abstractmethod

should_apply_slow_start(items_since_resume: int) -> tuple[bool, float]

Check if slow-start delay should be applied.

Parameters:

Name	Type	Description	Default
items_since_resume	int	Number of items processed since last rate limit	required

Returns:

Type	Description
tuple[bool, float]	(should_delay, delay_seconds)

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

@abstractmethod
def should_apply_slow_start(self, items_since_resume: int) -> tuple[bool, float]:
    """
    Check if slow-start delay should be applied.

    Args:
        items_since_resume: Number of items processed since last rate limit

    Returns:
        (should_delay, delay_seconds)
    """
    ...

ExponentialBackoffStrategy

llm_minder.ExponentialBackoffStrategy

ExponentialBackoffStrategy(initial_cooldown: float = 300.0, max_cooldown: float = 600.0, backoff_multiplier: float = 1.5, slow_start_items: int = 50, slow_start_initial_delay: float = 2.0, slow_start_final_delay: float = 0.1)

Bases: RateLimitStrategy

Exponential backoff with progressive slow-start.

Initialize exponential backoff strategy.

Parameters:

Name	Type	Description	Default
initial_cooldown	float	Initial cooldown duration in seconds	300.0
max_cooldown	float	Maximum cooldown duration in seconds	600.0
backoff_multiplier	float	Multiplier for exponential backoff	1.5
slow_start_items	int	Number of items for slow start after cooldown	50
slow_start_initial_delay	float	Initial delay between items during slow start	2.0
slow_start_final_delay	float	Final delay between items during slow start	0.1

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

def __init__(
    self,
    initial_cooldown: float = 300.0,
    max_cooldown: float = 600.0,
    backoff_multiplier: float = 1.5,
    slow_start_items: int = 50,
    slow_start_initial_delay: float = 2.0,
    slow_start_final_delay: float = 0.1,
):
    """
    Initialize exponential backoff strategy.

    Args:
        initial_cooldown: Initial cooldown duration in seconds
        max_cooldown: Maximum cooldown duration in seconds
        backoff_multiplier: Multiplier for exponential backoff
        slow_start_items: Number of items for slow start after cooldown
        slow_start_initial_delay: Initial delay between items during slow start
        slow_start_final_delay: Final delay between items during slow start
    """
    self.initial_cooldown = initial_cooldown
    self.max_cooldown = max_cooldown
    self.backoff_multiplier = backoff_multiplier
    self.slow_start_items = slow_start_items
    self.slow_start_initial_delay = slow_start_initial_delay
    self.slow_start_final_delay = slow_start_final_delay

on_rate_limit async

on_rate_limit(worker_id: int, consecutive_limit_count: int) -> float

Calculate exponential backoff cooldown.

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

async def on_rate_limit(self, worker_id: int, consecutive_limit_count: int) -> float:
    """Calculate exponential backoff cooldown."""
    cooldown = capped_backoff(
        self.initial_cooldown,
        self.backoff_multiplier,
        consecutive_limit_count - 1,
        self.max_cooldown,
    )
    logger.info(
        f"Worker {worker_id} hit rate limit #{consecutive_limit_count}, "
        f"cooling down for {cooldown:.1f}s"
    )
    return cooldown

should_apply_slow_start

should_apply_slow_start(items_since_resume: int) -> tuple[bool, float]

Apply progressive slow-start after rate limit.

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

def should_apply_slow_start(self, items_since_resume: int) -> tuple[bool, float]:
    """Apply progressive slow-start after rate limit."""
    if items_since_resume >= self.slow_start_items:
        return (False, 0.0)

    # Progressive delay from initial to final
    progress = items_since_resume / self.slow_start_items
    delay = max(
        self.slow_start_final_delay,
        self.slow_start_initial_delay
        - (self.slow_start_initial_delay - self.slow_start_final_delay) * progress,
    )
    return (True, delay)

FixedDelayStrategy

llm_minder.FixedDelayStrategy

FixedDelayStrategy(cooldown: float = 300.0, delay_between_requests: float = 1.0)

Bases: RateLimitStrategy

Simple fixed delay between requests after rate limit.

Initialize fixed delay strategy.

Parameters:

Name	Type	Description	Default
cooldown	float	Fixed cooldown duration in seconds	300.0
delay_between_requests	float	Fixed delay between each request	1.0

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

def __init__(self, cooldown: float = 300.0, delay_between_requests: float = 1.0):
    """
    Initialize fixed delay strategy.

    Args:
        cooldown: Fixed cooldown duration in seconds
        delay_between_requests: Fixed delay between each request
    """
    self.cooldown = cooldown
    self.delay = delay_between_requests

on_rate_limit async

on_rate_limit(worker_id: int, consecutive_limit_count: int) -> float

Return fixed cooldown duration.

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

async def on_rate_limit(self, worker_id: int, consecutive_limit_count: int) -> float:
    """Return fixed cooldown duration."""
    logger.info(f"Worker {worker_id} hit rate limit, cooling down for {self.cooldown:.1f}s")
    return self.cooldown

should_apply_slow_start

should_apply_slow_start(items_since_resume: int) -> tuple[bool, float]

Always apply fixed delay.

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

def should_apply_slow_start(self, items_since_resume: int) -> tuple[bool, float]:
    """Always apply fixed delay."""
    return (True, self.delay)

Deadlines, aborts and interruptions

llm_minder.ItemDeadlineExceeded

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

Bases: TimeoutError, LLMMinderError

The end-to-end monotonic deadline for one logical item expired.

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

llm_minder.BatchDeadlineExceeded

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

Bases: TimeoutError, LLMMinderError

The batch deadline 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

llm_minder.BatchAbortedError

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

Bases: RuntimeError, LLMMinderError

An accepted collateral item was stopped by configured fail-fast.

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

llm_minder.BatchInterruptedError

Bases: RuntimeError, LLMMinderError

A batch worker was cancelled before all accepted work drained.

Raised by process_all() when another task shuts down the processor or cancels a worker. The caller itself was not cancelled; its own cancellation still propagates as asyncio.CancelledError.

llm_minder.FrameworkTimeoutError

FrameworkTimeoutError(message: str, *, item_id: str | None = None, elapsed: float | None = None, timeout_limit: float | None = None)

Bases: TimeoutError, LLMMinderError

Timeout enforced by the batch-llm framework (asyncio.wait_for).

This distinguishes framework-level timeouts from API-level timeouts. Framework timeouts indicate the configured attempt_timeout was exceeded, whereas API timeouts indicate the LLM provider returned a timeout error.

Attributes:

Name	Type	Description
item_id		ID of the work item that timed out (if available)
elapsed		Actual time elapsed in seconds
timeout_limit		Configured timeout limit in seconds

Initialize FrameworkTimeoutError with structured context (v0.4.0).

Parameters:

Name	Type	Description	Default
message	str	Human-readable error message	required
item_id	str | None	ID of the work item that timed out	None
elapsed	float | None	Actual time elapsed before timeout	None
timeout_limit	float | None	The timeout limit that was exceeded	None

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

def __init__(
    self,
    message: str,
    *,
    item_id: str | None = None,
    elapsed: float | None = None,
    timeout_limit: float | None = None,
):
    """
    Initialize FrameworkTimeoutError with structured context (v0.4.0).

    Args:
        message: Human-readable error message
        item_id: ID of the work item that timed out
        elapsed: Actual time elapsed before timeout
        timeout_limit: The timeout limit that was exceeded
    """
    super().__init__(message)
    self.item_id = item_id
    self.elapsed = elapsed
    self.timeout_limit = timeout_limit

llm_minder.RateLimitRetriesExceeded

RateLimitRetriesExceeded(message: str, *, item_id: str | None = None, rate_limit_retries: int | None = None)

Bases: LLMMinderError

A work item was retried after rate limits more than max_rate_limit_retries.

Rate-limit errors don't consume the max_attempts budget (they're a "wait and try again", not a failed attempt), but they ARE bounded separately by RetryConfig.max_rate_limit_retries so an endpoint that throttles forever can't hang the batch indefinitely. When that bound is exceeded the framework fails the item with this exception.

Attributes:

Name	Type	Description
item_id		ID of the work item that exhausted its rate-limit retries.
rate_limit_retries		How many rate-limit retries were attempted.

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

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

Provider responses and structured output

llm_minder.EmptyResponseError

EmptyResponseError(message: str, *, token_usage: TokenUsage | dict[str, int] | None = None, reason: EmptyResponseReason = 'unknown', provider_reason: str | None = None)

Bases: ValueError, LLMMinderError

The provider returned a billed response with no usable text.

Raised by the built-in models when the API call succeeded (and was billed) but produced no content — e.g. a Gemini safety block, or an OpenAI response whose finish_reason is length/content_filter or a tool call.

Subclasses ValueError so existing handlers and classifiers keep treating it as a deterministic, non-retryable failure. The tokens the provider already billed are attached as _failed_token_usage so failed-attempt accounting (WorkItemResult.token_usage) reflects the real spend.

Attributes:

Name	Type	Description
reason	EmptyResponseReason	Why the response was unusable, in provider-independent terms: "max_tokens" (the output bound was reached; raised as :class:TruncatedResponseError), "content_filter" (a safety or policy block), "refusal", "tool_call" (the model answered with a tool call instead of text), "no_content" (the response finished normally but had no text) or "unknown". Match on this rather than the message text.
provider_reason		The provider's own value, such as "MAX_TOKENS", "length" or "max_output_tokens", for logging; None when the provider gave none.

Added in v0.16.0; reason and provider_reason are new in 1.0 and provisional.

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

def __init__(
    self,
    message: str,
    *,
    token_usage: TokenUsage | dict[str, int] | None = None,
    reason: EmptyResponseReason = "unknown",
    provider_reason: str | None = None,
):
    super().__init__(message)
    self.reason: EmptyResponseReason = reason
    self.provider_reason = provider_reason
    if token_usage:
        self._failed_token_usage = dict(token_usage)

llm_minder.TruncatedResponseError

TruncatedResponseError(message: str, *, token_usage: TokenUsage | dict[str, int] | None = None, provider_reason: str | None = None)

Bases: EmptyResponseError

The response stopped at its output-token bound before producing usable text.

The EmptyResponseError raised when reason == "max_tokens", so a classifier or on_error() can match it with isinstance. The built-in classifiers give it the non-retryable response_truncated category: retrying the same request would usually truncate again. To retry with a larger bound, classify it as retryable in a custom classifier and raise the bound in ModelStrategy.attempt_generation_config().

New in 1.0 and provisional.

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

def __init__(
    self,
    message: str,
    *,
    token_usage: TokenUsage | dict[str, int] | None = None,
    provider_reason: str | None = None,
):
    super().__init__(
        message,
        token_usage=token_usage,
        reason="max_tokens",
        provider_reason=provider_reason,
    )

llm_minder.ProviderResponseError

ProviderResponseError(message: str, *, code: int | None = None, provider_error: Any = None, token_usage: TokenUsage | dict[str, int] | None = None)

Bases: LLMMinderError

Provider signaled failure inside an HTTP-200 response body.

Some gateways (notably OpenRouter) report upstream failures — no provider available, upstream 5xx, upstream rate limits — as HTTP 200 with an error object in the body and no choices, so the SDK never raises. These are typically transient routing failures: classifiers treat them as retryable, and as rate limits when the embedded code is 429.

Attributes:

Name	Type	Description
code		Numeric error code embedded in the body, if any.
provider_error		The raw error payload from the response body.

Added in v0.16.0.

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

def __init__(
    self,
    message: str,
    *,
    code: int | None = None,
    provider_error: Any = None,
    token_usage: TokenUsage | dict[str, int] | None = None,
):
    super().__init__(message)
    self.code = code
    self.provider_error = provider_error
    if token_usage:
        self._failed_token_usage = dict(token_usage)

llm_minder.StructuredOutputSchemaError

StructuredOutputSchemaError(message: str, *, token_usage: TokenUsage | dict[str, int] | None = None)

Bases: ValueError, LLMMinderError

A provider rejected a requested structured-output schema.

This is a deterministic request/schema compatibility failure, distinct from a model response that was accepted by the provider but failed local output validation. Classifiers do not retry it.

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

def __init__(
    self,
    message: str,
    *,
    token_usage: TokenUsage | dict[str, int] | None = None,
) -> None:
    super().__init__(message)
    if token_usage:
        self._failed_token_usage = dict(token_usage)

llm_minder.StructuredOutputValidationError

Bases: ValueError, LLMMinderError

A provider returned output that failed local structured validation.

Provider-enforced output should normally make this impossible, but a malformed or schema-invalid successful response is retryable because a subsequent model attempt may be usable.

llm_minder.TokenTrackingError

TokenTrackingError(message: str, *, token_usage: TokenUsage | dict[str, int] | None = None)

Bases: LLMMinderError

Wrapper exception that preserves token usage from failed LLM calls.

When an LLM call fails (e.g., validation error), we still want to track the tokens that were consumed. This wrapper attaches token usage to exceptions that don't natively support it (e.g., built-in exceptions without dict).

Attributes:

Name	Type	Description
token_usage		Dictionary with input_tokens, output_tokens, total_tokens, and optionally cached_input_tokens.

Example

try: ... # LLM call that fails validation ... output = parse_response(response) ... except Exception as e: ... wrapped = TokenTrackingError(str(e), token_usage=tokens) ... wrapped.cause = e ... raise wrapped from e

Initialize TokenTrackingError.

Parameters:

Name	Type	Description	Default
message	str	Human-readable error message	required
token_usage	TokenUsage | dict[str, int] | None	Token usage dict to preserve (input_tokens, output_tokens, etc.)	None

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

def __init__(self, message: str, *, token_usage: TokenUsage | dict[str, int] | None = None):
    """
    Initialize TokenTrackingError.

    Args:
        message: Human-readable error message
        token_usage: Token usage dict to preserve (input_tokens, output_tokens, etc.)
    """
    super().__init__(message)
    self._failed_token_usage = token_usage or {}

Configuration errors

llm_minder.MiddlewareContractError

Bases: ValueError, LLMMinderError

A preprocessing result violated accepted work-item identity or invariants.

llm_minder.QuotaScopeError

Bases: ValueError, LLMMinderError

A strategy quota scope could not be resolved safely before admission.

llm_minder.TokenEstimationError

Bases: LLMMinderError

Framework-owned, non-retryable token-estimation failure.

llm_minder.TokenEstimatorRequired

Bases: TokenEstimationError

TPM admission was enabled but no estimator resolved for an item.

llm_minder.TokenEstimateExceedsLimit

Bases: TokenEstimationError

One attempt's estimate cannot fit in the configured TPM bucket.