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

Fixing context summarization tests

Browse Source
This commit is contained in:
filipi87
2026-02-27 18:42:50 -03:00
parent 0839e3813f
commit d077a810ae
2 changed files with 232 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

104
tests/test_context_summarization.py
View File

@@ -14,8 +14,10 @@ from pipecat.frames.frames import LLMContextSummaryRequestFrame, LLMContextSumma
from pipecat.processors.aggregators.llm_context import LLMContext, LLMSpecificMessage
from pipecat.services.llm_service import LLMService
from pipecat.utils.context.llm_context_summarization import (
LLMAutoContextSummarizationConfig,
LLMContextSummarizationConfig,
LLMContextSummarizationUtil,
LLMContextSummaryConfig,
)
@@ -167,43 +169,109 @@ class TestContextSummarizationMixin(unittest.TestCase):
self.assertIn("USER: First part Second part", transcript)
class TestLLMContextSummarizationConfig(unittest.TestCase):
"""Tests for LLMContextSummarizationConfig."""
class TestLLMContextSummaryConfig(unittest.TestCase):
"""Tests for LLMContextSummaryConfig."""
def test_default_config(self):
"""Test default configuration values."""
config = LLMContextSummarizationConfig()
config = LLMContextSummaryConfig()
self.assertEqual(config.max_context_tokens, 8000)
self.assertEqual(config.max_unsummarized_messages, 20)
self.assertEqual(config.target_context_tokens, 6000)
self.assertEqual(config.min_messages_after_summary, 4)
self.assertIsNone(config.summarization_prompt)
def test_custom_config(self):
"""Test custom configuration."""
config = LLMContextSummarizationConfig(
max_context_tokens=2500,
config = LLMContextSummaryConfig(
target_context_tokens=2000,
max_unsummarized_messages=15,
min_messages_after_summary=4,
summarization_prompt="Custom prompt",
)
self.assertEqual(config.max_context_tokens, 2500)
self.assertEqual(config.target_context_tokens, 2000)
self.assertEqual(config.max_unsummarized_messages, 15)
self.assertEqual(config.min_messages_after_summary, 4)
self.assertEqual(config.summary_prompt, "Custom prompt")
def test_summary_prompt_property(self):
"""Test summary_prompt property uses default when None."""
config = LLMContextSummarizationConfig()
config = LLMContextSummaryConfig()
self.assertIn("summarizing a conversation", config.summary_prompt.lower())
config_with_custom = LLMContextSummarizationConfig(summarization_prompt="Custom")
config_with_custom = LLMContextSummaryConfig(summarization_prompt="Custom")
self.assertEqual(config_with_custom.summary_prompt, "Custom")
class TestLLMAutoContextSummarizationConfig(unittest.TestCase):
"""Tests for LLMAutoContextSummarizationConfig."""
def test_default_config(self):
"""Test default configuration values."""
config = LLMAutoContextSummarizationConfig()
self.assertEqual(config.max_context_tokens, 8000)
self.assertEqual(config.max_unsummarized_messages, 20)
self.assertEqual(config.summary_config.target_context_tokens, 6000)
self.assertEqual(config.summary_config.min_messages_after_summary, 4)
def test_custom_config(self):
"""Test custom configuration."""
config = LLMAutoContextSummarizationConfig(
max_context_tokens=2500,
max_unsummarized_messages=15,
summary_config=LLMContextSummaryConfig(
target_context_tokens=2000,
min_messages_after_summary=4,
summarization_prompt="Custom prompt",
),
)
self.assertEqual(config.max_context_tokens, 2500)
self.assertEqual(config.max_unsummarized_messages, 15)
self.assertEqual(config.summary_config.target_context_tokens, 2000)
self.assertEqual(config.summary_config.min_messages_after_summary, 4)
self.assertEqual(config.summary_config.summary_prompt, "Custom prompt")
def test_target_tokens_auto_adjusted(self):
"""Test that target_context_tokens is auto-adjusted when it exceeds max."""
config = LLMAutoContextSummarizationConfig(
max_context_tokens=1000,
summary_config=LLMContextSummaryConfig(target_context_tokens=9000),
)
self.assertLessEqual(config.summary_config.target_context_tokens, config.max_context_tokens)
class TestLLMContextSummarizationConfigDeprecated(unittest.TestCase):
"""Tests for deprecated LLMContextSummarizationConfig."""
def test_emits_deprecation_warning(self):
"""Test that instantiating the deprecated config emits a DeprecationWarning."""
with self.assertWarns(DeprecationWarning):
LLMContextSummarizationConfig()
def test_to_auto_config(self):
"""Test conversion to the new LLMAutoContextSummarizationConfig."""
import warnings
with warnings.catch_warnings():
warnings.simplefilter("ignore", DeprecationWarning)
old_config = LLMContextSummarizationConfig(
max_context_tokens=2500,
target_context_tokens=2000,
max_unsummarized_messages=15,
min_messages_after_summary=4,
summarization_prompt="Custom",
)
new_config = old_config.to_auto_config()
self.assertIsInstance(new_config, LLMAutoContextSummarizationConfig)
self.assertEqual(new_config.max_context_tokens, 2500)
self.assertEqual(new_config.max_unsummarized_messages, 15)
self.assertEqual(new_config.summary_config.target_context_tokens, 2000)
self.assertEqual(new_config.summary_config.min_messages_after_summary, 4)
self.assertEqual(new_config.summary_config.summarization_prompt, "Custom")
class TestFunctionCallHandling(unittest.TestCase):
"""Tests for function call handling in summarization."""
@@ -670,10 +738,12 @@ class TestDedicatedLLMSummarization(unittest.IsolatedAsyncioTestCase):
{"role": "user", "content": f"Test message {i} that adds tokens to context."}
)
config = LLMContextSummarizationConfig(
config = LLMAutoContextSummarizationConfig(
max_context_tokens=50, # Very low to trigger easily
llm=dedicated_llm,
summarization_timeout=5.0,
summary_config=LLMContextSummaryConfig(
llm=dedicated_llm,
summarization_timeout=5.0,
),
)
return context, config
@@ -736,7 +806,7 @@ class TestDedicatedLLMSummarization(unittest.IsolatedAsyncioTestCase):
dedicated_llm._generate_summary = slow_summary
context, config = self._create_context_and_config(dedicated_llm)
config.summarization_timeout = 0.1 # Very short timeout
config.summary_config.summarization_timeout = 0.1 # Very short timeout
summarizer = LLMContextSummarizer(context=context, config=config)
await summarizer.setup(self.task_manager)
@@ -826,7 +896,7 @@ class TestDedicatedLLMSummarization(unittest.IsolatedAsyncioTestCase):
{"role": "user", "content": f"Test message {i} that adds tokens to context."}
)
config = LLMContextSummarizationConfig(max_context_tokens=50)
config = LLMAutoContextSummarizationConfig(max_context_tokens=50)
summarizer = LLMContextSummarizer(context=context, config=config)
await summarizer.setup(self.task_manager)

162
tests/test_llm_context_summarizer.py
View File

@@ -12,6 +12,7 @@ from pipecat.frames.frames import (
LLMContextSummaryRequestFrame,
LLMContextSummaryResultFrame,
LLMFullResponseStartFrame,
LLMSummarizeContextFrame,
)
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_context_summarizer import (
@@ -19,7 +20,10 @@ from pipecat.processors.aggregators.llm_context_summarizer import (
SummaryAppliedEvent,
)
from pipecat.utils.asyncio.task_manager import TaskManager, TaskManagerParams
from pipecat.utils.context.llm_context_summarization import LLMContextSummarizationConfig
from pipecat.utils.context.llm_context_summarization import (
LLMAutoContextSummarizationConfig,
LLMContextSummaryConfig,
)
class TestLLMContextSummarizer(unittest.IsolatedAsyncioTestCase):
@@ -35,7 +39,7 @@ class TestLLMContextSummarizer(unittest.IsolatedAsyncioTestCase):
async def test_summarization_triggered_by_token_limit(self):
"""Test that summarization is triggered when token limit is reached."""
config = LLMContextSummarizationConfig(
config = LLMAutoContextSummarizationConfig(
max_context_tokens=100, # Very low to trigger easily
max_unsummarized_messages=100, # High so it doesn't trigger by message count
)
@@ -71,7 +75,7 @@ class TestLLMContextSummarizer(unittest.IsolatedAsyncioTestCase):
async def test_summarization_triggered_by_message_count(self):
"""Test that summarization is triggered when message count threshold is reached."""
config = LLMContextSummarizationConfig(
config = LLMAutoContextSummarizationConfig(
max_context_tokens=100000, # Very high so it doesn't trigger by tokens
max_unsummarized_messages=5, # Low to trigger easily
)
@@ -101,7 +105,7 @@ class TestLLMContextSummarizer(unittest.IsolatedAsyncioTestCase):
async def test_summarization_not_triggered_below_thresholds(self):
"""Test that summarization is not triggered when below thresholds."""
config = LLMContextSummarizationConfig(
config = LLMAutoContextSummarizationConfig(
max_context_tokens=10000,
max_unsummarized_messages=20,
)
@@ -130,7 +134,7 @@ class TestLLMContextSummarizer(unittest.IsolatedAsyncioTestCase):
async def test_summarization_in_progress_prevents_duplicate(self):
"""Test that a summarization in progress prevents triggering another."""
config = LLMContextSummarizationConfig(
config = LLMAutoContextSummarizationConfig(
max_context_tokens=50, # Very low
max_unsummarized_messages=100,
)
@@ -161,7 +165,10 @@ class TestLLMContextSummarizer(unittest.IsolatedAsyncioTestCase):
async def test_summary_result_handling(self):
"""Test that summary results are processed and applied correctly."""
config = LLMContextSummarizationConfig(max_context_tokens=50, min_messages_after_summary=2)
config = LLMAutoContextSummarizationConfig(
max_context_tokens=50,
summary_config=LLMContextSummaryConfig(min_messages_after_summary=2),
)
summarizer = LLMContextSummarizer(context=self.context, config=config)
await summarizer.setup(self.task_manager)
@@ -208,7 +215,7 @@ class TestLLMContextSummarizer(unittest.IsolatedAsyncioTestCase):
async def test_interruption_cancels_summarization(self):
"""Test that an interruption cancels pending summarization."""
config = LLMContextSummarizationConfig(max_context_tokens=50)
config = LLMAutoContextSummarizationConfig(max_context_tokens=50)
summarizer = LLMContextSummarizer(context=self.context, config=config)
await summarizer.setup(self.task_manager)
@@ -238,7 +245,10 @@ class TestLLMContextSummarizer(unittest.IsolatedAsyncioTestCase):
async def test_stale_summary_result_ignored(self):
"""Test that stale summary results are ignored."""
config = LLMContextSummarizationConfig(max_context_tokens=50, min_messages_after_summary=2)
config = LLMAutoContextSummarizationConfig(
max_context_tokens=50,
summary_config=LLMContextSummaryConfig(min_messages_after_summary=2),
)
summarizer = LLMContextSummarizer(context=self.context, config=config)
await summarizer.setup(self.task_manager)
@@ -294,9 +304,116 @@ class TestLLMContextSummarizer(unittest.IsolatedAsyncioTestCase):
await summarizer.cleanup()
async def test_manual_summarization_via_frame(self):
"""Test that LLMSummarizeContextFrame triggers summarization on demand."""
config = LLMAutoContextSummarizationConfig(
max_context_tokens=100000, # High — auto trigger would never fire
max_unsummarized_messages=100,
)
summarizer = LLMContextSummarizer(
context=self.context,
config=config,
auto_trigger=False, # Disable auto; only manual requests should work
)
await summarizer.setup(self.task_manager)
request_frame = None
@summarizer.event_handler("on_request_summarization")
async def on_request_summarization(summarizer, frame):
nonlocal request_frame
request_frame = frame
# Add messages
for i in range(5):
self.context.add_message({"role": "user", "content": f"Message {i}"})
# Auto-trigger should NOT fire even on LLMFullResponseStartFrame
await summarizer.process_frame(LLMFullResponseStartFrame())
self.assertIsNone(request_frame)
# Manual trigger via LLMSummarizeContextFrame should fire
await summarizer.process_frame(LLMSummarizeContextFrame())
self.assertIsNotNone(request_frame)
self.assertIsInstance(request_frame, LLMContextSummaryRequestFrame)
# The request must have a valid request_id and carry the current context
self.assertTrue(request_frame.request_id)
self.assertEqual(request_frame.context, self.context)
await summarizer.cleanup()
async def test_manual_summarization_with_config_override(self):
"""Test that LLMSummarizeContextFrame can override default summary config."""
config = LLMAutoContextSummarizationConfig(
max_context_tokens=100000,
summary_config=LLMContextSummaryConfig(
target_context_tokens=6000,
min_messages_after_summary=4,
),
)
summarizer = LLMContextSummarizer(context=self.context, config=config)
await summarizer.setup(self.task_manager)
request_frame = None
@summarizer.event_handler("on_request_summarization")
async def on_request_summarization(summarizer, frame):
nonlocal request_frame
request_frame = frame
for i in range(5):
self.context.add_message({"role": "user", "content": f"Message {i}"})
# Push a manual frame with custom config overrides
custom_config = LLMContextSummaryConfig(
target_context_tokens=500,
min_messages_after_summary=1,
)
await summarizer.process_frame(LLMSummarizeContextFrame(config=custom_config))
self.assertIsNotNone(request_frame)
# The request should use the overridden values
self.assertEqual(request_frame.target_context_tokens, 500)
self.assertEqual(request_frame.min_messages_to_keep, 1)
await summarizer.cleanup()
async def test_manual_summarization_blocked_when_in_progress(self):
"""Test that a second LLMSummarizeContextFrame is ignored while one is in progress."""
config = LLMAutoContextSummarizationConfig(max_context_tokens=100000)
summarizer = LLMContextSummarizer(context=self.context, config=config)
await summarizer.setup(self.task_manager)
request_count = 0
@summarizer.event_handler("on_request_summarization")
async def on_request_summarization(summarizer, frame):
nonlocal request_count
request_count += 1
for i in range(5):
self.context.add_message({"role": "user", "content": f"Message {i}"})
# First manual request
await summarizer.process_frame(LLMSummarizeContextFrame())
self.assertEqual(request_count, 1)
# Second manual request while first is in progress — should be ignored
await summarizer.process_frame(LLMSummarizeContextFrame())
self.assertEqual(request_count, 1)
await summarizer.cleanup()
async def test_summary_message_role_is_user(self):
"""Test that the summary message uses the user role."""
config = LLMContextSummarizationConfig(max_context_tokens=50, min_messages_after_summary=2)
config = LLMAutoContextSummarizationConfig(
max_context_tokens=50,
summary_config=LLMContextSummaryConfig(min_messages_after_summary=2),
)
summarizer = LLMContextSummarizer(context=self.context, config=config)
await summarizer.setup(self.task_manager)
@@ -335,7 +452,10 @@ class TestLLMContextSummarizer(unittest.IsolatedAsyncioTestCase):
async def test_summary_message_default_template(self):
"""Test that the default summary_message_template is used."""
config = LLMContextSummarizationConfig(max_context_tokens=50, min_messages_after_summary=2)
config = LLMAutoContextSummarizationConfig(
max_context_tokens=50,
summary_config=LLMContextSummaryConfig(min_messages_after_summary=2),
)
summarizer = LLMContextSummarizer(context=self.context, config=config)
await summarizer.setup(self.task_manager)
@@ -377,10 +497,12 @@ class TestLLMContextSummarizer(unittest.IsolatedAsyncioTestCase):
async def test_summary_message_custom_template(self):
"""Test that a custom summary_message_template is applied."""
config = LLMContextSummarizationConfig(
config = LLMAutoContextSummarizationConfig(
max_context_tokens=50,
min_messages_after_summary=2,
summary_message_template="<context_summary>\n{summary}\n</context_summary>",
summary_config=LLMContextSummaryConfig(
min_messages_after_summary=2,
summary_message_template="<context_summary>\n{summary}\n</context_summary>",
),
)
summarizer = LLMContextSummarizer(context=self.context, config=config)
@@ -420,7 +542,10 @@ class TestLLMContextSummarizer(unittest.IsolatedAsyncioTestCase):
async def test_on_summary_applied_event(self):
"""Test that on_summary_applied event fires with correct data."""
config = LLMContextSummarizationConfig(max_context_tokens=50, min_messages_after_summary=2)
config = LLMAutoContextSummarizationConfig(
max_context_tokens=50,
summary_config=LLMContextSummaryConfig(min_messages_after_summary=2),
)
summarizer = LLMContextSummarizer(context=self.context, config=config)
await summarizer.setup(self.task_manager)
@@ -474,7 +599,10 @@ class TestLLMContextSummarizer(unittest.IsolatedAsyncioTestCase):
async def test_on_summary_applied_not_fired_on_error(self):
"""Test that on_summary_applied event is NOT fired when summarization fails."""
config = LLMContextSummarizationConfig(max_context_tokens=50, min_messages_after_summary=2)
config = LLMAutoContextSummarizationConfig(
max_context_tokens=50,
summary_config=LLMContextSummaryConfig(min_messages_after_summary=2),
)
summarizer = LLMContextSummarizer(context=self.context, config=config)
await summarizer.setup(self.task_manager)
@@ -515,9 +643,9 @@ class TestLLMContextSummarizer(unittest.IsolatedAsyncioTestCase):
async def test_request_frame_includes_timeout(self):
"""Test that the request frame includes the configured summarization_timeout."""
config = LLMContextSummarizationConfig(
config = LLMAutoContextSummarizationConfig(
max_context_tokens=50,
summarization_timeout=60.0,
summary_config=LLMContextSummaryConfig(summarization_timeout=60.0),
)
summarizer = LLMContextSummarizer(context=self.context, config=config)