diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 012ab9af7..a9b58d986 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -1603,17 +1603,46 @@ class MixerEnableFrame(MixerControlFrame): @dataclass -class ServiceSwitcherFrame(ControlFrame): - """A base class for frames that control ServiceSwitcher behavior.""" +class ManuallySwitchServiceBaseFrame: + """Base class for frames that request a manual service switch from a ServiceSwitcher. + + Must be inherited alongside either ServiceSwitcherFrame or ServiceSwitcherControlFrame. + """ + + service: "FrameProcessor" + + +@dataclass +class ServiceSwitcherFrame(SystemFrame): + """A base class for frames that affect ServiceSwitcher behavior immediately.""" pass @dataclass -class ManuallySwitchServiceFrame(ServiceSwitcherFrame): - """A frame to request a manual switch in the active service in a ServiceSwitcher. +class ManuallySwitchServiceFrame(ServiceSwitcherFrame, ManuallySwitchServiceBaseFrame): + """A frame to request an immediate manual switch in the active service in a ServiceSwitcher. Handled by ServiceSwitcherStrategyManual to switch the active service. """ - service: "FrameProcessor" + pass + + +@dataclass +class ServiceSwitcherControlFrame(ControlFrame): + """A base class for frames that affect ServiceSwitcher behavior, in order.""" + + pass + + +@dataclass +class ManuallySwitchServiceControlFrame( + ServiceSwitcherControlFrame, ManuallySwitchServiceBaseFrame +): + """A frame to request a manual switch in the active service in a ServiceSwitcher, in order. + + Handled by ServiceSwitcherStrategyManual to switch the active service. + """ + + pass diff --git a/src/pipecat/pipeline/service_switcher.py b/src/pipecat/pipeline/service_switcher.py index 8dc7c0362..41efcb974 100644 --- a/src/pipecat/pipeline/service_switcher.py +++ b/src/pipecat/pipeline/service_switcher.py @@ -12,8 +12,10 @@ from typing import Any, Generic, List, Optional, Type, TypeVar from pipecat.frames.frames import ( ControlFrame, Frame, - ManuallySwitchServiceFrame, + ManuallySwitchServiceBaseFrame, + ServiceSwitcherControlFrame, ServiceSwitcherFrame, + SystemFrame, ) from pipecat.pipeline.parallel_pipeline import ParallelPipeline from pipecat.processors.filters.function_filter import FunctionFilter @@ -28,7 +30,9 @@ class ServiceSwitcherStrategy: self.services = services self.active_service: Optional[FrameProcessor] = None - def handle_frame(self, frame: ServiceSwitcherFrame, direction: FrameDirection): + def handle_frame( + self, frame: ServiceSwitcherFrame | ServiceSwitcherControlFrame, direction: FrameDirection + ): """Handle a frame that controls service switching. This method can be overridden by subclasses to implement specific logic @@ -53,14 +57,16 @@ class ServiceSwitcherStrategyManual(ServiceSwitcherStrategy): super().__init__(services) self.active_service = services[0] if services else None - def handle_frame(self, frame: ServiceSwitcherFrame, direction: FrameDirection): + def handle_frame( + self, frame: ServiceSwitcherFrame | ServiceSwitcherControlFrame, direction: FrameDirection + ): """Handle a frame that controls service switching. Args: frame: The frame to handle. direction: The direction of the frame (upstream or downstream). """ - if isinstance(frame, ManuallySwitchServiceFrame): + if isinstance(frame, ManuallySwitchServiceBaseFrame): self._set_active(frame.service) else: raise ValueError(f"Unsupported frame type: {type(frame)}") @@ -110,7 +116,7 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]): async def process_frame(self, frame, direction): """Process a frame through the filter, handling special internal filter-updating frames.""" - if isinstance(frame, ServiceSwitcher.ServiceSwitcherFilterFrame): + if isinstance(frame, ServiceSwitcher.ServiceSwitcherFilterBaseFrame): self.active_service = frame.active_service # Two ServiceSwitcherFilters "sandwich" a service. Push the # frame only to update the other side of the sandwich, but @@ -122,11 +128,23 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]): await super().process_frame(frame, direction) @dataclass - class ServiceSwitcherFilterFrame(ControlFrame): - """An internal frame used by ServiceSwitcher to filter frames based on active service.""" + class ServiceSwitcherFilterBaseFrame: + """Base class for internal frames used by ServiceSwitcher to filter frames based on active service.""" active_service: FrameProcessor + @dataclass + class ServiceSwitcherFilterFrame(SystemFrame, ServiceSwitcherFilterBaseFrame): + """An internal frame used by ServiceSwitcher to filter frames based on active service.""" + + pass + + @dataclass + class ServiceSwitcherFilterControlFrame(ControlFrame, ServiceSwitcherFilterBaseFrame): + """An internal control frame used by ServiceSwitcher to filter frames based on active service.""" + + pass + @staticmethod def _make_pipeline_definitions( services: List[FrameProcessor], strategy: ServiceSwitcherStrategy @@ -163,11 +181,18 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]): """ await super().process_frame(frame, direction) - if isinstance(frame, ServiceSwitcherFrame): + if isinstance(frame, (ServiceSwitcherFrame, ServiceSwitcherControlFrame)): self.strategy.handle_frame(frame, direction) - service_switcher_filter_frame = ServiceSwitcher.ServiceSwitcherFilterFrame( - active_service=self.strategy.active_service - ) + if isinstance(frame, ServiceSwitcherFrame): + # Apply immediate update (system frame) + service_switcher_filter_frame = ServiceSwitcher.ServiceSwitcherFilterFrame( + active_service=self.strategy.active_service + ) + else: + # Apply in-order update (control frame) + service_switcher_filter_frame = ServiceSwitcher.ServiceSwitcherFilterControlFrame( + active_service=self.strategy.active_service + ) # Queue frame that updates filters with new active service # (Hack: we need access ParallelPipeline internals here to queue the frame in each of the pipelines) for p in self._pipelines: diff --git a/tests/test_service_switcher.py b/tests/test_service_switcher.py index 9ee01ea57..b8a508c82 100644 --- a/tests/test_service_switcher.py +++ b/tests/test_service_switcher.py @@ -9,7 +9,12 @@ import asyncio import unittest -from pipecat.frames.frames import Frame, ManuallySwitchServiceFrame, TextFrame +from pipecat.frames.frames import ( + Frame, + ManuallySwitchServiceControlFrame, + ManuallySwitchServiceFrame, + TextFrame, +) from pipecat.pipeline.service_switcher import ServiceSwitcher, ServiceSwitcherStrategyManual from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.tests.utils import run_test @@ -72,7 +77,7 @@ class TestServiceSwitcherStrategyManual(unittest.IsolatedAsyncioTestCase): self.assertEqual(strategy.services, []) self.assertIsNone(strategy.active_service) - def test_handle_frame_manually_switch_service(self): + def test_handle_manually_switch_service_frame(self): """Test manual service switching with ManuallySwitchServiceFrame.""" strategy = ServiceSwitcherStrategyManual(self.services) @@ -96,6 +101,30 @@ class TestServiceSwitcherStrategyManual(unittest.IsolatedAsyncioTestCase): self.assertNotEqual(strategy.active_service, self.service2) self.assertEqual(strategy.active_service, self.service3) + def test_handle_manually_switch_service_control_frame(self): + """Test manual service switching with ManuallySwitchServiceControlFrame.""" + strategy = ServiceSwitcherStrategyManual(self.services) + + # Initially service1 should be active + self.assertEqual(strategy.active_service, self.service1) + self.assertNotEqual(strategy.active_service, self.service2) + + # Switch to service2 + switch_frame = ManuallySwitchServiceControlFrame(service=self.service2) + strategy.handle_frame(switch_frame, FrameDirection.DOWNSTREAM) + + self.assertNotEqual(strategy.active_service, self.service1) + self.assertEqual(strategy.active_service, self.service2) + self.assertNotEqual(strategy.active_service, self.service3) + + # Switch to service3 + switch_frame = ManuallySwitchServiceControlFrame(service=self.service3) + strategy.handle_frame(switch_frame, FrameDirection.DOWNSTREAM) + + self.assertNotEqual(strategy.active_service, self.service1) + self.assertNotEqual(strategy.active_service, self.service2) + self.assertEqual(strategy.active_service, self.service3) + def test_handle_frame_invalid_service(self): """Test that switching to an invalid service raises an error.""" strategy = ServiceSwitcherStrategyManual(self.services) @@ -178,8 +207,8 @@ class TestServiceSwitcher(unittest.IsolatedAsyncioTestCase): for i, frame in enumerate(text_frames): self.assertEqual(frame.text, f"Hello {i + 1}") - async def test_service_switching(self): - """Test that after service switching the new active service receives frames while others don't.""" + async def test_service_switching_in_order(self): + """Test that after service switching using ManuallySwitchServiceControlFrame, the new active service receives frames while others don't.""" switcher = ServiceSwitcher(self.services, ServiceSwitcherStrategyManual) # Reset counters @@ -191,10 +220,10 @@ class TestServiceSwitcher(unittest.IsolatedAsyncioTestCase): switcher, frames_to_send=[ TextFrame("Frame for service1"), - ManuallySwitchServiceFrame(service=self.service2), + ManuallySwitchServiceControlFrame(service=self.service2), TextFrame("Frame for service2"), ], - expected_down_frames=[TextFrame, ManuallySwitchServiceFrame, TextFrame], + expected_down_frames=[TextFrame, ManuallySwitchServiceControlFrame, TextFrame], ) # Verify service2 received the frame @@ -214,6 +243,43 @@ class TestServiceSwitcher(unittest.IsolatedAsyncioTestCase): self.assertEqual(service1_text_frames[0].text, "Frame for service1") self.assertEqual(service2_text_frames[0].text, "Frame for service2") + async def test_service_switching_immediate(self): + """Test that after service switching using ManuallySwitchServiceFrame, the new active service receives frames while others don't.""" + switcher = ServiceSwitcher(self.services, ServiceSwitcherStrategyManual) + + # Reset counters + for service in self.services: + service.reset_counters() + + # Send a test frame, a switch frame, and another test frame + await run_test( + switcher, + frames_to_send=[ + # Out of order on purpose - ManuallySwitchServiceFrame should jump the queue + TextFrame("Frame 1 for service2"), + ManuallySwitchServiceFrame(service=self.service2), + TextFrame("Frame 2 for service2"), + ], + expected_down_frames=[ManuallySwitchServiceFrame, TextFrame, TextFrame], + ) + + # Verify service2 received the frame + service1_text_frames = [ + f for f in self.service1.processed_frames if isinstance(f, TextFrame) + ] + service2_text_frames = [ + f for f in self.service2.processed_frames if isinstance(f, TextFrame) + ] + service3_text_frames = [ + f for f in self.service3.processed_frames if isinstance(f, TextFrame) + ] + + self.assertEqual(len(service1_text_frames), 0) + self.assertEqual(len(service2_text_frames), 2) + self.assertEqual(len(service3_text_frames), 0) + self.assertEqual(service2_text_frames[0].text, "Frame 1 for service2") + self.assertEqual(service2_text_frames[1].text, "Frame 2 for service2") + if __name__ == "__main__": unittest.main()