diff --git a/CHANGELOG.md b/CHANGELOG.md index 3d585698b..b12667248 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -27,6 +27,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Fixed an issue that would cause wrong user/assistant context ordering when + using interruption strategies. + - Fixed RTVI incoming message handling, broken in 0.0.87. ## [0.0.87] - 2025-10-02 diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 2a9651b83..a871fe74b 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -849,13 +849,15 @@ class FrameProcessorResumeUrgentFrame(SystemFrame): class InterruptionFrame(SystemFrame): """Frame indicating user started speaking (interruption detected). - Emitted by the BaseInputTransport to indicate that a user has started - speaking (i.e. is interrupting). This is similar to - UserStartedSpeakingFrame except that it should be pushed concurrently - with other frames (so the order is not guaranteed). + Usually emitted by the pipeline task to indicate that all frame processors + should be interrrupted. + + Parameters: + pushed_by_task: Whether this interruption was pushed from the pipeline + task. """ - pass + pushed_by_task: bool = False @dataclass @@ -886,6 +888,17 @@ class StartInterruptionFrame(InterruptionFrame): ) +@dataclass +class InterruptionCompletedFrame(SystemFrame): + """Frame indicating that the whole pipeline has been interrupted. + + This is emitted by the pipeline task when an InterruptionFrame has made it + all the way to the end of pipeline, interrupting all the frame processors. + """ + + pass + + @dataclass class UserStartedSpeakingFrame(SystemFrame): """Frame indicating user has started speaking. diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 9ce3baf7f..d040beac1 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -30,6 +30,7 @@ from pipecat.frames.frames import ( ErrorFrame, Frame, HeartbeatFrame, + InterruptionCompletedFrame, InterruptionFrame, InterruptionTaskFrame, MetricsFrame, @@ -691,8 +692,11 @@ class PipelineTask(BasePipelineTask): # bypassing the push queue and directly queue into the # 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()) + interruption_frame = InterruptionFrame(pushed_by_task=True) + logger.debug( + f"{self}: received interruption task frame {frame}, pushing {interruption_frame}" + ) + await self._pipeline.queue_frame(interruption_frame) elif isinstance(frame, ErrorFrame): if frame.fatal: logger.error(f"A fatal error occurred: {frame}") @@ -736,6 +740,14 @@ class PipelineTask(BasePipelineTask): self._pipeline_end_event.set() elif isinstance(frame, CancelFrame): self._pipeline_end_event.set() + elif isinstance(frame, InterruptionFrame) and frame.pushed_by_task: + # If an interruption frame made it all the way to the end of the + # pipeline, send an InterruptionCompleteFrame. Note that we are + # bypassing the push queue and directly queue into the + # 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}: interruption completed with {frame}") + await self._pipeline.queue_frame(InterruptionCompletedFrame()) elif isinstance(frame, HeartbeatFrame): await self._heartbeat_queue.put(frame) diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 0b55082aa..25dc2716e 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -28,6 +28,7 @@ from pipecat.frames.frames import ( FrameProcessorPauseUrgentFrame, FrameProcessorResumeFrame, FrameProcessorResumeUrgentFrame, + InterruptionCompletedFrame, InterruptionFrame, InterruptionTaskFrame, StartFrame, @@ -571,7 +572,9 @@ class FrameProcessor(BaseObject): # frames and we will process the frame right away. This is because a # previous system frame might be waiting for the interruption frame and # it's blocking the input task. - if self._wait_for_interruption and isinstance(frame, InterruptionFrame): + if self._wait_for_interruption and isinstance( + frame, (InterruptionFrame, InterruptionCompletedFrame) + ): await self.__process_frame(frame, direction, callback) return @@ -661,18 +664,17 @@ class FrameProcessor(BaseObject): 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): + # If we are waiting for an interruption and one completed, then we can + # unblock `push_interruption_task_frame_and_wait()`. + if self._wait_for_interruption and isinstance(frame, InterruptionCompletedFrame): self._wait_interruption_event.set() async def push_interruption_task_frame_and_wait(self): """Push an interruption task frame upstream and wait for the interruption. This function sends an `InterruptionTaskFrame` upstream to the pipeline - task and waits to receive the corresponding `InterruptionFrame`. When - the function finishes it is guaranteed that the `InterruptionFrame` has - been pushed downstream. + task and waits to receive an `InterruptionCompletedFrame`. This + guarantees the whole pipeline has been interrupted. """ self._wait_for_interruption = True diff --git a/tests/test_context_aggregators.py b/tests/test_context_aggregators.py index 77b6acc87..accbfd669 100644 --- a/tests/test_context_aggregators.py +++ b/tests/test_context_aggregators.py @@ -5,6 +5,7 @@ # import json +import sys import unittest from typing import Any @@ -20,6 +21,7 @@ from pipecat.frames.frames import ( FunctionCallResultFrame, FunctionCallResultProperties, InterimTranscriptionFrame, + InterruptionCompletedFrame, InterruptionFrame, InterruptionTaskFrame, LLMFullResponseEndFrame, @@ -568,6 +570,7 @@ class BaseTestUserContextAggregator: BotStartedSpeakingFrame, UserStartedSpeakingFrame, InterruptionFrame, + InterruptionCompletedFrame, UserStoppedSpeakingFrame, *self.EXPECTED_CONTEXT_FRAMES, ] diff --git a/tests/test_dtmf_aggregator.py b/tests/test_dtmf_aggregator.py index c7590ae47..18082a992 100644 --- a/tests/test_dtmf_aggregator.py +++ b/tests/test_dtmf_aggregator.py @@ -10,6 +10,7 @@ from pipecat.audio.dtmf.types import KeypadEntry from pipecat.frames.frames import ( EndFrame, InputDTMFFrame, + InterruptionCompletedFrame, InterruptionFrame, TranscriptionFrame, ) @@ -30,6 +31,7 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase): expected_down_frames = [ InputDTMFFrame, InterruptionFrame, + InterruptionCompletedFrame, InputDTMFFrame, InputDTMFFrame, InputDTMFFrame, @@ -62,10 +64,12 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase): expected_down_frames = [ InputDTMFFrame, InterruptionFrame, + InterruptionCompletedFrame, InputDTMFFrame, TranscriptionFrame, # First aggregation "12" InputDTMFFrame, InterruptionFrame, + InterruptionCompletedFrame, TranscriptionFrame, # Second aggregation "3" ] @@ -98,11 +102,13 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase): expected_down_frames = [ InputDTMFFrame, InterruptionFrame, + InterruptionCompletedFrame, InputDTMFFrame, InputDTMFFrame, TranscriptionFrame, # "12#" InputDTMFFrame, InterruptionFrame, + InterruptionCompletedFrame, InputDTMFFrame, TranscriptionFrame, # "45" ] @@ -132,6 +138,7 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase): expected_down_frames = [ InputDTMFFrame, InterruptionFrame, + InterruptionCompletedFrame, InputDTMFFrame, TranscriptionFrame, # Should flush before EndFrame EndFrame, @@ -160,6 +167,7 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase): expected_down_frames = [ InputDTMFFrame, InterruptionFrame, + InterruptionCompletedFrame, InputDTMFFrame, TranscriptionFrame, ] @@ -187,6 +195,7 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase): expected_down_frames = [ InputDTMFFrame, InterruptionFrame, + InterruptionCompletedFrame, InputDTMFFrame, InputDTMFFrame, TranscriptionFrame, @@ -224,7 +233,7 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase): # All the InputDTMFFrames plus one TranscriptionFrame expected_down_frames = ( - [InputDTMFFrame, InterruptionFrame] + [InputDTMFFrame, InterruptionFrame, InterruptionCompletedFrame] + [InputDTMFFrame] * (len(frames_to_send) - 1) + [TranscriptionFrame] ) diff --git a/tests/test_frame_processor.py b/tests/test_frame_processor.py index d0072e5fb..5c1ee1b38 100644 --- a/tests/test_frame_processor.py +++ b/tests/test_frame_processor.py @@ -10,6 +10,7 @@ import unittest from pipecat.frames.frames import ( EndFrame, Frame, + InterruptionCompletedFrame, InterruptionFrame, OutputTransportMessageUrgentFrame, TextFrame, @@ -101,6 +102,7 @@ class TestFrameProcessor(unittest.IsolatedAsyncioTestCase): expected_down_frames = [ InterruptionFrame, InterruptionFrame, + InterruptionCompletedFrame, OutputTransportMessageUrgentFrame, EndFrame, ]