From de51637b7755b8b788afb7afc58b955fc14bc59d Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Fri, 12 Sep 2025 13:40:19 -0400 Subject: [PATCH] Update `ServiceSwitcher` so that `ServiceSwitcherFrame`s (which might update the currently active service) are processed and have an effect at the expected time. We should be able to, for example, queue: - A text frame - A `ManuallySwitchServiceFrame` (which is a `ServiceSwitcherFrame`) - Another text frame And expect that the first text frame be handled by the initially active service and the second text frame be handled by the newly active one. Previously, the `ManuallySwitchServiceFrame` would have an effect too early, causing both text frames to be handled by the newly active service. Why? Because the frame filtering condition was being updated *directly* by the `ServiceSwitcher`, which is upstream from the services it's switching between. It could therefore update the filters *before* the services received the prior frames. --- src/pipecat/pipeline/service_switcher.py | 87 +++++++++++++++--------- 1 file changed, 56 insertions(+), 31 deletions(-) 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)