diff --git a/CHANGELOG.md b/CHANGELOG.md index a07832525..3cbf4a57e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed -- `FunctionFilter` now has a `block_system_frames` arg, which controls whether +- `FunctionFilter` now has a `filter_system_frames` arg, which controls whether or not SystemFrames are filtered. - Upgraded `aws_sdk_bedrock_runtime` to v0.1.1 to resolve potential CPU issues diff --git a/examples/foundational/48-service-switcher.py b/examples/foundational/48-service-switcher.py index e003db7eb..d0e15d2d3 100644 --- a/examples/foundational/48-service-switcher.py +++ b/examples/foundational/48-service-switcher.py @@ -26,8 +26,9 @@ from pipecat.runner.utils import create_transport from pipecat.services.cartesia.stt import CartesiaSTTService from pipecat.services.cartesia.tts import CartesiaTTSService from pipecat.services.deepgram.stt import DeepgramSTTService +from pipecat.services.deepgram.tts import DeepgramTTSService +from pipecat.services.google.llm import GoogleLLMService from pipecat.services.openai.llm import OpenAILLMService -from pipecat.services.stt_service import STTService from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.daily.transport import DailyParams from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams @@ -68,12 +69,20 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): services=[stt_cartesia, stt_deepgram], strategy_type=ServiceSwitcherStrategyManual ) - tts = CartesiaTTSService( + tts_cartesia = CartesiaTTSService( api_key=os.getenv("CARTESIA_API_KEY"), - voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady + voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", + ) + tts_deepgram = DeepgramTTSService(api_key=os.getenv("DEEPGRAM_API_KEY")) + tts_switcher = ServiceSwitcher( + services=[tts_cartesia, tts_deepgram], strategy_type=ServiceSwitcherStrategyManual ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) + llm_openai = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) + llm_google = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY")) + llm_switcher = ServiceSwitcher( + services=[llm_openai, llm_google], strategy_type=ServiceSwitcherStrategyManual + ) messages = [ { @@ -90,8 +99,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): transport.input(), # Transport user input stt_switcher, context_aggregator.user(), # User responses - llm, # LLM - tts, # TTS + llm_switcher, # LLM + tts_switcher, # TTS transport.output(), # Transport bot output context_aggregator.assistant(), # Assistant spoken responses ] @@ -115,6 +124,12 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): await asyncio.sleep(15) print(f"Switching to {stt_deepgram}") await task.queue_frames([ManuallySwitchServiceFrame(service=stt_deepgram)]) + await asyncio.sleep(15) + print(f"Switching to {llm_google}") + await task.queue_frames([ManuallySwitchServiceFrame(service=llm_google)]) + await asyncio.sleep(15) + print(f"Switching to {tts_deepgram}") + await task.queue_frames([ManuallySwitchServiceFrame(service=tts_deepgram)]) @transport.event_handler("on_client_disconnected") async def on_client_disconnected(transport, client): diff --git a/src/pipecat/pipeline/service_switcher.py b/src/pipecat/pipeline/service_switcher.py index 3cfcd0fd8..8895d663c 100644 --- a/src/pipecat/pipeline/service_switcher.py +++ b/src/pipecat/pipeline/service_switcher.py @@ -144,7 +144,7 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]): async def filter(_: Frame) -> bool: return self._wrapped_service == self._active_service - super().__init__(filter, direction, block_system_frames=True) + super().__init__(filter, direction, filter_system_frames=True) async def process_frame(self, frame, direction): """Process a frame through the filter, handling special internal filter-updating frames.""" diff --git a/src/pipecat/processors/filters/function_filter.py b/src/pipecat/processors/filters/function_filter.py index 5bc6e1eb9..556f2bc87 100644 --- a/src/pipecat/processors/filters/function_filter.py +++ b/src/pipecat/processors/filters/function_filter.py @@ -28,7 +28,7 @@ class FunctionFilter(FrameProcessor): self, filter: Callable[[Frame], Awaitable[bool]], direction: FrameDirection = FrameDirection.DOWNSTREAM, - block_system_frames: bool = False, + filter_system_frames: bool = False, ): """Initialize the function filter. @@ -37,19 +37,17 @@ class FunctionFilter(FrameProcessor): frame should pass through, False otherwise. direction: The direction to apply filtering. Only frames moving in this direction will be filtered. Defaults to DOWNSTREAM. - block_system_frames: Whether to block system frames. Defaults to False. + filter_system_frames: Whether to filter system frames. Defaults to False. """ super().__init__() self._filter = filter self._direction = direction - self._block_system_frames = block_system_frames + self._filter_system_frames = filter_system_frames # # Frame processor # - # Ignore system frames, end frames and frames that are not following the - # direction of this gate def _should_passthrough_frame(self, frame, direction): """Check if a frame should pass through without filtering.""" # Always passthrough frames in the wrong direction @@ -60,8 +58,8 @@ class FunctionFilter(FrameProcessor): if isinstance(frame, (StartFrame, EndFrame, CancelFrame)): return True - # If not blocking system frames, passthrough all other system frames - if not self._block_system_frames and isinstance(frame, SystemFrame): + # If not filtering system frames, passthrough all other system frames + if not self._filter_system_frames and isinstance(frame, SystemFrame): return True return False