debug_log_observer

Debug logging observer for frame activity monitoring.

This module provides a debug observer that logs detailed frame activity to the console, making it useful for debugging pipeline behavior and understanding frame flow between processors.

class pipecat.observers.loggers.debug_log_observer.FrameEndpoint(*values)

Bases: Enum

Specifies which endpoint (source or destination) to filter on.

Parameters:

• SOURCE – Filter on the source component that is pushing the frame.

• DESTINATION – Filter on the destination component receiving the frame.

SOURCE = 1

DESTINATION = 2

class pipecat.observers.loggers.debug_log_observer.DebugLogObserver(frame_types: tuple[type[Frame], ...] | dict[type[Frame], tuple[type, FrameEndpoint] | None] | None = None, exclude_fields: set[str] | None = None, **kwargs)

Bases: BaseObserver

Observer that logs frame activity with detailed content to the console.

Automatically extracts and formats data from any frame type, making it useful for debugging pipeline behavior without needing frame-specific observers.

Examples

Log all frames from all services:

observers = DebugLogObserver()

Log specific frame types from any source/destination:

from pipecat.frames.frames import LLMTextFrame, TranscriptionFrame
observers=[
    DebugLogObserver(frame_types=(LLMTextFrame,TranscriptionFrame,)),
]

Log frames with specific source/destination filters:

from pipecat.frames.frames import InterruptionFrame, UserStartedSpeakingFrame, LLMTextFrame
from pipecat.observers.loggers.debug_log_observer import DebugLogObserver, FrameEndpoint
from pipecat.transports.base_output import BaseOutputTransport
from pipecat.services.stt_service import STTService

observers=[
    DebugLogObserver(
        frame_types={
            # Only log InterruptionFrame when source is BaseOutputTransport
            InterruptionFrame: (BaseOutputTransport, FrameEndpoint.SOURCE),
            # Only log UserStartedSpeakingFrame when destination is STTService
            UserStartedSpeakingFrame: (STTService, FrameEndpoint.DESTINATION),
            # Log LLMTextFrame regardless of source or destination type
            LLMTextFrame: None,
        }
    ),
]

__init__(frame_types: tuple[type[Frame], ...] | dict[type[Frame], tuple[type, FrameEndpoint] | None] | None = None, exclude_fields: set[str] | None = None, **kwargs)

Initialize the debug log observer.

Parameters:

• frame_types –

  Frame types to log. Can be:

  • Tuple of frame types to log all instances

  • Dict mapping frame types to filter configurations

  • None to log all frames

  Filter configurations can be None (log all instances) or a tuple of (service_type, endpoint) to filter on specific services.

• exclude_fields – Field names to exclude from logging. Defaults to excluding binary data fields like ‘audio’, ‘image’, ‘images’.

• **kwargs – Additional arguments passed to parent class.

async on_push_frame(data: FramePushed)

Process a frame being pushed into the pipeline.

Logs frame details to the console with all relevant fields and values.

Parameters:

data – Event data containing the frame, source, destination, direction, and timestamp.