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

STTMuteFilter: use FunctionCallsStartedFrame and support multiple function calls

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2025-12-27 13:47:58 -08:00
parent 43fc26cf0e
commit a962c4eeba
3 changed files with 61 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

97
tests/test_stt_mute_filter.py
View File

@@ -9,8 +9,10 @@ import unittest
from pipecat.frames.frames import (
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
FunctionCallFromLLM,
FunctionCallInProgressFrame,
FunctionCallResultFrame,
FunctionCallsStartedFrame,
InputAudioRawFrame,
InterimTranscriptionFrame,
TranscriptionFrame,
@@ -148,54 +150,59 @@ class TestSTTMuteFilter(unittest.IsolatedAsyncioTestCase):
expected_down_frames=expected_returned_frames,
)
# TODO: Revisit once we figure out how to test SystemFrames and DataFrames
# async def test_function_call_strategy(self):
# filter = STTMuteFilter(config=STTMuteConfig(strategies={STTMuteStrategy.FUNCTION_CALL}))
async def test_function_call_strategy(self):
filter = STTMuteFilter(config=STTMuteConfig(strategies={STTMuteStrategy.FUNCTION_CALL}))
# frames_to_send = [
# VADUserStartedSpeakingFrame(), # Should pass through initially
# UserStartedSpeakingFrame(), # Should pass through initially
# VADUserStoppedSpeakingFrame(),
# UserStoppedSpeakingFrame(),
# FunctionCallInProgressFrame(
# function_name="get_weather",
# tool_call_id="call_123",
# arguments='{"location": "San Francisco"}',
# ), # Start function call
# VADUserStartedSpeakingFrame(), # Should be suppressed
# UserStartedSpeakingFrame(), # Should be suppressed
# VADUserStoppedSpeakingFrame(), # Should be suppressed
# UserStoppedSpeakingFrame(), # Should be suppressed
# FunctionCallResultFrame(
# function_name="get_weather",
# tool_call_id="call_123",
# arguments='{"location": "San Francisco"}',
# result={"temperature": 22},
# ), # End function call
# VADUserStartedSpeakingFrame(), # Should pass through again
# UserStartedSpeakingFrame(), # Should pass through again
# VADUserStoppedSpeakingFrame(),
# UserStoppedSpeakingFrame(),
# ]
frames_to_send = [
VADUserStartedSpeakingFrame(), # Should pass through initially
UserStartedSpeakingFrame(), # Should pass through initially
VADUserStoppedSpeakingFrame(), # Should pass through initially
UserStoppedSpeakingFrame(), # Should pass through initially
FunctionCallsStartedFrame(
function_calls=[
FunctionCallFromLLM(
function_name="get_weather",
tool_call_id="call_123",
arguments='{"location": "San Francisco"}',
context=None,
)
]
), # Start function call
VADUserStartedSpeakingFrame(), # Should be suppressed
UserStartedSpeakingFrame(), # Should be suppressed
VADUserStoppedSpeakingFrame(), # Should be suppressed
UserStoppedSpeakingFrame(), # Should be suppressed
FunctionCallResultFrame(
function_name="get_weather",
tool_call_id="call_123",
arguments='{"location": "San Francisco"}',
result={"temperature": 22},
), # End function call
SleepFrame(),
VADUserStartedSpeakingFrame(), # Should pass through again
UserStartedSpeakingFrame(), # Should pass through again
VADUserStoppedSpeakingFrame(),
UserStoppedSpeakingFrame(),
]
# expected_returned_frames = [
# VADUserStartedSpeakingFrame,
# UserStartedSpeakingFrame,
# VADUserStoppedSpeakingFrame,
# UserStoppedSpeakingFrame,
# FunctionCallInProgressFrame,
# FunctionCallResultFrame,
# VADUserStartedSpeakingFrame,
# UserStartedSpeakingFrame,
# VADUserStoppedSpeakingFrame,
# UserStoppedSpeakingFrame,
# ]
expected_returned_frames = [
VADUserStartedSpeakingFrame,
UserStartedSpeakingFrame,
VADUserStoppedSpeakingFrame,
UserStoppedSpeakingFrame,
FunctionCallsStartedFrame,
FunctionCallResultFrame,
VADUserStartedSpeakingFrame,
UserStartedSpeakingFrame,
VADUserStoppedSpeakingFrame,
UserStoppedSpeakingFrame,
]
# await run_test(
# filter,
# frames_to_send=frames_to_send,
# expected_down_frames=expected_returned_frames,
# )
await run_test(
filter,
frames_to_send=frames_to_send,
expected_down_frames=expected_returned_frames,
)
async def test_mute_until_first_bot_complete_strategy(self):
filter = STTMuteFilter(