From 7e69288898b99d7e2b627572771d8a04250403ed Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 8 Dec 2025 18:22:58 -0800 Subject: [PATCH] tests: added bot turn start strategies unit tests --- pyproject.toml | 2 +- tests/test_bot_turn_start_strategy.py | 476 ++++++++++++++++++++++++++ tests/test_langchain.py | 14 +- uv.lock | 9 +- 4 files changed, 491 insertions(+), 10 deletions(-) create mode 100644 tests/test_bot_turn_start_strategy.py diff --git a/pyproject.toml b/pyproject.toml index 23ac8490c..b927bafe4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -125,7 +125,7 @@ dev = [ "pre-commit~=4.2.0", "pyright>=1.1.404,<1.2", "pytest~=8.4.1", - "pytest-asyncio~=1.1.0", + "pytest-asyncio~=1.3.0", "pytest-aiohttp==1.1.0", "ruff>=0.12.11,<1", "setuptools~=78.1.1", diff --git a/tests/test_bot_turn_start_strategy.py b/tests/test_bot_turn_start_strategy.py new file mode 100644 index 000000000..781688f21 --- /dev/null +++ b/tests/test_bot_turn_start_strategy.py @@ -0,0 +1,476 @@ +# +# Copyright (c) 2024-2025 Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import unittest + +from pipecat.frames.frames import ( + InterimTranscriptionFrame, + TranscriptionFrame, + VADUserStartedSpeakingFrame, + VADUserStoppedSpeakingFrame, +) +from pipecat.turns.bot.transcription_bot_turn_start_strategy import ( + TranscriptionBotTurnStartStrategy, +) +from pipecat.utils.asyncio.task_manager import TaskManager, TaskManagerParams + +AGGREGATION_TIMEOUT = 0.1 + + +class TestTranscriptionBotTurnStartStrategy(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self) -> None: + self.task_manager = TaskManager() + self.task_manager.setup(TaskManagerParams(loop=asyncio.get_running_loop())) + + async def test_ste(self): + strategy = TranscriptionBotTurnStartStrategy() + + should_start = None + + @strategy.event_handler("on_bot_turn_started") + async def on_bot_turn_started(strategy): + nonlocal should_start + should_start = True + + # S + await strategy.process_frame(VADUserStartedSpeakingFrame()) + self.assertIsNone(should_start) + + # T + await strategy.process_frame(TranscriptionFrame(text="Hello!", user_id="cat", timestamp="")) + self.assertIsNone(should_start) + + # E + await strategy.process_frame(VADUserStoppedSpeakingFrame()) + + # Transcription comes in between user started/stopped and there are not + # interim, we just trigger bot speech. + self.assertTrue(should_start) + + async def test_site(self): + strategy = TranscriptionBotTurnStartStrategy() + + should_start = None + + @strategy.event_handler("on_bot_turn_started") + async def on_bot_turn_started(strategy): + nonlocal should_start + should_start = True + + # S + await strategy.process_frame(VADUserStartedSpeakingFrame()) + self.assertIsNone(should_start) + + # I + await strategy.process_frame( + InterimTranscriptionFrame(text="Hello!", user_id="cat", timestamp="") + ) + self.assertIsNone(should_start) + + # T + await strategy.process_frame(TranscriptionFrame(text="Hello!", user_id="cat", timestamp="")) + self.assertIsNone(should_start) + + # E + await strategy.process_frame(VADUserStoppedSpeakingFrame()) + + # Transcription comes in between user started/stopped, so we trigger + # speech right away. + self.assertTrue(should_start) + + async def test_st1iest2e(self): + strategy = TranscriptionBotTurnStartStrategy() + + should_start = None + + @strategy.event_handler("on_bot_turn_started") + async def on_bot_turn_started(strategy): + nonlocal should_start + should_start = True + + # S + await strategy.process_frame(VADUserStartedSpeakingFrame()) + self.assertIsNone(should_start) + + # T1 + await strategy.process_frame(TranscriptionFrame(text="Hello!", user_id="cat", timestamp="")) + self.assertIsNone(should_start) + + # I + await strategy.process_frame( + InterimTranscriptionFrame(text="Hello!", user_id="cat", timestamp="") + ) + self.assertIsNone(should_start) + + # E + await strategy.process_frame(VADUserStoppedSpeakingFrame()) + self.assertIsNone(should_start) + + # S + await strategy.process_frame(VADUserStartedSpeakingFrame()) + self.assertIsNone(should_start) + + # T2 + await strategy.process_frame( + TranscriptionFrame(text="How are you?", user_id="cat", timestamp="") + ) + self.assertIsNone(should_start) + + # E + await strategy.process_frame(VADUserStoppedSpeakingFrame()) + + # There was an interim before the first user stopped speaking, then we + # got a transcription comes in between user started/stopped, so we + # trigger speech right away. + self.assertTrue(should_start) + + async def test_siet(self): + strategy = TranscriptionBotTurnStartStrategy(timeout=AGGREGATION_TIMEOUT) + await strategy.setup(self.task_manager) + + should_start = None + + @strategy.event_handler("on_bot_turn_started") + async def on_bot_turn_started(strategy): + nonlocal should_start + should_start = True + + # S + await strategy.process_frame(VADUserStartedSpeakingFrame()) + self.assertIsNone(should_start) + + # I + await strategy.process_frame( + InterimTranscriptionFrame(text="How", user_id="cat", timestamp="") + ) + self.assertIsNone(should_start) + + # E + await strategy.process_frame(VADUserStoppedSpeakingFrame()) + self.assertIsNone(should_start) + + # T + await strategy.process_frame(TranscriptionFrame(text="Hello!", user_id="cat", timestamp="")) + self.assertIsNone(should_start) + + # Transcription comes after user stopped speaking, we need to wait for + # at least the aggregation timeout. + await asyncio.sleep(AGGREGATION_TIMEOUT + 0.1) + self.assertTrue(should_start) + + async def test_sieit(self): + strategy = TranscriptionBotTurnStartStrategy(timeout=AGGREGATION_TIMEOUT) + await strategy.setup(self.task_manager) + + should_start = None + + @strategy.event_handler("on_bot_turn_started") + async def on_bot_turn_started(strategy): + nonlocal should_start + should_start = True + + # S + await strategy.process_frame(VADUserStartedSpeakingFrame()) + self.assertIsNone(should_start) + + # I + await strategy.process_frame( + InterimTranscriptionFrame(text="How", user_id="cat", timestamp="") + ) + self.assertIsNone(should_start) + + # E + await strategy.process_frame(VADUserStoppedSpeakingFrame()) + self.assertIsNone(should_start) + + # I + await strategy.process_frame( + InterimTranscriptionFrame(text="are you?", user_id="cat", timestamp="") + ) + self.assertIsNone(should_start) + + # T + await strategy.process_frame( + TranscriptionFrame(text="How are you?", user_id="cat", timestamp="") + ) + self.assertIsNone(should_start) + + # Transcription comes after user stopped speaking, we need to wait for + # at least the aggregation timeout. + await asyncio.sleep(AGGREGATION_TIMEOUT + 0.1) + self.assertTrue(should_start) + + async def test_set(self): + strategy = TranscriptionBotTurnStartStrategy(timeout=AGGREGATION_TIMEOUT) + await strategy.setup(self.task_manager) + + should_start = None + + @strategy.event_handler("on_bot_turn_started") + async def on_bot_turn_started(strategy): + nonlocal should_start + should_start = True + + # S + await strategy.process_frame(VADUserStartedSpeakingFrame()) + self.assertIsNone(should_start) + + # E + await strategy.process_frame(VADUserStoppedSpeakingFrame()) + self.assertIsNone(should_start) + + # T + await strategy.process_frame( + TranscriptionFrame(text="How are you?", user_id="cat", timestamp="") + ) + self.assertIsNone(should_start) + + # Transcription comes after user stopped speaking, we need to wait for + # at least the aggregation timeout. + await asyncio.sleep(AGGREGATION_TIMEOUT + 0.1) + self.assertTrue(should_start) + + async def test_seit(self): + strategy = TranscriptionBotTurnStartStrategy(timeout=AGGREGATION_TIMEOUT) + await strategy.setup(self.task_manager) + + should_start = None + + @strategy.event_handler("on_bot_turn_started") + async def on_bot_turn_started(strategy): + nonlocal should_start + should_start = True + + # S + await strategy.process_frame(VADUserStartedSpeakingFrame()) + self.assertIsNone(should_start) + + # E + await strategy.process_frame(VADUserStoppedSpeakingFrame()) + self.assertIsNone(should_start) + + # I + await strategy.process_frame( + InterimTranscriptionFrame(text="How", user_id="cat", timestamp="") + ) + self.assertIsNone(should_start) + + # T + await strategy.process_frame( + TranscriptionFrame(text="How are you?", user_id="cat", timestamp="") + ) + self.assertIsNone(should_start) + + # Transcription comes after user stopped speaking, we need to wait for + # at least the aggregation timeout. + await asyncio.sleep(AGGREGATION_TIMEOUT + 0.1) + self.assertTrue(should_start) + + async def test_st1et2(self): + strategy = TranscriptionBotTurnStartStrategy(timeout=AGGREGATION_TIMEOUT) + await strategy.setup(self.task_manager) + + should_start = None + + @strategy.event_handler("on_bot_turn_started") + async def on_bot_turn_started(strategy): + nonlocal should_start + should_start = True + + # S + await strategy.process_frame(VADUserStartedSpeakingFrame()) + self.assertIsNone(should_start) + + # T1 + await strategy.process_frame(TranscriptionFrame(text="Hello!", user_id="cat", timestamp="")) + self.assertIsNone(should_start) + + # E + await strategy.process_frame(VADUserStoppedSpeakingFrame()) + + # Transcription comes between user start/stopped speaking, we need to + # trigger speech right away. + self.assertTrue(should_start) + should_start = None + + # T2 + await strategy.process_frame( + TranscriptionFrame(text="How are you?", user_id="cat", timestamp="") + ) + self.assertIsNone(should_start) + + # Transcription comes after user stopped speaking, we need to wait for + # at least the aggregation timeout. + await asyncio.sleep(AGGREGATION_TIMEOUT + 0.1) + self.assertTrue(should_start) + + async def test_set1t2(self): + strategy = TranscriptionBotTurnStartStrategy(timeout=AGGREGATION_TIMEOUT) + await strategy.setup(self.task_manager) + + should_start = None + + @strategy.event_handler("on_bot_turn_started") + async def on_bot_turn_started(strategy): + nonlocal should_start + should_start = True + + # S + await strategy.process_frame(VADUserStartedSpeakingFrame()) + self.assertIsNone(should_start) + + # E + await strategy.process_frame(VADUserStoppedSpeakingFrame()) + self.assertIsNone(should_start) + + # T1 + await strategy.process_frame(TranscriptionFrame(text="Hello!", user_id="cat", timestamp="")) + self.assertIsNone(should_start) + + # T2 + await strategy.process_frame( + TranscriptionFrame(text="How are you?", user_id="cat", timestamp="") + ) + self.assertIsNone(should_start) + + # Transcription comes after user stopped speaking, we need to wait for + # at least the aggregation timeout. + await asyncio.sleep(AGGREGATION_TIMEOUT + 0.1) + self.assertTrue(should_start) + + async def test_siet1it2(self): + strategy = TranscriptionBotTurnStartStrategy(timeout=AGGREGATION_TIMEOUT) + await strategy.setup(self.task_manager) + + should_start = None + + @strategy.event_handler("on_bot_turn_started") + async def on_bot_turn_started(strategy): + nonlocal should_start + should_start = True + + # S + await strategy.process_frame(VADUserStartedSpeakingFrame()) + self.assertIsNone(should_start) + + # I + await strategy.process_frame( + InterimTranscriptionFrame(text="Hello!", user_id="cat", timestamp="") + ) + + # E + await strategy.process_frame(VADUserStoppedSpeakingFrame()) + self.assertIsNone(should_start) + + # T1 + await strategy.process_frame(TranscriptionFrame(text="Hello!", user_id="cat", timestamp="")) + self.assertIsNone(should_start) + + # I + await strategy.process_frame( + InterimTranscriptionFrame(text="How", user_id="cat", timestamp="") + ) + self.assertIsNone(should_start) + + # T2 + await strategy.process_frame( + TranscriptionFrame(text="How are you?", user_id="cat", timestamp="") + ) + self.assertIsNone(should_start) + + # Transcription comes after user stopped speaking, we need to wait for + # at least the aggregation timeout. + await asyncio.sleep(AGGREGATION_TIMEOUT + 0.1) + self.assertTrue(should_start) + + async def test_t(self): + strategy = TranscriptionBotTurnStartStrategy(timeout=AGGREGATION_TIMEOUT) + await strategy.setup(self.task_manager) + + should_start = None + + @strategy.event_handler("on_bot_turn_started") + async def on_bot_turn_started(strategy): + nonlocal should_start + should_start = True + + # T + await strategy.process_frame(TranscriptionFrame(text="Hello!", user_id="cat", timestamp="")) + self.assertIsNone(should_start) + + # Transcription comes after user stopped speaking, we need to wait for + # at least the aggregation timeout. + await asyncio.sleep(AGGREGATION_TIMEOUT + 0.1) + self.assertTrue(should_start) + + async def test_it(self): + strategy = TranscriptionBotTurnStartStrategy(timeout=AGGREGATION_TIMEOUT) + await strategy.setup(self.task_manager) + + should_start = None + + @strategy.event_handler("on_bot_turn_started") + async def on_bot_turn_started(strategy): + nonlocal should_start + should_start = True + + # I + await strategy.process_frame( + InterimTranscriptionFrame(text="Hello!", user_id="cat", timestamp="") + ) + + # T + await strategy.process_frame(TranscriptionFrame(text="Hello!", user_id="cat", timestamp="")) + self.assertIsNone(should_start) + + # Transcription comes after user stopped speaking, we need to wait for + # at least the aggregation timeout. + await asyncio.sleep(AGGREGATION_TIMEOUT + 0.1) + self.assertTrue(should_start) + + async def test_sie_delay_it(self): + strategy = TranscriptionBotTurnStartStrategy(timeout=AGGREGATION_TIMEOUT) + await strategy.setup(self.task_manager) + + should_start = None + + @strategy.event_handler("on_bot_turn_started") + async def on_bot_turn_started(strategy): + nonlocal should_start + should_start = True + + # S + await strategy.process_frame(VADUserStartedSpeakingFrame()) + self.assertIsNone(should_start) + + # I + await strategy.process_frame( + InterimTranscriptionFrame(text="Hello!", user_id="cat", timestamp="") + ) + + # E + await strategy.process_frame(VADUserStoppedSpeakingFrame()) + self.assertIsNone(should_start) + + # Delay + await asyncio.sleep(AGGREGATION_TIMEOUT + 0.1) + + # I + await strategy.process_frame( + InterimTranscriptionFrame(text="How", user_id="cat", timestamp="") + ) + + # T + await strategy.process_frame( + TranscriptionFrame(text="How are you?", user_id="cat", timestamp="") + ) + self.assertIsNone(should_start) + + # Transcription comes after user stopped speaking, we need to wait for + # at least the aggregation timeout. + await asyncio.sleep(AGGREGATION_TIMEOUT + 0.1) + self.assertTrue(should_start) diff --git a/tests/test_langchain.py b/tests/test_langchain.py index 4e197b2aa..21d8428eb 100644 --- a/tests/test_langchain.py +++ b/tests/test_langchain.py @@ -10,6 +10,7 @@ from langchain.prompts import ChatPromptTemplate from langchain_core.language_models import FakeStreamingListLLM from pipecat.frames.frames import ( + InterruptionFrame, LLMContextAssistantTimestampFrame, LLMContextFrame, LLMFullResponseEndFrame, @@ -18,12 +19,11 @@ from pipecat.frames.frames import ( TranscriptionFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, + VADUserStartedSpeakingFrame, + VADUserStoppedSpeakingFrame, ) from pipecat.pipeline.pipeline import Pipeline from pipecat.processors.aggregators.llm_context import LLMContext -from pipecat.processors.aggregators.llm_response import ( - LLMAssistantAggregatorParams, -) from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair from pipecat.processors.frame_processor import FrameProcessor from pipecat.processors.frameworks.langchain import LangchainProcessor @@ -72,13 +72,17 @@ class TestLangchain(unittest.IsolatedAsyncioTestCase): ) frames_to_send = [ - UserStartedSpeakingFrame(), + VADUserStartedSpeakingFrame(), TranscriptionFrame(text="Hi World", user_id="user", timestamp="now"), SleepFrame(), - UserStoppedSpeakingFrame(), + VADUserStoppedSpeakingFrame(), + SleepFrame(sleep=1.0), ] expected_down_frames = [ + VADUserStartedSpeakingFrame, UserStartedSpeakingFrame, + InterruptionFrame, + VADUserStoppedSpeakingFrame, UserStoppedSpeakingFrame, LLMContextFrame, LLMContextAssistantTimestampFrame, diff --git a/uv.lock b/uv.lock index cede216c2..f9b826937 100644 --- a/uv.lock +++ b/uv.lock @@ -4191,7 +4191,7 @@ dev = [ { name = "pyright", specifier = ">=1.1.404,<1.2" }, { name = "pytest", specifier = "~=8.4.1" }, { name = "pytest-aiohttp", specifier = "==1.1.0" }, - { name = "pytest-asyncio", specifier = "~=1.1.0" }, + { name = "pytest-asyncio", specifier = "~=1.3.0" }, { name = "python-dotenv", specifier = ">=1.0.1,<2.0.0" }, { name = "ruff", specifier = ">=0.12.11,<1" }, { name = "setuptools", specifier = "~=78.1.1" }, @@ -4828,15 +4828,16 @@ wheels = [ [[package]] name = "pytest-asyncio" -version = "1.1.1" +version = "1.3.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "backports-asyncio-runner", marker = "python_full_version < '3.11'" }, { name = "pytest" }, + { name = "typing-extensions", marker = "python_full_version < '3.13'" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/8d/1e/2aa43805d4a320a9489d2b99f7877b69f9094c79aa0732159a1415dd6cd4/pytest_asyncio-1.1.1.tar.gz", hash = "sha256:b72d215c38e2c91dbb32f275e0b5be69602d7869910e109360e375129960a649", size = 46590, upload-time = "2025-09-12T06:36:20.834Z" } +sdist = { url = "https://files.pythonhosted.org/packages/90/2c/8af215c0f776415f3590cac4f9086ccefd6fd463befeae41cd4d3f193e5a/pytest_asyncio-1.3.0.tar.gz", hash = "sha256:d7f52f36d231b80ee124cd216ffb19369aa168fc10095013c6b014a34d3ee9e5", size = 50087, upload-time = "2025-11-10T16:07:47.256Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/28/de/aba79e9ccdb51b5d0d65c67dd857bd78b00c64723df16b9fc800d8b94ce6/pytest_asyncio-1.1.1-py3-none-any.whl", hash = "sha256:726339d30fcfde24691f589445b9b67d058b311ac632b1d704e97f20f1d878da", size = 14719, upload-time = "2025-09-12T06:36:19.726Z" }, + { url = "https://files.pythonhosted.org/packages/e5/35/f8b19922b6a25bc0880171a2f1a003eaeb93657475193ab516fd87cac9da/pytest_asyncio-1.3.0-py3-none-any.whl", hash = "sha256:611e26147c7f77640e6d0a92a38ed17c3e9848063698d5c93d5aa7aa11cebff5", size = 15075, upload-time = "2025-11-10T16:07:45.537Z" }, ] [[package]]