Merge branch 'main' into piper-tts
# Conflicts: # test-requirements.txt
This commit is contained in:
Filipi Fuchter
@@ -1,33 +0,0 @@
|
||||
import asyncio
|
||||
import os
|
||||
import unittest
|
||||
|
||||
from openai.types.chat import (
|
||||
ChatCompletionSystemMessageParam,
|
||||
)
|
||||
|
||||
from pipecat.processors.aggregators.openai_llm_context import (
|
||||
OpenAILLMContext,
|
||||
OpenAILLMContextFrame,
|
||||
)
|
||||
from pipecat.services.azure import AzureLLMService
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@unittest.skip("Skip azure integration test")
|
||||
async def test_chat():
|
||||
llm = AzureLLMService(
|
||||
api_key=os.getenv("AZURE_CHATGPT_API_KEY"),
|
||||
endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"),
|
||||
model=os.getenv("AZURE_CHATGPT_MODEL"),
|
||||
)
|
||||
context = OpenAILLMContext()
|
||||
message: ChatCompletionSystemMessageParam = ChatCompletionSystemMessageParam(
|
||||
content="Please tell the world hello.", name="system", role="system"
|
||||
)
|
||||
context.add_message(message)
|
||||
frame = OpenAILLMContextFrame(context)
|
||||
async for s in llm.process_frame(frame):
|
||||
print(s)
|
||||
|
||||
asyncio.run(test_chat())
|
||||
@@ -1,28 +0,0 @@
|
||||
import asyncio
|
||||
import unittest
|
||||
|
||||
from openai.types.chat import (
|
||||
ChatCompletionSystemMessageParam,
|
||||
)
|
||||
|
||||
from pipecat.processors.aggregators.openai_llm_context import (
|
||||
OpenAILLMContext,
|
||||
OpenAILLMContextFrame,
|
||||
)
|
||||
from pipecat.services.ollama import OLLamaLLMService
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@unittest.skip("Skip azure integration test")
|
||||
async def test_chat():
|
||||
llm = OLLamaLLMService()
|
||||
context = OpenAILLMContext()
|
||||
message: ChatCompletionSystemMessageParam = ChatCompletionSystemMessageParam(
|
||||
content="Please tell the world hello.", name="system", role="system"
|
||||
)
|
||||
context.add_message(message)
|
||||
frame = OpenAILLMContextFrame(context)
|
||||
async for s in llm.process_frame(frame):
|
||||
print(s)
|
||||
|
||||
asyncio.run(test_chat())
|
||||
@@ -1,128 +0,0 @@
|
||||
import asyncio
|
||||
import json
|
||||
import os
|
||||
from typing import List
|
||||
|
||||
from openai.types.chat import (
|
||||
ChatCompletionSystemMessageParam,
|
||||
ChatCompletionToolParam,
|
||||
ChatCompletionUserMessageParam,
|
||||
)
|
||||
|
||||
from pipecat.frames.frames import LLMFullResponseEndFrame, LLMFullResponseStartFrame, TextFrame
|
||||
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
|
||||
from pipecat.services.openai import OpenAILLMContext, OpenAILLMContextFrame, OpenAILLMService
|
||||
from pipecat.utils.test_frame_processor import TestFrameProcessor
|
||||
|
||||
tools = [
|
||||
ChatCompletionToolParam(
|
||||
type="function",
|
||||
function={
|
||||
"name": "get_current_weather",
|
||||
"description": "Get the current weather",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"location": {
|
||||
"type": "string",
|
||||
"description": "The city and state, e.g. San Francisco, CA",
|
||||
},
|
||||
"format": {
|
||||
"type": "string",
|
||||
"enum": ["celsius", "fahrenheit"],
|
||||
"description": "The temperature unit to use. Infer this from the users location.",
|
||||
},
|
||||
},
|
||||
"required": ["location", "format"],
|
||||
},
|
||||
},
|
||||
)
|
||||
]
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
async def test_simple_functions():
|
||||
async def get_weather_from_api(llm, args):
|
||||
return json.dumps({"conditions": "nice", "temperature": "75"})
|
||||
|
||||
api_key = os.getenv("OPENAI_API_KEY")
|
||||
|
||||
llm = OpenAILLMService(
|
||||
api_key=api_key or "",
|
||||
model="gpt-4-1106-preview",
|
||||
)
|
||||
|
||||
llm.register_function("get_current_weather", get_weather_from_api)
|
||||
t = TestFrameProcessor([LLMFullResponseStartFrame, TextFrame, LLMFullResponseEndFrame])
|
||||
llm.link(t)
|
||||
|
||||
context = OpenAILLMContext(tools=tools)
|
||||
system_message: ChatCompletionSystemMessageParam = ChatCompletionSystemMessageParam(
|
||||
content="Ask the user to ask for a weather report", name="system", role="system"
|
||||
)
|
||||
user_message: ChatCompletionUserMessageParam = ChatCompletionUserMessageParam(
|
||||
content="Could you tell me the weather for Boulder, Colorado",
|
||||
name="user",
|
||||
role="user",
|
||||
)
|
||||
context.add_message(system_message)
|
||||
context.add_message(user_message)
|
||||
frame = OpenAILLMContextFrame(context)
|
||||
await llm.process_frame(frame, FrameDirection.DOWNSTREAM)
|
||||
|
||||
async def test_advanced_functions():
|
||||
async def get_weather_from_api(llm, args):
|
||||
return [
|
||||
{
|
||||
"role": "system",
|
||||
"content": "The user has asked for live weather. Respond by telling them we don't currently support live weather for that area, but it's coming soon.",
|
||||
}
|
||||
]
|
||||
|
||||
api_key = os.getenv("OPENAI_API_KEY")
|
||||
|
||||
llm = OpenAILLMService(
|
||||
api_key=api_key or "",
|
||||
model="gpt-4-1106-preview",
|
||||
)
|
||||
|
||||
llm.register_function("get_current_weather", get_weather_from_api)
|
||||
t = TestFrameProcessor([LLMFullResponseStartFrame, TextFrame, LLMFullResponseEndFrame])
|
||||
llm.link(t)
|
||||
|
||||
context = OpenAILLMContext(tools=tools)
|
||||
system_message: ChatCompletionSystemMessageParam = ChatCompletionSystemMessageParam(
|
||||
content="Ask the user to ask for a weather report", name="system", role="system"
|
||||
)
|
||||
user_message: ChatCompletionUserMessageParam = ChatCompletionUserMessageParam(
|
||||
content="Could you tell me the weather for Boulder, Colorado",
|
||||
name="user",
|
||||
role="user",
|
||||
)
|
||||
context.add_message(system_message)
|
||||
context.add_message(user_message)
|
||||
frame = OpenAILLMContextFrame(context)
|
||||
await llm.process_frame(frame, FrameDirection.DOWNSTREAM)
|
||||
|
||||
async def test_chat():
|
||||
api_key = os.getenv("OPENAI_API_KEY")
|
||||
t = TestFrameProcessor([LLMFullResponseStartFrame, TextFrame, LLMFullResponseEndFrame])
|
||||
llm = OpenAILLMService(
|
||||
api_key=api_key or "",
|
||||
model="gpt-4o",
|
||||
)
|
||||
llm.link(t)
|
||||
context = OpenAILLMContext()
|
||||
message: ChatCompletionSystemMessageParam = ChatCompletionSystemMessageParam(
|
||||
content="Please tell the world hello.", name="system", role="system"
|
||||
)
|
||||
context.add_message(message)
|
||||
frame = OpenAILLMContextFrame(context)
|
||||
await llm.process_frame(frame, FrameDirection.DOWNSTREAM)
|
||||
|
||||
async def run_tests():
|
||||
await test_simple_functions()
|
||||
await test_advanced_functions()
|
||||
await test_chat()
|
||||
|
||||
asyncio.run(run_tests())
|
||||
@@ -0,0 +1,99 @@
|
||||
#
|
||||
# Copyright (c) 2024-2025 Daily
|
||||
#
|
||||
# SPDX-License-Identifier: BSD 2-Clause License
|
||||
#
|
||||
|
||||
import os
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import pytest
|
||||
from dotenv import load_dotenv
|
||||
|
||||
from pipecat.adapters.schemas.function_schema import FunctionSchema
|
||||
from pipecat.adapters.schemas.tools_schema import ToolsSchema
|
||||
from pipecat.pipeline.pipeline import Pipeline
|
||||
from pipecat.services.ai_services import LLMService
|
||||
from pipecat.services.anthropic import AnthropicLLMService
|
||||
from pipecat.services.google import GoogleLLMService
|
||||
from pipecat.services.openai import OpenAILLMContext, OpenAILLMContextFrame, OpenAILLMService
|
||||
from pipecat.tests.utils import run_test
|
||||
|
||||
load_dotenv(override=True)
|
||||
|
||||
|
||||
def standard_tools() -> ToolsSchema:
|
||||
weather_function = FunctionSchema(
|
||||
name="get_current_weather",
|
||||
description="Get the current weather",
|
||||
properties={
|
||||
"location": {
|
||||
"type": "string",
|
||||
"description": "The city and state, e.g. San Francisco, CA",
|
||||
},
|
||||
"format": {
|
||||
"type": "string",
|
||||
"enum": ["celsius", "fahrenheit"],
|
||||
"description": "The temperature unit to use. Infer this from the user's location.",
|
||||
},
|
||||
},
|
||||
required=["location"],
|
||||
)
|
||||
tools_def = ToolsSchema(standard_tools=[weather_function])
|
||||
return tools_def
|
||||
|
||||
|
||||
async def _test_llm_function_calling(llm: LLMService):
|
||||
# Create an AsyncMock for the function
|
||||
mock_fetch_weather = AsyncMock()
|
||||
|
||||
llm.register_function(None, mock_fetch_weather)
|
||||
|
||||
messages = [
|
||||
{
|
||||
"role": "system",
|
||||
"content": "You are a helpful assistant who can report the weather in any location in the universe. Respond concisely. Your response will be turned into speech so use only simple words and punctuation.",
|
||||
},
|
||||
{"role": "user", "content": " How is the weather today in San Francisco, California?"},
|
||||
]
|
||||
context = OpenAILLMContext(messages, standard_tools())
|
||||
# This is done by default inside the create_context_aggregator
|
||||
context.set_llm_adapter(llm.get_llm_adapter())
|
||||
|
||||
pipeline = Pipeline([llm])
|
||||
|
||||
frames_to_send = [OpenAILLMContextFrame(context)]
|
||||
await run_test(
|
||||
pipeline,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=None,
|
||||
)
|
||||
|
||||
# Assert that the mock function was called
|
||||
mock_fetch_weather.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.skipif(os.getenv("OPENAI_API_KEY") is None, reason="OPENAI_API_KEY is not set")
|
||||
@pytest.mark.asyncio
|
||||
async def test_unified_function_calling_openai():
|
||||
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
|
||||
# This will fail if an exception is raised
|
||||
await _test_llm_function_calling(llm)
|
||||
|
||||
|
||||
@pytest.mark.skipif(os.getenv("GOOGLE_API_KEY") is None, reason="GOOGLE_API_KEY is not set")
|
||||
@pytest.mark.asyncio
|
||||
async def test_unified_function_calling_gemini():
|
||||
llm = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY"), model="gemini-2.0-flash-001")
|
||||
# This will fail if an exception is raised
|
||||
await _test_llm_function_calling(llm)
|
||||
|
||||
|
||||
@pytest.mark.skipif(os.getenv("ANTHROPIC_API_KEY") is None, reason="ANTHROPIC_API_KEY is not set")
|
||||
@pytest.mark.asyncio
|
||||
async def test_unified_function_calling_anthropic():
|
||||
llm = AnthropicLLMService(
|
||||
api_key=os.getenv("ANTHROPIC_API_KEY"), model="claude-3-5-sonnet-20240620"
|
||||
)
|
||||
# This will fail if an exception is raised
|
||||
await _test_llm_function_calling(llm)
|
||||
@@ -29,9 +29,13 @@ class TestSentenceAggregator(unittest.IsolatedAsyncioTestCase):
|
||||
for word in sentence.split(" "):
|
||||
frames_to_send.append(TextFrame(text=word + " "))
|
||||
|
||||
expected_returned_frames = [TextFrame, TextFrame, TextFrame]
|
||||
expected_down_frames = [TextFrame, TextFrame, TextFrame]
|
||||
|
||||
(received_down, _) = await run_test(aggregator, frames_to_send, expected_returned_frames)
|
||||
(received_down, _) = await run_test(
|
||||
aggregator,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
assert received_down[-3].text == "Hello, world. "
|
||||
assert received_down[-2].text == "How are you? "
|
||||
assert received_down[-1].text == "I am fine! "
|
||||
@@ -55,7 +59,7 @@ class TestGatedAggregator(unittest.IsolatedAsyncioTestCase):
|
||||
LLMFullResponseEndFrame(),
|
||||
]
|
||||
|
||||
expected_returned_frames = [
|
||||
expected_down_frames = [
|
||||
OutputImageRawFrame,
|
||||
LLMFullResponseStartFrame,
|
||||
TextFrame,
|
||||
@@ -66,5 +70,7 @@ class TestGatedAggregator(unittest.IsolatedAsyncioTestCase):
|
||||
]
|
||||
|
||||
(received_down, _) = await run_test(
|
||||
gated_aggregator, frames_to_send, expected_returned_frames
|
||||
gated_aggregator,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
|
||||
738
tests/test_context_aggregators.py
Normal file
738
tests/test_context_aggregators.py
Normal file
@@ -0,0 +1,738 @@
|
||||
#
|
||||
# Copyright (c) 2024-2025 Daily
|
||||
#
|
||||
# SPDX-License-Identifier: BSD 2-Clause License
|
||||
#
|
||||
|
||||
import json
|
||||
import unittest
|
||||
from typing import Any
|
||||
|
||||
import google.ai.generativelanguage as glm
|
||||
|
||||
from pipecat.frames.frames import (
|
||||
EmulateUserStartedSpeakingFrame,
|
||||
EmulateUserStoppedSpeakingFrame,
|
||||
FunctionCallInProgressFrame,
|
||||
FunctionCallResultFrame,
|
||||
FunctionCallResultProperties,
|
||||
InterimTranscriptionFrame,
|
||||
LLMFullResponseEndFrame,
|
||||
LLMFullResponseStartFrame,
|
||||
OpenAILLMContextAssistantTimestampFrame,
|
||||
StartInterruptionFrame,
|
||||
TextFrame,
|
||||
TranscriptionFrame,
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
)
|
||||
from pipecat.processors.aggregators.llm_response import LLMUserContextAggregator
|
||||
from pipecat.processors.aggregators.openai_llm_context import (
|
||||
OpenAILLMContext,
|
||||
OpenAILLMContextFrame,
|
||||
)
|
||||
from pipecat.services.anthropic import (
|
||||
AnthropicAssistantContextAggregator,
|
||||
AnthropicLLMContext,
|
||||
AnthropicUserContextAggregator,
|
||||
)
|
||||
from pipecat.services.google.google import (
|
||||
GoogleAssistantContextAggregator,
|
||||
GoogleLLMContext,
|
||||
GoogleUserContextAggregator,
|
||||
)
|
||||
from pipecat.services.openai import OpenAIAssistantContextAggregator, OpenAIUserContextAggregator
|
||||
from pipecat.tests.utils import SleepFrame, run_test
|
||||
|
||||
AGGREGATION_TIMEOUT = 0.1
|
||||
AGGREGATION_SLEEP = 0.15
|
||||
|
||||
|
||||
class BaseTestUserContextAggregator:
|
||||
CONTEXT_CLASS = None # To be set in subclasses
|
||||
AGGREGATOR_CLASS = None # To be set in subclasses
|
||||
EXPECTED_CONTEXT_FRAMES = [OpenAILLMContextFrame]
|
||||
|
||||
def check_message_content(self, context: OpenAILLMContext, index: int, content: str):
|
||||
assert context.messages[index]["content"] == content
|
||||
|
||||
def check_message_multi_content(
|
||||
self, context: OpenAILLMContext, content_index: int, index: int, content: str
|
||||
):
|
||||
assert context.messages[index]["content"] == content
|
||||
|
||||
async def test_se(self):
|
||||
assert self.CONTEXT_CLASS is not None, "CONTEXT_CLASS must be set in a subclass"
|
||||
assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass"
|
||||
|
||||
context = self.CONTEXT_CLASS()
|
||||
aggregator = self.AGGREGATOR_CLASS(context)
|
||||
frames_to_send = [UserStartedSpeakingFrame(), UserStoppedSpeakingFrame()]
|
||||
expected_down_frames = [UserStartedSpeakingFrame, UserStoppedSpeakingFrame]
|
||||
await run_test(
|
||||
aggregator,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
|
||||
async def test_ste(self):
|
||||
assert self.CONTEXT_CLASS is not None, "CONTEXT_CLASS must be set in a subclass"
|
||||
assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass"
|
||||
|
||||
context = self.CONTEXT_CLASS()
|
||||
aggregator = self.AGGREGATOR_CLASS(context)
|
||||
frames_to_send = [
|
||||
UserStartedSpeakingFrame(),
|
||||
TranscriptionFrame(text="Hello!", user_id="cat", timestamp=""),
|
||||
SleepFrame(),
|
||||
UserStoppedSpeakingFrame(),
|
||||
]
|
||||
expected_down_frames = [
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
*self.EXPECTED_CONTEXT_FRAMES,
|
||||
]
|
||||
await run_test(
|
||||
aggregator,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
self.check_message_content(context, 0, "Hello!")
|
||||
|
||||
async def test_site(self):
|
||||
assert self.CONTEXT_CLASS is not None, "CONTEXT_CLASS must be set in a subclass"
|
||||
assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass"
|
||||
|
||||
context = self.CONTEXT_CLASS()
|
||||
aggregator = self.AGGREGATOR_CLASS(context)
|
||||
frames_to_send = [
|
||||
UserStartedSpeakingFrame(),
|
||||
InterimTranscriptionFrame(text="Hello", user_id="cat", timestamp=""),
|
||||
TranscriptionFrame(text="Hello Pipecat!", user_id="cat", timestamp=""),
|
||||
SleepFrame(),
|
||||
UserStoppedSpeakingFrame(),
|
||||
]
|
||||
expected_down_frames = [
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
*self.EXPECTED_CONTEXT_FRAMES,
|
||||
]
|
||||
await run_test(
|
||||
aggregator,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
self.check_message_content(context, 0, "Hello Pipecat!")
|
||||
|
||||
async def test_st1iest2e(self):
|
||||
assert self.CONTEXT_CLASS is not None, "CONTEXT_CLASS must be set in a subclass"
|
||||
assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass"
|
||||
|
||||
context = self.CONTEXT_CLASS()
|
||||
aggregator = self.AGGREGATOR_CLASS(context)
|
||||
frames_to_send = [
|
||||
UserStartedSpeakingFrame(),
|
||||
TranscriptionFrame(text="Hello Pipecat!", user_id="cat", timestamp=""),
|
||||
InterimTranscriptionFrame(text="How ", user_id="cat", timestamp=""),
|
||||
SleepFrame(),
|
||||
UserStoppedSpeakingFrame(),
|
||||
UserStartedSpeakingFrame(),
|
||||
TranscriptionFrame(text="How are you?", user_id="cat", timestamp=""),
|
||||
SleepFrame(),
|
||||
UserStoppedSpeakingFrame(),
|
||||
]
|
||||
expected_down_frames = [
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
*self.EXPECTED_CONTEXT_FRAMES,
|
||||
]
|
||||
await run_test(
|
||||
aggregator,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
self.check_message_content(context, 0, "Hello Pipecat! How are you?")
|
||||
|
||||
async def test_siet(self):
|
||||
assert self.CONTEXT_CLASS is not None, "CONTEXT_CLASS must be set in a subclass"
|
||||
assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass"
|
||||
|
||||
context = self.CONTEXT_CLASS()
|
||||
aggregator = self.AGGREGATOR_CLASS(context, aggregation_timeout=AGGREGATION_TIMEOUT)
|
||||
frames_to_send = [
|
||||
UserStartedSpeakingFrame(),
|
||||
InterimTranscriptionFrame(text="How ", user_id="cat", timestamp=""),
|
||||
SleepFrame(),
|
||||
UserStoppedSpeakingFrame(),
|
||||
TranscriptionFrame(text="How are you?", user_id="cat", timestamp=""),
|
||||
SleepFrame(sleep=AGGREGATION_SLEEP),
|
||||
]
|
||||
expected_down_frames = [
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
*self.EXPECTED_CONTEXT_FRAMES,
|
||||
]
|
||||
await run_test(
|
||||
aggregator,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
self.check_message_content(context, 0, "How are you?")
|
||||
|
||||
async def test_sieit(self):
|
||||
assert self.CONTEXT_CLASS is not None, "CONTEXT_CLASS must be set in a subclass"
|
||||
assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass"
|
||||
|
||||
context = self.CONTEXT_CLASS()
|
||||
aggregator = self.AGGREGATOR_CLASS(context, aggregation_timeout=AGGREGATION_TIMEOUT)
|
||||
frames_to_send = [
|
||||
UserStartedSpeakingFrame(),
|
||||
InterimTranscriptionFrame(text="How ", user_id="cat", timestamp=""),
|
||||
SleepFrame(),
|
||||
UserStoppedSpeakingFrame(),
|
||||
InterimTranscriptionFrame(text="are you?", user_id="cat", timestamp=""),
|
||||
TranscriptionFrame(text="How are you?", user_id="cat", timestamp=""),
|
||||
SleepFrame(sleep=AGGREGATION_SLEEP),
|
||||
]
|
||||
expected_down_frames = [
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
*self.EXPECTED_CONTEXT_FRAMES,
|
||||
]
|
||||
await run_test(
|
||||
aggregator,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
self.check_message_content(context, 0, "How are you?")
|
||||
|
||||
async def test_set(self):
|
||||
assert self.CONTEXT_CLASS is not None, "CONTEXT_CLASS must be set in a subclass"
|
||||
assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass"
|
||||
|
||||
context = self.CONTEXT_CLASS()
|
||||
aggregator = self.AGGREGATOR_CLASS(context, aggregation_timeout=AGGREGATION_TIMEOUT)
|
||||
frames_to_send = [
|
||||
UserStartedSpeakingFrame(),
|
||||
UserStoppedSpeakingFrame(),
|
||||
TranscriptionFrame(text="How are you?", user_id="cat", timestamp=""),
|
||||
SleepFrame(sleep=AGGREGATION_SLEEP),
|
||||
]
|
||||
expected_down_frames = [
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
*self.EXPECTED_CONTEXT_FRAMES,
|
||||
]
|
||||
await run_test(
|
||||
aggregator,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
self.check_message_content(context, 0, "How are you?")
|
||||
|
||||
async def test_seit(self):
|
||||
assert self.CONTEXT_CLASS is not None, "CONTEXT_CLASS must be set in a subclass"
|
||||
assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass"
|
||||
|
||||
context = self.CONTEXT_CLASS()
|
||||
aggregator = self.AGGREGATOR_CLASS(context, aggregation_timeout=AGGREGATION_TIMEOUT)
|
||||
frames_to_send = [
|
||||
UserStartedSpeakingFrame(),
|
||||
UserStoppedSpeakingFrame(),
|
||||
InterimTranscriptionFrame(text="How ", user_id="cat", timestamp=""),
|
||||
TranscriptionFrame(text="How are you?", user_id="cat", timestamp=""),
|
||||
SleepFrame(sleep=AGGREGATION_SLEEP),
|
||||
]
|
||||
expected_down_frames = [
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
*self.EXPECTED_CONTEXT_FRAMES,
|
||||
]
|
||||
await run_test(
|
||||
aggregator,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
self.check_message_content(context, 0, "How are you?")
|
||||
|
||||
async def test_st1et2(self):
|
||||
assert self.CONTEXT_CLASS is not None, "CONTEXT_CLASS must be set in a subclass"
|
||||
assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass"
|
||||
|
||||
context = self.CONTEXT_CLASS()
|
||||
aggregator = self.AGGREGATOR_CLASS(context, aggregation_timeout=AGGREGATION_TIMEOUT)
|
||||
frames_to_send = [
|
||||
UserStartedSpeakingFrame(),
|
||||
TranscriptionFrame(text="Hello Pipecat!", user_id="cat", timestamp=""),
|
||||
SleepFrame(),
|
||||
UserStoppedSpeakingFrame(),
|
||||
TranscriptionFrame(text="How are you?", user_id="cat", timestamp=""),
|
||||
SleepFrame(sleep=AGGREGATION_SLEEP),
|
||||
]
|
||||
expected_down_frames = [
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
*self.EXPECTED_CONTEXT_FRAMES,
|
||||
*self.EXPECTED_CONTEXT_FRAMES,
|
||||
]
|
||||
await run_test(
|
||||
aggregator,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
self.check_message_multi_content(context, 0, 0, "Hello Pipecat!")
|
||||
self.check_message_multi_content(context, 0, 1, "How are you?")
|
||||
|
||||
async def test_set1t2(self):
|
||||
assert self.CONTEXT_CLASS is not None, "CONTEXT_CLASS must be set in a subclass"
|
||||
assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass"
|
||||
|
||||
context = self.CONTEXT_CLASS()
|
||||
aggregator = self.AGGREGATOR_CLASS(context, aggregation_timeout=AGGREGATION_TIMEOUT)
|
||||
frames_to_send = [
|
||||
UserStartedSpeakingFrame(),
|
||||
UserStoppedSpeakingFrame(),
|
||||
TranscriptionFrame(text="Hello Pipecat!", user_id="cat", timestamp=""),
|
||||
TranscriptionFrame(text="How are you?", user_id="cat", timestamp=""),
|
||||
SleepFrame(sleep=AGGREGATION_SLEEP),
|
||||
]
|
||||
expected_down_frames = [
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
*self.EXPECTED_CONTEXT_FRAMES,
|
||||
]
|
||||
await run_test(
|
||||
aggregator,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
self.check_message_content(context, 0, "Hello Pipecat! How are you?")
|
||||
|
||||
async def test_siet1it2(self):
|
||||
assert self.CONTEXT_CLASS is not None, "CONTEXT_CLASS must be set in a subclass"
|
||||
assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass"
|
||||
|
||||
context = self.CONTEXT_CLASS()
|
||||
aggregator = self.AGGREGATOR_CLASS(context, aggregation_timeout=AGGREGATION_TIMEOUT)
|
||||
frames_to_send = [
|
||||
UserStartedSpeakingFrame(),
|
||||
InterimTranscriptionFrame(text="Hello ", user_id="cat", timestamp=""),
|
||||
SleepFrame(),
|
||||
UserStoppedSpeakingFrame(),
|
||||
TranscriptionFrame(text="Hello Pipecat!", user_id="cat", timestamp=""),
|
||||
InterimTranscriptionFrame(text="How ", user_id="cat", timestamp=""),
|
||||
TranscriptionFrame(text="How are you?", user_id="cat", timestamp=""),
|
||||
SleepFrame(sleep=AGGREGATION_SLEEP),
|
||||
]
|
||||
expected_down_frames = [
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
*self.EXPECTED_CONTEXT_FRAMES,
|
||||
]
|
||||
await run_test(
|
||||
aggregator,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
self.check_message_content(context, 0, "Hello Pipecat! How are you?")
|
||||
|
||||
async def test_t(self):
|
||||
assert self.CONTEXT_CLASS is not None, "CONTEXT_CLASS must be set in a subclass"
|
||||
assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass"
|
||||
|
||||
context = self.CONTEXT_CLASS()
|
||||
aggregator = self.AGGREGATOR_CLASS(context, aggregation_timeout=AGGREGATION_TIMEOUT)
|
||||
frames_to_send = [
|
||||
TranscriptionFrame(text="Hello!", user_id="cat", timestamp=""),
|
||||
SleepFrame(sleep=AGGREGATION_SLEEP),
|
||||
]
|
||||
expected_down_frames = [*self.EXPECTED_CONTEXT_FRAMES]
|
||||
expected_up_frames = [EmulateUserStartedSpeakingFrame, EmulateUserStoppedSpeakingFrame]
|
||||
await run_test(
|
||||
aggregator,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
expected_up_frames=expected_up_frames,
|
||||
)
|
||||
self.check_message_content(context, 0, "Hello!")
|
||||
|
||||
async def test_it(self):
|
||||
assert self.CONTEXT_CLASS is not None, "CONTEXT_CLASS must be set in a subclass"
|
||||
assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass"
|
||||
|
||||
context = self.CONTEXT_CLASS()
|
||||
aggregator = self.AGGREGATOR_CLASS(context, aggregation_timeout=AGGREGATION_TIMEOUT)
|
||||
frames_to_send = [
|
||||
InterimTranscriptionFrame(text="Hello ", user_id="cat", timestamp=""),
|
||||
SleepFrame(),
|
||||
TranscriptionFrame(text="Hello Pipecat!", user_id="cat", timestamp=""),
|
||||
SleepFrame(sleep=AGGREGATION_SLEEP),
|
||||
]
|
||||
expected_down_frames = [*self.EXPECTED_CONTEXT_FRAMES]
|
||||
expected_up_frames = [EmulateUserStartedSpeakingFrame, EmulateUserStoppedSpeakingFrame]
|
||||
await run_test(
|
||||
aggregator,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
expected_up_frames=expected_up_frames,
|
||||
)
|
||||
self.check_message_content(context, 0, "Hello Pipecat!")
|
||||
|
||||
async def test_sie_delay_it(self):
|
||||
assert self.CONTEXT_CLASS is not None, "CONTEXT_CLASS must be set in a subclass"
|
||||
assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass"
|
||||
|
||||
context = self.CONTEXT_CLASS()
|
||||
aggregator = self.AGGREGATOR_CLASS(
|
||||
context,
|
||||
aggregation_timeout=AGGREGATION_TIMEOUT,
|
||||
)
|
||||
frames_to_send = [
|
||||
UserStartedSpeakingFrame(),
|
||||
InterimTranscriptionFrame(text="How ", user_id="cat", timestamp=""),
|
||||
SleepFrame(),
|
||||
UserStoppedSpeakingFrame(),
|
||||
SleepFrame(AGGREGATION_SLEEP),
|
||||
InterimTranscriptionFrame(text="are you?", user_id="cat", timestamp=""),
|
||||
TranscriptionFrame(text="How are you?", user_id="cat", timestamp=""),
|
||||
SleepFrame(sleep=AGGREGATION_SLEEP),
|
||||
]
|
||||
expected_down_frames = [
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
*self.EXPECTED_CONTEXT_FRAMES,
|
||||
]
|
||||
await run_test(
|
||||
aggregator,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
self.check_message_content(context, 0, "How are you?")
|
||||
|
||||
|
||||
class BaseTestAssistantContextAggreagator:
|
||||
CONTEXT_CLASS = None # To be set in subclasses
|
||||
AGGREGATOR_CLASS = None # To be set in subclasses
|
||||
EXPECTED_CONTEXT_FRAMES = None # To be set in subclasses
|
||||
|
||||
def check_message_content(self, context: OpenAILLMContext, index: int, content: str):
|
||||
assert context.messages[index]["content"] == content
|
||||
|
||||
def check_message_multi_content(
|
||||
self, context: OpenAILLMContext, content_index: int, index: int, content: str
|
||||
):
|
||||
assert context.messages[index]["content"] == content
|
||||
|
||||
def check_function_call_result(self, context: OpenAILLMContext, index: int, content: str):
|
||||
assert json.loads(context.messages[index]["content"]) == content
|
||||
|
||||
async def test_empty(self):
|
||||
assert self.CONTEXT_CLASS is not None, "CONTEXT_CLASS must be set in a subclass"
|
||||
assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass"
|
||||
|
||||
context = self.CONTEXT_CLASS()
|
||||
aggregator = self.AGGREGATOR_CLASS(context)
|
||||
frames_to_send = [LLMFullResponseStartFrame(), LLMFullResponseEndFrame()]
|
||||
expected_down_frames = []
|
||||
await run_test(
|
||||
aggregator,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
|
||||
async def test_single_text(self):
|
||||
assert self.CONTEXT_CLASS is not None, "CONTEXT_CLASS must be set in a subclass"
|
||||
assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass"
|
||||
|
||||
context = self.CONTEXT_CLASS()
|
||||
aggregator = self.AGGREGATOR_CLASS(context)
|
||||
frames_to_send = [
|
||||
LLMFullResponseStartFrame(),
|
||||
TextFrame(text="Hello Pipecat!"),
|
||||
LLMFullResponseEndFrame(),
|
||||
]
|
||||
expected_down_frames = [*self.EXPECTED_CONTEXT_FRAMES]
|
||||
await run_test(
|
||||
aggregator,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
self.check_message_content(context, 0, "Hello Pipecat!")
|
||||
|
||||
async def test_multiple_text(self):
|
||||
assert self.CONTEXT_CLASS is not None, "CONTEXT_CLASS must be set in a subclass"
|
||||
assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass"
|
||||
|
||||
context = self.CONTEXT_CLASS()
|
||||
aggregator = self.AGGREGATOR_CLASS(context, expect_stripped_words=False)
|
||||
frames_to_send = [
|
||||
LLMFullResponseStartFrame(),
|
||||
TextFrame(text="Hello "),
|
||||
TextFrame(text="Pipecat. "),
|
||||
TextFrame(text="How are "),
|
||||
TextFrame(text="you?"),
|
||||
LLMFullResponseEndFrame(),
|
||||
]
|
||||
expected_down_frames = [*self.EXPECTED_CONTEXT_FRAMES]
|
||||
await run_test(
|
||||
aggregator,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
self.check_message_content(context, 0, "Hello Pipecat. How are you?")
|
||||
|
||||
async def test_multiple_text_stripped(self):
|
||||
assert self.CONTEXT_CLASS is not None, "CONTEXT_CLASS must be set in a subclass"
|
||||
assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass"
|
||||
|
||||
context = self.CONTEXT_CLASS()
|
||||
aggregator = self.AGGREGATOR_CLASS(context)
|
||||
frames_to_send = [
|
||||
LLMFullResponseStartFrame(),
|
||||
TextFrame(text="Hello"),
|
||||
TextFrame(text="Pipecat."),
|
||||
TextFrame(text="How are"),
|
||||
TextFrame(text="you?"),
|
||||
LLMFullResponseEndFrame(),
|
||||
]
|
||||
expected_down_frames = [*self.EXPECTED_CONTEXT_FRAMES]
|
||||
await run_test(
|
||||
aggregator,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
self.check_message_content(context, 0, "Hello Pipecat. How are you?")
|
||||
|
||||
async def test_multiple_llm_responses(self):
|
||||
assert self.CONTEXT_CLASS is not None, "CONTEXT_CLASS must be set in a subclass"
|
||||
assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass"
|
||||
|
||||
context = self.CONTEXT_CLASS()
|
||||
aggregator = self.AGGREGATOR_CLASS(context, expect_stripped_words=False)
|
||||
frames_to_send = [
|
||||
LLMFullResponseStartFrame(),
|
||||
TextFrame(text="Hello "),
|
||||
TextFrame(text="Pipecat."),
|
||||
LLMFullResponseEndFrame(),
|
||||
LLMFullResponseStartFrame(),
|
||||
TextFrame(text="How are "),
|
||||
TextFrame(text="you?"),
|
||||
LLMFullResponseEndFrame(),
|
||||
]
|
||||
expected_down_frames = [*self.EXPECTED_CONTEXT_FRAMES, *self.EXPECTED_CONTEXT_FRAMES]
|
||||
await run_test(
|
||||
aggregator,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
self.check_message_multi_content(context, 0, 0, "Hello Pipecat.")
|
||||
self.check_message_multi_content(context, 0, 1, "How are you?")
|
||||
|
||||
async def test_multiple_llm_responses_interruption(self):
|
||||
assert self.CONTEXT_CLASS is not None, "CONTEXT_CLASS must be set in a subclass"
|
||||
assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass"
|
||||
|
||||
context = self.CONTEXT_CLASS()
|
||||
aggregator = self.AGGREGATOR_CLASS(context, expect_stripped_words=False)
|
||||
frames_to_send = [
|
||||
LLMFullResponseStartFrame(),
|
||||
TextFrame(text="Hello "),
|
||||
TextFrame(text="Pipecat."),
|
||||
LLMFullResponseEndFrame(),
|
||||
SleepFrame(AGGREGATION_SLEEP),
|
||||
StartInterruptionFrame(),
|
||||
LLMFullResponseStartFrame(),
|
||||
TextFrame(text="How are "),
|
||||
TextFrame(text="you?"),
|
||||
LLMFullResponseEndFrame(),
|
||||
]
|
||||
expected_down_frames = [
|
||||
*self.EXPECTED_CONTEXT_FRAMES,
|
||||
StartInterruptionFrame,
|
||||
*self.EXPECTED_CONTEXT_FRAMES,
|
||||
]
|
||||
await run_test(
|
||||
aggregator,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
self.check_message_multi_content(context, 0, 0, "Hello Pipecat.")
|
||||
self.check_message_multi_content(context, 0, 1, "How are you?")
|
||||
|
||||
async def test_function_call(self):
|
||||
assert self.CONTEXT_CLASS is not None, "CONTEXT_CLASS must be set in a subclass"
|
||||
assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass"
|
||||
|
||||
context = self.CONTEXT_CLASS()
|
||||
aggregator = self.AGGREGATOR_CLASS(context)
|
||||
frames_to_send = [
|
||||
FunctionCallInProgressFrame(
|
||||
function_name="get_weather",
|
||||
tool_call_id="1",
|
||||
arguments={"location": "Los Angeles"},
|
||||
cancel_on_interruption=False,
|
||||
),
|
||||
SleepFrame(),
|
||||
FunctionCallResultFrame(
|
||||
function_name="get_weather",
|
||||
tool_call_id="1",
|
||||
arguments={"location": "Los Angeles"},
|
||||
result={"conditions": "Sunny"},
|
||||
),
|
||||
]
|
||||
expected_down_frames = []
|
||||
await run_test(
|
||||
aggregator,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
self.check_function_call_result(context, -1, {"conditions": "Sunny"})
|
||||
|
||||
async def test_function_call_on_context_updated(self):
|
||||
assert self.CONTEXT_CLASS is not None, "CONTEXT_CLASS must be set in a subclass"
|
||||
assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass"
|
||||
|
||||
context_updated = False
|
||||
|
||||
async def on_context_updated():
|
||||
nonlocal context_updated
|
||||
context_updated = True
|
||||
|
||||
context = self.CONTEXT_CLASS()
|
||||
aggregator = self.AGGREGATOR_CLASS(context)
|
||||
frames_to_send = [
|
||||
FunctionCallInProgressFrame(
|
||||
function_name="get_weather",
|
||||
tool_call_id="1",
|
||||
arguments={"location": "Los Angeles"},
|
||||
cancel_on_interruption=False,
|
||||
),
|
||||
SleepFrame(),
|
||||
FunctionCallResultFrame(
|
||||
function_name="get_weather",
|
||||
tool_call_id="1",
|
||||
arguments={"location": "Los Angeles"},
|
||||
result={"conditions": "Sunny"},
|
||||
properties=FunctionCallResultProperties(on_context_updated=on_context_updated),
|
||||
),
|
||||
SleepFrame(),
|
||||
]
|
||||
expected_down_frames = []
|
||||
await run_test(
|
||||
aggregator,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
self.check_function_call_result(context, -1, {"conditions": "Sunny"})
|
||||
assert context_updated
|
||||
|
||||
|
||||
#
|
||||
# LLMUserContextAggregator
|
||||
#
|
||||
|
||||
|
||||
class TestLLMUserContextAggregator(BaseTestUserContextAggregator, unittest.IsolatedAsyncioTestCase):
|
||||
CONTEXT_CLASS = OpenAILLMContext
|
||||
AGGREGATOR_CLASS = LLMUserContextAggregator
|
||||
|
||||
|
||||
#
|
||||
# OpenAI
|
||||
#
|
||||
|
||||
|
||||
class TestOpenAIUserContextAggregator(
|
||||
BaseTestUserContextAggregator, unittest.IsolatedAsyncioTestCase
|
||||
):
|
||||
CONTEXT_CLASS = OpenAILLMContext
|
||||
AGGREGATOR_CLASS = OpenAIUserContextAggregator
|
||||
|
||||
|
||||
class TestOpenAIAssistantContextAggregator(
|
||||
BaseTestAssistantContextAggreagator, unittest.IsolatedAsyncioTestCase
|
||||
):
|
||||
CONTEXT_CLASS = OpenAILLMContext
|
||||
AGGREGATOR_CLASS = OpenAIAssistantContextAggregator
|
||||
EXPECTED_CONTEXT_FRAMES = [OpenAILLMContextFrame, OpenAILLMContextAssistantTimestampFrame]
|
||||
|
||||
|
||||
#
|
||||
# Anthropic
|
||||
#
|
||||
|
||||
|
||||
class TestAnthropicUserContextAggregator(
|
||||
BaseTestUserContextAggregator, unittest.IsolatedAsyncioTestCase
|
||||
):
|
||||
CONTEXT_CLASS = AnthropicLLMContext
|
||||
AGGREGATOR_CLASS = AnthropicUserContextAggregator
|
||||
|
||||
def check_message_multi_content(
|
||||
self, context: OpenAILLMContext, content_index: int, index: int, content: str
|
||||
):
|
||||
messages = context.messages[content_index]
|
||||
assert messages["content"][index]["text"] == content
|
||||
|
||||
|
||||
class TestAnthropicAssistantContextAggregator(
|
||||
BaseTestAssistantContextAggreagator, unittest.IsolatedAsyncioTestCase
|
||||
):
|
||||
CONTEXT_CLASS = AnthropicLLMContext
|
||||
AGGREGATOR_CLASS = AnthropicAssistantContextAggregator
|
||||
EXPECTED_CONTEXT_FRAMES = [OpenAILLMContextFrame, OpenAILLMContextAssistantTimestampFrame]
|
||||
|
||||
def check_message_multi_content(
|
||||
self, context: OpenAILLMContext, content_index: int, index: int, content: str
|
||||
):
|
||||
messages = context.messages[content_index]
|
||||
assert messages["content"][index]["text"] == content
|
||||
|
||||
def check_function_call_result(self, context: OpenAILLMContext, index: int, content: Any):
|
||||
assert context.messages[index]["content"][0]["content"] == json.dumps(content)
|
||||
|
||||
|
||||
#
|
||||
# Google
|
||||
#
|
||||
|
||||
|
||||
class TestGoogleUserContextAggregator(
|
||||
BaseTestUserContextAggregator, unittest.IsolatedAsyncioTestCase
|
||||
):
|
||||
CONTEXT_CLASS = GoogleLLMContext
|
||||
AGGREGATOR_CLASS = GoogleUserContextAggregator
|
||||
|
||||
def check_message_content(self, context: OpenAILLMContext, index: int, content: str):
|
||||
obj = glm.Content.to_dict(context.messages[index])
|
||||
assert obj["parts"][0]["text"] == content
|
||||
|
||||
def check_message_multi_content(
|
||||
self, context: OpenAILLMContext, content_index: int, index: int, content: str
|
||||
):
|
||||
obj = glm.Content.to_dict(context.messages[index])
|
||||
assert obj["parts"][0]["text"] == content
|
||||
|
||||
|
||||
class TestGoogleAssistantContextAggregator(
|
||||
BaseTestAssistantContextAggreagator, unittest.IsolatedAsyncioTestCase
|
||||
):
|
||||
CONTEXT_CLASS = GoogleLLMContext
|
||||
AGGREGATOR_CLASS = GoogleAssistantContextAggregator
|
||||
EXPECTED_CONTEXT_FRAMES = [OpenAILLMContextFrame, OpenAILLMContextAssistantTimestampFrame]
|
||||
|
||||
def check_message_content(self, context: OpenAILLMContext, index: int, content: str):
|
||||
obj = glm.Content.to_dict(context.messages[index])
|
||||
assert obj["parts"][0]["text"] == content
|
||||
|
||||
def check_message_multi_content(
|
||||
self, context: OpenAILLMContext, content_index: int, index: int, content: str
|
||||
):
|
||||
obj = glm.Content.to_dict(context.messages[index])
|
||||
assert obj["parts"][0]["text"] == content
|
||||
|
||||
def check_function_call_result(self, context: OpenAILLMContext, index: int, content: Any):
|
||||
obj = glm.Content.to_dict(context.messages[index])
|
||||
assert obj["parts"][0]["function_response"]["response"]["value"] == json.dumps(content)
|
||||
@@ -4,7 +4,6 @@
|
||||
# SPDX-License-Identifier: BSD 2-Clause License
|
||||
#
|
||||
|
||||
import asyncio
|
||||
import unittest
|
||||
|
||||
from pipecat.frames.frames import (
|
||||
@@ -19,35 +18,52 @@ from pipecat.processors.filters.frame_filter import FrameFilter
|
||||
from pipecat.processors.filters.function_filter import FunctionFilter
|
||||
from pipecat.processors.filters.identity_filter import IdentityFilter
|
||||
from pipecat.processors.filters.wake_check_filter import WakeCheckFilter
|
||||
from pipecat.tests.utils import EndTestFrame, run_test
|
||||
from pipecat.tests.utils import run_test
|
||||
|
||||
|
||||
class TestIdentifyFilter(unittest.IsolatedAsyncioTestCase):
|
||||
async def test_identity(self):
|
||||
filter = IdentityFilter()
|
||||
frames_to_send = [UserStartedSpeakingFrame(), UserStoppedSpeakingFrame()]
|
||||
expected_returned_frames = [UserStartedSpeakingFrame, UserStoppedSpeakingFrame]
|
||||
await run_test(filter, frames_to_send, expected_returned_frames)
|
||||
expected_down_frames = [UserStartedSpeakingFrame, UserStoppedSpeakingFrame]
|
||||
await run_test(
|
||||
filter,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
|
||||
|
||||
class TestFrameFilter(unittest.IsolatedAsyncioTestCase):
|
||||
async def test_text_frame(self):
|
||||
filter = FrameFilter(types=(TextFrame, EndTestFrame))
|
||||
filter = FrameFilter(types=(TextFrame,))
|
||||
frames_to_send = [TextFrame(text="Hello Pipecat!")]
|
||||
expected_returned_frames = [TextFrame]
|
||||
await run_test(filter, frames_to_send, expected_returned_frames)
|
||||
expected_down_frames = [TextFrame]
|
||||
await run_test(
|
||||
filter,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
|
||||
async def test_end_frame(self):
|
||||
filter = FrameFilter(types=(EndFrame, EndTestFrame))
|
||||
filter = FrameFilter(types=(EndFrame,))
|
||||
frames_to_send = [EndFrame()]
|
||||
expected_returned_frames = [EndFrame]
|
||||
await run_test(filter, frames_to_send, expected_returned_frames)
|
||||
expected_down_frames = [EndFrame]
|
||||
await run_test(
|
||||
filter,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
send_end_frame=False,
|
||||
)
|
||||
|
||||
async def test_system_frame(self):
|
||||
filter = FrameFilter(types=(EndTestFrame,))
|
||||
filter = FrameFilter(types=())
|
||||
frames_to_send = [UserStartedSpeakingFrame()]
|
||||
expected_returned_frames = [UserStartedSpeakingFrame]
|
||||
await run_test(filter, frames_to_send, expected_returned_frames)
|
||||
expected_down_frames = [UserStartedSpeakingFrame]
|
||||
await run_test(
|
||||
filter,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
|
||||
|
||||
class TestFunctionFilter(unittest.IsolatedAsyncioTestCase):
|
||||
@@ -57,8 +73,12 @@ class TestFunctionFilter(unittest.IsolatedAsyncioTestCase):
|
||||
|
||||
filter = FunctionFilter(filter=passthrough)
|
||||
frames_to_send = [TextFrame(text="Hello Pipecat!")]
|
||||
expected_returned_frames = [TextFrame]
|
||||
await run_test(filter, frames_to_send, expected_returned_frames)
|
||||
expected_down_frames = [TextFrame]
|
||||
await run_test(
|
||||
filter,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
|
||||
async def test_no_passthrough(self):
|
||||
async def no_passthrough(frame: Frame):
|
||||
@@ -66,22 +86,24 @@ class TestFunctionFilter(unittest.IsolatedAsyncioTestCase):
|
||||
|
||||
filter = FunctionFilter(filter=no_passthrough)
|
||||
frames_to_send = [TextFrame(text="Hello Pipecat!")]
|
||||
expected_returned_frames = [TextFrame]
|
||||
try:
|
||||
await asyncio.wait_for(
|
||||
run_test(filter, frames_to_send, expected_returned_frames), timeout=0.5
|
||||
)
|
||||
assert False
|
||||
except asyncio.TimeoutError:
|
||||
pass
|
||||
expected_down_frames = []
|
||||
await run_test(
|
||||
filter,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
|
||||
|
||||
class TestWakeCheckFilter(unittest.IsolatedAsyncioTestCase):
|
||||
async def test_no_wake_word(self):
|
||||
filter = WakeCheckFilter(wake_phrases=["Hey, Pipecat"])
|
||||
frames_to_send = [TranscriptionFrame(user_id="test", text="Phrase 1", timestamp="")]
|
||||
expected_returned_frames = []
|
||||
await run_test(filter, frames_to_send, expected_returned_frames)
|
||||
expected_down_frames = []
|
||||
await run_test(
|
||||
filter,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
|
||||
async def test_wake_word(self):
|
||||
filter = WakeCheckFilter(wake_phrases=["Hey, Pipecat"])
|
||||
@@ -89,6 +111,10 @@ class TestWakeCheckFilter(unittest.IsolatedAsyncioTestCase):
|
||||
TranscriptionFrame(user_id="test", text="Hey, Pipecat", timestamp=""),
|
||||
TranscriptionFrame(user_id="test", text="Phrase 1", timestamp=""),
|
||||
]
|
||||
expected_returned_frames = [TranscriptionFrame, TranscriptionFrame]
|
||||
(received_down, _) = await run_test(filter, frames_to_send, expected_returned_frames)
|
||||
expected_down_frames = [TranscriptionFrame, TranscriptionFrame]
|
||||
(received_down, _) = await run_test(
|
||||
filter,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
assert received_down[-1].text == "Phrase 1"
|
||||
|
||||
176
tests/test_function_calling_adapters.py
Normal file
176
tests/test_function_calling_adapters.py
Normal file
@@ -0,0 +1,176 @@
|
||||
#
|
||||
# Copyright (c) 2024–2025, Daily
|
||||
#
|
||||
# SPDX-License-Identifier: BSD 2-Clause License
|
||||
#
|
||||
|
||||
import unittest
|
||||
|
||||
from openai.types.chat import ChatCompletionToolParam
|
||||
|
||||
from pipecat.adapters.schemas.function_schema import FunctionSchema
|
||||
from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema
|
||||
from pipecat.adapters.services.anthropic_adapter import AnthropicLLMAdapter
|
||||
from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter
|
||||
from pipecat.adapters.services.open_ai_adapter import OpenAILLMAdapter
|
||||
from pipecat.adapters.services.open_ai_realtime_adapter import OpenAIRealtimeLLMAdapter
|
||||
|
||||
|
||||
class TestFunctionAdapters(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
"""Sets up a common tools schema for all tests."""
|
||||
function_def = FunctionSchema(
|
||||
name="get_weather",
|
||||
description="Get the weather in a given location",
|
||||
properties={
|
||||
"location": {"type": "string", "description": "The city, e.g. San Francisco"},
|
||||
"format": {
|
||||
"type": "string",
|
||||
"enum": ["celsius", "fahrenheit"],
|
||||
"description": "The temperature unit to use.",
|
||||
},
|
||||
},
|
||||
required=["location", "format"],
|
||||
)
|
||||
self.tools_def = ToolsSchema(standard_tools=[function_def])
|
||||
|
||||
def test_openai_adapter(self):
|
||||
"""Test OpenAI adapter format transformation."""
|
||||
expected = [
|
||||
ChatCompletionToolParam(
|
||||
type="function",
|
||||
function={
|
||||
"name": "get_weather",
|
||||
"description": "Get the weather in a given location",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"location": {
|
||||
"type": "string",
|
||||
"description": "The city, e.g. San Francisco",
|
||||
},
|
||||
"format": {
|
||||
"type": "string",
|
||||
"enum": ["celsius", "fahrenheit"],
|
||||
"description": "The temperature unit to use.",
|
||||
},
|
||||
},
|
||||
"required": ["location", "format"],
|
||||
},
|
||||
},
|
||||
)
|
||||
]
|
||||
assert OpenAILLMAdapter().to_provider_tools_format(self.tools_def) == expected
|
||||
|
||||
def test_anthropic_adapter(self):
|
||||
"""Test Anthropic adapter format transformation."""
|
||||
expected = [
|
||||
{
|
||||
"name": "get_weather",
|
||||
"description": "Get the weather in a given location",
|
||||
"input_schema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"location": {
|
||||
"type": "string",
|
||||
"description": "The city, e.g. San Francisco",
|
||||
},
|
||||
"format": {
|
||||
"type": "string",
|
||||
"enum": ["celsius", "fahrenheit"],
|
||||
"description": "The temperature unit to use.",
|
||||
},
|
||||
},
|
||||
"required": ["location", "format"],
|
||||
},
|
||||
}
|
||||
]
|
||||
assert AnthropicLLMAdapter().to_provider_tools_format(self.tools_def) == expected
|
||||
|
||||
def test_gemini_adapter(self):
|
||||
"""Test Gemini adapter format transformation."""
|
||||
expected = [
|
||||
{
|
||||
"function_declarations": [
|
||||
{
|
||||
"name": "get_weather",
|
||||
"description": "Get the weather in a given location",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"location": {
|
||||
"type": "string",
|
||||
"description": "The city, e.g. San Francisco",
|
||||
},
|
||||
"format": {
|
||||
"type": "string",
|
||||
"enum": ["celsius", "fahrenheit"],
|
||||
"description": "The temperature unit to use.",
|
||||
},
|
||||
},
|
||||
"required": ["location", "format"],
|
||||
},
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
assert GeminiLLMAdapter().to_provider_tools_format(self.tools_def) == expected
|
||||
|
||||
def test_openai_realtime_adapter(self):
|
||||
"""Test Anthropic adapter format transformation."""
|
||||
expected = [
|
||||
{
|
||||
"type": "function",
|
||||
"name": "get_weather",
|
||||
"description": "Get the weather in a given location",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"location": {
|
||||
"type": "string",
|
||||
"description": "The city, e.g. San Francisco",
|
||||
},
|
||||
"format": {
|
||||
"type": "string",
|
||||
"enum": ["celsius", "fahrenheit"],
|
||||
"description": "The temperature unit to use.",
|
||||
},
|
||||
},
|
||||
"required": ["location", "format"],
|
||||
},
|
||||
}
|
||||
]
|
||||
assert OpenAIRealtimeLLMAdapter().to_provider_tools_format(self.tools_def) == expected
|
||||
|
||||
def test_gemini_adapter_with_custom_tools(self):
|
||||
"""Test Gemini adapter format transformation."""
|
||||
search_tool = {"google_search": {}}
|
||||
expected = [
|
||||
{
|
||||
"function_declarations": [
|
||||
{
|
||||
"name": "get_weather",
|
||||
"description": "Get the weather in a given location",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"location": {
|
||||
"type": "string",
|
||||
"description": "The city, e.g. San Francisco",
|
||||
},
|
||||
"format": {
|
||||
"type": "string",
|
||||
"enum": ["celsius", "fahrenheit"],
|
||||
"description": "The temperature unit to use.",
|
||||
},
|
||||
},
|
||||
"required": ["location", "format"],
|
||||
},
|
||||
}
|
||||
]
|
||||
},
|
||||
search_tool,
|
||||
]
|
||||
tools_def = self.tools_def
|
||||
tools_def.custom_tools = {AdapterType.GEMINI: [search_tool]}
|
||||
assert GeminiLLMAdapter().to_provider_tools_format(tools_def) == expected
|
||||
@@ -10,23 +10,22 @@ from langchain.prompts import ChatPromptTemplate
|
||||
from langchain_core.language_models import FakeStreamingListLLM
|
||||
|
||||
from pipecat.frames.frames import (
|
||||
EndFrame,
|
||||
LLMFullResponseEndFrame,
|
||||
LLMFullResponseStartFrame,
|
||||
LLMMessagesFrame,
|
||||
TextFrame,
|
||||
TranscriptionFrame,
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
)
|
||||
from pipecat.pipeline.pipeline import Pipeline
|
||||
from pipecat.pipeline.runner import PipelineRunner
|
||||
from pipecat.pipeline.task import PipelineParams, PipelineTask
|
||||
from pipecat.processors.aggregators.llm_response import (
|
||||
LLMAssistantResponseAggregator,
|
||||
LLMUserResponseAggregator,
|
||||
)
|
||||
from pipecat.processors.frame_processor import FrameProcessor
|
||||
from pipecat.processors.frameworks.langchain import LangchainProcessor
|
||||
from pipecat.tests.utils import SleepFrame, run_test
|
||||
|
||||
|
||||
class TestLangchain(unittest.IsolatedAsyncioTestCase):
|
||||
@@ -64,31 +63,26 @@ class TestLangchain(unittest.IsolatedAsyncioTestCase):
|
||||
self.mock_proc = self.MockProcessor("token_collector")
|
||||
|
||||
tma_in = LLMUserResponseAggregator(messages)
|
||||
tma_out = LLMAssistantResponseAggregator(messages)
|
||||
tma_out = LLMAssistantResponseAggregator(messages, expect_stripped_words=False)
|
||||
|
||||
pipeline = Pipeline(
|
||||
[
|
||||
tma_in,
|
||||
proc,
|
||||
self.mock_proc,
|
||||
tma_out,
|
||||
]
|
||||
pipeline = Pipeline([tma_in, proc, self.mock_proc, tma_out])
|
||||
|
||||
frames_to_send = [
|
||||
UserStartedSpeakingFrame(),
|
||||
TranscriptionFrame(text="Hi World", user_id="user", timestamp="now"),
|
||||
SleepFrame(),
|
||||
UserStoppedSpeakingFrame(),
|
||||
]
|
||||
expected_down_frames = [
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
LLMMessagesFrame,
|
||||
]
|
||||
await run_test(
|
||||
pipeline,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
|
||||
task = PipelineTask(pipeline, PipelineParams(allow_interruptions=False))
|
||||
await task.queue_frames(
|
||||
[
|
||||
UserStartedSpeakingFrame(),
|
||||
TranscriptionFrame(text="Hi World", user_id="user", timestamp="now"),
|
||||
UserStoppedSpeakingFrame(),
|
||||
EndFrame(),
|
||||
]
|
||||
)
|
||||
|
||||
runner = PipelineRunner()
|
||||
await runner.run(task)
|
||||
self.assertEqual("".join(self.mock_proc.token), self.expected_response)
|
||||
# TODO: Address this issue
|
||||
# This next one would fail with:
|
||||
# AssertionError: ' H e l l o d e a r h u m a n' != 'Hello dear human'
|
||||
# self.assertEqual(tma_out.messages[-1]["content"], self.expected_response)
|
||||
self.assertEqual(tma_out.messages[-1]["content"], self.expected_response)
|
||||
|
||||
136
tests/test_llm_response.py
Normal file
136
tests/test_llm_response.py
Normal file
@@ -0,0 +1,136 @@
|
||||
#
|
||||
# Copyright (c) 2024-2025 Daily
|
||||
#
|
||||
# SPDX-License-Identifier: BSD 2-Clause License
|
||||
#
|
||||
|
||||
import unittest
|
||||
|
||||
from pipecat.frames.frames import (
|
||||
LLMFullResponseEndFrame,
|
||||
LLMFullResponseStartFrame,
|
||||
LLMTextFrame,
|
||||
StartInterruptionFrame,
|
||||
)
|
||||
from pipecat.processors.aggregators.llm_response import LLMFullResponseAggregator
|
||||
from pipecat.tests.utils import SleepFrame, run_test
|
||||
|
||||
|
||||
class TestLLMFullResponseAggregator(unittest.IsolatedAsyncioTestCase):
|
||||
async def test_empty(self):
|
||||
completion_ok = False
|
||||
|
||||
aggregator = LLMFullResponseAggregator()
|
||||
|
||||
@aggregator.event_handler("on_completion")
|
||||
async def on_completion(aggregator, completion, completed):
|
||||
nonlocal completion_ok
|
||||
completion_ok = completion == "" and completed
|
||||
|
||||
frames_to_send = [LLMFullResponseStartFrame(), LLMFullResponseEndFrame()]
|
||||
expected_down_frames = [LLMFullResponseStartFrame, LLMFullResponseEndFrame]
|
||||
await run_test(
|
||||
aggregator,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
assert completion_ok
|
||||
|
||||
async def test_simple(self):
|
||||
completion_ok = False
|
||||
|
||||
aggregator = LLMFullResponseAggregator()
|
||||
|
||||
@aggregator.event_handler("on_completion")
|
||||
async def on_completion(aggregator, completion, completed):
|
||||
nonlocal completion_ok
|
||||
completion_ok = completion == "Hello from Pipecat!" and completed
|
||||
|
||||
frames_to_send = [
|
||||
LLMFullResponseStartFrame(),
|
||||
LLMTextFrame("Hello from Pipecat!"),
|
||||
LLMFullResponseEndFrame(),
|
||||
]
|
||||
expected_down_frames = [LLMFullResponseStartFrame, LLMTextFrame, LLMFullResponseEndFrame]
|
||||
await run_test(
|
||||
aggregator,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
assert completion_ok
|
||||
|
||||
async def test_multiple(self):
|
||||
completion_ok = False
|
||||
|
||||
aggregator = LLMFullResponseAggregator()
|
||||
|
||||
@aggregator.event_handler("on_completion")
|
||||
async def on_completion(aggregator, completion, completed):
|
||||
nonlocal completion_ok
|
||||
completion_ok = completion == "Hello from Pipecat!" and completed
|
||||
|
||||
frames_to_send = [
|
||||
LLMFullResponseStartFrame(),
|
||||
LLMTextFrame("Hello "),
|
||||
LLMTextFrame("from "),
|
||||
LLMTextFrame("Pipecat!"),
|
||||
LLMFullResponseEndFrame(),
|
||||
]
|
||||
expected_down_frames = [
|
||||
LLMFullResponseStartFrame,
|
||||
LLMTextFrame,
|
||||
LLMTextFrame,
|
||||
LLMTextFrame,
|
||||
LLMFullResponseEndFrame,
|
||||
]
|
||||
await run_test(
|
||||
aggregator,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
assert completion_ok
|
||||
|
||||
async def test_interruption(self):
|
||||
completion_ok = True
|
||||
|
||||
completion_result = [("Hello ", False), ("Hello there!", True)]
|
||||
completion_index = 0
|
||||
|
||||
aggregator = LLMFullResponseAggregator()
|
||||
|
||||
@aggregator.event_handler("on_completion")
|
||||
async def on_completion(aggregator, completion, completed):
|
||||
nonlocal completion_result, completion_index, completion_ok
|
||||
(completion_expected, completion_completed) = completion_result[completion_index]
|
||||
completion_ok = (
|
||||
completion_ok
|
||||
and completion == completion_expected
|
||||
and completed == completion_completed
|
||||
)
|
||||
completion_index += 1
|
||||
|
||||
frames_to_send = [
|
||||
LLMFullResponseStartFrame(),
|
||||
LLMTextFrame("Hello "),
|
||||
SleepFrame(),
|
||||
StartInterruptionFrame(),
|
||||
LLMFullResponseStartFrame(),
|
||||
LLMTextFrame("Hello "),
|
||||
LLMTextFrame("there!"),
|
||||
LLMFullResponseEndFrame(),
|
||||
]
|
||||
expected_down_frames = [
|
||||
LLMFullResponseStartFrame,
|
||||
LLMTextFrame,
|
||||
StartInterruptionFrame,
|
||||
LLMFullResponseStartFrame,
|
||||
LLMTextFrame,
|
||||
LLMTextFrame,
|
||||
LLMFullResponseEndFrame,
|
||||
]
|
||||
await run_test(
|
||||
aggregator,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
assert completion_ok
|
||||
230
tests/test_markdown_text_filter.py
Normal file
230
tests/test_markdown_text_filter.py
Normal file
@@ -0,0 +1,230 @@
|
||||
#
|
||||
# Copyright (c) 2024-2025 Daily
|
||||
#
|
||||
# SPDX-License-Identifier: BSD 2-Clause License
|
||||
#
|
||||
|
||||
|
||||
import unittest
|
||||
|
||||
from pipecat.utils.text.markdown_text_filter import MarkdownTextFilter
|
||||
|
||||
|
||||
class TestMarkdownTextFilter(unittest.IsolatedAsyncioTestCase):
|
||||
def setUp(self):
|
||||
self.filter = MarkdownTextFilter()
|
||||
|
||||
async def test_basic_markdown_removal(self):
|
||||
"""Test removal of basic Markdown formatting while preserving content."""
|
||||
input_text = """
|
||||
**Bold text** and *italic text*
|
||||
1. Numbered list item
|
||||
- Bullet point
|
||||
Some `inline code` here
|
||||
"""
|
||||
|
||||
expected_text = """
|
||||
Bold text and italic text
|
||||
1. Numbered list item
|
||||
- Bullet point
|
||||
Some inline code here
|
||||
"""
|
||||
|
||||
result = self.filter.filter(input_text)
|
||||
self.assertEqual(result.strip(), expected_text.strip())
|
||||
|
||||
async def test_space_preservation(self):
|
||||
"""Test preservation of leading and trailing spaces (for
|
||||
word-by-word streaming in bot-tts-text).
|
||||
"""
|
||||
input_text = [
|
||||
" Leading spaces",
|
||||
"Trailing spaces ",
|
||||
" Both ends ",
|
||||
" Multiple spaces between words ",
|
||||
]
|
||||
|
||||
for text in input_text:
|
||||
result = self.filter.filter(text)
|
||||
self.assertEqual(
|
||||
len(result), len(text), f"Space preservation failed for: '{text}'\nGot: '{result}'"
|
||||
)
|
||||
# Check if spaces are in the same positions
|
||||
for i, char in enumerate(text):
|
||||
if char == " ":
|
||||
self.assertEqual(
|
||||
result[i], " ", f"Space at position {i} was not preserved in: '{text}'"
|
||||
)
|
||||
|
||||
async def test_repeated_character_removal(self):
|
||||
"""Test removal of repeated character sequences (5 or more)."""
|
||||
test_cases = {
|
||||
"Hello!!!!!World": "HelloWorld", # 5 exclamations removed
|
||||
"Test####ing": "Test####ing", # 4 hashes preserved
|
||||
"Normal text": "Normal text", # No repeated chars
|
||||
"!!!!!": "", # All repeated chars removed
|
||||
"Mixed!!!!!...../////": "Mixed", # Multiple repeated sequences
|
||||
"Text^^^^test": "Text^^^^test", # 4 carets preserved
|
||||
"Text^^^^^test": "Texttest", # 5 carets removed
|
||||
"Dots....here": "Dots....here", # 4 dots preserved
|
||||
"Dots.....here": "Dotshere", # 5 dots removed
|
||||
}
|
||||
|
||||
for input_text, expected in test_cases.items():
|
||||
result = self.filter.filter(input_text)
|
||||
self.assertEqual(
|
||||
result,
|
||||
expected,
|
||||
f"Failed to handle repeated characters in: '{input_text}'\nExpected: '{expected}'\nGot: '{result}'",
|
||||
)
|
||||
|
||||
async def test_numbered_list_preservation(self):
|
||||
"""Test that numbered lists are preserved correctly."""
|
||||
input_text = """1. First item
|
||||
2. Second item
|
||||
3. Third item with **bold**"""
|
||||
|
||||
expected = """1. First item
|
||||
2. Second item
|
||||
3. Third item with bold"""
|
||||
|
||||
result = self.filter.filter(input_text)
|
||||
self.assertEqual(
|
||||
result.strip(),
|
||||
expected.strip(),
|
||||
f"Numbered list preservation failed.\nExpected:\n{expected}\nGot:\n{result}",
|
||||
)
|
||||
|
||||
async def test_html_entity_conversion(self):
|
||||
"""Test conversion of HTML entities to their plain text equivalents."""
|
||||
test_cases = {
|
||||
"This & that": "This & that",
|
||||
"1 < 2": "1 < 2",
|
||||
"2 > 1": "2 > 1",
|
||||
"Line break": "Line break",
|
||||
"Mixed & <entities>": "Mixed & <entities>",
|
||||
}
|
||||
|
||||
for input_text, expected in test_cases.items():
|
||||
result = self.filter.filter(input_text)
|
||||
self.assertEqual(result, expected, f"HTML entity conversion failed for: '{input_text}'")
|
||||
|
||||
async def test_asterisk_removal(self):
|
||||
"""Test removal of Markdown asterisk formatting."""
|
||||
test_cases = {
|
||||
"**bold text**": "bold text", # Double asterisks
|
||||
"*italic text*": "italic text", # Single asterisks
|
||||
"**bold** and *italic*": "bold and italic", # Mixed
|
||||
"multiple**bold**words": "multipleboldwords", # No spaces
|
||||
"edge**cases***here*": "edgecaseshere", # Adjacent asterisks
|
||||
}
|
||||
|
||||
for input_text, expected in test_cases.items():
|
||||
result = self.filter.filter(input_text)
|
||||
self.assertEqual(result, expected, f"Asterisk removal failed for: '{input_text}'")
|
||||
|
||||
async def test_newline_handling(self):
|
||||
"""Test handling of empty and whitespace-only lines."""
|
||||
test_cases = {
|
||||
"Line 1\n\nLine 2": "Line 1\n Line 2", # Empty line becomes space
|
||||
"Line 1\n \nLine 2": "Line 1\n Line 2", # Whitespace line becomes single space
|
||||
"Text\n\n\nMore": "Text\n More", # Multiple empty lines become spaces
|
||||
}
|
||||
|
||||
for input_text, expected in test_cases.items():
|
||||
result = self.filter.filter(input_text)
|
||||
self.assertEqual(
|
||||
result, expected, f"Newline handling failed for:\n{input_text}\nGot:\n{result}"
|
||||
)
|
||||
|
||||
async def test_numbered_list_marker_handling(self):
|
||||
"""Test handling of numbered lists with the special §NUM§ marker."""
|
||||
test_cases = {
|
||||
"1. First\n2. Second": "1. First\n2. Second", # Basic numbered list
|
||||
" 1. Indented": " 1. Indented", # Indented numbered list
|
||||
"1. Item\nText\n2. Item": "1. Item\nText\n2. Item", # Text between items
|
||||
"1.No space": "1.No space", # Not a list item (no space)
|
||||
"12. Large number": "12. Large number", # Multi-digit numbers
|
||||
}
|
||||
|
||||
for input_text, expected in test_cases.items():
|
||||
result = self.filter.filter(input_text)
|
||||
self.assertEqual(
|
||||
result,
|
||||
expected,
|
||||
f"Numbered list handling failed for:\n{input_text}\nGot:\n{result}",
|
||||
)
|
||||
|
||||
async def test_inline_code_handling(self):
|
||||
"""Test handling of inline code with backticks."""
|
||||
test_cases = {
|
||||
"`code`": "code", # Basic inline code
|
||||
"Text `code` more": "Text code more", # Inline code within text
|
||||
"``nested`code``": "nested`code", # Nested backticks
|
||||
"`code1` and `code2`": "code1 and code2", # Multiple inline codes
|
||||
"No``space``between": "Nospacebetween", # No spaces around backticks
|
||||
}
|
||||
|
||||
for input_text, expected in test_cases.items():
|
||||
result = self.filter.filter(input_text)
|
||||
self.assertEqual(result, expected, f"Inline code handling failed for: '{input_text}'")
|
||||
|
||||
async def test_simple_table_removal(self):
|
||||
"""Test removal of a simple markdown table."""
|
||||
filter = MarkdownTextFilter(params=MarkdownTextFilter.InputParams(filter_tables=True))
|
||||
|
||||
input_text = "| Column 1 | Column 2 |\n|----------|----------|\n| Cell 1 | Cell 2 |"
|
||||
|
||||
expected = ""
|
||||
|
||||
result = filter.filter(input_text)
|
||||
self.assertEqual(
|
||||
result.strip(),
|
||||
expected.strip(),
|
||||
f"Simple table removal failed.\nExpected:\n{expected}\nGot:\n{result}",
|
||||
)
|
||||
|
||||
async def test_feature_toggles(self):
|
||||
"""Test enabling/disabling specific filter features."""
|
||||
# Create a filter with all features disabled
|
||||
filter = MarkdownTextFilter(
|
||||
params=MarkdownTextFilter.InputParams(
|
||||
enable_text_filter=False,
|
||||
filter_code=False,
|
||||
filter_tables=False,
|
||||
)
|
||||
)
|
||||
|
||||
# Test with text filtering disabled
|
||||
text_with_markdown = "**bold** and *italic* with `code`"
|
||||
self.assertEqual(
|
||||
filter.filter(text_with_markdown),
|
||||
text_with_markdown,
|
||||
"Disabled filter should not modify text",
|
||||
)
|
||||
|
||||
# Enable just text filtering
|
||||
filter.update_settings({"enable_text_filter": True})
|
||||
self.assertEqual(
|
||||
filter.filter(text_with_markdown),
|
||||
"bold and italic with code",
|
||||
"Enabled filter should remove markdown",
|
||||
)
|
||||
|
||||
async def test_settings_update(self):
|
||||
"""Test that filter settings can be updated at runtime."""
|
||||
filter = MarkdownTextFilter()
|
||||
|
||||
# Initial state - formatting should be removed
|
||||
input_text = "**bold** and *italic*"
|
||||
self.assertEqual(filter.filter(input_text), "bold and italic")
|
||||
|
||||
# Disable text filtering
|
||||
filter.update_settings({"enable_text_filter": False})
|
||||
self.assertEqual(filter.filter(input_text), input_text, "Text filtering should be disabled")
|
||||
|
||||
# Re-enable text filtering
|
||||
filter.update_settings({"enable_text_filter": True})
|
||||
self.assertEqual(
|
||||
filter.filter(input_text), "bold and italic", "Text filtering should be re-enabled"
|
||||
)
|
||||
147
tests/test_pattern_pair_aggregator.py
Normal file
147
tests/test_pattern_pair_aggregator.py
Normal file
@@ -0,0 +1,147 @@
|
||||
#
|
||||
# Copyright (c) 2024-2025 Daily
|
||||
#
|
||||
# SPDX-License-Identifier: BSD 2-Clause License
|
||||
#
|
||||
|
||||
import unittest
|
||||
from unittest.mock import Mock
|
||||
|
||||
from pipecat.utils.text.pattern_pair_aggregator import PatternMatch, PatternPairAggregator
|
||||
|
||||
|
||||
class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
|
||||
def setUp(self):
|
||||
self.aggregator = PatternPairAggregator()
|
||||
self.test_handler = Mock()
|
||||
|
||||
# Add a test pattern
|
||||
self.aggregator.add_pattern_pair(
|
||||
pattern_id="test_pattern",
|
||||
start_pattern="<test>",
|
||||
end_pattern="</test>",
|
||||
remove_match=True,
|
||||
)
|
||||
|
||||
# Register the mock handler
|
||||
self.aggregator.on_pattern_match("test_pattern", self.test_handler)
|
||||
|
||||
async def test_pattern_match_and_removal(self):
|
||||
# First part doesn't complete the pattern
|
||||
result = self.aggregator.aggregate("Hello <test>pattern")
|
||||
self.assertIsNone(result)
|
||||
self.assertEqual(self.aggregator.text, "Hello <test>pattern")
|
||||
|
||||
# Second part completes the pattern and includes an exclamation point
|
||||
result = self.aggregator.aggregate(" content</test>!")
|
||||
|
||||
# Verify the handler was called with correct PatternMatch object
|
||||
self.test_handler.assert_called_once()
|
||||
call_args = self.test_handler.call_args[0][0]
|
||||
self.assertIsInstance(call_args, PatternMatch)
|
||||
self.assertEqual(call_args.pattern_id, "test_pattern")
|
||||
self.assertEqual(call_args.full_match, "<test>pattern content</test>")
|
||||
self.assertEqual(call_args.content, "pattern content")
|
||||
|
||||
# The exclamation point should be treated as a sentence boundary,
|
||||
# so the result should include just text up to and including "!"
|
||||
self.assertEqual(result, "Hello !")
|
||||
|
||||
# Next sentence should be processed separately
|
||||
result = self.aggregator.aggregate(" This is another sentence.")
|
||||
self.assertEqual(result, " This is another sentence.")
|
||||
|
||||
# Buffer should be empty after returning a complete sentence
|
||||
self.assertEqual(self.aggregator.text, "")
|
||||
|
||||
async def test_incomplete_pattern(self):
|
||||
# Add text with incomplete pattern
|
||||
result = self.aggregator.aggregate("Hello <test>pattern content")
|
||||
|
||||
# No complete pattern yet, so nothing should be returned
|
||||
self.assertIsNone(result)
|
||||
|
||||
# The handler should not be called yet
|
||||
self.test_handler.assert_not_called()
|
||||
|
||||
# Buffer should contain the incomplete text
|
||||
self.assertEqual(self.aggregator.text, "Hello <test>pattern content")
|
||||
|
||||
# Reset and confirm buffer is cleared
|
||||
self.aggregator.reset()
|
||||
self.assertEqual(self.aggregator.text, "")
|
||||
|
||||
async def test_multiple_patterns(self):
|
||||
# Set up multiple patterns and handlers
|
||||
voice_handler = Mock()
|
||||
emphasis_handler = Mock()
|
||||
|
||||
self.aggregator.add_pattern_pair(
|
||||
pattern_id="voice", start_pattern="<voice>", end_pattern="</voice>", remove_match=True
|
||||
)
|
||||
|
||||
self.aggregator.add_pattern_pair(
|
||||
pattern_id="emphasis",
|
||||
start_pattern="<em>",
|
||||
end_pattern="</em>",
|
||||
remove_match=False, # Keep emphasis tags
|
||||
)
|
||||
|
||||
self.aggregator.on_pattern_match("voice", voice_handler)
|
||||
self.aggregator.on_pattern_match("emphasis", emphasis_handler)
|
||||
|
||||
# Test with multiple patterns in one text block
|
||||
text = "Hello <voice>female</voice> I am <em>very</em> excited to meet you!"
|
||||
result = self.aggregator.aggregate(text)
|
||||
|
||||
# Both handlers should be called with correct data
|
||||
voice_handler.assert_called_once()
|
||||
voice_match = voice_handler.call_args[0][0]
|
||||
self.assertEqual(voice_match.pattern_id, "voice")
|
||||
self.assertEqual(voice_match.content, "female")
|
||||
|
||||
emphasis_handler.assert_called_once()
|
||||
emphasis_match = emphasis_handler.call_args[0][0]
|
||||
self.assertEqual(emphasis_match.pattern_id, "emphasis")
|
||||
self.assertEqual(emphasis_match.content, "very")
|
||||
|
||||
# Voice pattern should be removed, emphasis pattern should remain
|
||||
self.assertEqual(result, "Hello I am <em>very</em> excited to meet you!")
|
||||
|
||||
# Buffer should be empty
|
||||
self.assertEqual(self.aggregator.text, "")
|
||||
|
||||
async def test_handle_interruption(self):
|
||||
# Start with incomplete pattern
|
||||
result = self.aggregator.aggregate("Hello <test>pattern")
|
||||
self.assertIsNone(result)
|
||||
|
||||
# Simulate interruption
|
||||
self.aggregator.handle_interruption()
|
||||
|
||||
# Buffer should be cleared
|
||||
self.assertEqual(self.aggregator.text, "")
|
||||
|
||||
# Handler should not have been called
|
||||
self.test_handler.assert_not_called()
|
||||
|
||||
async def test_pattern_across_sentences(self):
|
||||
# Test pattern that spans multiple sentences
|
||||
result = self.aggregator.aggregate("Hello <test>This is sentence one.")
|
||||
|
||||
# First sentence contains start of pattern but no end, so no complete pattern yet
|
||||
self.assertIsNone(result)
|
||||
|
||||
# Add second part with pattern end
|
||||
result = self.aggregator.aggregate(" This is sentence two.</test> Final sentence.")
|
||||
|
||||
# Handler should be called with entire content
|
||||
self.test_handler.assert_called_once()
|
||||
call_args = self.test_handler.call_args[0][0]
|
||||
self.assertEqual(call_args.content, "This is sentence one. This is sentence two.")
|
||||
|
||||
# Pattern should be removed, resulting in text with sentences merged
|
||||
self.assertEqual(result, "Hello Final sentence.")
|
||||
|
||||
# Buffer should be empty
|
||||
self.assertEqual(self.aggregator.text, "")
|
||||
@@ -5,14 +5,15 @@
|
||||
#
|
||||
|
||||
import asyncio
|
||||
import time
|
||||
import unittest
|
||||
|
||||
from pipecat.frames.frames import EndFrame, HeartbeatFrame, TextFrame
|
||||
from pipecat.frames.frames import EndFrame, HeartbeatFrame, StartFrame, TextFrame
|
||||
from pipecat.pipeline.parallel_pipeline import ParallelPipeline
|
||||
from pipecat.pipeline.pipeline import Pipeline
|
||||
from pipecat.pipeline.task import PipelineParams, PipelineTask
|
||||
from pipecat.processors.filters.identity_filter import IdentityFilter
|
||||
from pipecat.processors.frame_processor import FrameProcessor
|
||||
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
|
||||
from pipecat.tests.utils import HeartbeatsObserver, run_test
|
||||
|
||||
|
||||
@@ -21,8 +22,12 @@ class TestPipeline(unittest.IsolatedAsyncioTestCase):
|
||||
pipeline = Pipeline([IdentityFilter()])
|
||||
|
||||
frames_to_send = [TextFrame(text="Hello from Pipecat!")]
|
||||
expected_returned_frames = [TextFrame]
|
||||
await run_test(pipeline, frames_to_send, expected_returned_frames)
|
||||
expected_down_frames = [TextFrame]
|
||||
await run_test(
|
||||
pipeline,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
|
||||
async def test_pipeline_multiple(self):
|
||||
identity1 = IdentityFilter()
|
||||
@@ -32,8 +37,26 @@ class TestPipeline(unittest.IsolatedAsyncioTestCase):
|
||||
pipeline = Pipeline([identity1, identity2, identity3])
|
||||
|
||||
frames_to_send = [TextFrame(text="Hello from Pipecat!")]
|
||||
expected_returned_frames = [TextFrame]
|
||||
await run_test(pipeline, frames_to_send, expected_returned_frames)
|
||||
expected_down_frames = [TextFrame]
|
||||
await run_test(
|
||||
pipeline,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
|
||||
async def test_pipeline_start_metadata(self):
|
||||
pipeline = Pipeline([IdentityFilter()])
|
||||
|
||||
frames_to_send = []
|
||||
expected_down_frames = [StartFrame]
|
||||
(received_down, _) = await run_test(
|
||||
pipeline,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
ignore_start=False,
|
||||
start_metadata={"foo": "bar"},
|
||||
)
|
||||
assert "foo" in received_down[-1].metadata
|
||||
|
||||
|
||||
class TestParallelPipeline(unittest.IsolatedAsyncioTestCase):
|
||||
@@ -41,16 +64,24 @@ class TestParallelPipeline(unittest.IsolatedAsyncioTestCase):
|
||||
pipeline = ParallelPipeline([IdentityFilter()])
|
||||
|
||||
frames_to_send = [TextFrame(text="Hello from Pipecat!")]
|
||||
expected_returned_frames = [TextFrame]
|
||||
await run_test(pipeline, frames_to_send, expected_returned_frames)
|
||||
expected_down_frames = [TextFrame]
|
||||
await run_test(
|
||||
pipeline,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
|
||||
async def test_parallel_multiple(self):
|
||||
"""Should only passthrough one instance of TextFrame."""
|
||||
pipeline = ParallelPipeline([IdentityFilter()], [IdentityFilter()])
|
||||
|
||||
frames_to_send = [TextFrame(text="Hello from Pipecat!")]
|
||||
expected_returned_frames = [TextFrame]
|
||||
await run_test(pipeline, frames_to_send, expected_returned_frames)
|
||||
expected_down_frames = [TextFrame]
|
||||
await run_test(
|
||||
pipeline,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
|
||||
|
||||
class TestPipelineTask(unittest.IsolatedAsyncioTestCase):
|
||||
@@ -64,6 +95,42 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase):
|
||||
await task.run()
|
||||
assert task.has_finished()
|
||||
|
||||
async def test_task_event_handlers(self):
|
||||
upstream_received = False
|
||||
downstream_received = False
|
||||
|
||||
identity = IdentityFilter()
|
||||
pipeline = Pipeline([identity])
|
||||
task = PipelineTask(pipeline, cancel_on_idle_timeout=False)
|
||||
task.set_event_loop(asyncio.get_event_loop())
|
||||
task.set_reached_upstream_filter((TextFrame,))
|
||||
task.set_reached_downstream_filter((TextFrame,))
|
||||
|
||||
@task.event_handler("on_frame_reached_upstream")
|
||||
async def on_frame_reached_upstream(task, frame):
|
||||
nonlocal upstream_received
|
||||
if isinstance(frame, TextFrame) and frame.text == "Hello Upstream!":
|
||||
upstream_received = True
|
||||
|
||||
@task.event_handler("on_frame_reached_downstream")
|
||||
async def on_frame_reached_downstream(task, frame):
|
||||
nonlocal downstream_received
|
||||
if isinstance(frame, TextFrame) and frame.text == "Hello Downstream!":
|
||||
downstream_received = True
|
||||
await identity.push_frame(
|
||||
TextFrame(text="Hello Upstream!"), FrameDirection.UPSTREAM
|
||||
)
|
||||
|
||||
await task.queue_frame(TextFrame(text="Hello Downstream!"))
|
||||
|
||||
try:
|
||||
await asyncio.wait_for(asyncio.shield(task.run()), timeout=1.0)
|
||||
except asyncio.TimeoutError:
|
||||
pass
|
||||
|
||||
assert upstream_received
|
||||
assert downstream_received
|
||||
|
||||
async def test_task_heartbeats(self):
|
||||
heartbeats_counter = 0
|
||||
|
||||
@@ -79,8 +146,11 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase):
|
||||
task = PipelineTask(
|
||||
pipeline,
|
||||
params=PipelineParams(
|
||||
enable_heartbeats=True, heartbeats_period_secs=0.2, observers=[heartbeats_observer]
|
||||
enable_heartbeats=True,
|
||||
heartbeats_period_secs=0.2,
|
||||
),
|
||||
observers=[heartbeats_observer],
|
||||
cancel_on_idle_timeout=False,
|
||||
)
|
||||
task.set_event_loop(asyncio.get_event_loop())
|
||||
|
||||
@@ -88,7 +158,90 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase):
|
||||
|
||||
await task.queue_frame(TextFrame(text="Hello!"))
|
||||
try:
|
||||
await asyncio.wait_for(task.run(), timeout=1.0)
|
||||
await asyncio.wait_for(asyncio.shield(task.run()), timeout=1.0)
|
||||
except asyncio.TimeoutError:
|
||||
pass
|
||||
assert heartbeats_counter == expected_heartbeats
|
||||
|
||||
async def test_idle_task(self):
|
||||
identity = IdentityFilter()
|
||||
pipeline = Pipeline([identity])
|
||||
task = PipelineTask(pipeline, idle_timeout_secs=0.2)
|
||||
task.set_event_loop(asyncio.get_event_loop())
|
||||
await task.run()
|
||||
assert True
|
||||
|
||||
async def test_no_idle_task(self):
|
||||
identity = IdentityFilter()
|
||||
pipeline = Pipeline([identity])
|
||||
task = PipelineTask(pipeline, idle_timeout_secs=0.2, cancel_on_idle_timeout=False)
|
||||
task.set_event_loop(asyncio.get_event_loop())
|
||||
try:
|
||||
await asyncio.wait_for(asyncio.shield(task.run()), timeout=0.3)
|
||||
except asyncio.TimeoutError:
|
||||
assert True
|
||||
else:
|
||||
assert False
|
||||
|
||||
async def test_idle_task_heartbeats(self):
|
||||
identity = IdentityFilter()
|
||||
pipeline = Pipeline([identity])
|
||||
task = PipelineTask(
|
||||
pipeline,
|
||||
params=PipelineParams(
|
||||
enable_heartbeats=True,
|
||||
heartbeats_period_secs=0.1,
|
||||
),
|
||||
idle_timeout_secs=0.3,
|
||||
)
|
||||
task.set_event_loop(asyncio.get_event_loop())
|
||||
await task.run()
|
||||
assert True
|
||||
|
||||
async def test_idle_task_event_handler(self):
|
||||
identity = IdentityFilter()
|
||||
pipeline = Pipeline([identity])
|
||||
task = PipelineTask(pipeline, idle_timeout_secs=0.2, cancel_on_idle_timeout=False)
|
||||
task.set_event_loop(asyncio.get_event_loop())
|
||||
|
||||
idle_timeout = False
|
||||
|
||||
@task.event_handler("on_idle_timeout")
|
||||
async def on_idle_timeout(task: PipelineTask):
|
||||
nonlocal idle_timeout
|
||||
idle_timeout = True
|
||||
await task.cancel()
|
||||
|
||||
await task.run()
|
||||
assert True
|
||||
|
||||
async def test_idle_task_frames(self):
|
||||
idle_timeout_secs = 0.2
|
||||
sleep_time_secs = idle_timeout_secs / 2
|
||||
|
||||
identity = IdentityFilter()
|
||||
pipeline = Pipeline([identity])
|
||||
task = PipelineTask(
|
||||
pipeline,
|
||||
idle_timeout_secs=idle_timeout_secs,
|
||||
idle_timeout_frames=(TextFrame,),
|
||||
)
|
||||
task.set_event_loop(asyncio.get_event_loop())
|
||||
|
||||
async def delayed_frames():
|
||||
await asyncio.sleep(sleep_time_secs)
|
||||
await task.queue_frame(TextFrame("Hello Pipecat!"))
|
||||
await asyncio.sleep(sleep_time_secs)
|
||||
await task.queue_frame(TextFrame("Hello Pipecat!"))
|
||||
await asyncio.sleep(sleep_time_secs)
|
||||
await task.queue_frame(TextFrame("Hello Pipecat!"))
|
||||
|
||||
start_time = time.time()
|
||||
|
||||
tasks = {asyncio.create_task(task.run()), asyncio.create_task(delayed_frames())}
|
||||
|
||||
await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
|
||||
|
||||
diff_time = time.time() - start_time
|
||||
|
||||
self.assertGreater(diff_time, sleep_time_secs * 3)
|
||||
|
||||
29
tests/test_simple_text_aggregator.py
Normal file
29
tests/test_simple_text_aggregator.py
Normal file
@@ -0,0 +1,29 @@
|
||||
#
|
||||
# Copyright (c) 2024-2025 Daily
|
||||
#
|
||||
# SPDX-License-Identifier: BSD 2-Clause License
|
||||
#
|
||||
|
||||
import unittest
|
||||
|
||||
from pipecat.utils.text.simple_text_aggregator import SimpleTextAggregator
|
||||
|
||||
|
||||
class TestSimpleTextAggregator(unittest.IsolatedAsyncioTestCase):
|
||||
def setUp(self):
|
||||
self.aggregator = SimpleTextAggregator()
|
||||
|
||||
async def test_reset_aggregations(self):
|
||||
assert self.aggregator.aggregate("Hello ") == None
|
||||
assert self.aggregator.text == "Hello "
|
||||
self.aggregator.reset()
|
||||
assert self.aggregator.text == ""
|
||||
|
||||
async def test_simple_sentence(self):
|
||||
assert self.aggregator.aggregate("Hello ") == None
|
||||
assert self.aggregator.aggregate("Pipecat!") == "Hello Pipecat!"
|
||||
assert self.aggregator.text == ""
|
||||
|
||||
async def test_multiple_sentences(self):
|
||||
assert self.aggregator.aggregate("Hello Pipecat! How are ") == "Hello Pipecat!"
|
||||
assert self.aggregator.aggregate("you?") == " How are you?"
|
||||
54
tests/test_skip_tags_aggregator.py
Normal file
54
tests/test_skip_tags_aggregator.py
Normal file
@@ -0,0 +1,54 @@
|
||||
#
|
||||
# Copyright (c) 2024-2025 Daily
|
||||
#
|
||||
# SPDX-License-Identifier: BSD 2-Clause License
|
||||
#
|
||||
|
||||
import unittest
|
||||
|
||||
from pipecat.utils.text.skip_tags_aggregator import SkipTagsAggregator
|
||||
|
||||
|
||||
class TestSkipTagsAggregator(unittest.IsolatedAsyncioTestCase):
|
||||
def setUp(self):
|
||||
self.aggregator = SkipTagsAggregator([("<spell>", "</spell>")])
|
||||
|
||||
async def test_no_tags(self):
|
||||
self.aggregator.reset()
|
||||
|
||||
# No tags involved, aggregate at end of sentence.
|
||||
result = self.aggregator.aggregate("Hello Pipecat!")
|
||||
self.assertEqual(result, "Hello Pipecat!")
|
||||
self.assertEqual(self.aggregator.text, "")
|
||||
|
||||
async def test_basic_tags(self):
|
||||
self.aggregator.reset()
|
||||
|
||||
# Tags involved, avoid aggregation during tags.
|
||||
result = self.aggregator.aggregate("My email is <spell>foo@pipecat.ai</spell>.")
|
||||
self.assertEqual(result, "My email is <spell>foo@pipecat.ai</spell>.")
|
||||
self.assertEqual(self.aggregator.text, "")
|
||||
|
||||
async def test_streaming_tags(self):
|
||||
self.aggregator.reset()
|
||||
|
||||
# Tags involved, stream small chunk of texts.
|
||||
result = self.aggregator.aggregate("My email is <sp")
|
||||
self.assertIsNone(result)
|
||||
self.assertEqual(self.aggregator.text, "My email is <sp")
|
||||
|
||||
result = self.aggregator.aggregate("ell>foo.")
|
||||
self.assertIsNone(result)
|
||||
self.assertEqual(self.aggregator.text, "My email is <spell>foo.")
|
||||
|
||||
result = self.aggregator.aggregate("bar@pipecat.")
|
||||
self.assertIsNone(result)
|
||||
self.assertEqual(self.aggregator.text, "My email is <spell>foo.bar@pipecat.")
|
||||
|
||||
result = self.aggregator.aggregate("ai</spe")
|
||||
self.assertIsNone(result)
|
||||
self.assertEqual(self.aggregator.text, "My email is <spell>foo.bar@pipecat.ai</spe")
|
||||
|
||||
result = self.aggregator.aggregate("ll>.")
|
||||
self.assertEqual(result, "My email is <spell>foo.bar@pipecat.ai</spell>.")
|
||||
self.assertEqual(self.aggregator.text, "")
|
||||
248
tests/test_stt_mute_filter.py
Normal file
248
tests/test_stt_mute_filter.py
Normal file
@@ -0,0 +1,248 @@
|
||||
#
|
||||
# Copyright (c) 2024-2025 Daily
|
||||
#
|
||||
# SPDX-License-Identifier: BSD 2-Clause License
|
||||
#
|
||||
|
||||
import unittest
|
||||
|
||||
from pipecat.frames.frames import (
|
||||
BotStartedSpeakingFrame,
|
||||
BotStoppedSpeakingFrame,
|
||||
FunctionCallInProgressFrame,
|
||||
FunctionCallResultFrame,
|
||||
InputAudioRawFrame,
|
||||
STTMuteFrame,
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
)
|
||||
from pipecat.processors.filters.stt_mute_filter import STTMuteConfig, STTMuteFilter, STTMuteStrategy
|
||||
from pipecat.tests.utils import SleepFrame, run_test
|
||||
|
||||
|
||||
class TestSTTMuteFilter(unittest.IsolatedAsyncioTestCase):
|
||||
async def test_first_speech_strategy(self):
|
||||
filter = STTMuteFilter(config=STTMuteConfig(strategies={STTMuteStrategy.FIRST_SPEECH}))
|
||||
|
||||
frames_to_send = [
|
||||
BotStartedSpeakingFrame(), # First bot speech starts
|
||||
UserStartedSpeakingFrame(), # Should be suppressed
|
||||
InputAudioRawFrame(
|
||||
audio=b"", sample_rate=16000, num_channels=1
|
||||
), # Should be suppressed
|
||||
UserStoppedSpeakingFrame(), # Should be suppressed
|
||||
BotStoppedSpeakingFrame(), # First bot speech ends
|
||||
BotStartedSpeakingFrame(), # Second bot speech
|
||||
UserStartedSpeakingFrame(), # Should pass through
|
||||
InputAudioRawFrame(audio=b"", sample_rate=16000, num_channels=1), # Should pass through
|
||||
UserStoppedSpeakingFrame(), # Should pass through
|
||||
BotStoppedSpeakingFrame(),
|
||||
]
|
||||
|
||||
expected_returned_frames = [
|
||||
BotStartedSpeakingFrame,
|
||||
STTMuteFrame, # mute=True
|
||||
BotStoppedSpeakingFrame,
|
||||
STTMuteFrame, # mute=False
|
||||
BotStartedSpeakingFrame,
|
||||
UserStartedSpeakingFrame, # Now passes through
|
||||
InputAudioRawFrame, # Now passes through
|
||||
UserStoppedSpeakingFrame, # Now passes through
|
||||
BotStoppedSpeakingFrame,
|
||||
]
|
||||
|
||||
await run_test(
|
||||
filter,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_returned_frames,
|
||||
)
|
||||
|
||||
async def test_always_strategy(self):
|
||||
filter = STTMuteFilter(config=STTMuteConfig(strategies={STTMuteStrategy.ALWAYS}))
|
||||
|
||||
frames_to_send = [
|
||||
BotStartedSpeakingFrame(), # First speech starts
|
||||
UserStartedSpeakingFrame(), # Should be suppressed
|
||||
InputAudioRawFrame(
|
||||
audio=b"", sample_rate=16000, num_channels=1
|
||||
), # Should be suppressed
|
||||
UserStoppedSpeakingFrame(), # Should be suppressed
|
||||
BotStoppedSpeakingFrame(), # First speech ends
|
||||
UserStartedSpeakingFrame(), # Should pass through
|
||||
InputAudioRawFrame(audio=b"", sample_rate=16000, num_channels=1), # Should pass through
|
||||
UserStoppedSpeakingFrame(), # Should pass through
|
||||
BotStartedSpeakingFrame(), # Second speech starts
|
||||
UserStartedSpeakingFrame(), # Should be suppressed again
|
||||
InputAudioRawFrame(
|
||||
audio=b"", sample_rate=16000, num_channels=1
|
||||
), # Should be suppressed again
|
||||
UserStoppedSpeakingFrame(), # Should be suppressed again
|
||||
BotStoppedSpeakingFrame(), # Second speech ends
|
||||
]
|
||||
|
||||
expected_returned_frames = [
|
||||
BotStartedSpeakingFrame,
|
||||
STTMuteFrame, # mute=True
|
||||
BotStoppedSpeakingFrame,
|
||||
STTMuteFrame, # mute=False
|
||||
UserStartedSpeakingFrame,
|
||||
InputAudioRawFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
BotStartedSpeakingFrame,
|
||||
STTMuteFrame, # mute=True
|
||||
BotStoppedSpeakingFrame,
|
||||
STTMuteFrame, # mute=False
|
||||
]
|
||||
|
||||
await run_test(
|
||||
filter,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_returned_frames,
|
||||
)
|
||||
|
||||
# TODO: Revisit once we figure out how to test SystemFrames and DataFrames
|
||||
# async def test_function_call_strategy(self):
|
||||
# filter = STTMuteFilter(config=STTMuteConfig(strategies={STTMuteStrategy.FUNCTION_CALL}))
|
||||
|
||||
# frames_to_send = [
|
||||
# UserStartedSpeakingFrame(), # Should pass through initially
|
||||
# UserStoppedSpeakingFrame(),
|
||||
# FunctionCallInProgressFrame(
|
||||
# function_name="get_weather",
|
||||
# tool_call_id="call_123",
|
||||
# arguments='{"location": "San Francisco"}',
|
||||
# ), # Start function call
|
||||
# UserStartedSpeakingFrame(), # Should be suppressed
|
||||
# UserStoppedSpeakingFrame(), # Should be suppressed
|
||||
# FunctionCallResultFrame(
|
||||
# function_name="get_weather",
|
||||
# tool_call_id="call_123",
|
||||
# arguments='{"location": "San Francisco"}',
|
||||
# result={"temperature": 22},
|
||||
# ), # End function call
|
||||
# UserStartedSpeakingFrame(), # Should pass through again
|
||||
# UserStoppedSpeakingFrame(),
|
||||
# ]
|
||||
|
||||
# expected_returned_frames = [
|
||||
# UserStartedSpeakingFrame,
|
||||
# UserStoppedSpeakingFrame,
|
||||
# FunctionCallInProgressFrame,
|
||||
# STTMuteFrame, # mute=True
|
||||
# FunctionCallResultFrame,
|
||||
# STTMuteFrame, # mute=False
|
||||
# UserStartedSpeakingFrame,
|
||||
# UserStoppedSpeakingFrame,
|
||||
# ]
|
||||
|
||||
# await run_test(
|
||||
# filter,
|
||||
# frames_to_send=frames_to_send,
|
||||
# expected_down_frames=expected_returned_frames,
|
||||
# )
|
||||
|
||||
async def test_mute_until_first_bot_complete_strategy(self):
|
||||
filter = STTMuteFilter(
|
||||
config=STTMuteConfig(strategies={STTMuteStrategy.MUTE_UNTIL_FIRST_BOT_COMPLETE})
|
||||
)
|
||||
|
||||
frames_to_send = [
|
||||
UserStartedSpeakingFrame(), # Should be suppressed (starts muted)
|
||||
InputAudioRawFrame(
|
||||
audio=b"", sample_rate=16000, num_channels=1
|
||||
), # Should be suppressed
|
||||
UserStoppedSpeakingFrame(), # Should be suppressed
|
||||
BotStartedSpeakingFrame(), # First bot speech
|
||||
UserStartedSpeakingFrame(), # Should be suppressed
|
||||
InputAudioRawFrame(
|
||||
audio=b"", sample_rate=16000, num_channels=1
|
||||
), # Should be suppressed
|
||||
UserStoppedSpeakingFrame(), # Should be suppressed
|
||||
BotStoppedSpeakingFrame(), # First speech ends, unmutes
|
||||
UserStartedSpeakingFrame(), # Should pass through
|
||||
InputAudioRawFrame(audio=b"", sample_rate=16000, num_channels=1), # Should pass through
|
||||
UserStoppedSpeakingFrame(), # Should pass through
|
||||
BotStartedSpeakingFrame(), # Second speech
|
||||
UserStartedSpeakingFrame(), # Should pass through
|
||||
InputAudioRawFrame(audio=b"", sample_rate=16000, num_channels=1), # Should pass through
|
||||
UserStoppedSpeakingFrame(), # Should pass through
|
||||
BotStoppedSpeakingFrame(),
|
||||
]
|
||||
|
||||
expected_returned_frames = [
|
||||
STTMuteFrame, # mute=True after first speech
|
||||
BotStartedSpeakingFrame,
|
||||
BotStoppedSpeakingFrame,
|
||||
STTMuteFrame, # mute=False after first speech
|
||||
UserStartedSpeakingFrame,
|
||||
InputAudioRawFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
BotStartedSpeakingFrame,
|
||||
UserStartedSpeakingFrame,
|
||||
InputAudioRawFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
BotStoppedSpeakingFrame,
|
||||
]
|
||||
|
||||
await run_test(
|
||||
filter,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_returned_frames,
|
||||
)
|
||||
|
||||
async def test_incompatible_strategies(self):
|
||||
with self.assertRaises(ValueError):
|
||||
STTMuteFilter(
|
||||
config=STTMuteConfig(
|
||||
strategies={
|
||||
STTMuteStrategy.FIRST_SPEECH,
|
||||
STTMuteStrategy.MUTE_UNTIL_FIRST_BOT_COMPLETE,
|
||||
}
|
||||
)
|
||||
)
|
||||
|
||||
async def test_custom_strategy(self):
|
||||
async def custom_mute_logic(processor: STTMuteFilter) -> bool:
|
||||
return processor._bot_is_speaking
|
||||
|
||||
filter = STTMuteFilter(
|
||||
config=STTMuteConfig(
|
||||
strategies={STTMuteStrategy.CUSTOM},
|
||||
should_mute_callback=custom_mute_logic,
|
||||
)
|
||||
)
|
||||
|
||||
frames_to_send = [
|
||||
UserStartedSpeakingFrame(), # Should pass through
|
||||
InputAudioRawFrame(audio=b"", sample_rate=16000, num_channels=1), # Should pass through
|
||||
UserStoppedSpeakingFrame(), # Should pass through
|
||||
BotStartedSpeakingFrame(), # Bot starts speaking
|
||||
UserStartedSpeakingFrame(), # Should be suppressed
|
||||
InputAudioRawFrame(
|
||||
audio=b"", sample_rate=16000, num_channels=1
|
||||
), # Should be suppressed
|
||||
UserStoppedSpeakingFrame(), # Should be suppressed
|
||||
BotStoppedSpeakingFrame(), # Bot stops speaking
|
||||
UserStartedSpeakingFrame(), # Should pass through
|
||||
InputAudioRawFrame(audio=b"", sample_rate=16000, num_channels=1), # Should pass through
|
||||
UserStoppedSpeakingFrame(), # Should pass through
|
||||
]
|
||||
|
||||
expected_returned_frames = [
|
||||
UserStartedSpeakingFrame,
|
||||
InputAudioRawFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
BotStartedSpeakingFrame,
|
||||
STTMuteFrame, # mute=True
|
||||
BotStoppedSpeakingFrame,
|
||||
STTMuteFrame, # mute=False
|
||||
UserStartedSpeakingFrame,
|
||||
InputAudioRawFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
]
|
||||
|
||||
await run_test(
|
||||
filter,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_returned_frames,
|
||||
)
|
||||
480
tests/test_transcript_processor.py
Normal file
480
tests/test_transcript_processor.py
Normal file
@@ -0,0 +1,480 @@
|
||||
#
|
||||
# Copyright (c) 2024-2025 Daily
|
||||
#
|
||||
# SPDX-License-Identifier: BSD 2-Clause License
|
||||
#
|
||||
|
||||
|
||||
import asyncio
|
||||
import unittest
|
||||
from datetime import datetime, timezone
|
||||
from typing import List, Tuple, cast
|
||||
|
||||
from pipecat.frames.frames import (
|
||||
BotStartedSpeakingFrame,
|
||||
BotStoppedSpeakingFrame,
|
||||
CancelFrame,
|
||||
StartInterruptionFrame,
|
||||
TranscriptionFrame,
|
||||
TranscriptionMessage,
|
||||
TranscriptionUpdateFrame,
|
||||
TTSTextFrame,
|
||||
)
|
||||
from pipecat.processors.transcript_processor import (
|
||||
AssistantTranscriptProcessor,
|
||||
UserTranscriptProcessor,
|
||||
)
|
||||
from pipecat.tests.utils import SleepFrame, run_test
|
||||
|
||||
|
||||
class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
|
||||
"""Tests for UserTranscriptProcessor"""
|
||||
|
||||
async def test_basic_transcription(self):
|
||||
"""Test basic transcription frame processing"""
|
||||
# Create processor
|
||||
processor = UserTranscriptProcessor()
|
||||
|
||||
# Create test timestamp
|
||||
timestamp = datetime.now(timezone.utc).isoformat()
|
||||
|
||||
# Create frames to send
|
||||
frames_to_send = [
|
||||
TranscriptionFrame(text="Hello, world!", user_id="test_user", timestamp=timestamp)
|
||||
]
|
||||
|
||||
# Expected frames downstream - note the order:
|
||||
# 1. TranscriptionUpdateFrame (processor emits the update first)
|
||||
# 2. TranscriptionFrame (original frame is passed through)
|
||||
expected_down_frames = [TranscriptionUpdateFrame, TranscriptionFrame]
|
||||
|
||||
# Run test
|
||||
received_frames, _ = await run_test(
|
||||
processor,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
|
||||
# Verify the content of the TranscriptionUpdateFrame
|
||||
update_frame = cast(
|
||||
TranscriptionUpdateFrame, received_frames[0]
|
||||
) # Note: now checking first frame
|
||||
self.assertIsInstance(update_frame, TranscriptionUpdateFrame)
|
||||
self.assertEqual(len(update_frame.messages), 1)
|
||||
message = update_frame.messages[0]
|
||||
self.assertEqual(message.role, "user")
|
||||
self.assertEqual(message.content, "Hello, world!")
|
||||
self.assertEqual(message.timestamp, timestamp)
|
||||
|
||||
async def test_event_handler(self):
|
||||
"""Test that event handlers are called with transcript updates"""
|
||||
# Create processor
|
||||
processor = UserTranscriptProcessor()
|
||||
|
||||
# Track received updates
|
||||
received_updates: List[TranscriptionMessage] = []
|
||||
|
||||
# Register event handler
|
||||
@processor.event_handler("on_transcript_update")
|
||||
async def handle_update(proc, frame: TranscriptionUpdateFrame):
|
||||
received_updates.extend(frame.messages)
|
||||
|
||||
# Create test data
|
||||
timestamp = datetime.now(timezone.utc).isoformat()
|
||||
frames_to_send = [
|
||||
TranscriptionFrame(text="First message", user_id="test_user", timestamp=timestamp),
|
||||
TranscriptionFrame(text="Second message", user_id="test_user", timestamp=timestamp),
|
||||
]
|
||||
|
||||
expected_down_frames = [
|
||||
TranscriptionUpdateFrame,
|
||||
TranscriptionFrame, # First message
|
||||
TranscriptionUpdateFrame,
|
||||
TranscriptionFrame, # Second message
|
||||
]
|
||||
|
||||
# Run test
|
||||
await run_test(
|
||||
processor,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
|
||||
# Verify event handler received updates
|
||||
self.assertEqual(len(received_updates), 2)
|
||||
|
||||
# Check first message
|
||||
self.assertEqual(received_updates[0].role, "user")
|
||||
self.assertEqual(received_updates[0].content, "First message")
|
||||
self.assertEqual(received_updates[0].timestamp, timestamp)
|
||||
|
||||
# Check second message
|
||||
self.assertEqual(received_updates[1].role, "user")
|
||||
self.assertEqual(received_updates[1].content, "Second message")
|
||||
self.assertEqual(received_updates[1].timestamp, timestamp)
|
||||
|
||||
async def test_text_aggregation(self):
|
||||
"""Test that TTSTextFrames are properly aggregated into a single message"""
|
||||
# Create processor
|
||||
processor = AssistantTranscriptProcessor()
|
||||
|
||||
# Track received updates
|
||||
received_updates: List[TranscriptionUpdateFrame] = []
|
||||
|
||||
@processor.event_handler("on_transcript_update")
|
||||
async def handle_update(proc, frame: TranscriptionUpdateFrame):
|
||||
received_updates.append(frame)
|
||||
|
||||
# Create test frames simulating bot speaking multiple text chunks
|
||||
frames_to_send = [
|
||||
BotStartedSpeakingFrame(),
|
||||
SleepFrame(sleep=0.1), # Wait for StartedSpeaking to process
|
||||
TTSTextFrame(text="Hello"),
|
||||
TTSTextFrame(text="world!"),
|
||||
TTSTextFrame(text="How"),
|
||||
TTSTextFrame(text="are"),
|
||||
TTSTextFrame(text="you?"),
|
||||
SleepFrame(sleep=0.1), # Wait for text frames to queue
|
||||
BotStoppedSpeakingFrame(),
|
||||
]
|
||||
|
||||
# Expected order:
|
||||
# 1. BotStartedSpeakingFrame (system frame, immediate)
|
||||
# 2. All queued TTSTextFrames
|
||||
# 3. BotStoppedSpeakingFrame (system frame, immediate)
|
||||
# 4. TranscriptionUpdateFrame (after aggregation)
|
||||
expected_down_frames = [
|
||||
BotStartedSpeakingFrame,
|
||||
TTSTextFrame,
|
||||
TTSTextFrame,
|
||||
TTSTextFrame,
|
||||
TTSTextFrame,
|
||||
TTSTextFrame,
|
||||
BotStoppedSpeakingFrame,
|
||||
TranscriptionUpdateFrame,
|
||||
]
|
||||
|
||||
# Run test
|
||||
received_frames, _ = await run_test(
|
||||
processor,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
|
||||
# Verify update was received
|
||||
self.assertEqual(len(received_updates), 1)
|
||||
|
||||
# Get the update frame
|
||||
update_frame = received_updates[0]
|
||||
|
||||
# Should have one aggregated message
|
||||
self.assertEqual(len(update_frame.messages), 1)
|
||||
|
||||
message = update_frame.messages[0]
|
||||
self.assertEqual(message.role, "assistant")
|
||||
self.assertEqual(message.content, "Hello world! How are you?")
|
||||
|
||||
# Verify timestamp exists
|
||||
self.assertIsNotNone(message.timestamp)
|
||||
|
||||
# All frames should be passed through in order, with update at end
|
||||
downstream_update = cast(TranscriptionUpdateFrame, received_frames[-1])
|
||||
self.assertEqual(downstream_update.messages[0].content, "Hello world! How are you?")
|
||||
|
||||
async def test_empty_text_handling(self):
|
||||
"""Test that empty messages are not emitted"""
|
||||
processor = AssistantTranscriptProcessor()
|
||||
|
||||
received_updates: List[TranscriptionUpdateFrame] = []
|
||||
|
||||
@processor.event_handler("on_transcript_update")
|
||||
async def handle_update(proc, frame: TranscriptionUpdateFrame):
|
||||
received_updates.append(frame)
|
||||
|
||||
frames_to_send = [
|
||||
BotStartedSpeakingFrame(),
|
||||
SleepFrame(sleep=0.1),
|
||||
TTSTextFrame(text=""), # Empty text
|
||||
TTSTextFrame(text=" "), # Just whitespace
|
||||
TTSTextFrame(text="\n"), # Just newline
|
||||
BotStoppedSpeakingFrame(),
|
||||
# Pipeline ends here; run_test will automatically send EndFrame
|
||||
]
|
||||
|
||||
# From our earlier tests, we know BotStoppedSpeakingFrame comes before TTSTextFrames
|
||||
expected_down_frames = [
|
||||
BotStartedSpeakingFrame,
|
||||
BotStoppedSpeakingFrame,
|
||||
TTSTextFrame, # empty
|
||||
TTSTextFrame, # whitespace
|
||||
TTSTextFrame, # newline
|
||||
# No TranscriptionUpdateFrame since content is empty after stripping
|
||||
]
|
||||
|
||||
await run_test(
|
||||
processor,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
|
||||
self.assertEqual(len(received_updates), 0, "No updates should be emitted for empty content")
|
||||
|
||||
async def test_interruption_handling(self):
|
||||
"""Test that messages are properly captured when bot is interrupted"""
|
||||
processor = AssistantTranscriptProcessor()
|
||||
|
||||
# Track received updates
|
||||
received_updates: List[TranscriptionUpdateFrame] = []
|
||||
|
||||
@processor.event_handler("on_transcript_update")
|
||||
async def handle_update(proc, frame: TranscriptionUpdateFrame):
|
||||
received_updates.append(frame)
|
||||
|
||||
# Simulate bot being interrupted mid-sentence
|
||||
frames_to_send = [
|
||||
BotStartedSpeakingFrame(),
|
||||
SleepFrame(sleep=0.1),
|
||||
TTSTextFrame(text="Hello"),
|
||||
TTSTextFrame(text="world!"),
|
||||
SleepFrame(sleep=0.1),
|
||||
StartInterruptionFrame(), # User interrupts here
|
||||
BotStartedSpeakingFrame(),
|
||||
SleepFrame(sleep=0.1),
|
||||
TTSTextFrame(text="New"),
|
||||
TTSTextFrame(text="response"),
|
||||
SleepFrame(sleep=0.1),
|
||||
BotStoppedSpeakingFrame(),
|
||||
]
|
||||
|
||||
# Actual order of frames:
|
||||
expected_down_frames = [
|
||||
BotStartedSpeakingFrame,
|
||||
TTSTextFrame, # "Hello"
|
||||
TTSTextFrame, # "world!"
|
||||
TranscriptionUpdateFrame, # First message (emitted due to interruption)
|
||||
StartInterruptionFrame, # Interruption frame comes after the update
|
||||
BotStartedSpeakingFrame,
|
||||
TTSTextFrame, # "New"
|
||||
TTSTextFrame, # "response"
|
||||
BotStoppedSpeakingFrame,
|
||||
TranscriptionUpdateFrame, # Second message
|
||||
]
|
||||
|
||||
# Run test
|
||||
received_frames, _ = await run_test(
|
||||
processor,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
|
||||
# Should have received two updates
|
||||
self.assertEqual(len(received_updates), 2)
|
||||
|
||||
# First update should be interrupted message
|
||||
first_message = received_updates[0].messages[0]
|
||||
self.assertEqual(first_message.role, "assistant")
|
||||
self.assertEqual(first_message.content, "Hello world!")
|
||||
self.assertIsNotNone(first_message.timestamp)
|
||||
|
||||
# Second update should be new response
|
||||
second_message = received_updates[1].messages[0]
|
||||
self.assertEqual(second_message.role, "assistant")
|
||||
self.assertEqual(second_message.content, "New response")
|
||||
self.assertIsNotNone(second_message.timestamp)
|
||||
|
||||
# Verify timestamps are different
|
||||
self.assertNotEqual(first_message.timestamp, second_message.timestamp)
|
||||
|
||||
async def test_end_frame_handling(self):
|
||||
"""Test that final messages are captured when pipeline ends normally"""
|
||||
processor = AssistantTranscriptProcessor()
|
||||
|
||||
received_updates: List[TranscriptionUpdateFrame] = []
|
||||
|
||||
@processor.event_handler("on_transcript_update")
|
||||
async def handle_update(proc, frame: TranscriptionUpdateFrame):
|
||||
received_updates.append(frame)
|
||||
|
||||
frames_to_send = [
|
||||
BotStartedSpeakingFrame(),
|
||||
SleepFrame(sleep=0.1),
|
||||
TTSTextFrame(text="Hello"),
|
||||
TTSTextFrame(text="world"),
|
||||
# Pipeline ends here; run_test will automatically send EndFrame
|
||||
]
|
||||
|
||||
expected_down_frames = [
|
||||
BotStartedSpeakingFrame,
|
||||
TTSTextFrame,
|
||||
TTSTextFrame,
|
||||
TranscriptionUpdateFrame, # Final message emitted due to EndFrame
|
||||
]
|
||||
|
||||
# Run test - EndFrame will be sent automatically
|
||||
received_frames, _ = await run_test(
|
||||
processor,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
|
||||
self.assertEqual(len(received_updates), 1)
|
||||
message = received_updates[0].messages[0]
|
||||
self.assertEqual(message.role, "assistant")
|
||||
self.assertEqual(message.content, "Hello world")
|
||||
|
||||
async def test_cancel_frame_handling(self):
|
||||
"""Test that messages are properly captured when pipeline is cancelled"""
|
||||
processor = AssistantTranscriptProcessor()
|
||||
|
||||
# Track updates with timestamps to verify order
|
||||
received_updates: List[Tuple[str, float]] = []
|
||||
|
||||
@processor.event_handler("on_transcript_update")
|
||||
async def handle_update(proc, frame: TranscriptionUpdateFrame):
|
||||
# Record message content and time received
|
||||
received_updates.append((frame.messages[0].content, asyncio.get_event_loop().time()))
|
||||
|
||||
frames_to_send = [
|
||||
BotStartedSpeakingFrame(),
|
||||
SleepFrame(sleep=0.1),
|
||||
TTSTextFrame(text="Hello"),
|
||||
TTSTextFrame(text="world"),
|
||||
SleepFrame(sleep=0.1), # Ensure messages are processed
|
||||
CancelFrame(),
|
||||
]
|
||||
|
||||
# We don't need to verify frame order, just that CancelFrame triggers message emission
|
||||
expected_down_frames = [
|
||||
BotStartedSpeakingFrame,
|
||||
TTSTextFrame,
|
||||
TTSTextFrame,
|
||||
CancelFrame,
|
||||
]
|
||||
|
||||
await run_test(
|
||||
processor,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
send_end_frame=False,
|
||||
)
|
||||
|
||||
# Verify that we received an update
|
||||
self.assertEqual(len(received_updates), 1, "Should receive one update before cancellation")
|
||||
content, _ = received_updates[0]
|
||||
self.assertEqual(content, "Hello world")
|
||||
|
||||
async def test_transcript_processor_factory(self):
|
||||
"""Test that factory properly manages processors and event handlers"""
|
||||
from pipecat.processors.transcript_processor import TranscriptProcessor
|
||||
|
||||
factory = TranscriptProcessor()
|
||||
received_updates: List[TranscriptionMessage] = []
|
||||
|
||||
# Register handler with factory
|
||||
@factory.event_handler("on_transcript_update")
|
||||
async def handle_update(proc, frame: TranscriptionUpdateFrame):
|
||||
received_updates.extend(frame.messages)
|
||||
|
||||
# Get processors and verify they're reused
|
||||
user_proc1 = factory.user()
|
||||
user_proc2 = factory.user()
|
||||
self.assertIs(user_proc1, user_proc2, "User processor should be reused")
|
||||
|
||||
asst_proc1 = factory.assistant()
|
||||
asst_proc2 = factory.assistant()
|
||||
self.assertIs(asst_proc1, asst_proc2, "Assistant processor should be reused")
|
||||
|
||||
# Test user processor
|
||||
timestamp = datetime.now(timezone.utc).isoformat()
|
||||
frames_to_send = [
|
||||
TranscriptionFrame(text="User message", user_id="user1", timestamp=timestamp)
|
||||
]
|
||||
|
||||
await run_test(
|
||||
user_proc1,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=[TranscriptionUpdateFrame, TranscriptionFrame],
|
||||
)
|
||||
|
||||
# Test assistant processor
|
||||
frames_to_send = [
|
||||
BotStartedSpeakingFrame(),
|
||||
SleepFrame(sleep=0.1),
|
||||
TTSTextFrame(text="Assistant"),
|
||||
TTSTextFrame(text="message"),
|
||||
BotStoppedSpeakingFrame(),
|
||||
]
|
||||
|
||||
# The actual order we see in the output:
|
||||
await run_test(
|
||||
asst_proc1,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=[
|
||||
BotStartedSpeakingFrame,
|
||||
BotStoppedSpeakingFrame,
|
||||
TTSTextFrame,
|
||||
TTSTextFrame,
|
||||
TranscriptionUpdateFrame,
|
||||
],
|
||||
)
|
||||
|
||||
# Verify both processors triggered the same handler
|
||||
self.assertEqual(len(received_updates), 2)
|
||||
self.assertEqual(received_updates[0].role, "user")
|
||||
self.assertEqual(received_updates[0].content, "User message")
|
||||
self.assertEqual(received_updates[1].role, "assistant")
|
||||
self.assertEqual(received_updates[1].content, "Assistant message")
|
||||
|
||||
async def test_text_fragments_with_spaces(self):
|
||||
"""Test aggregating text fragments with various spacing patterns"""
|
||||
processor = AssistantTranscriptProcessor()
|
||||
|
||||
# Track received updates
|
||||
received_updates = []
|
||||
|
||||
@processor.event_handler("on_transcript_update")
|
||||
async def handle_update(proc, frame: TranscriptionUpdateFrame):
|
||||
received_updates.append(frame)
|
||||
|
||||
# Test the specific pattern shared
|
||||
frames_to_send = [
|
||||
BotStartedSpeakingFrame(),
|
||||
SleepFrame(sleep=0.1),
|
||||
TTSTextFrame(text="Hello"),
|
||||
TTSTextFrame(text=" there"),
|
||||
TTSTextFrame(text="!"),
|
||||
TTSTextFrame(text=" How"),
|
||||
TTSTextFrame(text="'s"),
|
||||
TTSTextFrame(text=" it"),
|
||||
TTSTextFrame(text=" going"),
|
||||
TTSTextFrame(text="?"),
|
||||
BotStoppedSpeakingFrame(),
|
||||
]
|
||||
|
||||
expected_down_frames = [
|
||||
BotStartedSpeakingFrame,
|
||||
BotStoppedSpeakingFrame,
|
||||
TTSTextFrame,
|
||||
TTSTextFrame,
|
||||
TTSTextFrame,
|
||||
TTSTextFrame,
|
||||
TTSTextFrame,
|
||||
TTSTextFrame,
|
||||
TTSTextFrame,
|
||||
TTSTextFrame,
|
||||
TranscriptionUpdateFrame,
|
||||
]
|
||||
|
||||
# Run test
|
||||
received_frames, _ = await run_test(
|
||||
processor,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
|
||||
# Verify result
|
||||
self.assertEqual(len(received_updates), 1)
|
||||
message = received_updates[0].messages[0]
|
||||
self.assertEqual(message.role, "assistant")
|
||||
# Should be properly joined without extra spaces
|
||||
self.assertEqual(message.content, "Hello there! How's it going?")
|
||||
216
tests/test_user_idle_processor.py
Normal file
216
tests/test_user_idle_processor.py
Normal file
@@ -0,0 +1,216 @@
|
||||
#
|
||||
# Copyright (c) 2024-2025 Daily
|
||||
#
|
||||
# SPDX-License-Identifier: BSD 2-Clause License
|
||||
#
|
||||
|
||||
import asyncio
|
||||
import unittest
|
||||
|
||||
from pipecat.frames.frames import (
|
||||
BotSpeakingFrame,
|
||||
BotStartedSpeakingFrame,
|
||||
BotStoppedSpeakingFrame,
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
)
|
||||
from pipecat.processors.user_idle_processor import UserIdleProcessor
|
||||
from pipecat.tests.utils import SleepFrame, run_test
|
||||
|
||||
|
||||
class TestUserIdleProcessor(unittest.IsolatedAsyncioTestCase):
|
||||
async def test_basic_idle_detection(self):
|
||||
"""Test that idle callback is triggered after timeout when user stops speaking."""
|
||||
callback_called = asyncio.Event()
|
||||
|
||||
async def idle_callback(processor: UserIdleProcessor) -> None:
|
||||
callback_called.set()
|
||||
|
||||
# Create processor with a short timeout for testing
|
||||
processor = UserIdleProcessor(callback=idle_callback, timeout=0.1) # 100ms timeout
|
||||
|
||||
frames_to_send = [
|
||||
# Start conversation
|
||||
UserStartedSpeakingFrame(),
|
||||
UserStoppedSpeakingFrame(),
|
||||
# Wait 200ms - double the idle timeout to ensure it triggers
|
||||
SleepFrame(sleep=0.2),
|
||||
]
|
||||
|
||||
expected_down_frames = [
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
]
|
||||
|
||||
await run_test(
|
||||
processor,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
|
||||
assert callback_called.is_set(), "Idle callback was not called"
|
||||
|
||||
async def test_active_listening_resets_idle(self):
|
||||
"""Test that bot speaking frames reset the idle timer because user is actively listening."""
|
||||
callback_called = asyncio.Event()
|
||||
|
||||
async def idle_callback(processor: UserIdleProcessor) -> None:
|
||||
callback_called.set()
|
||||
|
||||
processor = UserIdleProcessor(callback=idle_callback, timeout=0.2)
|
||||
|
||||
frames_to_send = [
|
||||
# Start conversation
|
||||
UserStartedSpeakingFrame(),
|
||||
UserStoppedSpeakingFrame(),
|
||||
# Wait almost long enough for idle timeout
|
||||
SleepFrame(sleep=0.1),
|
||||
# Bot speaking frame should reset idle timer
|
||||
BotSpeakingFrame(),
|
||||
# Wait almost long enough for idle timeout again
|
||||
SleepFrame(sleep=0.1),
|
||||
# Another bot speaking frame resets timer again
|
||||
BotSpeakingFrame(),
|
||||
]
|
||||
|
||||
expected_down_frames = [
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
BotSpeakingFrame,
|
||||
BotSpeakingFrame,
|
||||
]
|
||||
|
||||
await run_test(
|
||||
processor,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
|
||||
assert not callback_called.is_set(), (
|
||||
"Idle callback was called even though bot speaking frames reset the timer"
|
||||
)
|
||||
|
||||
async def test_idle_retry_callback(self):
|
||||
"""Test that retry count increases until user activity resets it."""
|
||||
retry_counts = []
|
||||
|
||||
async def retry_callback(processor: UserIdleProcessor, retry_count: int) -> bool:
|
||||
retry_counts.append(retry_count)
|
||||
return True # Keep monitoring for idle events
|
||||
|
||||
processor = UserIdleProcessor(callback=retry_callback, timeout=0.4)
|
||||
|
||||
frames_to_send = [
|
||||
# Start conversation
|
||||
UserStartedSpeakingFrame(),
|
||||
UserStoppedSpeakingFrame(),
|
||||
# Wait for first idle timeout (count=1)
|
||||
SleepFrame(sleep=0.5),
|
||||
# Wait for second idle timeout (count=2)
|
||||
SleepFrame(sleep=0.5),
|
||||
# User activity resets the count
|
||||
UserStartedSpeakingFrame(),
|
||||
UserStoppedSpeakingFrame(),
|
||||
# Wait for new idle timeout (count should be 1 again)
|
||||
SleepFrame(sleep=0.5),
|
||||
]
|
||||
|
||||
expected_down_frames = [
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
]
|
||||
|
||||
await run_test(
|
||||
processor,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
|
||||
assert retry_counts == [1, 2, 1], f"Expected retry counts [1, 2, 1], got {retry_counts}"
|
||||
|
||||
async def test_idle_monitoring_stops_on_false_return(self):
|
||||
"""Test that idle monitoring stops when callback returns False."""
|
||||
retry_counts = []
|
||||
|
||||
async def retry_callback(processor: UserIdleProcessor, retry_count: int) -> bool:
|
||||
retry_counts.append(retry_count)
|
||||
return retry_count < 2 # Stop after second retry
|
||||
|
||||
processor = UserIdleProcessor(callback=retry_callback, timeout=0.4)
|
||||
|
||||
frames_to_send = [
|
||||
UserStartedSpeakingFrame(),
|
||||
UserStoppedSpeakingFrame(),
|
||||
SleepFrame(sleep=0.5), # First retry (count=1, returns True)
|
||||
SleepFrame(sleep=0.5), # Second retry (count=2, returns False)
|
||||
SleepFrame(sleep=0.5), # Should not trigger callback
|
||||
]
|
||||
|
||||
expected_down_frames = [
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
]
|
||||
|
||||
await run_test(
|
||||
processor,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
|
||||
assert retry_counts == [1, 2], f"Expected retry counts [1, 2], got {retry_counts}"
|
||||
|
||||
async def test_no_idle_before_conversation(self):
|
||||
"""Test that idle monitoring doesn't start before first conversation activity."""
|
||||
callback_called = asyncio.Event()
|
||||
|
||||
async def idle_callback(processor: UserIdleProcessor) -> None:
|
||||
callback_called.set()
|
||||
|
||||
processor = UserIdleProcessor(callback=idle_callback, timeout=0.1)
|
||||
|
||||
frames_to_send = [
|
||||
SleepFrame(sleep=0.2), # Should not trigger callback
|
||||
# No conversation activity yet
|
||||
]
|
||||
|
||||
expected_down_frames = []
|
||||
|
||||
await run_test(
|
||||
processor,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
|
||||
assert not callback_called.is_set(), "Idle callback was called before conversation started"
|
||||
|
||||
async def test_idle_starts_with_bot_speech(self):
|
||||
"""Test that monitoring starts with bot speaking frames, not just user speech."""
|
||||
callback_called = asyncio.Event()
|
||||
|
||||
async def idle_callback(processor: UserIdleProcessor) -> None:
|
||||
callback_called.set()
|
||||
|
||||
processor = UserIdleProcessor(callback=idle_callback, timeout=0.1)
|
||||
|
||||
frames_to_send = [
|
||||
BotStartedSpeakingFrame(),
|
||||
BotSpeakingFrame(),
|
||||
BotStoppedSpeakingFrame(),
|
||||
SleepFrame(sleep=0.2),
|
||||
]
|
||||
|
||||
expected_down_frames = [
|
||||
BotStartedSpeakingFrame,
|
||||
BotSpeakingFrame,
|
||||
BotStoppedSpeakingFrame,
|
||||
]
|
||||
|
||||
await run_test(
|
||||
processor,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
)
|
||||
|
||||
assert callback_called.is_set(), "Idle callback not called after bot speech"
|
||||
34
tests/test_utils_network.py
Normal file
34
tests/test_utils_network.py
Normal file
@@ -0,0 +1,34 @@
|
||||
#
|
||||
# Copyright (c) 2024-2025 Daily
|
||||
#
|
||||
# SPDX-License-Identifier: BSD 2-Clause License
|
||||
#
|
||||
|
||||
import unittest
|
||||
|
||||
from pipecat.utils.network import exponential_backoff_time
|
||||
|
||||
|
||||
class TestUtilsNetwork(unittest.IsolatedAsyncioTestCase):
|
||||
async def test_exponential_backoff_time(self):
|
||||
# min_wait=4, max_wait=10, multiplier=1
|
||||
assert exponential_backoff_time(attempt=1, min_wait=4, max_wait=10, multiplier=1) == 4
|
||||
assert exponential_backoff_time(attempt=2, min_wait=4, max_wait=10, multiplier=1) == 4
|
||||
assert exponential_backoff_time(attempt=3, min_wait=4, max_wait=10, multiplier=1) == 4
|
||||
assert exponential_backoff_time(attempt=4, min_wait=4, max_wait=10, multiplier=1) == 8
|
||||
assert exponential_backoff_time(attempt=5, min_wait=4, max_wait=10, multiplier=1) == 10
|
||||
assert exponential_backoff_time(attempt=6, min_wait=4, max_wait=10, multiplier=1) == 10
|
||||
# min_wait=1, max_wait=10, multiplier=1
|
||||
assert exponential_backoff_time(attempt=1, min_wait=1, max_wait=10, multiplier=1) == 1
|
||||
assert exponential_backoff_time(attempt=2, min_wait=1, max_wait=10, multiplier=1) == 2
|
||||
assert exponential_backoff_time(attempt=3, min_wait=1, max_wait=10, multiplier=1) == 4
|
||||
assert exponential_backoff_time(attempt=4, min_wait=1, max_wait=10, multiplier=1) == 8
|
||||
assert exponential_backoff_time(attempt=5, min_wait=1, max_wait=10, multiplier=1) == 10
|
||||
assert exponential_backoff_time(attempt=6, min_wait=1, max_wait=10, multiplier=1) == 10
|
||||
# min_wait=1, max_wait=20, multiplier=2
|
||||
assert exponential_backoff_time(attempt=1, min_wait=1, max_wait=20, multiplier=2) == 2
|
||||
assert exponential_backoff_time(attempt=2, min_wait=1, max_wait=20, multiplier=2) == 4
|
||||
assert exponential_backoff_time(attempt=3, min_wait=1, max_wait=20, multiplier=2) == 8
|
||||
assert exponential_backoff_time(attempt=4, min_wait=1, max_wait=20, multiplier=2) == 16
|
||||
assert exponential_backoff_time(attempt=5, min_wait=1, max_wait=20, multiplier=2) == 20
|
||||
assert exponential_backoff_time(attempt=6, min_wait=1, max_wait=20, multiplier=2) == 20
|
||||
@@ -6,16 +6,27 @@
|
||||
|
||||
import unittest
|
||||
|
||||
from pipecat.utils.string import match_endofsentence
|
||||
from pipecat.utils.string import match_endofsentence, parse_start_end_tags
|
||||
|
||||
|
||||
class TestUtilsString(unittest.IsolatedAsyncioTestCase):
|
||||
async def test_endofsentence(self):
|
||||
assert match_endofsentence("This is a sentence.")
|
||||
assert match_endofsentence("This is a sentence! ")
|
||||
assert match_endofsentence("This is a sentence?")
|
||||
assert match_endofsentence("This is a sentence:")
|
||||
assert match_endofsentence("This is a sentence;")
|
||||
assert match_endofsentence("This is a sentence.") == 19
|
||||
assert match_endofsentence("This is a sentence!") == 19
|
||||
assert match_endofsentence("This is a sentence?") == 19
|
||||
assert match_endofsentence("This is a sentence;") == 19
|
||||
assert match_endofsentence("This is a sentence...") == 21
|
||||
assert match_endofsentence("This is a sentence . . .") == 24
|
||||
assert match_endofsentence("This is a sentence. ..") == 22
|
||||
assert match_endofsentence("This is for Mr. and Mrs. Jones.") == 31
|
||||
assert match_endofsentence("U.S.A and U.S.A..") == 17
|
||||
assert match_endofsentence("My emails are foo@pipecat.ai and bar@pipecat.ai.") == 48
|
||||
assert match_endofsentence("My email is foo.bar@pipecat.ai.") == 31
|
||||
assert match_endofsentence("My email is spell(foo.bar@pipecat.ai).") == 38
|
||||
assert match_endofsentence("My email is <spell>foo.bar@pipecat.ai</spell>.") == 46
|
||||
assert match_endofsentence("The number pi is 3.14159.") == 25
|
||||
assert match_endofsentence("Valid scientific notation 1.23e4.") == 33
|
||||
assert match_endofsentence("Valid scientific notation 0.e4.") == 31
|
||||
assert not match_endofsentence("This is not a sentence")
|
||||
assert not match_endofsentence("This is not a sentence,")
|
||||
assert not match_endofsentence("This is not a sentence, ")
|
||||
@@ -26,6 +37,8 @@ class TestUtilsString(unittest.IsolatedAsyncioTestCase):
|
||||
assert not match_endofsentence("Heute ist Dienstag, der 3.") # 3. Juli 2024
|
||||
assert not match_endofsentence("America, or the U.") # U.S.A.
|
||||
assert not match_endofsentence("It still early, it's 3:00 a.") # 3:00 a.m.
|
||||
assert not match_endofsentence("My emails are foo@pipecat.ai and bar@pipecat.ai")
|
||||
assert not match_endofsentence("The number pi is 3.14159")
|
||||
|
||||
async def test_endofsentence_zh(self):
|
||||
chinese_sentences = [
|
||||
@@ -33,7 +46,6 @@ class TestUtilsString(unittest.IsolatedAsyncioTestCase):
|
||||
"你好！",
|
||||
"吃了吗？",
|
||||
"安全第一；",
|
||||
"他说：",
|
||||
]
|
||||
for i in chinese_sentences:
|
||||
assert match_endofsentence(i)
|
||||
@@ -49,3 +61,50 @@ class TestUtilsString(unittest.IsolatedAsyncioTestCase):
|
||||
for i in hindi_sentences:
|
||||
assert match_endofsentence(i)
|
||||
assert not match_endofsentence("हैलो，")
|
||||
|
||||
|
||||
class TestStartEndTags(unittest.IsolatedAsyncioTestCase):
|
||||
async def test_empty(self):
|
||||
assert parse_start_end_tags("", [], None, 0) == (None, 0)
|
||||
assert parse_start_end_tags("Hello from Pipecat!", [], None, 0) == (None, 0)
|
||||
|
||||
async def test_simple(self):
|
||||
# (<a>, </a>)
|
||||
assert parse_start_end_tags("Hello from <a>Pipecat</a>!", [("<a>", "</a>")], None, 0) == (
|
||||
None,
|
||||
26,
|
||||
)
|
||||
assert parse_start_end_tags("Hello from <a>Pipecat", [("<a>", "</a>")], None, 0) == (
|
||||
("<a>", "</a>"),
|
||||
21,
|
||||
)
|
||||
assert parse_start_end_tags("Hello from <a>Pipecat", [("<a>", "</a>")], None, 6) == (
|
||||
("<a>", "</a>"),
|
||||
21,
|
||||
)
|
||||
|
||||
# (spell(, ))
|
||||
assert parse_start_end_tags("Hello from spell(Pipecat)!", [("spell(", ")")], None, 0) == (
|
||||
None,
|
||||
26,
|
||||
)
|
||||
assert parse_start_end_tags("Hello from spell(Pipecat", [("spell(", ")")], None, 0) == (
|
||||
("spell(", ")"),
|
||||
24,
|
||||
)
|
||||
|
||||
async def test_multiple(self):
|
||||
# (<a>, </a>)
|
||||
assert parse_start_end_tags(
|
||||
"Hello from <a>Pipecat</a>! Hello <a>World</a>!", [("<a>", "</a>")], None, 0
|
||||
) == (
|
||||
None,
|
||||
46,
|
||||
)
|
||||
|
||||
assert parse_start_end_tags(
|
||||
"Hello from <a>Pipecat</a>! Hello <a>World", [("<a>", "</a>")], None, 0
|
||||
) == (
|
||||
("<a>", "</a>"),
|
||||
41,
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user