wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Simplify, undoing the change allowing controlling ServiceSwitcher with immediate frames (SystemFrames). Service switcher frames are ControlFrames, which are easier to reason about. We can always build the immediate option later if needed (i.e. if there's sufficient user pull for it)

Browse Source
This commit is contained in:
Paul Kompfner
2025-09-16 14:32:03 -04:00
parent b814b70e1e
commit a12392182c
3 changed files with 30 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
src/pipecat/frames/frames.py
View File

@@ -1603,46 +1603,17 @@ class MixerEnableFrame(MixerControlFrame):
@dataclass @dataclass
class ManuallySwitchServiceBaseFrame: class ServiceSwitcherFrame(ControlFrame):
"""Base class for frames that request a manual service switch from a ServiceSwitcher. """A base class for frames that affect ServiceSwitcher behavior."""
Must be inherited alongside either ServiceSwitcherFrame or ServiceSwitcherControlFrame. pass
@dataclass
class ManuallySwitchServiceFrame(ServiceSwitcherFrame):
"""A frame to request a manual switch in the active service in a ServiceSwitcher.
Handled by ServiceSwitcherStrategyManual to switch the active service.
""" """
service: "FrameProcessor" service: "FrameProcessor"
@dataclass
class ServiceSwitcherFrame(SystemFrame):
"""A base class for frames that affect ServiceSwitcher behavior immediately."""
pass
@dataclass
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.
"""
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

50
src/pipecat/pipeline/service_switcher.py
View File

@@ -12,10 +12,8 @@ from typing import Any, Generic, List, Optional, Type, TypeVar
from pipecat.frames.frames import ( from pipecat.frames.frames import (
ControlFrame, ControlFrame,
Frame, Frame,
ManuallySwitchServiceBaseFrame, ManuallySwitchServiceFrame,
ServiceSwitcherControlFrame,
ServiceSwitcherFrame, ServiceSwitcherFrame,
SystemFrame,
) )
from pipecat.pipeline.parallel_pipeline import ParallelPipeline from pipecat.pipeline.parallel_pipeline import ParallelPipeline
from pipecat.processors.filters.function_filter import FunctionFilter from pipecat.processors.filters.function_filter import FunctionFilter
@@ -30,9 +28,7 @@ class ServiceSwitcherStrategy:
self.services = services self.services = services
self.active_service: Optional[FrameProcessor] = None self.active_service: Optional[FrameProcessor] = None
def handle_frame( def handle_frame(self, frame: ServiceSwitcherFrame, direction: FrameDirection):
self, frame: ServiceSwitcherFrame | ServiceSwitcherControlFrame, direction: FrameDirection
):
"""Handle a frame that controls service switching. """Handle a frame that controls service switching.
This method can be overridden by subclasses to implement specific logic This method can be overridden by subclasses to implement specific logic
@@ -57,16 +53,14 @@ class ServiceSwitcherStrategyManual(ServiceSwitcherStrategy):
super().__init__(services) super().__init__(services)
self.active_service = services[0] if services else None self.active_service = services[0] if services else None
def handle_frame( def handle_frame(self, frame: ServiceSwitcherFrame, direction: FrameDirection):
self, frame: ServiceSwitcherFrame | ServiceSwitcherControlFrame, direction: FrameDirection
):
"""Handle a frame that controls service switching. """Handle a frame that controls service switching.
Args: Args:
frame: The frame to handle. frame: The frame to handle.
direction: The direction of the frame (upstream or downstream). direction: The direction of the frame (upstream or downstream).
""" """
if isinstance(frame, ManuallySwitchServiceBaseFrame): if isinstance(frame, ManuallySwitchServiceFrame):
self._set_active(frame.service) self._set_active(frame.service)
else: else:
raise ValueError(f"Unsupported frame type: {type(frame)}") raise ValueError(f"Unsupported frame type: {type(frame)}")
@@ -116,7 +110,7 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]):
async def process_frame(self, frame, direction): async def process_frame(self, frame, direction):
"""Process a frame through the filter, handling special internal filter-updating frames.""" """Process a frame through the filter, handling special internal filter-updating frames."""
if isinstance(frame, ServiceSwitcher.ServiceSwitcherFilterBaseFrame): if isinstance(frame, ServiceSwitcher.ServiceSwitcherFilterFrame):
self.active_service = frame.active_service self.active_service = frame.active_service
# Two ServiceSwitcherFilters "sandwich" a service. Push the # Two ServiceSwitcherFilters "sandwich" a service. Push the
# frame only to update the other side of the sandwich, but # frame only to update the other side of the sandwich, but
@@ -128,22 +122,10 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]):
await super().process_frame(frame, direction) await super().process_frame(frame, direction)
@dataclass @dataclass
class ServiceSwitcherFilterBaseFrame: class ServiceSwitcherFilterFrame(ControlFrame):
"""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.""" """An internal frame used by ServiceSwitcher to filter frames based on active service."""
pass active_service: FrameProcessor
@dataclass
class ServiceSwitcherFilterControlFrame(ControlFrame, ServiceSwitcherFilterBaseFrame):
"""An internal control frame used by ServiceSwitcher to filter frames based on active service."""
pass
@staticmethod @staticmethod
def _make_pipeline_definitions( def _make_pipeline_definitions(
@@ -181,19 +163,11 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]):
""" """
await super().process_frame(frame, direction) await super().process_frame(frame, direction)
if isinstance(frame, (ServiceSwitcherFrame, ServiceSwitcherControlFrame)): if isinstance(frame, ServiceSwitcherFrame):
self.strategy.handle_frame(frame, direction) self.strategy.handle_frame(frame, direction)
if isinstance(frame, ServiceSwitcherFrame): service_switcher_filter_frame = ServiceSwitcher.ServiceSwitcherFilterFrame(
# Apply immediate update (system frame) active_service=self.strategy.active_service
service_switcher_filter_frame = ServiceSwitcher.ServiceSwitcherFilterFrame( )
active_service=self.strategy.active_service # Hack: we need access ParallelPipeline internals here to queue the frame in each of the pipelines
)
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: for p in self._pipelines:
await p.queue_frame(service_switcher_filter_frame, direction) await p.queue_frame(service_switcher_filter_frame, direction)

78
tests/test_service_switcher.py
View File

@@ -6,12 +6,10 @@
"""Unit tests for ServiceSwitcher and related components.""" """Unit tests for ServiceSwitcher and related components."""
import asyncio
import unittest import unittest
from pipecat.frames.frames import ( from pipecat.frames.frames import (
Frame, Frame,
ManuallySwitchServiceControlFrame,
ManuallySwitchServiceFrame, ManuallySwitchServiceFrame,
TextFrame, TextFrame,
) )
@@ -101,30 +99,6 @@ class TestServiceSwitcherStrategyManual(unittest.IsolatedAsyncioTestCase):
self.assertNotEqual(strategy.active_service, self.service2) self.assertNotEqual(strategy.active_service, self.service2)
self.assertEqual(strategy.active_service, self.service3) 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): def test_handle_frame_invalid_service(self):
"""Test that switching to an invalid service raises an error.""" """Test that switching to an invalid service raises an error."""
strategy = ServiceSwitcherStrategyManual(self.services) strategy = ServiceSwitcherStrategyManual(self.services)
@@ -207,8 +181,8 @@ class TestServiceSwitcher(unittest.IsolatedAsyncioTestCase):
for i, frame in enumerate(text_frames): for i, frame in enumerate(text_frames):
self.assertEqual(frame.text, f"Hello {i + 1}") self.assertEqual(frame.text, f"Hello {i + 1}")
async def test_service_switching_in_order(self): async def test_service_switching(self):
"""Test that after service switching using ManuallySwitchServiceControlFrame, the new active service receives frames while others don't.""" """Test that after service switching using ManuallySwitchServiceFrame, the new active service receives frames while others don't."""
switcher = ServiceSwitcher(self.services, ServiceSwitcherStrategyManual) switcher = ServiceSwitcher(self.services, ServiceSwitcherStrategyManual)
# Reset counters # Reset counters
@@ -219,11 +193,11 @@ class TestServiceSwitcher(unittest.IsolatedAsyncioTestCase):
await run_test( await run_test(
switcher, switcher,
frames_to_send=[ frames_to_send=[
TextFrame("Frame for service1"), TextFrame("Hello 1"),
ManuallySwitchServiceControlFrame(service=self.service2), ManuallySwitchServiceFrame(service=self.service2),
TextFrame("Frame for service2"), TextFrame("Hello 2"),
], ],
expected_down_frames=[TextFrame, ManuallySwitchServiceControlFrame, TextFrame], expected_down_frames=[TextFrame, ManuallySwitchServiceFrame, TextFrame],
) )
# Verify service2 received the frame # Verify service2 received the frame
@@ -240,45 +214,9 @@ class TestServiceSwitcher(unittest.IsolatedAsyncioTestCase):
self.assertEqual(len(service1_text_frames), 1) self.assertEqual(len(service1_text_frames), 1)
self.assertEqual(len(service2_text_frames), 1) self.assertEqual(len(service2_text_frames), 1)
self.assertEqual(len(service3_text_frames), 0) self.assertEqual(len(service3_text_frames), 0)
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): self.assertEqual(service1_text_frames[0].text, "Hello 1")
"""Test that after service switching using ManuallySwitchServiceFrame, the new active service receives frames while others don't.""" self.assertEqual(service2_text_frames[0].text, "Hello 2")
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__": if __name__ == "__main__":