diff --git a/CHANGELOG.md b/CHANGELOG.md index 5bbbe624a..0edd55d07 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,19 @@ All notable changes to **Pipecat** will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +### Added + +- Added `on_pipeline_error` event to `PipelineTask`. This event will get fired + when an `ErrorFrame` is pushed (use `FrameProcessor.push_error()`). + + ```python + @task.event_handler("on_pipeline_error") + async def on_pipeline_error(task: PipelineTask, frame: ErrorFrame): + ... + ``` + ## [0.0.89] - 2025-10-07 ### Fixed diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 9ce3baf7f..f6d17262e 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -138,6 +138,8 @@ class PipelineTask(BasePipelineTask): Use this event for cleanup, logging, or post-processing tasks. Users can inspect the frame if they need to handle specific cases. + - on_pipeline_error: Called when an error occurs with ErrorFrame + Example:: @task.event_handler("on_frame_reached_upstream") @@ -148,9 +150,17 @@ class PipelineTask(BasePipelineTask): async def on_pipeline_idle_timeout(task): ... + @task.event_handler("on_pipeline_started") + async def on_pipeline_started(task, frame): + ... + @task.event_handler("on_pipeline_finished") async def on_pipeline_finished(task, frame): ... + + @task.event_handler("on_pipeline_error") + async def on_pipeline_error(task, frame): + ... """ def __init__( @@ -288,6 +298,7 @@ class PipelineTask(BasePipelineTask): self._register_event_handler("on_pipeline_ended") self._register_event_handler("on_pipeline_cancelled") self._register_event_handler("on_pipeline_finished") + self._register_event_handler("on_pipeline_error") @property def params(self) -> PipelineParams: @@ -694,12 +705,11 @@ class PipelineTask(BasePipelineTask): logger.debug(f"{self}: received interruption task frame {frame}") await self._pipeline.queue_frame(InterruptionFrame()) elif isinstance(frame, ErrorFrame): + await self._call_event_handler("on_pipeline_error", frame) if frame.fatal: logger.error(f"A fatal error occurred: {frame}") # Cancel all tasks downstream. await self.queue_frame(CancelFrame()) - # Tell the task we should stop. - await self.queue_frame(StopTaskFrame()) else: logger.warning(f"{self}: Something went wrong: {frame}") diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index fda6fa63e..df4af49b9 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -11,6 +11,7 @@ import unittest from pipecat.frames.frames import ( CancelFrame, EndFrame, + ErrorFrame, Frame, HeartbeatFrame, InputAudioRawFrame, @@ -450,3 +451,34 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase): await task.run(PipelineTaskParams(loop=asyncio.get_event_loop())) except asyncio.CancelledError: assert cancelled + + async def test_task_error(self): + class ErrorProcessor(FrameProcessor): + def __init__(self, **kwargs): + super().__init__(**kwargs) + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if isinstance(frame, TextFrame): + await self.push_error(ErrorFrame("Boo!")) + + await self.push_frame(frame, direction) + + error_received = False + + pipeline = Pipeline([ErrorProcessor()]) + task = PipelineTask(pipeline) + + @task.event_handler("on_pipeline_error") + async def on_pipeline_error(task: PipelineTask, frame: ErrorFrame): + nonlocal error_received + error_received = True + await task.cancel() + + await task.queue_frame(TextFrame(text="Hello from Pipecat!")) + + try: + await task.run(PipelineTaskParams(loop=asyncio.get_event_loop())) + except asyncio.CancelledError: + assert error_received