diff --git a/CHANGELOG.md b/CHANGELOG.md index 056912f56..df75b8f7d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added new `on_process_frame()` observer method which makes it possible to know + when a frame is being processed. + - Added new `FrameProcessor.entry_processor()` method. This allows you to access the first non-compound processor in a pipeline. diff --git a/src/pipecat/observers/base_observer.py b/src/pipecat/observers/base_observer.py index d8a67a373..a5c31b81c 100644 --- a/src/pipecat/observers/base_observer.py +++ b/src/pipecat/observers/base_observer.py @@ -11,7 +11,6 @@ processors without modifying the pipeline structure. Observers can be used for logging, debugging, analytics, and monitoring pipeline behavior. """ -from abc import abstractmethod from dataclasses import dataclass from typing_extensions import TYPE_CHECKING @@ -23,6 +22,28 @@ if TYPE_CHECKING: from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +@dataclass +class FrameProcessed: + """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 + + @dataclass class FramePushed: """Event data for frame transfers between processors in the pipeline. @@ -56,7 +77,18 @@ class BaseObserver(BaseObject): performance analysis, and analytics collection. """ - @abstractmethod + async def on_process_frame(self, data: FrameProcessed): + """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. + + Args: + data: The event data containing details about the frame processing. + """ + pass + async def on_push_frame(self, data: FramePushed): """Handle the event when a frame is pushed from one processor to another. diff --git a/src/pipecat/pipeline/task_observer.py b/src/pipecat/pipeline/task_observer.py index 532ef7977..6e4e44742 100644 --- a/src/pipecat/pipeline/task_observer.py +++ b/src/pipecat/pipeline/task_observer.py @@ -13,11 +13,11 @@ the main pipeline execution. import asyncio import inspect -from typing import Dict, List, Optional +from typing import Any, Dict, List, Optional from attr import dataclass -from pipecat.observers.base_observer import BaseObserver, FramePushed +from pipecat.observers.base_observer import BaseObserver, FrameProcessed, FramePushed from pipecat.utils.asyncio.task_manager import BaseTaskManager from pipecat.utils.asyncio.watchdog_queue import WatchdogQueue @@ -120,14 +120,21 @@ class TaskObserver(BaseObserver): for proxy in self._proxies.values(): await self._task_manager.cancel_task(proxy.task) + async def on_process_frame(self, data: FramePushed): + """Queue frame data for all managed observers. + + Args: + data: The frame push event data to distribute to observers. + """ + await self._send_to_proxy(data) + async def on_push_frame(self, data: FramePushed): """Queue frame data for all managed observers. Args: data: The frame push event data to distribute to observers. """ - for proxy in self._proxies.values(): - await proxy.queue.put(data) + await self._send_to_proxy(data) def _started(self) -> bool: """Check if the task observer has been started.""" @@ -151,6 +158,10 @@ class TaskObserver(BaseObserver): proxies[observer] = proxy return proxies + async def _send_to_proxy(self, data: Any): + for proxy in self._proxies.values(): + await proxy.queue.put(data) + async def _proxy_task_handler(self, queue: asyncio.Queue, observer: BaseObserver): """Handle frame processing for a single observer.""" on_push_frame_deprecated = False @@ -169,11 +180,15 @@ class TaskObserver(BaseObserver): while True: data = await queue.get() - if on_push_frame_deprecated: - await observer.on_push_frame( - data.src, data.dst, data.frame, data.direction, data.timestamp - ) - else: - await observer.on_push_frame(data) + + if isinstance(data, FramePushed): + if on_push_frame_deprecated: + await observer.on_push_frame( + data.src, data.dst, data.frame, data.direction, data.timestamp + ) + else: + await observer.on_push_frame(data) + elif isinstance(data, FrameProcessed): + await observer.on_process_frame(data) queue.task_done() diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index fdc160514..47612ed89 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -34,7 +34,7 @@ from pipecat.frames.frames import ( SystemFrame, ) from pipecat.metrics.metrics import LLMTokenUsage, MetricsData -from pipecat.observers.base_observer import BaseObserver, FramePushed +from pipecat.observers.base_observer import BaseObserver, FrameProcessed, FramePushed from pipecat.processors.metrics.frame_processor_metrics import FrameProcessorMetrics from pipecat.utils.asyncio.task_manager import BaseTaskManager from pipecat.utils.asyncio.watchdog_event import WatchdogEvent @@ -594,6 +594,16 @@ class FrameProcessor(BaseObject): frame: The frame to process. direction: The direction of frame flow. """ + if self._observer: + timestamp = self._clock.get_time() if self._clock else 0 + data = FrameProcessed( + processor=self, + frame=frame, + direction=direction, + timestamp=timestamp, + ) + await self._observer.on_process_frame(data) + if isinstance(frame, StartFrame): await self.__start(frame) elif isinstance(frame, StartInterruptionFrame):