BaseObserver: added new on_process_frame

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.
 

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.
 

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 self._send_to_proxy(data)
-            await proxy.queue.put(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(
+            if isinstance(data, FramePushed):
-                    data.src, data.dst, data.frame, data.direction, data.timestamp
+                if on_push_frame_deprecated:
-                )
+                    await observer.on_push_frame(
-            else:
+                        data.src, data.dst, data.frame, data.direction, data.timestamp
-                await observer.on_push_frame(data)
+                    )
+                else:
+                    await observer.on_push_frame(data)
+            elif isinstance(data, FrameProcessed):
+                await observer.on_process_frame(data)
 
             queue.task_done()

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):