Observers API Reference

ProcessorObserver

async_batch_llm.observers.ProcessorObserver

Bases: ABC

Abstract base class for processor event observers.

on_event abstractmethod async

on_event(event: ProcessingEvent, data: dict[str, Any]) -> None

Handle processor event.

Parameters:

Name	Type	Description	Default
event	ProcessingEvent	The event type	required
data	dict[str, Any]	Event-specific data	required

Source code in src/async_batch_llm/observers/base.py

@abstractmethod
async def on_event(
    self,
    event: ProcessingEvent,
    data: dict[str, Any],
) -> None:
    """
    Handle processor event.

    Args:
        event: The event type
        data: Event-specific data
    """
    pass

BaseObserver

async_batch_llm.observers.BaseObserver

Bases: ProcessorObserver

Base observer with no-op implementation.

on_event async

on_event(event: ProcessingEvent, data: dict[str, Any]) -> None

Default: do nothing.

Source code in src/async_batch_llm/observers/base.py

async def on_event(
    self,
    event: ProcessingEvent,
    data: dict[str, Any],
) -> None:
    """Default: do nothing."""
    pass

MetricsObserver

async_batch_llm.observers.MetricsObserver

MetricsObserver(*, max_processing_samples: int = 100)

Bases: BaseObserver

Collect metrics for monitoring (thread-safe).

Initialize metrics collector.

Source code in src/async_batch_llm/observers/metrics.py

def __init__(self, *, max_processing_samples: int = 100):
    """Initialize metrics collector."""
    if max_processing_samples <= 0:
        raise ValueError("max_processing_samples must be positive")
    self.metrics: dict[str, Any] = _initial_metrics()
    self._processing_times: list[float] = []
    self._quota_scopes: set[int] = set()
    self._max_processing_samples = max_processing_samples
    self._lock = asyncio.Lock()

export_dict async

export_dict() -> dict[str, Any]

Export metrics as a dictionary.

Returns:

Type	Description
dict[str, Any]	Dictionary containing all metrics and computed statistics

Example

observer = MetricsObserver()

... process items ...

data = await observer.export_dict() print(data["success_rate"])

Source code in src/async_batch_llm/observers/metrics.py

async def export_dict(self) -> dict[str, Any]:
    """Export metrics as a dictionary.

    Returns:
        Dictionary containing all metrics and computed statistics

    Example:
        >>> observer = MetricsObserver()
        >>> # ... process items ...
        >>> data = await observer.export_dict()
        >>> print(data["success_rate"])
    """
    return await self.get_metrics()

export_json async

export_json() -> str

Export metrics as JSON string.

Returns:

Type	Description
str	JSON string containing all metrics and computed statistics

Example

observer = MetricsObserver()

... process items ...

json_str = await observer.export_json() print(json_str)

Source code in src/async_batch_llm/observers/metrics.py

async def export_json(self) -> str:
    """Export metrics as JSON string.

    Returns:
        JSON string containing all metrics and computed statistics

    Example:
        >>> observer = MetricsObserver()
        >>> # ... process items ...
        >>> json_str = await observer.export_json()
        >>> print(json_str)
    """
    metrics = await self.get_metrics()
    # Convert processing_times list to just count for cleaner export
    export_data = {
        **{k: v for k, v in metrics.items() if k != "processing_times"},
        "processing_times_count": metrics.get("processing_times_count", 0),
    }
    return json.dumps(export_data, indent=2)

export_prometheus async

export_prometheus() -> str

Export metrics in Prometheus text format.

Returns:

Type	Description
str	Prometheus-formatted metrics string

Example

observer = MetricsObserver()

... process items ...

prom_text = await observer.export_prometheus() print(prom_text)

HELP async_batch_llm_items_processed Total items processed

TYPE async_batch_llm_items_processed counter

async_batch_llm_items_processed 100 ...

Source code in src/async_batch_llm/observers/metrics.py

async def export_prometheus(self) -> str:
    """Export metrics in Prometheus text format.

    Returns:
        Prometheus-formatted metrics string

    Example:
        >>> observer = MetricsObserver()
        >>> # ... process items ...
        >>> prom_text = await observer.export_prometheus()
        >>> print(prom_text)
        # HELP async_batch_llm_items_processed Total items processed
        # TYPE async_batch_llm_items_processed counter
        async_batch_llm_items_processed 100
        ...
    """
    metrics = await self.get_metrics()

    lines = []

    # Counter metrics
    counters = [
        ("items_processed", "Total items processed"),
        ("items_succeeded", "Total items succeeded"),
        ("items_failed", "Total items failed"),
        ("rate_limits_hit", "Total rate limits encountered"),
        ("structured_output_recoveries", "Total structured outputs recovered"),
        ("items_replayed", "Total terminal items replayed from artifacts"),
        ("items_aborted", "Total collateral items aborted by guardrails"),
        ("batches_aborted", "Total batches stopped by guardrails"),
        ("quota_admitted_attempts", "Total live quota reservations admitted"),
        ("reserved_tokens", "Total tokens reserved for TPM admission"),
        ("reported_reconciliation_tokens", "Total known tokens used for reconciliation"),
        ("refunded_tokens", "Total unused reserved tokens refunded"),
        ("underestimated_tokens", "Total token underestimation charged as debt"),
        ("unknown_usage_attempts", "Started quota attempts with unknown usage"),
        ("known_zero_usage_attempts", "Started quota attempts reporting zero usage"),
        ("token_estimation_failures", "Total framework token-estimation failures"),
        (
            "structured_output_retries_avoided",
            "Total validation retries avoided by structured-output recovery",
        ),
    ]

    for metric_name, help_text in counters:
        lines.append(f"# HELP async_batch_llm_{metric_name} {help_text}")
        lines.append(f"# TYPE async_batch_llm_{metric_name} counter")
        lines.append(f"async_batch_llm_{metric_name} {metrics.get(metric_name, 0)}")
        lines.append("")

    # Gauge metrics
    gauges = [
        ("avg_processing_time", "Average processing time in seconds"),
        ("success_rate", "Success rate (0.0 to 1.0)"),
        ("total_cooldown_time", "Total time spent in rate limit cooldown (seconds)"),
        ("processing_times_count", "Number of recorded processing time samples"),
        ("admission_wait_count", "Number of provider-capacity admissions"),
        ("admission_wait_seconds_sum", "Total provider-capacity wait time in seconds"),
        ("admission_wait_seconds_max", "Maximum provider-capacity wait time in seconds"),
        ("avg_admission_wait_seconds", "Average provider-capacity wait time in seconds"),
        ("quota_wait_seconds_sum", "Total RPM/TPM quota wait time in seconds"),
        ("quota_wait_seconds_max", "Maximum RPM/TPM quota wait time in seconds"),
        ("avg_quota_wait_seconds", "Average RPM/TPM quota wait time in seconds"),
        ("estimated_input_tokens", "Total estimated input tokens"),
        ("estimated_output_tokens", "Total estimated output tokens"),
        ("quota_scope_count", "Number of run-local quota scopes observed"),
    ]

    for metric_name, help_text in gauges:
        lines.append(f"# HELP async_batch_llm_{metric_name} {help_text}")
        lines.append(f"# TYPE async_batch_llm_{metric_name} gauge")
        lines.append(f"async_batch_llm_{metric_name} {metrics.get(metric_name, 0)}")
        lines.append("")

    # Error counts as labeled counter
    error_counts = metrics.get("error_counts", {})
    if error_counts:
        lines.append("# HELP async_batch_llm_errors_total Total errors by type")
        lines.append("# TYPE async_batch_llm_errors_total counter")
        for error_type, count in error_counts.items():
            # Sanitize error type for Prometheus label
            safe_type = error_type.replace('"', '\\"')
            lines.append(f'async_batch_llm_errors_total{{error_type="{safe_type}"}} {count}')
        lines.append("")

    recovery_reasons = metrics.get("structured_output_recovery_reasons", {})
    if recovery_reasons:
        lines.append(
            "# HELP async_batch_llm_structured_output_recoveries_by_reason "
            "Structured-output recoveries by reason"
        )
        lines.append("# TYPE async_batch_llm_structured_output_recoveries_by_reason counter")
        for reason, count in recovery_reasons.items():
            safe_reason = reason.replace('"', '\\"')
            lines.append(
                "async_batch_llm_structured_output_recoveries_by_reason"
                f'{{reason="{safe_reason}"}} {count}'
            )
        lines.append("")

    return "\n".join(lines)

get_metrics async

get_metrics() -> dict[str, Any]

Get collected metrics with computed statistics (thread-safe).

Source code in src/async_batch_llm/observers/metrics.py

async def get_metrics(self) -> dict[str, Any]:
    """Get collected metrics with computed statistics (thread-safe)."""
    async with self._lock:
        return {
            **self.metrics,
            "processing_times": list(self._processing_times),
            "avg_processing_time": (
                self.metrics["processing_times_sum"] / self.metrics["processing_times_count"]
                if self.metrics["processing_times_count"] > 0
                else 0
            ),
            "avg_admission_wait_seconds": (
                self.metrics["admission_wait_seconds_sum"]
                / self.metrics["admission_wait_count"]
                if self.metrics["admission_wait_count"] > 0
                else 0
            ),
            "avg_quota_wait_seconds": (
                self.metrics["quota_wait_seconds_sum"] / self.metrics["quota_admitted_attempts"]
                if self.metrics["quota_admitted_attempts"] > 0
                else 0
            ),
            "success_rate": (
                self.metrics["items_succeeded"] / self.metrics["items_processed"]
                if self.metrics["items_processed"] > 0
                else 0
            ),
        }

on_event async

on_event(event: ProcessingEvent, data: dict[str, Any]) -> None

Collect metrics from events (thread-safe).

Source code in src/async_batch_llm/observers/metrics.py

async def on_event(
    self,
    event: ProcessingEvent,
    data: dict[str, Any],
) -> None:
    """Collect metrics from events (thread-safe)."""
    async with self._lock:
        if event == ProcessingEvent.ITEM_COMPLETED:
            self.metrics["items_processed"] += 1
            self.metrics["items_succeeded"] += 1
            if "duration" in data:
                duration = float(data["duration"])
                self.metrics["processing_times_sum"] += duration
                self.metrics["processing_times_count"] += 1
                self._processing_times.append(duration)
                if len(self._processing_times) > self._max_processing_samples:
                    self._processing_times.pop(0)
            if data.get("structured_output_recovered") is True:
                self.metrics["structured_output_recoveries"] += 1
                retries_avoided = data.get("structured_output_retries_avoided", 0)
                if (
                    not isinstance(retries_avoided, bool)
                    and isinstance(retries_avoided, int)
                    and retries_avoided > 0
                ):
                    self.metrics["structured_output_retries_avoided"] += retries_avoided
                reason = data.get("structured_output_recovery_reason")
                if isinstance(reason, str) and reason:
                    reasons = self.metrics["structured_output_recovery_reasons"]
                    reasons[reason] = reasons.get(reason, 0) + 1

        elif event == ProcessingEvent.ITEM_ADMITTED:
            wait_seconds = float(data.get("wait_seconds", 0.0))
            self.metrics["admission_wait_count"] += 1
            self.metrics["admission_wait_seconds_sum"] += wait_seconds
            self.metrics["admission_wait_seconds_max"] = max(
                self.metrics["admission_wait_seconds_max"], wait_seconds
            )

        elif event == ProcessingEvent.QUOTA_ADMITTED:
            wait_seconds = float(data.get("wait_seconds", 0.0))
            self.metrics["quota_admitted_attempts"] += 1
            self.metrics["quota_wait_seconds_sum"] += wait_seconds
            self.metrics["quota_wait_seconds_max"] = max(
                self.metrics["quota_wait_seconds_max"], wait_seconds
            )
            for key in (
                "estimated_input_tokens",
                "estimated_output_tokens",
                "reserved_tokens",
            ):
                value = data.get(key, 0)
                if not isinstance(value, bool) and isinstance(value, int):
                    self.metrics[key] += value
            scope_id = data.get("quota_scope_id")
            if not isinstance(scope_id, bool) and isinstance(scope_id, int):
                self._quota_scopes.add(scope_id)
                self.metrics["quota_scope_count"] = len(self._quota_scopes)

        elif event == ProcessingEvent.QUOTA_RECONCILED:
            reported = data.get("reported_tokens")
            delta = data.get("delta_tokens")
            reserved = data.get("reserved_tokens", 0)
            if not isinstance(reported, bool) and isinstance(reported, int):
                self.metrics["reported_reconciliation_tokens"] += reported
                if reported == 0 and reserved:
                    self.metrics["known_zero_usage_attempts"] += 1
            elif reserved:
                self.metrics["unknown_usage_attempts"] += 1
            if not isinstance(delta, bool) and isinstance(delta, int):
                if delta < 0:
                    self.metrics["refunded_tokens"] += -delta
                elif delta > 0:
                    self.metrics["underestimated_tokens"] += delta

        elif event == ProcessingEvent.ITEM_FAILED:
            self.metrics["items_processed"] += 1
            self.metrics["items_failed"] += 1
            if data.get("error_category") in ABORT_RESULT_CATEGORIES:
                self.metrics["items_aborted"] += 1
            if "error_type" in data:
                error_type = data["error_type"]
                self.metrics["error_counts"][error_type] = (
                    self.metrics["error_counts"].get(error_type, 0) + 1
                )
            if data.get("error_category") in {
                "token_estimation_error",
                "token_estimator_required",
                "token_estimate_exceeds_limit",
            }:
                self.metrics["token_estimation_failures"] += 1

        elif event == ProcessingEvent.ITEM_REPLAYED:
            self.metrics["items_processed"] += 1
            self.metrics["items_replayed"] += 1
            if data.get("success") is True:
                self.metrics["items_succeeded"] += 1
            else:
                self.metrics["items_failed"] += 1

            if not data.get("success") and data.get("error_type") is not None:
                error_type = data["error_type"]
                self.metrics["error_counts"][error_type] = (
                    self.metrics["error_counts"].get(error_type, 0) + 1
                )

        elif event == ProcessingEvent.BATCH_ABORTED:
            self.metrics["batches_aborted"] += 1

        elif event == ProcessingEvent.RATE_LIMIT_HIT:
            self.metrics["rate_limits_hit"] += 1

        elif event == ProcessingEvent.COOLDOWN_ENDED:
            if "duration" in data:
                self.metrics["total_cooldown_time"] += data["duration"]

reset async

reset() -> None

Reset all metrics (thread-safe).

Async since v0.16: the reset acquires the same lock as on_event, so counts from an in-flight event can no longer land in the discarded pre-reset dict.

Source code in src/async_batch_llm/observers/metrics.py

async def reset(self) -> None:
    """Reset all metrics (thread-safe).

    Async since v0.16: the reset acquires the same lock as
    ``on_event``, so counts from an in-flight event can no longer land
    in the discarded pre-reset dict.
    """
    async with self._lock:
        self.metrics = _initial_metrics()
        self._processing_times = []
        self._quota_scopes = set()

ProcessingEvent

async_batch_llm.ProcessingEvent

Bases: Enum

Events that can be observed during processing.

Event contracts

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, total_admission_wait_seconds, max_admission_wait_seconds, total_quota_wait_seconds, max_quota_wait_seconds, quota_wait_p50/p95/p99_seconds, estimated_input_tokens, estimated_output_tokens, reserved_tokens, reported_reconciliation_tokens, refunded_tokens, underestimated_tokens, unknown_usage_attempts, known_zero_usage_attempts, token_estimation_failures, quota_scope_count, admission_wait_p50/p95/p99_seconds, execution_p50/p95/p99_seconds, structured_output_recoveries, structured_output_retries_avoided, structured_output_recovery_reasons, input_queue_high_water_mark, result_queue_high_water_mark, duration}
• BATCH_ABORTED: {kind, reason, error_category, triggering_item_id, abort_mode}
• WORKER_STARTED / WORKER_STOPPED: {worker_id}
• ITEM_STARTED: {item_id, worker_id}
• ITEM_ADMITTED: {item_id, worker_id, attempt, wait_seconds, capacity, startup_ramp_wait_seconds}
• QUOTA_ADMITTED: {item_id, worker_id, attempt, try_number, quota_scope_id, wait_seconds, request_reserved, estimated_input_tokens, estimated_output_tokens, estimated_total_tokens, reserved_tokens, rpm_configured, tpm_configured, limited_by}
• QUOTA_RECONCILED: {item_id, worker_id, attempt, try_number, quota_scope_id, reserved_tokens, reported_tokens, known_usage, delta_tokens, disposition}
• ITEM_COMPLETED: {item_id, duration, tokens, admission_wait_seconds, structured_output_recovered, structured_output_recovery_reason, structured_output_retries_avoided}
• ITEM_FAILED: {item_id, submission_index, error_type, error_category}
• ITEM_REPLAYED: {item_id, submission_index, success, error_type, error_category}
• ITEM_DEADLINE_EXCEEDED: {item_id}
• RATE_LIMIT_HIT: {item_id, worker_id, quota_scope_id, strategy_type}
• COOLDOWN_STARTED: {worker_id, duration, consecutive, quota_scope_id?, strategy_type?}
• COOLDOWN_ENDED: {duration, error?}

A key marked ? may be absent. The keys listed here are the documented payload that API stability covers; an event may carry more keys, which are diagnostic and can change in a minor release.

add_work() raises public BatchAdmissionClosedError (a RuntimeError) when finish, shutdown, or batch abort has stopped admission. Rejected items receive no submission index. Use a new processor for additional work.

Observers may be duck-typed objects with a callable on_event(event, data); subclassing ProcessorObserver is optional. The callback must return an awaitable. Invalid callbacks are rejected at processor construction. Observer exceptions and callback timeouts are logged without failing the item.

Each finalized item emits exactly one of ITEM_COMPLETED, ITEM_FAILED, or ITEM_REPLAYED, after its checkpoint and before any fail-fast BATCH_ABORTED event. A budget stop is different: the provider attempt that reaches the budget emits BATCH_ABORTED (kind="budget_exceeded") before that item's own terminal event. Middleware-filtered items count as failed; middleware recovery and after_process changes determine the final event. Fatal checkpoint failures and cancellation before finalization produce no terminal item event or processed-stat increment. Early shutdown may abandon accepted items, so processed counts can be lower than total. Replay does not emit an additional success/failure event or consume live tokens. For live provider successes, duration and tokens retain final-attempt values; new recovery-only success events use the final result's timing and token totals.

Cleanup note:

• Preferred: wrap ParallelBatchProcessor in async with so the ordered close runs automatically.
• If you do not use a context manager, call await processor.shutdown() after process_all(). It stops workers and callbacks, releases admission state, runs strategy cleanups, and closes the artifact store. Repeated calls are safe and retry only what did not complete.

---