> ## Documentation Index
> Fetch the complete documentation index at: https://daily-ms-ws-body-url-encode.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# Observer Pattern

> The Pipecat observer pattern: watch frames flow through a pipeline without modifying them, and write your own observers.

The Observer pattern in Pipecat allows non-intrusive monitoring of frames as they flow through the pipeline. Observers can watch frame traffic without affecting the pipeline's core functionality.

## Base Observer

All observers must inherit from `BaseObserver` and can implement these methods:

* `on_push_frame(data: FramePushed)`: Called when a frame is pushed from one processor to another
* `on_process_frame(data: FrameProcessed)`: Called when a frame is being processed by a processor
* `on_pipeline_started()`: Called after the `StartFrame` has been processed by all processors in the pipeline
* `on_processor_setup(data: ProcessorSetUp)`: Called once each processor has been set up
* `on_startup_warmup(data: StartupWarmup)`: **Deprecated since 1.12.0.** Never called. Will be removed in 2.0.0.

`ProcessorSetUp` carries the `processor` and the `started_at_ns` / `finished_at_ns` bracketing its `setup()`. Services connect during setup, so this is where that cost can be measured. Processors are set up concurrently, so these arrive in completion order rather than pipeline order, and the times come from `time.monotonic_ns()` because the pipeline clock isn't running yet.

### Frame Push Behavior

A frame is pushed again by every processor that passes it along. By default, an observer receives every push. An observer that handles a frame once (such as one that reports the moment a frame represents) should set `observe_every_push=False` and will be told about a frame only on its first push.

The `FramePushed` event includes a `first_push` field that identifies whether this is the first time a frame is being pushed, allowing observers to distinguish the initial push from subsequent relays.

```python theme={null}
from pipecat.observers.base_observer import BaseObserver, FramePushed, FrameProcessed

class CustomObserver(BaseObserver):
    def __init__(self):
        # observe_every_push=True by default
        super().__init__()

    async def on_push_frame(self, data: FramePushed):
        # data.first_push tells you if this is the first push
        # Your frame observation logic here
        pass

    async def on_process_frame(self, data: FrameProcessed):
        # Your frame processing observation logic here
        pass

    async def on_pipeline_started(self):
        # Called when the pipeline has fully started
        pass
```

**Example: Observer that handles each frame once**

```python theme={null}
class EventObserver(BaseObserver):
    def __init__(self):
        # Only observe the first push of each frame
        super().__init__(observe_every_push=False)

    async def on_push_frame(self, data: FramePushed):
        # This is called once per frame, on its first push
        if isinstance(data.frame, BotStartedSpeakingFrame):
            logger.info("Bot started speaking")
```

## Lifecycle and Cleanup

The pipeline calls `setup()` on each observer when it starts and `cleanup()` when it stops. If your observer spawns background work, use `self.create_task()` so the task is tracked by the pipeline's task manager, and override `cleanup()` to cancel it. Always call `super().cleanup()`, which waits for any in-flight event handlers to finish.

```python theme={null}
class CustomObserver(BaseObserver):
    def __init__(self):
        super().__init__()
        self._task = None

    async def on_pipeline_started(self):
        # Use create_task (not asyncio.create_task) so the task is tracked.
        self._task = self.create_task(self._run())

    async def _run(self):
        # Your long-running background logic here.
        pass

    async def cleanup(self):
        if self._task:
            await self.cancel_task(self._task)
        await super().cleanup()
```

<Note>
  If you give your observer a custom `__init__`, you must call
  `super().__init__()`. Skipping it leaves the observer partially initialized
  and raises errors such as `'CustomObserver' object has no attribute '_name'`
  at runtime.
</Note>

## Available Observers

Pipecat provides several built-in observers:

* **LLMLogObserver**: Logs LLM activity and responses
* **TranscriptionLogObserver**: Logs speech-to-text transcription events
* **RTVIObserver**: Converts internal frames to RTVI protocol messages for server to client messaging
* **[ServiceMetricsObserver](/api-reference/server/utilities/observers/service-metrics-observer)**: Reports service latency and usage metrics as structured records
* **[StartupTimingObserver](/api-reference/server/utilities/observers/startup-timing-observer)**: Measures processor startup times and transport readiness
* **[UserBotLatencyObserver](/api-reference/server/utilities/observers/user-bot-latency-observer)**: Measures user-to-bot response latency
* **[TurnTrackingObserver](/api-reference/server/utilities/observers/turn-tracking-observer)**: Tracks conversation turns and events

## Using Multiple Observers

You can attach multiple observers to a pipeline worker. Each observer will be notified of all frames:

```python theme={null}
worker = PipelineWorker(
    pipeline,
    observers=[LLMLogObserver(), TranscriptionLogObserver(), CustomObserver()],
)
```

## Example: Debug Observer

Here's an example observer that logs interruptions and bot speaking events:

```python theme={null}
from pipecat.observers.base_observer import BaseObserver, FramePushed, FrameProcessed
from pipecat.frames.frames import (
    InterruptionFrame,
    BotStartedSpeakingFrame,
    BotStoppedSpeakingFrame,
)
from pipecat.processors.frame_processor import FrameDirection
from loguru import logger

class DebugObserver(BaseObserver):
    """Observer to log interruptions and bot speaking events to the console.

    Logs all frame instances of:
    - InterruptionFrame
    - BotStartedSpeakingFrame
    - BotStoppedSpeakingFrame

    This allows you to see the frame flow from processor to processor through the pipeline for these frames.
    Log format: [EVENT TYPE]: [source processor] → [destination processor] at [timestamp]s
    """

    async def on_push_frame(self, data: FramePushed):
        time_sec = data.timestamp / 1_000_000_000
        arrow = "→" if data.direction == FrameDirection.DOWNSTREAM else "←"

        if isinstance(data.frame, InterruptionFrame):
            logger.info(f"⚡ INTERRUPTION START: {data.source} {arrow} {data.destination} at {time_sec:.2f}s")
        elif isinstance(data.frame, BotStartedSpeakingFrame):
            logger.info(f"🤖 BOT START SPEAKING: {data.source} {arrow} {data.destination} at {time_sec:.2f}s")
        elif isinstance(data.frame, BotStoppedSpeakingFrame):
            logger.info(f"🤖 BOT STOP SPEAKING: {data.source} {arrow} {data.destination} at {time_sec:.2f}s")
```

## Common Use Cases

Observers are particularly useful for:

* Debugging frame flow
* Logging specific events
* Monitoring pipeline behavior
* Collecting metrics
* Converting internal frames to external messages