The DebugLogObserver
provides detailed logging of frame activity in your Pipecat pipeline, with full visibility into frame content and flexible filtering options.
Features
- Log all frame types and their content
- Filter by specific frame types
- Filter by source or destination components
- Automatic formatting of frame fields
- Special handling for complex data structures
Usage
Log All Frames
Log all frames passing through the pipeline:
from pipecat.observers.loggers.debug_log_observer import DebugLogObserver
task = PipelineTask(
pipeline,
params=PipelineParams(
observers=[DebugLogObserver()],
),
)
Filter by Frame Types
Log only specific frame types:
from pipecat.frames.frames import TranscriptionFrame, InterimTranscriptionFrame
from pipecat.observers.loggers.debug_log_observer import DebugLogObserver
task = PipelineTask(
pipeline,
params=PipelineParams(
observers=[
DebugLogObserver(frame_types=(
TranscriptionFrame,
InterimTranscriptionFrame
))
],
),
)
Advanced Source/Destination Filtering
Filter frames based on their type and source/destination:
from pipecat.frames.frames import StartInterruptionFrame, UserStartedSpeakingFrame, LLMTextFrame
from pipecat.observers.loggers.debug_log_observer import DebugLogObserver, FrameEndpoint
from pipecat.transports.base_output_transport import BaseOutputTransport
from pipecat.services.stt_service import STTService
task = PipelineTask(
pipeline,
params=PipelineParams(
observers=[
DebugLogObserver(frame_types={
# Only log StartInterruptionFrame when source is BaseOutputTransport
StartInterruptionFrame: (BaseOutputTransport, FrameEndpoint.SOURCE),
# Only log UserStartedSpeakingFrame when destination is STTService
UserStartedSpeakingFrame: (STTService, FrameEndpoint.DESTINATION),
# Log LLMTextFrame regardless of source or destination
LLMTextFrame: None
})
],
),
)
The observer logs each frame with its complete details:
[Source] → [Destination]: [FrameType] [field1: value1, field2: value2, ...] at [timestamp]s
For example:
OpenAILLMService#0 → DailyTransport#0: LLMTextFrame text: 'Hello, how can I help you today?' at 1.24s
Configuration Options
Parameter | Type | Description |
---|
frame_types | Tuple[Type[Frame], ...] or Dict[Type[Frame], Optional[Tuple[Type, FrameEndpoint]]] | Frame types to log, with optional source/destination filtering |
exclude_fields | Set[str] | Field names to exclude from logging (defaults to binary fields) |
FrameEndpoint Enum
The FrameEndpoint
enum is used for source/destination filtering:
FrameEndpoint.SOURCE
: Filter by source component
FrameEndpoint.DESTINATION
: Filter by destination component