diff --git a/src/pipecat/pipeline/service_switcher.py b/src/pipecat/pipeline/service_switcher.py index 7dd36c503..b8b1e7c41 100644 --- a/src/pipecat/pipeline/service_switcher.py +++ b/src/pipecat/pipeline/service_switcher.py @@ -6,9 +6,15 @@ """Service switcher for switching between different services at runtime, with different switching strategies.""" +from dataclasses import dataclass from typing import Any, Generic, List, Optional, Type, TypeVar -from pipecat.frames.frames import Frame, ManuallySwitchServiceFrame, ServiceSwitcherFrame +from pipecat.frames.frames import ( + ControlFrame, + Frame, + ManuallySwitchServiceFrame, + ServiceSwitcherFrame, +) from pipecat.pipeline.parallel_pipeline import ParallelPipeline from pipecat.processors.filters.function_filter import FunctionFilter from pipecat.processors.frame_processor import FrameDirection, FrameProcessor @@ -22,19 +28,6 @@ class ServiceSwitcherStrategy: self.services = services self.active_service: Optional[FrameProcessor] = None - def is_active(self, service: FrameProcessor) -> bool: - """Determine if the given service is the currently active one. - - This method should be overridden by subclasses to implement specific logic. - - Args: - service: The service to check. - - Returns: - True if the given service is the active one, False otherwise. - """ - raise NotImplementedError("Subclasses must implement this method.") - def handle_frame(self, frame: ServiceSwitcherFrame, direction: FrameDirection): """Handle a frame that controls service switching. @@ -60,17 +53,6 @@ class ServiceSwitcherStrategyManual(ServiceSwitcherStrategy): super().__init__(services) self.active_service = services[0] if services else None - def is_active(self, service: FrameProcessor) -> bool: - """Check if the given service is the currently active one. - - Args: - service: The service to check. - - Returns: - True if the given service is the active one, False otherwise. - """ - return service == self.active_service - def handle_frame(self, frame: ServiceSwitcherFrame, direction: FrameDirection): """Handle a frame that controls service switching. @@ -108,6 +90,38 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]): self.services = services self.strategy = strategy + class ServiceSwitcherFilter(FunctionFilter): + """An internal filter that allows frames to pass through to the wrapped service only if it's the active service.""" + + def __init__( + self, + wrapped_service: FrameProcessor, + active_service: FrameProcessor, + direction: FrameDirection, + ): + """Initialize the service switcher filter with a strategy and direction.""" + + async def filter(_: Frame) -> bool: + return self.wrapped_service == self.active_service + + super().__init__(filter, direction) + self.wrapped_service = wrapped_service + self.active_service = active_service + + async def process_frame(self, frame, direction): + """Process a frame through the filter, handling special internal filter-updating frames.""" + if isinstance(frame, ServiceSwitcher.ServiceSwitcherFilterFrame): + self.active_service = frame.active_service + return + + await super().process_frame(frame, direction) + + @dataclass + class ServiceSwitcherFilterFrame(ControlFrame): + """An internal frame used by ServiceSwitcher to filter frames based on active service.""" + + active_service: FrameProcessor + @staticmethod def _make_pipeline_definitions( services: List[FrameProcessor], strategy: ServiceSwitcherStrategy @@ -121,14 +135,18 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]): def _make_pipeline_definition( service: FrameProcessor, strategy: ServiceSwitcherStrategy ) -> Any: - async def filter(frame) -> bool: - _ = frame - return strategy.is_active(service) - return [ - FunctionFilter(filter, direction=FrameDirection.DOWNSTREAM), + ServiceSwitcher.ServiceSwitcherFilter( + wrapped_service=service, + active_service=strategy.active_service, + direction=FrameDirection.DOWNSTREAM, + ), service, - FunctionFilter(filter, direction=FrameDirection.UPSTREAM), + ServiceSwitcher.ServiceSwitcherFilter( + wrapped_service=service, + active_service=strategy.active_service, + direction=FrameDirection.UPSTREAM, + ), ] async def process_frame(self, frame: Frame, direction: FrameDirection): @@ -142,3 +160,10 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]): if isinstance(frame, ServiceSwitcherFrame): self.strategy.handle_frame(frame, direction) + service_switcher_filter_frame = ServiceSwitcher.ServiceSwitcherFilterFrame( + 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: + await p.queue_frame(service_switcher_filter_frame, direction)