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

tests: added bot turn start strategies unit tests

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2025-12-08 18:22:58 -08:00
parent 76561da850
commit 7e69288898
4 changed files with 491 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
pyproject.toml
View File

@@ -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",

476
tests/test_bot_turn_start_strategy.py Normal file
View File

@@ -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)

14
tests/test_langchain.py
View File

@@ -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,

9
uv.lock generated
View File

@@ -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]]