wx44wx/ai-video-fullstack
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: release input mute after async client tools

Browse Source
This commit is contained in:
Xin Wang
2026-08-06 00:12:34 +08:00
parent 0c0a51da95
commit 90bb27a377
2 changed files with 123 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

101
backend/tests/test_tool_policy.py
View File

@@ -8,6 +8,7 @@ from pipecat.frames.frames import (
FunctionCallResultFrame,
FunctionCallsStartedFrame,
)
from pipecat.services.llm_service import FunctionCallResultProperties
from services.pipecat.processors import ToolInterruptionUserMuteStrategy
from services.tool_policy import policy_for_tool
@@ -143,6 +144,106 @@ class ToolInterruptionStrategyTests(unittest.IsolatedAsyncioTestCase):
await strategy.process_frame(BotStoppedSpeakingFrame())
)
async def test_async_tool_only_result_releases_without_followup_speech(self):
strategy = ToolInterruptionUserMuteStrategy(
{"set_photo_button_visible": "async"}
)
await strategy.process_frame(
FunctionCallsStartedFrame(
function_calls=[
SimpleNamespace(
function_name="set_photo_button_visible",
tool_call_id="call_4",
)
]
)
)
self.assertFalse(
await strategy.process_frame(
FunctionCallResultFrame(
function_name="set_photo_button_visible",
tool_call_id="call_4",
arguments={"visible": True},
result={"status": "ok"},
properties=FunctionCallResultProperties(run_llm=False),
)
)
)
self.assertFalse(strategy.is_muted)
async def test_async_tool_only_result_waits_for_active_preamble(self):
strategy = ToolInterruptionUserMuteStrategy(
{"set_photo_button_visible": "async"}
)
await strategy.process_frame(BotStartedSpeakingFrame())
await strategy.process_frame(
FunctionCallsStartedFrame(
function_calls=[
SimpleNamespace(
function_name="set_photo_button_visible",
tool_call_id="call_5",
)
]
)
)
self.assertTrue(
await strategy.process_frame(
FunctionCallResultFrame(
function_name="set_photo_button_visible",
tool_call_id="call_5",
arguments={"visible": True},
result={"status": "ok"},
properties=FunctionCallResultProperties(run_llm=False),
)
)
)
self.assertFalse(
await strategy.process_frame(BotStoppedSpeakingFrame())
)
async def test_async_intermediate_result_does_not_release_mute(self):
strategy = ToolInterruptionUserMuteStrategy(
{"set_photo_button_visible": "async"}
)
await strategy.process_frame(
FunctionCallsStartedFrame(
function_calls=[
SimpleNamespace(
function_name="set_photo_button_visible",
tool_call_id="call_6",
)
]
)
)
self.assertTrue(
await strategy.process_frame(
FunctionCallResultFrame(
function_name="set_photo_button_visible",
tool_call_id="call_6",
arguments={"visible": True},
result={"status": "pending"},
properties=FunctionCallResultProperties(
run_llm=False,
is_final=False,
),
)
)
)
self.assertFalse(
await strategy.process_frame(
FunctionCallResultFrame(
function_name="set_photo_button_visible",
tool_call_id="call_6",
arguments={"visible": True},
result={"status": "ok"},
properties=FunctionCallResultProperties(run_llm=False),
)
)
)
if __name__ == "__main__":
unittest.main()