Errors, Classification and Rate Limits

Every exception type the library exports subclasses AsyncBatchLLMError 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

async_batch_llm.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/async_batch_llm/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

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

async_batch_llm.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/async_batch_llm/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

async_batch_llm.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/async_batch_llm/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

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

async_batch_llm.OpenRouterErrorClassifier

Bases: OpenAIErrorClassifier

OpenAI-compatible classifier with OpenRouter-specific overrides.

PydanticAIErrorClassifier

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

async_batch_llm.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/async_batch_llm/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/async_batch_llm/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

async_batch_llm.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/async_batch_llm/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/async_batch_llm/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/async_batch_llm/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

async_batch_llm.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/async_batch_llm/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/async_batch_llm/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/async_batch_llm/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

async_batch_llm.ItemDeadlineExceeded

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

Bases: TimeoutError, AsyncBatchLLMError

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

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

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

async_batch_llm.BatchDeadlineExceeded

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

Bases: TimeoutError, AsyncBatchLLMError

The batch deadline stopped an accepted item before completion.

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

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

async_batch_llm.BatchAbortedError

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

Bases: RuntimeError, AsyncBatchLLMError

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

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

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

async_batch_llm.BatchInterruptedError

Bases: RuntimeError, AsyncBatchLLMError

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.

async_batch_llm.FrameworkTimeoutError

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

Bases: TimeoutError, AsyncBatchLLMError

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

async_batch_llm.RateLimitRetriesExceeded

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

Bases: AsyncBatchLLMError

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

async_batch_llm.EmptyResponseError

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

Bases: ValueError, AsyncBatchLLMError

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.

Added in v0.16.0.

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

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

async_batch_llm.ProviderResponseError

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

Bases: AsyncBatchLLMError

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/async_batch_llm/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)

async_batch_llm.StructuredOutputSchemaError

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

Bases: ValueError, AsyncBatchLLMError

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/async_batch_llm/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)

async_batch_llm.StructuredOutputValidationError

Bases: ValueError, AsyncBatchLLMError

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.

async_batch_llm.TokenTrackingError

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

Bases: AsyncBatchLLMError

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

async_batch_llm.MiddlewareContractError

Bases: ValueError, AsyncBatchLLMError

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

async_batch_llm.QuotaScopeError

Bases: ValueError, AsyncBatchLLMError

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

async_batch_llm.TokenEstimationError

Bases: AsyncBatchLLMError

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

async_batch_llm.TokenEstimatorRequired

Bases: TokenEstimationError

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

async_batch_llm.TokenEstimateExceedsLimit

Bases: TokenEstimationError

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