Rename to filter_system_frames

This commit is contained in:
Mark Backman
2025-10-22 10:59:43 -04:00
parent 5b921fc054
commit ec890a834f
4 changed files with 28 additions and 15 deletions

View File

@@ -9,7 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Changed ### 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. or not SystemFrames are filtered.
- Upgraded `aws_sdk_bedrock_runtime` to v0.1.1 to resolve potential CPU issues - Upgraded `aws_sdk_bedrock_runtime` to v0.1.1 to resolve potential CPU issues

View File

@@ -26,8 +26,9 @@ from pipecat.runner.utils import create_transport
from pipecat.services.cartesia.stt import CartesiaSTTService from pipecat.services.cartesia.stt import CartesiaSTTService
from pipecat.services.cartesia.tts import CartesiaTTSService from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService 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.openai.llm import OpenAILLMService
from pipecat.services.stt_service import STTService
from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams from pipecat.transports.daily.transport import DailyParams
from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams 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 services=[stt_cartesia, stt_deepgram], strategy_type=ServiceSwitcherStrategyManual
) )
tts = CartesiaTTSService( tts_cartesia = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"), 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 = [ messages = [
{ {
@@ -90,8 +99,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
transport.input(), # Transport user input transport.input(), # Transport user input
stt_switcher, stt_switcher,
context_aggregator.user(), # User responses context_aggregator.user(), # User responses
llm, # LLM llm_switcher, # LLM
tts, # TTS tts_switcher, # TTS
transport.output(), # Transport bot output transport.output(), # Transport bot output
context_aggregator.assistant(), # Assistant spoken responses context_aggregator.assistant(), # Assistant spoken responses
] ]
@@ -115,6 +124,12 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
await asyncio.sleep(15) await asyncio.sleep(15)
print(f"Switching to {stt_deepgram}") print(f"Switching to {stt_deepgram}")
await task.queue_frames([ManuallySwitchServiceFrame(service=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") @transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client): async def on_client_disconnected(transport, client):

View File

@@ -144,7 +144,7 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]):
async def filter(_: Frame) -> bool: async def filter(_: Frame) -> bool:
return self._wrapped_service == self._active_service 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): 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."""

View File

@@ -28,7 +28,7 @@ class FunctionFilter(FrameProcessor):
self, self,
filter: Callable[[Frame], Awaitable[bool]], filter: Callable[[Frame], Awaitable[bool]],
direction: FrameDirection = FrameDirection.DOWNSTREAM, direction: FrameDirection = FrameDirection.DOWNSTREAM,
block_system_frames: bool = False, filter_system_frames: bool = False,
): ):
"""Initialize the function filter. """Initialize the function filter.
@@ -37,19 +37,17 @@ class FunctionFilter(FrameProcessor):
frame should pass through, False otherwise. frame should pass through, False otherwise.
direction: The direction to apply filtering. Only frames moving in direction: The direction to apply filtering. Only frames moving in
this direction will be filtered. Defaults to DOWNSTREAM. 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__() super().__init__()
self._filter = filter self._filter = filter
self._direction = direction self._direction = direction
self._block_system_frames = block_system_frames self._filter_system_frames = filter_system_frames
# #
# Frame processor # 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): def _should_passthrough_frame(self, frame, direction):
"""Check if a frame should pass through without filtering.""" """Check if a frame should pass through without filtering."""
# Always passthrough frames in the wrong direction # Always passthrough frames in the wrong direction
@@ -60,8 +58,8 @@ class FunctionFilter(FrameProcessor):
if isinstance(frame, (StartFrame, EndFrame, CancelFrame)): if isinstance(frame, (StartFrame, EndFrame, CancelFrame)):
return True return True
# If not blocking system frames, passthrough all other system frames # If not filtering system frames, passthrough all other system frames
if not self._block_system_frames and isinstance(frame, SystemFrame): if not self._filter_system_frames and isinstance(frame, SystemFrame):
return True return True
return False return False