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)

Bases: BatchProcessor[TInput, TOutput, TContext], Generic[TInput, TOutput, TContext]

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. "
            "Use ProcessorConfig(max_workers=...) instead.",
            DeprecationWarning,
            stacklevel=2,
        )
    if timeout_per_item is not None:
        warnings.warn(
            "The 'timeout_per_item' parameter is deprecated. "
            "Use ProcessorConfig(attempt_timeout=...) instead.",
            DeprecationWarning,
            stacklevel=2,
        )
    if rate_limit_cooldown is not None:
        warnings.warn(
            "The 'rate_limit_cooldown' parameter is deprecated. "
            "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 = dataclasses.replace(config, **overrides)

    config.validate()
    # 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.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

    # 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._capacity_checked_strategy_ids: set[int] = set()
    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 = 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,
    )
    # 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,
    )
    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.

__aexit__ async

__aexit__(exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None) -> bool

Context manager exit - ensures cleanup of strategies and resources.

Calls cleanup() on all prepared strategies, then delegates to parent cleanup.

Parameters:

Name	Type	Description	Default
exc_type	type[BaseException] | None	Exception type (if any exception occurred)	required
exc_val	BaseException | None	Exception value (if any exception occurred)	required
exc_tb	TracebackType | None	Exception traceback (if any exception occurred)	required

Returns:

Type	Description
bool	False to indicate exceptions should not be suppressed

Source code in src/async_batch_llm/parallel.py

async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: "TracebackType | None",
) -> bool:
    """
    Context manager exit - ensures cleanup of strategies and resources.

    Calls cleanup() on all prepared strategies, then delegates to parent cleanup.

    Args:
        exc_type: Exception type (if any exception occurred)
        exc_val: Exception value (if any exception occurred)
        exc_tb: Exception traceback (if any exception occurred)

    Returns:
        False to indicate exceptions should not be suppressed
    """
    # Stop workers before closing strategies or their artifact sink: a
    # cancelled/early stream may still have an in-flight terminal write.
    errors: list[BaseException] = []
    for cleanup in (self.cleanup, self._cleanup_strategies):
        try:
            await cleanup()
        except BaseException as error:
            errors.append(error)
    if self.artifact_store is not None:
        try:
            await self.artifact_store.close()
        except BaseException as error:
            errors.append(error)
    if errors and exc_val is None:
        raise errors[0]
    return False  # Don't suppress exceptions

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."""
    if self.artifact_store is not None:
        work_item._artifact_key = await self.artifact_store.prepare_item(work_item)

    strategy_id = id(work_item.strategy)
    if strategy_id not in self._capacity_checked_strategy_ids:
        if self.config.concurrency is not None:
            # Let built-in models right-size
            # their connection pools before the first request. Runs before
            # the capacity check so a successful resize (advertised
            # capacity == concurrency == workers) produces no warning;
            # an explicit smaller max_connections refuses the resize and
            # the warning below surfaces the real contradiction.
            hook = getattr(work_item.strategy, "request_concurrency", None)
            if hook is not None:
                try:
                    await hook(self.config.concurrency)
                except Exception as e:
                    logger.warning(
                        "request_concurrency(%s) failed for %s: %s",
                        self.config.concurrency,
                        type(work_item.strategy).__name__,
                        e,
                    )
        warn_if_worker_capacity_exceeded(
            strategy=work_item.strategy,
            max_workers=cast(int, self.config.max_workers),
            surface="ParallelBatchProcessor",
            stacklevel=3,
        )
        self._capacity_checked_strategy_ids.add(strategy_id)
    assert self._abort_controller is not None
    if self._abort_controller.aborted:
        raise BatchAdmissionStopped("Batch is no longer accepting work")
    admission_state = self._admission_registry.resolve(work_item.strategy)
    if not self._compatibility_scope_bound:
        self._rate_limit_coord = admission_state.cooldown
        self._compatibility_scope_bound = True
    if self._streaming and self._guardrails_started:
        acceptance = asyncio.create_task(super().add_work(work_item))
        abort_wait = asyncio.create_task(self._abort_controller.event.wait())
        try:
            done, _ = await asyncio.wait(
                {acceptance, abort_wait}, return_when=asyncio.FIRST_COMPLETED
            )
            # An acceptance that completed concurrently wins: the item is
            # now owned by the queue and must receive a terminal result.
            if acceptance in done:
                await acceptance
            else:
                acceptance.cancel()
                with contextlib.suppress(asyncio.CancelledError):
                    await acceptance
                raise BatchAdmissionStopped("Batch stopped accepting work")
        finally:
            abort_wait.cancel()
            await asyncio.gather(abort_wait, return_exceptions=True)
    else:
        await super().add_work(work_item)

cleanup async

cleanup() -> None

Cancel workers, timers, and every quota-scoped admission resource.

Source code in src/async_batch_llm/parallel.py

async def cleanup(self) -> None:
    """Cancel workers, timers, and every quota-scoped admission resource."""
    errors: list[BaseException] = []
    for cleanup in (
        self._cancel_batch_timeout,
        super().cleanup,
        self._admission_registry.shutdown,
        self._compatibility_rate_limit_coord.shutdown,
    ):
        try:
            await cleanup()
        except BaseException as error:
            errors.append(error)
    self._classifier_resolver.clear()
    if errors:
        raise errors[0]

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:
        return self._stats.copy()

shutdown async

shutdown()

Clean up resources: flush observers and cancel pending tasks.

Source code in src/async_batch_llm/parallel.py

async def shutdown(self):
    """Clean up resources: flush observers and cancel pending tasks."""
    errors: list[BaseException] = []
    for cleanup in (self.cleanup, self._cleanup_strategies):
        try:
            await cleanup()
        except BaseException as error:
            errors.append(error)
    if errors:
        raise errors[0]

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."""
    self._start_guardrail_run()
    super().start()

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)

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."""
    if not self.item_id or not isinstance(self.item_id, str):
        raise ValueError(
            f"item_id must be a non-empty string (got {type(self.item_id).__name__}: {repr(self.item_id)}). "
            f"Provide a unique string identifier for this work item."
        )
    if not self.item_id.strip():
        raise ValueError(
            f"item_id cannot be whitespace only (got {repr(self.item_id)}). "
            f"Provide a non-whitespace string identifier."
        )
    if self.strategy is None:
        raise ValueError(
            "strategy must not be None. "
            "Pass an LLMCallStrategy instance (e.g., PydanticAIStrategy, GeminiStrategy, "
            "or your custom subclass)."
        )
    if not isinstance(self.prompt, str):
        raise TypeError(
            f"prompt must be a string (got {type(self.prompt).__name__}: {repr(self.prompt)[:80]}). "
            f"If you need to pass structured data, serialize it to a string first."
        )

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. To be removed in a future release.
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. Its traceback is detached before storage (the full failure is already logged at the failure site) so accumulated failed results don't pin frame locals; a re-raise gets a fresh 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)

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 the next major release.",
            DeprecationWarning,
            stacklevel=3,
        )
        self.attempt_timeout = alias_value
    if self.attempt_timeout is None:
        self.attempt_timeout = 120.0
    # 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."""
    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:
        if (
            isinstance(self.max_requests_per_minute, bool)
            or not isinstance(self.max_requests_per_minute, (int, float))
            or not math.isfinite(self.max_requests_per_minute)
            or self.max_requests_per_minute <= 0
        ):
            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)."
            )
    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."
        )

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)

Optional end-to-end deadlines and terminal-category fail-fast policy.

AbortMode

async_batch_llm.AbortMode

Bases: str, Enum

How a controlled batch abort treats provider calls already running.

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. When omitted (None) it defaults to CachedTokenRates.GEMINI (0.10) for backward compatibility — pre-v0.9.0 versions hardcoded this value. Pass an explicit rate when working with non-Gemini providers to get accurate numbers; relying on the implicit default while cached tokens are present emits a UserWarning, since the Gemini rate is wrong for e.g. OpenAI (~0.50).	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. When omitted (``None``) it defaults to
            ``CachedTokenRates.GEMINI`` (0.10) for backward compatibility —
            pre-v0.9.0 versions hardcoded this value. **Pass an explicit
            rate when working with non-Gemini providers** to get accurate
            numbers; relying on the implicit default while cached tokens
            are present emits a ``UserWarning``, since the Gemini rate is
            wrong for e.g. OpenAI (~0.50).

    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:
        # Implicit default. Only nudge when it actually changes the answer
        # (i.e. there are cached tokens to discount) — silent for the common
        # no-cache case so we don't cry wolf.
        if self.total_cached_tokens > 0:
            import warnings

            warnings.warn(
                "effective_input_tokens() called without an explicit "
                "cached_token_rate; defaulting to the Gemini rate "
                "(CachedTokenRates.GEMINI = 0.10). This is wrong for other "
                "providers (OpenAI is ~0.50). Pass an explicit "
                "CachedTokenRates constant to silence this warning.",
                UserWarning,
                stacklevel=2,
            )
        cached_token_rate = CachedTokenRates.GEMINI

    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). When None it defaults to the Gemini rate and emits a UserWarning if cached tokens are present — pass an explicit rate for other providers.	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`). When ``None`` it
            defaults to the Gemini rate and emits a ``UserWarning`` if cached
            tokens are present — pass an explicit rate for other providers.

    Returns:
        Estimated total cost: ``effective_input / 1e6 * input_per_mtok +
        output / 1e6 * output_per_mtok``.
    """
    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}")

    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', '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 "",
    )