base_observer

Base observer classes for monitoring frame flow in the Pipecat pipeline.

This module provides the foundation for observing frame transfers between processors without modifying the pipeline structure. Observers can be used for logging, debugging, analytics, and monitoring pipeline behavior.

class pipecat.observers.base_observer.FrameProcessed(processor: FrameProcessor, frame: Frame, direction: FrameDirection, timestamp: int)[source]

Bases: object

Event data for frame processing in the pipeline.

Represents an event where a frame is being processed by a processor. This data structure is typically used by observers to track the flow of frames through the pipeline for logging, debugging, or analytics purposes.

Parameters:

processor – The processor processing the frame.
frame – The frame being processed.
direction – The direction of the frame (e.g., downstream or upstream).
timestamp – The time when the frame was pushed, based on the pipeline clock.

processor: FrameProcessor

frame: Frame

direction: FrameDirection

timestamp: int

class pipecat.observers.base_observer.FramePushed(source: FrameProcessor, destination: FrameProcessor, frame: Frame, direction: FrameDirection, timestamp: int)[source]

Bases: object

Event data for frame transfers between processors in the pipeline.

Represents an event where a frame is pushed from one processor to another within the pipeline. This data structure is typically used by observers to track the flow of frames through the pipeline for logging, debugging, or analytics purposes.

Parameters:

source – The processor sending the frame.
destination – The processor receiving the frame.
frame – The frame being transferred.
direction – The direction of the transfer (e.g., downstream or upstream).
timestamp – The time when the frame was pushed, based on the pipeline clock.

source: FrameProcessor

destination: FrameProcessor

frame: Frame

direction: FrameDirection

timestamp: int

class pipecat.observers.base_observer.BaseObserver(*, name: str | None = None, **kwargs)[source]

Bases: BaseObject

Base class for pipeline frame observers.

Observers can view all frames that flow through the pipeline without needing to inject processors into the pipeline structure. This enables non-intrusive monitoring capabilities such as frame logging, debugging, performance analysis, and analytics collection.

async on_process_frame(data: FrameProcessed)[source]

Handle the event when a frame is being processed by a processor.

This method should be implemented by subclasses to define specific behavior (e.g., logging, monitoring, debugging) when a frame is being processed by a processor.

Parameters:: data – The event data containing details about the frame processing.

async on_push_frame(data: FramePushed)[source]

Handle the event when a frame is pushed from one processor to another.

This method should be implemented by subclasses to define specific behavior (e.g., logging, monitoring, debugging) when a frame is transferred through the pipeline.

Parameters:: data – The event data containing details about the frame transfer.

async on_pipeline_started()[source]

Called when the pipeline has fully started.

Fired after the StartFrame has been processed by all processors in the pipeline, including nested ParallelPipeline branches.