Core API Reference

ParallelBatchProcessor

async_batch_llm.ParallelBatchProcessor

ParallelBatchProcessor(max_workers: int | None = None, post_processor: PostProcessorFunc[TOutput, TContext] | None = None, timeout_per_item: float | None = None, rate_limit_cooldown: float | 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
max_workers	int | None	Maximum concurrent workers (deprecated, use config)	None
post_processor	PostProcessorFunc[TOutput, TContext] | None	Optional async function called after each successful item	None
timeout_per_item	float | None	Timeout per item in seconds (deprecated, use config)	None
rate_limit_cooldown	float | None	Cooldown duration (deprecated, use config)	None
config	ProcessorConfig | None	Processor configuration object (recommended)	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/async_batch_llm/parallel.py

def __init__(
    self,
    max_workers: int | None = None,
    post_processor: PostProcessorFunc[TOutput, TContext] | None = None,
    timeout_per_item: float | None = None,
    rate_limit_cooldown: float | None = None,  # Deprecated, use config
    # New parameters
    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:
        max_workers: Maximum concurrent workers (deprecated, use config)
        post_processor: Optional async function called after each successful item
        timeout_per_item: Timeout per item in seconds (deprecated, use config)
        rate_limit_cooldown: Cooldown duration (deprecated, use config)
        config: Processor configuration object (recommended)
        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
    """
    import warnings

    # Emit deprecation warnings for legacy parameters
    if max_workers is not None:
        warnings.warn(
            "The 'max_workers' parameter is deprecated and will be removed in 1.0. "
            "Use ProcessorConfig(max_workers=...) instead.",
            DeprecationWarning,
            stacklevel=2,
        )
    if timeout_per_item is not None:
        warnings.warn(
            "The 'timeout_per_item' parameter is deprecated and will be removed in 1.0. "
            "Use ProcessorConfig(attempt_timeout=...) instead.",
            DeprecationWarning,
            stacklevel=2,
        )
    if rate_limit_cooldown is not None:
        warnings.warn(
            "The 'rate_limit_cooldown' parameter is deprecated and will be removed in 1.0. "
            "Use ProcessorConfig(rate_limit=RateLimitConfig(cooldown_seconds=...)) instead.",
            DeprecationWarning,
            stacklevel=2,
        )

    # Handle backward compatibility
    if config is None:
        from .core import RateLimitConfig

        config = ProcessorConfig(
            max_workers=max_workers or 5,
            attempt_timeout=timeout_per_item or 120.0,
            rate_limit=RateLimitConfig(cooldown_seconds=rate_limit_cooldown or 300.0),
        )
    else:
        # Override config with explicit legacy parameters if provided — but
        # build a NEW config via dataclasses.replace rather than mutating the
        # caller's object (and its nested RateLimitConfig). Callers may reuse
        # the same ProcessorConfig across processors and shouldn't see it
        # silently rewritten under them.
        import dataclasses

        overrides: dict = {}
        if max_workers is not None:
            overrides["max_workers"] = max_workers
        if timeout_per_item is not None:
            overrides["attempt_timeout"] = timeout_per_item
        if rate_limit_cooldown is not None:
            overrides["rate_limit"] = dataclasses.replace(
                config.rate_limit, cooldown_seconds=rate_limit_cooldown
            )
        if overrides:
            config = replace_user_config(config, **overrides)

    # 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
    )

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

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

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

WorkItemResult

async_batch_llm.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, gemini_safety_ratings: dict[str, str] | 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 for legacy 2-tuple strategies. 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. (For Gemini safety ratings specifically, this replaces the older gemini_safety_ratings field — see below.)
gemini_safety_ratings	dict[str, str] | None	Deprecated. Use metadata['safety_ratings'] instead. Still populated when the underlying model surfaces them, for backward compat. Removed in 1.0.
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.

__post_init__

__post_init__()

Backfill gemini_safety_ratings from metadata['safety_ratings'] for backward compatibility. Once gemini_safety_ratings is removed, this method goes away. (Reads/writes go through __dict__ directly so the framework itself never triggers the deprecation warning.)

Source code in src/async_batch_llm/base.py

def __post_init__(self):
    """Backfill ``gemini_safety_ratings`` from ``metadata['safety_ratings']``
    for backward compatibility. Once ``gemini_safety_ratings`` is removed,
    this method goes away. (Reads/writes go through ``__dict__`` directly
    so the framework itself never triggers the deprecation warning.)
    """
    if (
        self.__dict__.get("gemini_safety_ratings") is None
        and self.metadata is not None
        and "safety_ratings" in self.metadata
    ):
        ratings = self.metadata["safety_ratings"]
        if isinstance(ratings, dict):
            self.__dict__["gemini_safety_ratings"] = ratings

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/async_batch_llm/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:~async_batch_llm.ResultSerializationError.

Source code in src/async_batch_llm/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:`~async_batch_llm.ResultSerializationError`.
    """
    from .serialization import work_item_result_to_dict

    return work_item_result_to_dict(self, encoder=encoder)

AttemptTiming

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

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

async_batch_llm.ProcessorConfig dataclass

ProcessorConfig(max_workers: int | None = None, timeout_per_item: float | 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, enable_detailed_logging: bool = False, 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 the deprecated timeout alias, then validate.

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

def __post_init__(self) -> None:
    """Resolve the deprecated timeout alias, then validate."""
    # Raw constructor value — the alias property's setter stores writes in
    # __dict__ directly, and its getter is bypassed here on purpose.
    alias_value = self.__dict__.get("timeout_per_item")
    if alias_value is not None:
        if self.attempt_timeout is not None:
            raise ValueError(
                "Pass attempt_timeout only — timeout_per_item is a deprecated "
                f"alias for it (got attempt_timeout={self.attempt_timeout}, "
                f"timeout_per_item={alias_value})."
            )
        warnings.warn(
            "ProcessorConfig(timeout_per_item=...) is deprecated; use "
            "attempt_timeout=... (same per-attempt semantics). "
            "timeout_per_item will be removed in 1.0.",
            DeprecationWarning,
            # caller -> keyword-checking wrapper -> __init__ -> __post_init__
            stacklevel=4,
        )
        self.attempt_timeout = alias_value
    if self.attempt_timeout is None:
        self.attempt_timeout = 120.0
    if self.enable_detailed_logging and not _COPYING_USER_CONFIG.get():
        warnings.warn(
            "ProcessorConfig(enable_detailed_logging=True) has no effect and is "
            "deprecated; it will be removed in 1.0. Set the level of the "
            "'async_batch_llm' logger instead.",
            DeprecationWarning,
            stacklevel=4,
        )
    # Normalize the alias slot so dataclasses.replace() re-passes None and
    # the resolved value travels via attempt_timeout alone.
    self.__dict__["timeout_per_item"] = None
    # 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/async_batch_llm/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

async_batch_llm.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/async_batch_llm/core/config.py

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

validate

validate() -> None

Validate retry configuration.

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

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

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

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:~async_batch_llm.AttemptUsage. A cost_function without max_total_cost only tracks cost. Budgets apply to processor runs; call(), call_result() and LLMCallPool reject them.

budget_configured property

budget_configured: bool

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

AbortMode

async_batch_llm.AbortMode

Bases: str, Enum

How a controlled batch abort treats provider calls already running.

AttemptUsage

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

BatchBudgetExceeded

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

StartupRampConfig

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

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

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, ...). The v0.18 method spelling batch.cache_hit_rate() still works via a transitional callable-float return value and emits a DeprecationWarning; drop the parentheses.

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

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/async_batch_llm/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 | None = None) -> int

Estimate billable input tokens after the cache discount.

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 | None	Fraction (0.0–1.0) of the normal input price paid for cached tokens. Omitting it is deprecated and it will be required in 1.0. When omitted it defaults to CachedTokenRates.GEMINI (0.10) and warns: a UserWarning when cached tokens are present (the Gemini rate is wrong for e.g. OpenAI, ~0.50), otherwise a DeprecationWarning.	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(), which means the returned billable estimate is
int	rounded up when the discount would have a fractional
int	part — a deliberately conservative choice for cost reporting
int	(your real bill is at most this number, never more).

Raises:

Type	Description
ValueError	If cached_token_rate is not in [0.0, 1.0].

Source code in src/async_batch_llm/base.py

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

    ``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. Omitting it is deprecated and it will
            be required in 1.0. When omitted it defaults to
            ``CachedTokenRates.GEMINI`` (0.10) and warns: a ``UserWarning``
            when cached tokens are present (the Gemini rate is wrong for
            e.g. OpenAI, ~0.50), otherwise a ``DeprecationWarning``.

    Returns:
        Effective input tokens billed. The discount is computed by
        truncating ``cached_tokens * (1 - rate)`` toward zero with
        ``int()``, which means the returned billable estimate is
        rounded **up** when the discount would have a fractional
        part — a deliberately conservative choice for cost reporting
        (your real bill is at most this number, never more).

    Raises:
        ValueError: If ``cached_token_rate`` is not in [0.0, 1.0].
    """
    if cached_token_rate is None:
        cached_token_rate = self._default_cached_token_rate("effective_input_tokens")

    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))
    return self.total_input_tokens - discount

estimated_cost

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

Estimate total spend from per-million-token prices.

Applies the cache discount 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 | None	Fraction of the normal input price paid for cached tokens (see :class:CachedTokenRates). Omitting it is deprecated and it will be required in 1.0; see :meth:effective_input_tokens for the default and warnings.	None

Returns:

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

Source code in src/async_batch_llm/base.py

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

    Applies the cache discount 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`). Omitting it is
            deprecated and it will be required in 1.0; see
            :meth:`effective_input_tokens` for the default and warnings.

    Returns:
        Estimated total cost: ``effective_input / 1e6 * input_per_mtok +
        output / 1e6 * output_per_mtok``.
    """
    if cached_token_rate is None:
        cached_token_rate = self._default_cached_token_rate("estimated_cost")
    billable_input = self.effective_input_tokens(cached_token_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/async_batch_llm/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/async_batch_llm/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/async_batch_llm/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/async_batch_llm/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/async_batch_llm/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/async_batch_llm/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, int, int]:
        return (
            sum(r.token_usage.get("input_tokens", 0) for r in results),
            sum(r.token_usage.get("cached_input_tokens", 0) for r in results),
            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, cached_tokens, output_tokens = _tokens(current)
    lines.append(
        f"Tokens:    in {input_tokens:,} (cached {cached_tokens:,}) · out {output_tokens:,}"
    )
    if replayed:
        r_in, r_cached, r_out = _tokens(replayed)
        lines.append(
            f"Replayed:  in {r_in:,} (cached {r_cached:,}) · 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/async_batch_llm/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/async_batch_llm/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/async_batch_llm/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

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

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

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

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

async_batch_llm.ErrorCategory

Bases: _StrValueEnum

Values of WorkItemResult.error_category produced by the library.

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.

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.

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

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

AsyncBatchLLMError

async_batch_llm.AsyncBatchLLMError

Bases: Exception

Base class for the exception types async-batch-llm 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 AsyncBatchLLMError catches the library's own errors and existing except TimeoutError (or RuntimeError, ValueError) clauses keep working.

Lifecycle exceptions

async_batch_llm.StreamFinalizationError

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

Bases: RuntimeError, AsyncBatchLLMError

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

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

async_batch_llm.CleanupInterruptedError

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

Bases: RuntimeError, AsyncBatchLLMError

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

async_batch_llm.BatchAdmissionClosedError

Bases: RuntimeError, AsyncBatchLLMError

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:

1. Runtime tasks and callbacks: the stream finalizer, workers, queued work, progress callbacks, background post-processors, and callback threads.
2. Admission resources (quota scopes and the rate-limit coordinator).
3. Prepared strategies (strategy.cleanup()).
4. 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 deprecated result.gemini_safety_ratings field is still backfilled for compatibility, and is 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)

---

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

Provisional. The shapes were adjusted in v0.27 for OpenAI's Responses API. The views and the reserved keys' shapes may be promoted to stable at 1.0.0rc1 if real-provider runs fill them in correctly; until then they can change in a minor release, and logprobs is likely to stay provisional after 1.0. The metadata dict channel itself is stable. If you persist these shapes, read them back defensively. See API stability.

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

async_batch_llm.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 (Gemini)

CachedTokenRates

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

Notes

• Anthropic prompt caching is asymmetric: cache reads are at ANTHROPIC_READ (10% of normal), but cache writes are billed at a 25% premium over normal input price. This helper covers read-side savings only; the write premium is a separate cost best computed at billing time from your usage logs.
• 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 (cache writes are billed at 1.25× normal — not modeled here).

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.

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.

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

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

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

async_batch_llm.PostProcessorFunc module-attribute

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

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

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

Deprecated names

BatchProcessor, ProcessingStats, LLMGateway and grounding_metadata_extractor still import from async_batch_llm but warn and leave the public API in 1.0. They aren't documented here; see the v0.27 migration guide for replacements.