Files
VoiceAgent/test_working_phrase.py
T
2026-08-07 18:15:36 -04:00

49 lines
1.9 KiB
Python

"""Silence during a slow tool call must be broken.
From a recorded session: four of nine turns went unanswered because tool-heavy
turns ran for over a minute with no sound, and speaking again to check whether
it was alive cancelled the turn in flight.
"""
import asyncio, sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent))
from claude_llm import ClaudeCodeLLM
from pipecat.frames.frames import TTSSpeakFrame
class Recorder(ClaudeCodeLLM):
def __init__(self, **kw):
super().__init__(options=None, **kw)
self.pushed = []
async def push_frame(self, frame, direction=None):
self.pushed.append(frame)
async def main():
# Tool runs before anything has been said: the user hears nothing otherwise.
llm = Recorder()
llm._said_working = False
await llm._say_working(spoken=[])
said = [f.text for f in llm.pushed if isinstance(f, TTSSpeakFrame)]
print(f" {'PASS' if said else 'FAIL'} speaks up when a tool runs first: {said}")
# Only once per turn, however many tools run.
await llm._say_working(spoken=[])
await llm._say_working(spoken=[])
said = [f.text for f in llm.pushed if isinstance(f, TTSSpeakFrame)]
print(f" {'PASS' if len(said) == 1 else 'FAIL'} says it only once per turn ({len(said)})")
# Already answering: adding a filler would talk over the real reply.
llm2 = Recorder()
llm2._said_working = False
await llm2._say_working(spoken=["Running that now."])
quiet = not [f for f in llm2.pushed if isinstance(f, TTSSpeakFrame)]
print(f" {'PASS' if quiet else 'FAIL'} stays quiet when it already narrated")
# Opted out.
llm3 = Recorder(working_phrase=None)
llm3._said_working = False
await llm3._say_working(spoken=[])
quiet = not [f for f in llm3.pushed if isinstance(f, TTSSpeakFrame)]
print(f" {'PASS' if quiet else 'FAIL'} respects working_phrase=None")
asyncio.run(main())