Error Observer

​

Features

• Reports every error at its origin
• Includes recovered errors that don’t reach the top of the pipeline
• Provides error categorization via ErrorCategory
• Tracks processor usability after failures
• Deduplicates errors that travel through multiple processors

​

Usage

​

Basic Setup

from pipecat.observers.error_observer import ErrorObserver

observer = ErrorObserver()

@observer.event_handler("on_error")
async def on_error(observer, event):
    logger.error(
        f"Error in {event.processor}: {event.message} "
        f"(category: {event.category}, usable: {event.processor_usable})"
    )

worker = PipelineWorker(
    pipeline,
    observers=[observer],
)

​

JSON Logging

@observer.event_handler("on_error")
async def on_error(observer, event):
    logger.info(event.model_dump_json())

​

Configuration

​

ErrorEvent

​

Event Handlers

​

Notes

• Errors are reported once per occurrence, even though they travel upstream through multiple processors
• An error is attributed to the processor that raised it, not the ones that passed it along
• processor_usable distinguishes transient failures (unreachable service) from permanent ones (invalid credentials)
• Errors recovered by a processor (e.g., service switcher failover) are still reported, unlike PipelineWorker.on_pipeline_error which only sees errors that reach the top