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

MinWordsUserTurnStartStrategy: single word interrupt if bot not speaking

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2025-12-22 14:32:21 -08:00
parent 08a9b434c1
commit a0d5ee3873
3 changed files with 80 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
tests/test_user_turn_start_strategy.py
View File

@@ -21,7 +21,7 @@ from pipecat.turns.user.vad_user_turn_start_strategy import VADUserTurnStartStra
class TestMinWordsInterruptionStrategy(unittest.IsolatedAsyncioTestCase):
async def test_only_transcriptions(self):
async def test_bot_speaking_transcriptions(self):
strategy = MinWordsUserTurnStartStrategy(min_words=2)
should_start = None
@@ -31,6 +31,7 @@ class TestMinWordsInterruptionStrategy(unittest.IsolatedAsyncioTestCase):
nonlocal should_start
should_start = True
await strategy.process_frame(BotStartedSpeakingFrame())
await strategy.process_frame(TranscriptionFrame(text="Hello", user_id="cat", timestamp=""))
self.assertFalse(should_start)
@@ -43,6 +44,7 @@ class TestMinWordsInterruptionStrategy(unittest.IsolatedAsyncioTestCase):
should_start = None
await strategy.reset()
await strategy.process_frame(BotStartedSpeakingFrame())
await strategy.process_frame(TranscriptionFrame(text="Hello!", user_id="cat", timestamp=""))
self.assertFalse(should_start)
@@ -51,7 +53,7 @@ class TestMinWordsInterruptionStrategy(unittest.IsolatedAsyncioTestCase):
)
self.assertTrue(should_start)
async def test_only_interim_transcriptions(self):
async def test_bot_speaking_interim_transcriptions(self):
strategy = MinWordsUserTurnStartStrategy(min_words=2)
should_start = None
@@ -61,17 +63,19 @@ class TestMinWordsInterruptionStrategy(unittest.IsolatedAsyncioTestCase):
nonlocal should_start
should_start = True
await strategy.process_frame(BotStartedSpeakingFrame())
await strategy.process_frame(
InterimTranscriptionFrame(text="Hello", user_id="cat", timestamp="")
)
self.assertFalse(should_start)
await strategy.process_frame(BotStartedSpeakingFrame())
await strategy.process_frame(
InterimTranscriptionFrame(text="Hello there!", user_id="cat", timestamp="")
)
self.assertTrue(should_start)
async def test_all_transcriptions(self):
async def test_bot_speaking_all_transcriptions(self):
strategy = MinWordsUserTurnStartStrategy(min_words=2)
should_start = None
@@ -81,6 +85,7 @@ class TestMinWordsInterruptionStrategy(unittest.IsolatedAsyncioTestCase):
nonlocal should_start
should_start = True
await strategy.process_frame(BotStartedSpeakingFrame())
await strategy.process_frame(
InterimTranscriptionFrame(text="Hello", user_id="cat", timestamp="")
)
@@ -91,6 +96,34 @@ class TestMinWordsInterruptionStrategy(unittest.IsolatedAsyncioTestCase):
)
self.assertTrue(should_start)
async def test_bot_not_speaking_transcriptions(self):
strategy = MinWordsUserTurnStartStrategy(min_words=2)
should_start = None
@strategy.event_handler("on_user_turn_started")
async def on_user_turn_started(strategy):
nonlocal should_start
should_start = True
await strategy.process_frame(TranscriptionFrame(text="Hello", user_id="cat", timestamp=""))
self.assertTrue(should_start)
async def test_bot_not_speaking_interim_transcriptions(self):
strategy = MinWordsUserTurnStartStrategy(min_words=2)
should_start = None
@strategy.event_handler("on_user_turn_started")
async def on_user_turn_started(strategy):
nonlocal should_start
should_start = True
await strategy.process_frame(
InterimTranscriptionFrame(text="Hello", user_id="cat", timestamp="")
)
self.assertTrue(should_start)
class TestVADUserTurnStartStrategy(unittest.IsolatedAsyncioTestCase):
async def test_vad_strategy(self):