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

allow custom interruption strategies

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2025-06-01 13:54:57 -07:00
parent 13546d5e8f
commit 5512de3221
10 changed files with 140 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
tests/test_interruption_strategies.py Normal file
View File

@@ -0,0 +1,24 @@
#
# Copyright (c) 2024-2025 Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import unittest
from pipecat.audio.interruptions.min_words_interruption_strategy import MinWordsInterruptionStrategy
class TestInterruptionStrategy(unittest.IsolatedAsyncioTestCase):
async def test_min_words(self):
strategy = MinWordsInterruptionStrategy(min_words=2)
await strategy.append_text("Hello")
self.assertEqual(await strategy.should_interrupt(), False)
await strategy.append_text(" there!")
self.assertEqual(await strategy.should_interrupt(), True)
# Reset and check again
await strategy.reset()
await strategy.append_text("Hello!")
self.assertEqual(await strategy.should_interrupt(), False)
await strategy.append_text(" How are you?")
self.assertEqual(await strategy.should_interrupt(), True)