diff --git a/CHANGELOG.md b/CHANGELOG.md index fc9502767..f596c3462 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added an asyncio event `finished_event` field to `InterruptionFrame`. When + assigned, the asyncio event will be set when the frame reaches the end of the + pipeline. You can use this field to know when an interruption made it all the + way to the end of a pipeline. The field has been also added to + `InterruptionTaskFrame`. + - Added support for loading external observers. You can now register custom pipeline observers by setting the `PIPECAT_OBSERVER_FILES` environment variable. This variable should contain a colon-separated list of Python files @@ -28,8 +34,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `CancelFrame` and `CancelTaskFrame` have an optional `reason` field to indicate why the pipeline is being canceled. This can be also specified when - you cancel a task with `PipelineTask.cancel(reason="cancellation your -reason")`. + you cancel a task with `PipelineTask.cancel(reason="cancellation reason")`. - Added `include_prob_metrics` parameter to Whisper STT services to enable access to probability metrics from transcription results. diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 53b8dba0c..791de0f34 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -862,7 +862,7 @@ class InterruptionFrame(SystemFrame): with other frames (so the order is not guaranteed). Parameters: - wait_event: If not None, the event will be set when the frame + finished_event: If not None, the event will be set when the frame reaches the end of the pipeline. """ diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 90976b52c..ad51db0ca 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -745,7 +745,7 @@ class PipelineTask(BasePipelineTask): # pipeline. This is in case the push task is blocked waiting for a # pipeline-ending frame to finish traversing the pipeline. logger.debug(f"{self}: received interruption task frame {frame}") - await self._pipeline.queue_frame(InterruptionFrame()) + await self._pipeline.queue_frame(InterruptionFrame(finished_event=frame.finished_event)) elif isinstance(frame, ErrorFrame): await self._call_event_handler("on_pipeline_error", frame) if frame.fatal: @@ -786,6 +786,10 @@ class PipelineTask(BasePipelineTask): self._pipeline_end_event.set() elif isinstance(frame, HeartbeatFrame): await self._heartbeat_queue.put(frame) + elif isinstance(frame, InterruptionFrame) and frame.finished_event: + # This should unblock any code waiting for the interruption to + # complete. + frame.finished_event.set() async def _heartbeat_push_handler(self): """Push heartbeat frames at regular intervals."""