diff --git a/CHANGELOG.md b/CHANGELOG.md index 121558825..1717449cd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Added `on_pipeline_finished` event to `PipelineTask`. This event will get + fired when the pipeline is done running. This can be the result of a + `StopFrame`, `CancelFrame` or `EndFrame`. + + ```python + @task.event_handler("on_pipeline_finished") + async def on_pipeline_finished(task: PipelineTask, frame: Frame): + ... + ``` + +### Deprecated + +- `PipelineTask` events `on_pipeline_stopped`, `on_pipeline_ended` and + `on_pipeline_cancelled` are now deprecated. Use `on_pipeline_finished` + instead. + ### Fixed - Fixed a `FastAPIWebsocketTransport` and `SmallWebRTCTransport` issue where diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 8eec80ce4..57e4621ce 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -115,9 +115,28 @@ class PipelineTask(BasePipelineTask): - on_frame_reached_downstream: Called when downstream frames reach the sink - on_idle_timeout: Called when pipeline is idle beyond timeout threshold - on_pipeline_started: Called when pipeline starts with StartFrame - - on_pipeline_stopped: Called when pipeline stops with StopFrame - - on_pipeline_ended: Called when pipeline ends with EndFrame - - on_pipeline_cancelled: Called when pipeline is cancelled + - on_pipeline_stopped: [deprecated] Called when pipeline stops with StopFrame + + .. deprecated:: 0.0.86 + Use `on_pipeline_finished` instead. + + - on_pipeline_ended: [deprecated] Called when pipeline ends with EndFrame + + .. deprecated:: 0.0.86 + Use `on_pipeline_finished` instead. + + - on_pipeline_cancelled: [deprecated] Called when pipeline is cancelled with CancelFrame + + .. deprecated:: 0.0.86 + Use `on_pipeline_finished` instead. + + - on_pipeline_finished: Called after the pipeline has reached any terminal state. + This includes: + - StopFrame: pipeline was stopped (processors keep connections open) + - EndFrame: pipeline ended normally + - CancelFrame: pipeline was cancelled + Use this event for cleanup, logging, or post-processing tasks. Users can inspect + the frame if they need to handle specific cases. Example:: @@ -128,6 +147,10 @@ class PipelineTask(BasePipelineTask): @task.event_handler("on_idle_timeout") async def on_pipeline_idle_timeout(task): ... + + @task.event_handler("on_pipeline_finished") + async def on_pipeline_finished(task, frame): + ... """ def __init__( @@ -264,6 +287,7 @@ class PipelineTask(BasePipelineTask): self._register_event_handler("on_pipeline_stopped") self._register_event_handler("on_pipeline_ended") self._register_event_handler("on_pipeline_cancelled") + self._register_event_handler("on_pipeline_finished") @property def params(self) -> PipelineParams: @@ -292,6 +316,27 @@ class PipelineTask(BasePipelineTask): """ return self._turn_trace_observer + def event_handler(self, event_name: str): + """Decorator for registering event handlers. + + Args: + event_name: The name of the event to handle. + + Returns: + The decorator function that registers the handler. + """ + if event_name in ["on_pipeline_stopped", "on_pipeline_ended", "on_pipeline_cancelled"]: + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + f"Event '{event_name}' is deprecated, use 'on_pipeline_finished' instead.", + DeprecationWarning, + ) + + return super().event_handler(event_name) + def add_observer(self, observer: BaseObserver): """Add an observer to monitor pipeline execution. @@ -534,6 +579,7 @@ class PipelineTask(BasePipelineTask): ) finally: await self._call_event_handler("on_pipeline_cancelled", frame) + await self._call_event_handler("on_pipeline_finished", frame) logger.debug(f"{self}: Closing. Waiting for {frame} to reach the end of the pipeline...") @@ -681,9 +727,11 @@ class PipelineTask(BasePipelineTask): self._pipeline_start_event.set() elif isinstance(frame, EndFrame): await self._call_event_handler("on_pipeline_ended", frame) + await self._call_event_handler("on_pipeline_finished", frame) self._pipeline_end_event.set() elif isinstance(frame, StopFrame): await self._call_event_handler("on_pipeline_stopped", frame) + await self._call_event_handler("on_pipeline_finished", frame) self._pipeline_end_event.set() elif isinstance(frame, CancelFrame): self._pipeline_end_event.set() diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index f36c0cf2f..fda6fa63e 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -196,10 +196,10 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase): nonlocal start_received start_received = True - @task.event_handler("on_pipeline_ended") - async def on_pipeline_ended(task, frame: EndFrame): + @task.event_handler("on_pipeline_finished") + async def on_pipeline_finished(task, frame: Frame): nonlocal end_received - end_received = True + end_received = isinstance(frame, EndFrame) await task.queue_frame(EndFrame()) await task.run(PipelineTaskParams(loop=asyncio.get_event_loop())) @@ -214,10 +214,10 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase): pipeline = Pipeline([identity]) task = PipelineTask(pipeline) - @task.event_handler("on_pipeline_stopped") - async def on_pipeline_ended(task, frame: StopFrame): + @task.event_handler("on_pipeline_finished") + async def on_pipeline_finished(task, frame: Frame): nonlocal stop_received - stop_received = True + stop_received = isinstance(frame, StopFrame) await task.queue_frame(StopFrame()) await task.run(PipelineTaskParams(loop=asyncio.get_event_loop())) @@ -441,10 +441,10 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase): async def on_pipeline_started(task: PipelineTask, frame: StartFrame): await task.cancel() - @task.event_handler("on_pipeline_cancelled") - async def on_pipeline_cancelled(task: PipelineTask, frame: CancelFrame): + @task.event_handler("on_pipeline_finished") + async def on_pipeline_finished(task: PipelineTask, frame: Frame): nonlocal cancelled - cancelled = True + cancelled = isinstance(frame, CancelFrame) try: await task.run(PipelineTaskParams(loop=asyncio.get_event_loop()))