async-batch-llm API Reference

Complete API documentation for async-batch-llm v0.16.0.

Table of Contents

• Core Classes
• LLMWorkItem
• WorkItemResult
• BatchResult
• ParallelBatchProcessor
• LLM Strategies
• LLMCallStrategy (Abstract)
• PydanticAIStrategy
• GeminiStrategy
• GeminiCachedModel
• Configuration
• ProcessorConfig
• RetryConfig
• RateLimitConfig
• Error Handling
• ErrorClassifier
• ErrorInfo
• RateLimitStrategy
• Middleware and Observers
• Middleware
• ProcessorObserver
• MetricsObserver
• Core Types
• TokenUsage
• FrameworkTimeoutError
• TokenTrackingError
• Type Aliases
• PostProcessorFunc
• ProgressCallbackFunc

---

Core Classes

LLMWorkItem

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

@dataclass
class LLMWorkItem(Generic[TInput, TOutput, TContext]):
    item_id: str
    strategy: LLMCallStrategy[TOutput]
    prompt: str = ""
    context: TContext | None = None

Type Parameters:

• TInput: Input data type (unused in v0.1, kept for backward compatibility)
• TOutput: Expected output type from the LLM
• TContext: Optional context data type passed through to results

Fields:

• item_id (str): Unique identifier for this work item. Must be non-empty.
• strategy (LLMCallStrategy[TOutput]): Strategy that encapsulates how to make the LLM call
• prompt (str, optional): The prompt/input to pass to the LLM. Default: ""
• context (TContext | None, optional): Optional context data passed through to results/post-processor

Example:

from async_batch_llm import LLMWorkItem, PydanticAIStrategy
from pydantic_ai import Agent

agent = Agent("openai:gpt-4", output_type=MyOutput)
strategy = PydanticAIStrategy(agent=agent)

work_item = LLMWorkItem(
    item_id="task_1",
    strategy=strategy,
    prompt="Analyze this text...",
    context={"user_id": 123}
)

Validation:

• Raises ValueError if item_id is empty or whitespace-only
• Raises ValueError if item_id is not a string

---

WorkItemResult

Result of processing a single work item.

@dataclass
class WorkItemResult(Generic[TOutput, TContext]):
    item_id: str
    success: bool
    output: TOutput | None = None
    error: str | None = None
    context: TContext | None = None
    token_usage: TokenUsage = field(
        default_factory=lambda: {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}
    )
    metadata: dict[str, Any] | None = None
    exception: Exception | None = None  # original exception on failure
    gemini_safety_ratings: dict[str, str] | None = None  # deprecated — warns on read

Fields:

• item_id (str): ID of the work item
• success (bool): Whether processing succeeded
• output (TOutput | None): LLM 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 statistics with optional fields:
• input_tokens (int): Number of tokens in the input/prompt
• output_tokens (int): Number of tokens in the output/completion
• total_tokens (int): Total tokens used (input + output)
• cached_input_tokens (int): Number of input tokens served from cache (Gemini context caching)
• metadata (dict[str, Any] | None): Provider metadata (provider name, finish reason, safety ratings, ...) forwarded from the strategy
• exception (Exception | None): The original exception when the item failed (what call() / LLMGateway.submit() re-raise); None on success
• gemini_safety_ratings (dict[str, str] | None): Deprecated. Reading it emits a DeprecationWarning; use result.metadata["safety_ratings"] instead

Example:

result = await processor.process_all()
for item_result in result.results:
    if item_result.success:
        print(f"✓ {item_result.item_id}: {item_result.output}")
        print(f"  Tokens: {item_result.token_usage}")
    else:
        print(f"✗ {item_result.item_id}: {item_result.error}")

---

BatchResult

Result of processing a batch of work items.

@dataclass
class BatchResult(Generic[TOutput, TContext]):
    results: list[WorkItemResult[TOutput, TContext]]
    # Derived summary fields (init=False) — computed from `results` in
    # __post_init__; they cannot be passed to the constructor.
    total_items: int
    succeeded: int
    failed: int
    total_input_tokens: int
    total_output_tokens: int
    total_cached_tokens: int

Fields:

• results (list[WorkItemResult]): List of individual work item results
• total_items (int): Total number of items processed
• 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 from Gemini context caching

Note: Only results is a constructor argument. The summary fields are init=False and calculated automatically in __post_init__ — construct with BatchResult(results=[...]).

Example:

result = await processor.process_all()

print(f"Processed {result.total_items} items")
print(f"Success: {result.succeeded}, Failed: {result.failed}")
print(f"Total tokens: {result.total_input_tokens + result.total_output_tokens}")

# Access individual results
for item_result in result.results:
    if item_result.success:
        process_output(item_result.output)

---

ParallelBatchProcessor

Main processor that executes work items in parallel.

class ParallelBatchProcessor(
    BatchProcessor[TInput, TOutput, TContext],
    Generic[TInput, TOutput, TContext]
):
    def __init__(
        self,
        max_workers: int | None = None,          # deprecated, use config
        post_processor: PostProcessorFunc[TOutput, TContext] | None = None,
        timeout_per_item: float | None = None,   # deprecated, use config
        rate_limit_cooldown: float | None = None,  # deprecated, use config
        config: ProcessorConfig | None = None,
        error_classifier: ErrorClassifier | None = None,
        rate_limit_strategy: RateLimitStrategy | None = None,
        middlewares: list[Middleware] | None = None,
        observers: list[ProcessorObserver] | None = None,
        progress_callback: ProgressCallbackFunc | None = None,
    )

Pass everything by keyword — the first positional parameter is the deprecated max_workers, not config.

Parameters:

• config (ProcessorConfig | None): Configuration for the processor (recommended)
• post_processor (PostProcessorFunc | None): Optional async function called after each item
• progress_callback (ProgressCallbackFunc | None): Optional callback for progress updates
• error_classifier (ErrorClassifier | None): Custom error classifier. Default: auto-selected from the work items' strategies (e.g. GeminiStrategy → GeminiErrorClassifier), falling back to DefaultErrorClassifier() when there is no recommendation or providers conflict
• rate_limit_strategy (RateLimitStrategy | None): Custom rate limit handling. Default: ExponentialBackoffStrategy()
• middlewares (list[Middleware] | None): List of middleware for pre/post processing
• observers (list[ProcessorObserver] | None): List of observers for monitoring events
• max_workers, timeout_per_item, rate_limit_cooldown: deprecated loose parameters; set them on ProcessorConfig instead

Post-processing: The optional post_processor runs inline on the worker as soon as an item finishes. It should hand off any heavy operations (long DB writes, expensive analytics, etc.) to another system; if the function takes too long the worker sits idle until the timeout triggers (ProcessorConfig.post_processor_timeout, default 90 s), reducing overall throughput.

Methods:

async def add_work(work_item: LLMWorkItem) -> None

Add a work item to the processing queue.

await processor.add_work(work_item)

Note: If max_queue_size is set and the queue is full, add_work raises ValueError in batch mode. Only streaming mode (start()/results()/finish()) applies backpressure by blocking until space is available.

async def process_all() -> BatchResult

Process all work items in the queue.

result = await processor.process_all()

Returns: BatchResult containing all results and statistics

Behavior:

1. Starts worker tasks (up to max_workers)
2. Workers process items from queue with retry logic
3. Waits for all work to complete
4. Returns aggregated results

async def cleanup() -> None

Clean up resources (cancel pending workers, clear queue).

await processor.cleanup()

Note: Automatically called when using async context manager.

Context Manager Support

async with ParallelBatchProcessor(config=config) as processor:
    await processor.add_work(item)
    result = await processor.process_all()
# Automatic cleanup

Example:

from async_batch_llm import ParallelBatchProcessor, ProcessorConfig, LLMWorkItem

config = ProcessorConfig(max_workers=5, timeout_per_item=60.0)

async with ParallelBatchProcessor(config=config) as processor:
    for i in range(100):
        work_item = LLMWorkItem(
            item_id=f"item_{i}",
            strategy=my_strategy,
            prompt=f"Task {i}"
        )
        await processor.add_work(work_item)

    result = await processor.process_all()
    print(f"Completed: {result.succeeded}/{result.total_items}")

---

LLM Strategies

LLMCallStrategy

Abstract base class for LLM call strategies.

class LLMCallStrategy(ABC, Generic[TOutput]):
    async def prepare(self) -> None: ...

    @abstractmethod
    async def execute(
        self,
        prompt: str,
        attempt: int,
        timeout: float,
        state: RetryState | None = None,
    ) -> tuple[TOutput, TokenUsage]: ...

    async def on_error(
        self,
        exception: Exception,
        attempt: int,
        state: RetryState | None = None,
    ) -> None: ...

    async def cleanup(self) -> None: ...

    async def dry_run(self, prompt: str) -> tuple[TOutput, TokenUsage]: ...

Lifecycle:

1. prepare() - Called once before any retry attempts
2. For each attempt (including retries):
3. execute() is called (or dry_run() if config.dry_run=True)
4. If execute() raises an exception, on_error() is called before retry logic
5. cleanup() - Called once after all attempts complete

Methods:

async def prepare() -> None

Initialize resources before making LLM calls (e.g., create caches, initialize clients).

Default: No-op

async def execute(prompt: str, attempt: int, timeout: float, state: RetryState | None = None) -> tuple[TOutput, TokenUsage]

Execute an LLM call.

Parameters:

• prompt (str): The prompt to send to the LLM
• attempt (int): Which retry attempt this is (1, 2, 3, ...)
• timeout (float): Maximum time to wait for response (seconds)
• Note: Timeout enforcement is handled by the framework wrapping this call in asyncio.wait_for()
• state (RetryState | None): Mutable per-work-item state provided by the framework so strategies can track partial progress across retries

Returns: Tuple of (output, token_usage)

• output (TOutput): The LLM response
• token_usage (TokenUsage): Token usage dict with optional keys: input_tokens, output_tokens, total_tokens, cached_input_tokens

Raises: Any exception to trigger retry (if retryable) or failure

async def dry_run(prompt: str) -> tuple[TOutput, TokenUsage]

Return mock output for dry-run mode (testing without API calls).

Called when ProcessorConfig(dry_run=True) is set. Override this method to provide realistic mock data for testing.

Parameters:

• prompt (str): The prompt that would have been sent to the LLM

Returns: Tuple of (mock_output, mock_token_usage)

Default behavior:

• Returns string "[DRY-RUN] Mock output for prompt: {prompt[:50]}..." as output
• Returns mock token usage: 100 input, 50 output, 150 total tokens

Example override:

class MyStrategy(LLMCallStrategy[Output]):
    async def dry_run(self, prompt: str) -> tuple[Output, TokenUsage]:
        # Return realistic mock data
        mock_output = Output(result="Test result")
        mock_tokens: TokenUsage = {
            "input_tokens": len(prompt.split()),
            "output_tokens": 50,
            "total_tokens": len(prompt.split()) + 50,
        }
        return mock_output, mock_tokens

async def on_error(exception: Exception, attempt: int, state: RetryState | None = None) -> None

Handle errors that occur during execute().

Called by the framework when execute() raises an exception, before deciding whether to retry. This allows strategies to:

• Inspect the error type to adjust retry behavior
• Store error information for use in the next attempt
• Modify prompts based on validation errors
• Track error patterns across attempts
• Make intelligent decisions (e.g., escalate to smarter model only on validation errors)

Parameters:

• exception (Exception): The exception that was raised during execute()
• attempt (int): Which attempt number failed (1, 2, 3, ...)
• state (RetryState | None): Retry state that persists across attempts (v0.3.0)

Default: No-op

Use Cases:

1. Smart Model Escalation - Only escalate to expensive models on validation errors, not network errors:

class SmartModelEscalationStrategy(LLMCallStrategy[Output]):
    def __init__(self):
        self.validation_failures = 0

    async def on_error(self, exception: Exception, attempt: int, state=None) -> None:
        if isinstance(exception, ValidationError):
            self.validation_failures += 1

    async def execute(self, prompt: str, attempt: int, timeout: float, state=None):
        # Only escalate model on validation errors
        model_index = min(self.validation_failures, len(MODELS) - 1)
        model = MODELS[model_index]
        # Make call with appropriate model...

1. Smart Retry with Partial Parsing - Build better retry prompts based on what failed:

class SmartRetryStrategy(LLMCallStrategy[Output]):
    def __init__(self):
        self.last_error = None
        self.last_response = None

    async def on_error(self, exception: Exception, attempt: int, state=None) -> None:
        if isinstance(exception, ValidationError):
            self.last_error = exception
            # last_response set in execute() before raising

    async def execute(self, prompt: str, attempt: int, timeout: float, state=None):
        if attempt > 1 and self.last_error:
            # Build smart retry prompt with partial parsing feedback
            prompt = self._create_retry_prompt_with_partial_data(prompt)
        # Make call with improved prompt...

1. Error Type Tracking - Distinguish between different error types:

class ErrorTrackingStrategy(LLMCallStrategy[Output]):
    def __init__(self):
        self.validation_errors = 0
        self.network_errors = 0
        self.rate_limit_errors = 0

    async def on_error(self, exception: Exception, attempt: int, state=None) -> None:
        if isinstance(exception, ValidationError):
            self.validation_errors += 1
        elif isinstance(exception, ConnectionError):
            self.network_errors += 1
        elif "429" in str(exception):
            self.rate_limit_errors += 1

Important Notes:

• Exceptions in on_error() are caught and logged by the framework - they won't crash processing
• on_error() is only called when execute() raises an exception, not on success
• The error is still propagated to the framework's retry logic after on_error() returns
• For stateful strategies, each work item should use a separate strategy instance

See Also:

• examples/example_smart_model_escalation.py - Complete smart model escalation example
• examples/example_gemini_smart_retry.py - Smart retry with partial parsing

async def cleanup() -> None

Clean up resources after all attempts complete (e.g., delete caches, close clients).

Default: No-op

Custom Strategy Example:

from async_batch_llm import LLMCallStrategy, TokenUsage

class MyCustomStrategy(LLMCallStrategy[str]):
    async def execute(
        self, prompt: str, attempt: int, timeout: float, state=None
    ) -> tuple[str, TokenUsage]:
        # Your custom LLM API call
        response = await my_llm_api.generate(prompt)

        tokens: TokenUsage = {
            "input_tokens": response.input_tokens,
            "output_tokens": response.output_tokens,
            "total_tokens": response.total_tokens,
        }

        return response.text, tokens

---

PydanticAIStrategy

Strategy for using PydanticAI agents.

class PydanticAIStrategy(LLMCallStrategy[TOutput]):
    def __init__(self, agent: Agent[None, TOutput])

Parameters:

• agent (Agent[None, TOutput]): Configured PydanticAI agent

Requires: pip install 'async-batch-llm[pydantic-ai]'

Example:

from async_batch_llm import PydanticAIStrategy, LLMWorkItem
from pydantic_ai import Agent
from pydantic import BaseModel

class BookSummary(BaseModel):
    title: str
    summary: str

agent = Agent("openai:gpt-4", output_type=BookSummary)
strategy = PydanticAIStrategy(agent=agent)

work_item = LLMWorkItem(
    item_id="book_1",
    strategy=strategy,
    prompt="Summarize: The Great Gatsby..."
)

---

GeminiStrategy

Strategy for calling Google Gemini API directly (without caching).

class GeminiStrategy(LLMCallStrategy[TOutput]):
    def __init__(
        self,
        model: LLMModel,
        response_parser: Callable[[LLMResponse], TOutput] | None = None,
        *,
        temperature: float | None = 0.0,
        generation_config: dict[str, Any] | None = None,
    )

Parameters:

• model (LLMModel): Model wrapper such as GeminiModel or GeminiCachedModel
• response_parser (Callable | None): Function to parse LLMResponse into TOutput. Defaults to returning response.text.
• temperature (float | None): Sampling temperature. Default: 0.0. Pass None to omit the parameter and use the provider default.
• generation_config (dict | None): Extra provider config merged into each generate() call (e.g. tools, response_mime_type, logprobs). Also available on OpenAIStrategy/OpenRouterStrategy/DeepSeekStrategy (shared ModelStrategy base).

Requires: pip install 'async-batch-llm[gemini]'

API key: Set GOOGLE_API_KEY (preferred) or the legacy GEMINI_API_KEY environment variable before running this example.

Example:

from async_batch_llm import GeminiModel, GeminiStrategy, LLMWorkItem
from google import genai

client = genai.Client(api_key=os.getenv("GOOGLE_API_KEY") or os.getenv("GEMINI_API_KEY"))
model = GeminiModel("gemini-2.5-flash", client)

strategy = GeminiStrategy(
    model=model,
)

work_item = LLMWorkItem(
    item_id="task_1",
    strategy=strategy,
    prompt="Explain quantum computing"
)

---

GeminiCachedModel

Model for calling Google Gemini with context caching. Since v0.6.0, caching is a property of the model, not the strategy — wrap a GeminiCachedModel in the ordinary GeminiStrategy. (The old GeminiCachedStrategy was removed; the model now owns the cache find/create/renew/delete lifecycle.)

class GeminiCachedModel:
    def __init__(
        self,
        model: str,
        client: genai.Client,
        cached_content: list[Content],
        *,
        cache_ttl_seconds: int = 3600,
        cache_renewal_buffer_seconds: int = 300,
        auto_renew: bool = True,
        cache_tags: dict[str, str] | None = None,
        safety_settings: list[dict[str, Any]] | None = None,
        metadata_extractors: list[MetadataExtractor] | None = None,
    )

Parameters:

• model (str): Model name
• client (genai.Client): Initialized Gemini client
• cached_content (list[Content]): Content to cache (system instructions, documents)
• cache_ttl_seconds (int): Cache TTL in seconds. Default: 3600 (1 hour)
• cache_renewal_buffer_seconds (int): Renew caches this many seconds before expiry (default 300)
• auto_renew (bool): Automatically renew caches when they near expiry. Default: True
• cache_tags (dict[str, str] | None): Optional metadata for precise cache matching/versioning (encoded into the cache's display_name)
• safety_settings (list[dict] | None): Optional default safety settings for all calls

Lifecycle (driven by the framework via GeminiStrategy):

• prepare(): Finds or creates the Gemini cache (once per shared instance)
• generate(): Uses the cache and auto-renews when enabled
• cleanup(): Runs once when the processor exits; by default caches are left alive so future batches can reuse them (call delete_cache() to remove immediately)

Requires: pip install 'async-batch-llm[gemini]'

API key: Same as above – GOOGLE_API_KEY is preferred, GEMINI_API_KEY also works.

Share one instance. Create a single GeminiCachedModel and reuse it across every work item that should share the cached context. Constructing a new instance per item defeats caching entirely and can cost ~10x more.

Example:

from async_batch_llm import GeminiCachedModel, GeminiStrategy, LLMWorkItem
from google import genai
from google.genai.types import Content

client = genai.Client(api_key=os.getenv("GOOGLE_API_KEY") or os.getenv("GEMINI_API_KEY"))

# Large document to cache
cached_content = [
    Content(role="system", parts=[{"text": "You are a helpful assistant."}]),
    Content(role="user", parts=[{"text": large_document}]),
]

cached_model = GeminiCachedModel(
    "gemini-2.5-flash",
    client,
    cached_content=cached_content,
    cache_ttl_seconds=3600,
)
strategy = GeminiStrategy(cached_model, response_parser=lambda r: r.text)

# Reuse the same strategy/model across work items to benefit from caching
for i in range(100):
    work_item = LLMWorkItem(
        item_id=f"task_{i}",
        strategy=strategy,  # Same strategy, shared cache
        prompt=f"Question {i} about the document"
    )
    await processor.add_work(work_item)

---

Configuration

ProcessorConfig

Complete configuration for batch processor.

@dataclass
class ProcessorConfig:
    max_workers: int = 5
    timeout_per_item: float = 120.0
    post_processor_timeout: float = 90.0
    concurrent_post_processing: bool = False
    retry: RetryConfig = field(default_factory=RetryConfig)
    rate_limit: RateLimitConfig = field(default_factory=RateLimitConfig)
    progress_interval: int = 10
    progress_callback_timeout: float | None = 5.0
    enable_detailed_logging: bool = False
    max_queue_size: int = 0
    max_requests_per_minute: float | None = None
    dry_run: bool = False

Fields:

• max_workers (int): Maximum number of concurrent workers. Default: 5
• timeout_per_item (float): Timeout applied to each execute() attempt in seconds (per-attempt, not a total budget across retries). Default: 120.0
• post_processor_timeout (float): Max seconds for the post_processor callback per item. Default: 90.0
• concurrent_post_processing (bool): Run the post-processor without holding the results lock. Default: False
• retry (RetryConfig): Retry configuration
• rate_limit (RateLimitConfig): Rate limit handling configuration
• progress_interval (int): Log progress every N items. Default: 10
• progress_callback_timeout (float | None): Max seconds to wait for progress callback. Default: 5.0. Set to None for no timeout.
• enable_detailed_logging (bool): Enable detailed debug logging. Default: False
• max_queue_size (int): Max queue size (0 = unlimited). Default: 0
• max_requests_per_minute (float | None): Optional proactive rate limiter that throttles requests before hitting provider limits
• dry_run (bool): Skip actual API calls, use mock data from strategy.dry_run(). Default: False

Example:

from async_batch_llm import ProcessorConfig, RetryConfig

config = ProcessorConfig(
    max_workers=10,
    timeout_per_item=60.0,
    retry=RetryConfig(max_attempts=5, initial_wait=2.0),
    progress_interval=20,
    max_queue_size=1000,
)

---

RetryConfig

Configuration for retry behavior.

@dataclass
class 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

Fields:

• max_attempts (int): Maximum attempts for content/transport failures (validation, timeout, connection, 5xx). Default: 3
• initial_wait (float): Initial wait time in seconds. Default: 1.0
• max_wait (float): Maximum wait time in seconds. Default: 60.0
• exponential_base (float): Exponential backoff base. Default: 2.0
• jitter (bool): Add random jitter to wait times. Default: True
• max_rate_limit_retries (int): Maximum times an item may be retried after a rate-limit/cooldown without consuming its max_attempts budget. Rate limits are retried at the same logical attempt number; exceeding this fails the item with a RateLimitRetriesExceeded error. 0 makes rate limits fail immediately. Default: 20

Validation:

• max_attempts must be >= 1
• initial_wait must be > 0
• max_wait must be >= initial_wait
• exponential_base must be >= 1
• max_rate_limit_retries must be >= 0

Example:

retry_config = RetryConfig(
    max_attempts=5,
    initial_wait=2.0,
    max_wait=120.0,
    exponential_base=2.0,
    jitter=True,
)

---

RateLimitConfig

Configuration for rate limit handling.

@dataclass
class 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 = 600.0

Fields:

• cooldown_seconds (float): Cooldown after rate limit. Default: 300.0 (5 minutes)
• slow_start_items (int): Number of items for slow start. Default: 50
• slow_start_initial_delay (float): Initial delay in slow start. Default: 2.0
• slow_start_final_delay (float): Final delay in slow start. Default: 0.1
• backoff_multiplier (float): Increase cooldown on repeated rate limits. Default: 1.5
• max_cooldown_seconds (float): Cap on the escalated cooldown (v0.16). Default: 600.0

Validation:

• cooldown_seconds must be >= 0
• slow_start_items must be >= 0
• slow_start_initial_delay must be >= slow_start_final_delay
• backoff_multiplier must be >= 1.0

---

Error Handling

ErrorClassifier

Interface for classifying errors as retryable or not.

class ErrorClassifier(ABC):
    @abstractmethod
    def classify(self, exception: Exception) -> ErrorInfo: ...

Built-in Implementations:

• DefaultErrorClassifier: Provider-agnostic classification based on exception types
• GeminiErrorClassifier: Specialized for Google Gemini API errors

Custom Example:

from async_batch_llm import ErrorClassifier, ErrorInfo

class MyErrorClassifier(ErrorClassifier):
    def classify(self, exception: Exception) -> ErrorInfo:
        error_str = str(exception).lower()

        if "rate limit" in error_str:
            return ErrorInfo(
                is_retryable=True,
                is_rate_limit=True,
                is_timeout=False,
                error_category="rate_limit",
            )
        elif "timeout" in error_str:
            return ErrorInfo(
                is_retryable=True,
                is_rate_limit=False,
                is_timeout=True,
                error_category="api_timeout",
            )
        else:
            return ErrorInfo(
                is_retryable=False,
                is_rate_limit=False,
                is_timeout=False,
                error_category="unknown",
            )

---

ErrorInfo

Information about a classified error.

@dataclass
class ErrorInfo:
    is_retryable: bool
    is_rate_limit: bool
    is_timeout: bool
    error_category: str
    suggested_wait: float | None = None
    hint: str | None = None

Fields:

• is_retryable (bool): Whether the error should trigger a retry
• is_rate_limit (bool): Whether this is a rate limit error (429, resource_exhausted, etc.)
• is_timeout (bool): Whether this is a timeout error (framework or API timeout)
• error_category (str): Error category for logging/metrics. Common values:
• "framework_timeout" - Framework timeout (exceeded timeout_per_item)
• "api_timeout" - API-level timeout
• "rate_limit" - Rate limit error
• "validation_error" - Pydantic validation error
• "insufficient_balance" - 402 Payment Required / balance exhausted (non-retryable)
• "client_error" - 4xx client error
• "server_error" - 5xx server error
• "connection_error" - Network connection error
• "unknown" - Unclassified error
• suggested_wait (float | None): Suggested wait time before retry (seconds). Used for rate limits.
• hint (str | None): Optional operator-facing remediation hint, surfaced in the logs at WARNING when a non-retryable error gives up (e.g. the 402 "top up your prepaid balance" guidance). None means no extra guidance.

Example:

from async_batch_llm import ErrorInfo

# Rate limit error
rate_limit_info = ErrorInfo(
    is_retryable=False,  # Don't retry via exponential backoff
    is_rate_limit=True,  # Trigger rate limit cooldown
    is_timeout=False,
    error_category="rate_limit",
    suggested_wait=300.0,  # 5 minute cooldown
)

# Framework timeout (retryable, might succeed if faster)
timeout_info = ErrorInfo(
    is_retryable=True,
    is_rate_limit=False,
    is_timeout=True,
    error_category="framework_timeout",
)

---

RateLimitStrategy

Interface for custom rate limit handling strategies.

class RateLimitStrategy(ABC):
    @abstractmethod
    async def on_rate_limit(
        self, worker_id: int, consecutive_limit_count: int
    ) -> float:
        """Called when a rate limit is detected. Returns the cooldown in seconds."""

    @abstractmethod
    def should_apply_slow_start(
        self, items_since_resume: int
    ) -> tuple[bool, float]:
        """Whether to delay the next item after a cooldown, and by how much."""

Built-in Implementations:

• ExponentialBackoffStrategy: Exponential backoff with configurable parameters
• FixedDelayStrategy: Fixed delay between retries

---

Middleware and Observers

Middleware

Interface for middleware that can modify work items before/after processing.

class Middleware(ABC):
    async def before_process(
        self, work_item: LLMWorkItem
    ) -> LLMWorkItem | None: ...

    async def after_process(
        self, result: WorkItemResult
    ) -> WorkItemResult: ...

    async def on_error(
        self, work_item: LLMWorkItem, error: Exception
    ) -> WorkItemResult | None: ...

Methods:

• before_process(): Modify work item before processing. Return None to skip the item (it is recorded as failed).
• after_process(): Modify the result after processing (takes only the result).
• on_error(): Handle errors. Return a WorkItemResult to substitute it for the error (the first middleware returning non-None wins), or None for default error handling.

All three are abstract — subclass BaseMiddleware for no-op defaults so you only override the hooks you need.

Example:

from async_batch_llm.middleware import BaseMiddleware

class LoggingMiddleware(BaseMiddleware):
    async def before_process(self, work_item):
        print(f"Processing {work_item.item_id}")
        return work_item

    async def after_process(self, result):
        print(f"Completed {result.item_id}: {result.success}")
        return result

---

ProcessorObserver

Interface for observers that monitor processing events.

class ProcessorObserver(ABC):
    @abstractmethod
    async def on_event(
        self, event: ProcessingEvent, data: dict[str, Any]
    ) -> None: ...

Events:

• BATCH_STARTED: {total, max_workers, start_time}
• BATCH_COMPLETED: {processed, succeeded, failed, total, total_tokens, cached_input_tokens, duration}
• WORKER_STARTED / WORKER_STOPPED: {worker_id}
• ITEM_STARTED: {item_id, worker_id}
• ITEM_COMPLETED: {item_id, duration, tokens}
• ITEM_FAILED: {item_id, error_type}
• RATE_LIMIT_HIT: {item_id, worker_id}
• COOLDOWN_STARTED: {worker_id, duration, consecutive}
• COOLDOWN_ENDED: {duration, error?}

Cleanup note:

• Preferred: wrap ParallelBatchProcessor in async with so strategy cleanup runs automatically.
• If you do not use a context manager, call await processor.shutdown() after process_all() to flush observers, stop workers, and run strategy cleanups.

---

MetricsObserver

Built-in observer for collecting metrics.

class MetricsObserver(BaseObserver):
    async def get_metrics(self) -> dict[str, Any]: ...
    async def export_json(self) -> str: ...
    async def export_prometheus(self) -> str: ...
    async def export_dict(self) -> dict[str, Any]: ...

Methods:

• get_metrics(): Get current metrics as dict
• export_json(): Export metrics as JSON string
• export_prometheus(): Export in Prometheus text format
• export_dict(): Export as dictionary

Example:

from async_batch_llm import MetricsObserver

metrics = MetricsObserver()
processor = ParallelBatchProcessor(config=config, observers=[metrics])

await processor.process_all()

# Get metrics
metrics_data = await metrics.get_metrics()
print(f"Items processed: {metrics_data['items_processed']}")
print(f"Success rate: {metrics_data['success_rate']:.1%}")

# Export for monitoring
prometheus_text = await metrics.export_prometheus()

---

Core Types

TokenUsage

TypedDict for token usage statistics from LLM API calls.

class TokenUsage(TypedDict, total=False):
    input_tokens: int
    output_tokens: int
    total_tokens: int
    cached_input_tokens: int

Fields (all optional):

• input_tokens (int): Number of tokens in the input/prompt
• output_tokens (int): Number of tokens in the output/completion
• total_tokens (int): Total tokens used (input + output)
• cached_input_tokens (int): Number of input tokens served from cache (Gemini context caching)

Notes:

• All fields are optional to accommodate different provider APIs
• Different providers may return different subsets of these fields
• Use .get() method for safe access: tokens.get("input_tokens", 0)

Example:

from async_batch_llm import TokenUsage

tokens: TokenUsage = {
    "input_tokens": 150,
    "output_tokens": 75,
    "total_tokens": 225,
}

# Safe access
input_tokens = tokens.get("input_tokens", 0)

# Gemini with caching
gemini_tokens: TokenUsage = {
    "input_tokens": 50,  # New tokens only
    "output_tokens": 75,
    "total_tokens": 125,
    "cached_input_tokens": 1000,  # Tokens served from cache
}

---

RetryState

Mutable per-work-item state that persists across retries. The framework creates a RetryState instance for each LLMWorkItem and passes it to both strategy.execute(...) and strategy.on_error(...) via the state parameter.

from dataclasses import dataclass, field

@dataclass
class RetryState:
    data: dict[str, Any] = field(default_factory=dict)

    def get(self, key: str, default: Any = None) -> Any: ...
    def set(self, key: str, value: Any) -> None: ...
    def delete(self, key: str, raise_if_missing: bool = False) -> None: ...
    def clear(self) -> None: ...

Typical uses:

• Track validation failures to escalate models only when schema validation fails
• Store partial results so retries request only the missing fields
• Record which advanced retry prompt should be used next

Example:

async def execute(
    self, prompt: str, attempt: int, timeout: float, state: RetryState | None = None
):
    state = state or RetryState()
    missing = state.get("missing_fields", ["name", "email"])
    response = await self.client.generate(prompt, focus=missing)
    result = parse(response)

    missing = [f for f in ALL_FIELDS if f not in result]
    if missing:
        state.set("missing_fields", missing)
        raise ValidationError("Still missing fields", result)
    state.delete("missing_fields", raise_if_missing=False)
    return result, extract_tokens(response)

Because the same RetryState instance is reused across attempts, each retry can build on the previous attempt’s context without relying on global variables.

---

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.

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

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)

Experimental. This surface is new (v0.16.0) and hasn't seen much real-world use yet — the reserved-key dict shapes and the typed views may change in a future minor release while they stabilize. The metadata dict channel itself is stable; if you persist these shapes, read them back defensively.

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

Key	Shape	Emitted by
grounding	{"sources": [{"uri", "title"}], "queries": [str], "supports": [dict]}	Gemini models, when the response carries google_search grounding
reasoning	str — the model's reasoning/thinking trace	OpenAI-compatible models (reasoning_content, e.g. DeepSeek, falling back to reasoning, e.g. OpenRouter)
tool_calls	[{"id": str\|None, "name": str, "arguments": str}] — arguments is the raw JSON string	OpenAI-compatible models
logprobs	provider-shaped dict/list (via model_dump())	OpenAI-compatible models, when requested

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). Covered for OpenAI-compatible providers only this phase (Gemini function-call parts are not extracted yet).
• A response whose content is null (e.g. a pure tool-call turn) still raises EmptyResponseError before any result exists, so tool_calls surfaces only when the model returned text alongside the calls.
• Auxiliary output does not survive empty/safety-blocked responses — the call fails first.

---

FrameworkTimeoutError

Exception raised when framework-level timeout is exceeded.

class FrameworkTimeoutError(TimeoutError):
    """
    Timeout enforced by the async-batch-llm framework (asyncio.wait_for).

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

Purpose:

Differentiates between:

• Framework timeout: asyncio.wait_for() timed out (exceeded timeout_per_item)
• API timeout: LLM provider returned timeout error (network issue, slow response)

Error Classification:

• is_retryable: True (might succeed if LLM is faster on retry)
• is_timeout: True
• error_category: "framework_timeout"

When to increase timeout_per_item:

If you see frequent FrameworkTimeoutError, it indicates:

1. LLM calls are taking longer than configured timeout
2. Retry delays don't fit within timeout window
3. Solution: Increase timeout_per_item or reduce retry configuration

Example:

from async_batch_llm import FrameworkTimeoutError

try:
    result = await processor.process_all()
except FrameworkTimeoutError as e:
    print(f"Framework timeout: {e}")
    print("Consider increasing timeout_per_item in config")

# Or check in results
for item_result in result.results:
    if not item_result.success and "FrameworkTimeoutError" in item_result.error:
        print(f"{item_result.item_id} exceeded timeout")

---

TokenTrackingError

Exception wrapper that preserves token usage from failed LLM calls.

class TokenTrackingError(Exception):
    """
    Wrapper exception that preserves token usage from failed LLM calls.

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

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

Purpose:

When an LLM call succeeds in getting a response but fails during parsing/validation, the tokens were still consumed and should be tracked for accurate cost accounting. This wrapper preserves that token usage even when the original exception doesn't have a __dict__ (like built-in exceptions).

Usage:

Strategies use this internally to wrap exceptions that don't support attribute assignment:

try:
    output = parse_response(response)
except Exception as e:
    if not hasattr(e, "__dict__"):
        wrapped = TokenTrackingError(str(e), token_usage=tokens)
        wrapped.__cause__ = e
        raise wrapped from e
    else:
        e.__dict__["_failed_token_usage"] = tokens
        raise

Catching TokenTrackingError:

from async_batch_llm import TokenTrackingError

for item_result in result.results:
    if not item_result.success:
        # Token usage is preserved even for failed items
        print(f"Failed: {item_result.item_id}")
        print(f"Tokens consumed: {item_result.token_usage}")

---

Type Aliases

PostProcessorFunc

Callback function called after each item (both successes and failures).

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

Example:

async def save_result(result: WorkItemResult):
    if result.success:
        await database.save(result.item_id, result.output)

processor = ParallelBatchProcessor(
    config=config,
    post_processor=save_result
)

ProgressCallbackFunc

Callback function for progress updates.

ProgressCallbackFunc = Callable[
    [int, int, str],  # (completed, total, current_item_id)
    Awaitable[None] | None
]

Example:

async def on_progress(completed: int, total: int, current_item: str):
    print(f"Progress: {completed}/{total} - {current_item}")

processor = ParallelBatchProcessor(
    config=config,
    progress_callback=on_progress
)

---

Complete Example

import asyncio
from async_batch_llm import (
    ParallelBatchProcessor,
    ProcessorConfig,
    LLMWorkItem,
    PydanticAIStrategy,
    MetricsObserver,
)
from pydantic_ai import Agent
from pydantic import BaseModel

class Summary(BaseModel):
    title: str
    summary: str

async def main():
    # Configure processor
    config = ProcessorConfig(
        max_workers=10,
        timeout_per_item=60.0,
        max_queue_size=100,
    )

    # Create strategy
    agent = Agent("openai:gpt-4", output_type=Summary)
    strategy = PydanticAIStrategy(agent=agent)

    # Add metrics
    metrics = MetricsObserver()

    # Create processor with context manager
    async with ParallelBatchProcessor(
        config=config,
        observers=[metrics]
    ) as processor:
        # Add work items
        for i in range(50):
            work_item = LLMWorkItem(
                item_id=f"doc_{i}",
                strategy=strategy,
                prompt=f"Summarize document {i}...",
            )
            await processor.add_work(work_item)

        # Process all
        result = await processor.process_all()

        # Report results
        print(f"Completed: {result.succeeded}/{result.total_items}")
        print(f"Tokens used: {result.total_input_tokens + result.total_output_tokens}")

        # Get metrics
        metrics_data = await metrics.get_metrics()
        print(f"Average processing time: {metrics_data['avg_processing_time']:.2f}s")

if __name__ == "__main__":
    asyncio.run(main())

---

See Also

• README.md - Getting started guide
• MIGRATION_V0_1.md - Migration guide from v0.0.x (strategy pattern)
• MIGRATION_V0_4.md - Migration guide to v0.4.0 (context managers)
• GEMINI_INTEGRATION.md - Detailed Gemini integration guide
• CHANGELOG.md - Version history