Merge pull request #2697 from pipecat-ai/aleix/frame-processor-before-after-events

FrameProcessor: add before/after events for processed/pushed frames

CHANGELOG.md

@@ -9,6 +9,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added `on_before_process_frame`, `on_after_process_frame`,
+  `on_before_push_frame` and `on_after_push_frame`. These are synchronous events
+  that get called before and after a frame is processed or pushed. Note that
+  these events are synchrnous so they should ideally perform lightweight tasks
+  in order to not block the pipeline. See
+  `examples/foundational/45-before-and-after-events.py`.
+
 - Added `on_before_leave` synchronous event to `DailyTransport`.
 
 - Added `on_before_disconnect` synchronous event to `LiveKitTransport`.

examples/foundational/45-before-and-after-events.py

@@ -0,0 +1,150 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import os
+from dataclasses import dataclass
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams
+from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.audio.vad.vad_analyzer import VADParams
+from pipecat.frames.frames import DataFrame, LLMRunFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.runner.types import RunnerArguments
+from pipecat.runner.utils import create_transport
+from pipecat.services.cartesia.tts import CartesiaTTSService
+from pipecat.services.deepgram.stt import DeepgramSTTService
+from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.daily.transport import DailyParams
+from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
+
+load_dotenv(override=True)
+
+
+@dataclass
+class CustomBeforeProcessFrame(DataFrame):
+    pass
+
+
+@dataclass
+class CustomAfterPushFrame(DataFrame):
+    pass
+
+
+# We store functions so objects (e.g. SileroVADAnalyzer) don't get
+# instantiated. The function will be called when the desired transport gets
+# selected.
+transport_params = {
+    "daily": lambda: DailyParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
+        turn_analyzer=LocalSmartTurnAnalyzerV3(params=SmartTurnParams()),
+    ),
+    "twilio": lambda: FastAPIWebsocketParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
+        turn_analyzer=LocalSmartTurnAnalyzerV3(params=SmartTurnParams()),
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
+        turn_analyzer=LocalSmartTurnAnalyzerV3(params=SmartTurnParams()),
+    ),
+}
+
+
+async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
+    logger.info(f"Starting bot")
+
+    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+    tts = CartesiaTTSService(
+        api_key=os.getenv("CARTESIA_API_KEY"),
+        voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
+    )
+
+    llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
+
+    messages = [
+        {
+            "role": "system",
+            "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
+        },
+    ]
+
+    context = OpenAILLMContext(messages)
+    context_aggregator = llm.create_context_aggregator(context)
+
+    pipeline = Pipeline(
+        [
+            transport.input(),  # Transport user input
+            stt,
+            context_aggregator.user(),  # User responses
+            llm,  # LLM
+            tts,  # TTS
+            transport.output(),  # Transport bot output
+            context_aggregator.assistant(),  # Assistant spoken responses
+        ]
+    )
+
+    task = PipelineTask(
+        pipeline,
+        params=PipelineParams(
+            enable_metrics=True,
+            enable_usage_metrics=True,
+        ),
+        idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
+    )
+
+    @transport.output().event_handler("on_before_process_frame")
+    async def transport_on_before_process_frame(transport, frame):
+        if isinstance(frame, CustomBeforeProcessFrame):
+            logger.info("*** CUSTOM FRAME RECEIVED. Bot is about to talk, let's do something!")
+
+    @transport.output().event_handler("on_after_push_frame")
+    async def transport_on_after_push_frame(transport, frame):
+        if isinstance(frame, CustomAfterPushFrame):
+            logger.info("*** CUSTOM FRAME RECEIVED. Bot has stopped talking, let's do something!")
+
+    @transport.event_handler("on_client_connected")
+    async def on_client_connected(transport, client):
+        logger.info(f"Client connected")
+        # Kick off the conversation.
+        messages.append({"role": "system", "content": "Please introduce yourself to the user."})
+        # CustomSynchronizationFrame is pushed in order so it will reach the
+        # `on_after_push_frame` handler before the bot has finished talking.
+        await task.queue_frames([CustomBeforeProcessFrame(), LLMRunFrame(), CustomAfterPushFrame()])
+
+    @transport.event_handler("on_client_disconnected")
+    async def on_client_disconnected(transport, client):
+        logger.info(f"Client disconnected")
+        await task.cancel()
+
+    runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
+
+    await runner.run(task)
+
+
+async def bot(runner_args: RunnerArguments):
+    """Main bot entry point compatible with Pipecat Cloud."""
+    transport = await create_transport(runner_args, transport_params)
+    await run_bot(transport, runner_args)
+
+
+if __name__ == "__main__":
+    from pipecat.runner.run import main
+
+    main()

src/pipecat/processors/frame_processor.py

@@ -141,6 +141,12 @@ class FrameProcessor(BaseObject):
     task. System frames are also processed in a separate task which guarantees
     frame priority.
 
+    Event handlers available:
+
+    - on_before_process_frame: Called before a frame is processed
+    - on_after_process_frame: Called after a frame is processed
+    - on_before_push_frame: Called before a frame is pushed
+    - on_after_push_frame: Called after a frame is pushed
     """
 
     def __init__(
@@ -228,6 +234,12 @@ class FrameProcessor(BaseObject):
         self._wait_for_interruption = False
         self._wait_interruption_event = asyncio.Event()
 
+        # Frame processor events.
+        self._register_event_handler("on_before_process_frame", sync=True)
+        self._register_event_handler("on_after_process_frame", sync=True)
+        self._register_event_handler("on_before_push_frame", sync=True)
+        self._register_event_handler("on_after_push_frame", sync=True)
+
     @property
     def id(self) -> int:
         """Get the unique identifier for this processor.
@@ -639,8 +651,12 @@ class FrameProcessor(BaseObject):
         if not self._check_started(frame):
             return
 
+        await self._call_event_handler("on_before_push_frame", frame)
+
         await self.__internal_push_frame(frame, direction)
 
+        await self._call_event_handler("on_after_push_frame", frame)
+
         # If we are waiting for an interruption and we get an interruption, then
         # we can unblock `push_interruption_task_frame_and_wait()`.
         if self._wait_for_interruption and isinstance(frame, InterruptionFrame):
@@ -836,11 +852,15 @@ class FrameProcessor(BaseObject):
         self, frame: Frame, direction: FrameDirection, callback: Optional[FrameCallback]
     ):
         try:
+            await self._call_event_handler("on_before_process_frame", frame)
+
             # Process the frame.
             await self.process_frame(frame, direction)
             # If this frame has an associated callback, call it now.
             if callback:
                 await callback(self, frame, direction)
+
+            await self._call_event_handler("on_after_process_frame", frame)
         except Exception as e:
             logger.exception(f"{self}: error processing frame: {e}")
             await self.push_error(ErrorFrame(str(e)))

tests/test_frame_processor.py

@@ -15,11 +15,56 @@ from pipecat.frames.frames import (
     TransportMessageUrgentFrame,
 )
 from pipecat.pipeline.pipeline import Pipeline
+from pipecat.processors.filters.identity_filter import IdentityFilter
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.tests.utils import SleepFrame, run_test
 
 
 class TestFrameProcessor(unittest.IsolatedAsyncioTestCase):
+    async def test_before_after_events(self):
+        identity = IdentityFilter()
+
+        before_process_called = False
+        after_process_called = False
+        before_push_called = False
+        after_push_called = False
+
+        @identity.event_handler("on_before_process_frame")
+        async def on_before_process_frame(filter, frame):
+            nonlocal before_process_called
+            before_process_called = True
+
+        @identity.event_handler("on_after_process_frame")
+        async def on_after_process_frame(filter, frame):
+            nonlocal after_process_called
+            after_process_called = True
+
+        @identity.event_handler("on_before_push_frame")
+        async def on_before_push_frame(filter, frame):
+            nonlocal before_push_called
+            before_push_called = True
+
+        @identity.event_handler("on_after_push_frame")
+        async def on_after_push_frame(filter, frame):
+            nonlocal after_push_called
+            after_push_called = True
+
+        pipeline = Pipeline([identity])
+
+        frames_to_send = [TextFrame(text="Hello cat!")]
+
+        expected_down_frames = [TextFrame]
+
+        await run_test(
+            pipeline,
+            frames_to_send=frames_to_send,
+            expected_down_frames=expected_down_frames,
+        )
+        assert before_process_called
+        assert after_process_called
+        assert before_push_called
+        assert after_push_called
+
     async def test_interruption_and_wait(self):
         class DelayFrameProcessor(FrameProcessor):
             """This processors just gives time to the event loop to change