Add custom automatic context summarization callback

changelog/5684.added.md

@@ -0,0 +1 @@
+- Added `should_summarize_callback` for custom automatic context-summarization policies without token estimation or threshold checks.

src/pipecat/processors/aggregators/llm_context_summarizer.py

@@ -8,6 +8,7 @@
 
 import asyncio
 import uuid
+from collections.abc import Callable
 from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
@@ -68,8 +69,9 @@ class LLMContextSummarizer(BaseObject):
 
     When ``auto_trigger=True`` (the default), summarization is triggered
     automatically based on the configured thresholds in
-    ``LLMAutoContextSummarizationConfig``. When ``auto_trigger=False``,
-    threshold checks are skipped and summarization only happens when an
+    ``LLMAutoContextSummarizationConfig`` or a custom
+    ``should_summarize_callback``. When ``auto_trigger=False``, automatic
+    checks are skipped and summarization only happens when an
     ``LLMSummarizeContextFrame`` is explicitly pushed into the pipeline.
 
     Both modes can coexist: set ``auto_trigger=True`` and also push
@@ -107,6 +109,7 @@ def __init__(
         context: LLMContext,
         config: LLMAutoContextSummarizationConfig | None = None,
         auto_trigger: bool = True,
+        should_summarize_callback: Callable[[LLMContext], bool] | None = None,
     ):
         """Initialize the context summarizer.
 
@@ -115,16 +118,20 @@ def __init__(
             config: Auto-summarization configuration controlling both trigger
                 thresholds and default summary generation parameters. If None,
                 uses default ``LLMAutoContextSummarizationConfig`` values.
-            auto_trigger: Whether to automatically trigger summarization when
-                thresholds are reached. When False, summarization only happens
-                when an ``LLMSummarizeContextFrame`` is pushed into the pipeline.
-                Defaults to True.
+            auto_trigger: Whether to automatically trigger summarization. When
+                False, summarization only happens when an
+                ``LLMSummarizeContextFrame`` is pushed into the pipeline. Defaults
+                to True.
+            should_summarize_callback: Optional predicate for automatic
+                summarization. When provided, it replaces threshold evaluation
+                after automatic triggering and in-progress guards are checked.
         """
         super().__init__()
 
         self._context = context
         self._auto_config = config or LLMAutoContextSummarizationConfig()
         self._auto_trigger = auto_trigger
+        self._should_summarize_callback = should_summarize_callback
 
         self._summarization_in_progress = False
         self._pending_summary_request_id: str | None = None
@@ -254,16 +261,16 @@ async def _clear_summarization_state(self):
     def _should_summarize(self) -> bool:
         """Determine if context summarization should be triggered.
 
-        Evaluates whether the current context has reached either the token
-        threshold or message count threshold that warrants compression.
-        Either threshold can be ``None`` to disable that check; at least one
-        must be set (enforced at config construction time).
+        Uses ``should_summarize_callback`` when configured. Otherwise, evaluates
+        whether the current context has reached either the token threshold or
+        message count threshold that warrants compression. Either threshold can
+        be ``None`` to disable that check; at least one must be set (enforced at
+        config construction time).
 
         Returns:
             True when ``auto_trigger`` is enabled, no summarization is in
-            progress, and either the token count exceeds ``max_context_tokens``
-            or the message count since the last summary exceeds
-            ``max_unsummarized_messages`` — whichever of the two is set.
+            progress, and the configured callback returns True or a configured
+            threshold is exceeded.
         """
         logger.trace(f"{self}: Checking if context summarization is needed")
 
@@ -274,6 +281,9 @@ def _should_summarize(self) -> bool:
             logger.debug(f"{self}: Summarization already in progress")
             return False
 
+        if self._should_summarize_callback is not None:
+            return self._should_summarize_callback(self._context)
+
         # Estimate tokens in context
         total_tokens = LLMContextSummarizationUtil.estimate_context_tokens(self._context)
         num_messages = len(self._context.messages)

src/pipecat/processors/aggregators/llm_response_universal.py

@@ -215,6 +215,9 @@ class LLMAssistantAggregatorParams:
             summarization. Controls trigger thresholds, message preservation, and
             summarization prompts. If None, uses default
             ``LLMAutoContextSummarizationConfig`` values.
+        should_summarize_callback: Optional predicate for automatic context
+            summarization. When provided, it replaces the configured threshold
+            checks after automatic triggering and in-progress guards are checked.
         add_tool_change_messages: When True, on each ``LLMSetToolsFrame`` the
             aggregator computes the diff against the currently advertised tools
             and appends a developer-role message to the context describing
@@ -242,6 +245,7 @@ class LLMAssistantAggregatorParams:
 
     enable_auto_context_summarization: bool = False
     auto_context_summarization_config: LLMAutoContextSummarizationConfig | None = None
+    should_summarize_callback: Callable[[LLMContext], bool] | None = None
     add_tool_change_messages: bool = False
 
     # Deprecated field names — kept for backward compatibility. See the
@@ -1513,6 +1517,7 @@ def __init__(
             context=self._context,
             config=self._params.auto_context_summarization_config,
             auto_trigger=self._params.enable_auto_context_summarization,
+            should_summarize_callback=self._params.should_summarize_callback,
         )
         self._summarizer.add_event_handler(
             "on_request_summarization", self._on_request_summarization

tests/test_llm_context_summarizer.py

@@ -6,6 +6,7 @@
 
 import asyncio
 import unittest
+from unittest.mock import patch
 
 from pipecat.frames.frames import (
     InterruptionFrame,
@@ -19,6 +20,11 @@
     LLMContextSummarizer,
     SummaryAppliedEvent,
 )
+from pipecat.processors.aggregators.llm_response_universal import (
+    LLMAssistantAggregator,
+    LLMAssistantAggregatorParams,
+)
+from pipecat.tests.utils import run_test
 from pipecat.utils.asyncio.task_manager import TaskManager
 from pipecat.utils.context.llm_context_summarization import (
     LLMAutoContextSummarizationConfig,
@@ -132,6 +138,149 @@ async def on_request_summarization(summarizer, frame):
 
         await summarizer.cleanup()
 
+    async def test_callback_triggers_summarization_without_threshold_evaluation(self):
+        """A callback can trigger automatic summarization below configured thresholds."""
+        config = LLMAutoContextSummarizationConfig(
+            max_context_tokens=100000,
+            max_unsummarized_messages=100,
+        )
+        callback_contexts = []
+        summarizer = LLMContextSummarizer(
+            context=self.context,
+            config=config,
+            should_summarize_callback=lambda context: callback_contexts.append(context) or True,
+        )
+        await summarizer.setup(frame_processor_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
+
+        with patch(
+            "pipecat.processors.aggregators.llm_context_summarizer."
+            "LLMContextSummarizationUtil.estimate_context_tokens",
+            side_effect=AssertionError("callback mode must not estimate tokens"),
+        ):
+            await summarizer.process_frame(LLMFullResponseStartFrame())
+
+        self.assertEqual(callback_contexts, [self.context])
+        self.assertIsNotNone(request_frame)
+        await summarizer.cleanup()
+
+    async def test_callback_can_suppress_exceeded_thresholds_without_token_estimation(self):
+        """A callback can suppress automatic summarization after thresholds are exceeded."""
+        config = LLMAutoContextSummarizationConfig(
+            max_context_tokens=1,
+            max_unsummarized_messages=1,
+        )
+        callback_contexts = []
+        summarizer = LLMContextSummarizer(
+            context=self.context,
+            config=config,
+            should_summarize_callback=lambda context: callback_contexts.append(context) or False,
+        )
+        await summarizer.setup(frame_processor_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
+
+        with patch(
+            "pipecat.processors.aggregators.llm_context_summarizer."
+            "LLMContextSummarizationUtil.estimate_context_tokens",
+            side_effect=AssertionError("callback mode must not estimate tokens"),
+        ):
+            await summarizer.process_frame(LLMFullResponseStartFrame())
+
+        self.assertEqual(callback_contexts, [self.context])
+        self.assertIsNone(request_frame)
+        await summarizer.cleanup()
+
+    async def test_callback_respects_auto_trigger_and_in_progress_guards(self):
+        """Automatic summary guards run before the callback."""
+        callback_calls = 0
+
+        def callback(context):
+            nonlocal callback_calls
+            callback_calls += 1
+            return True
+
+        summarizer = LLMContextSummarizer(
+            context=self.context,
+            auto_trigger=False,
+            should_summarize_callback=callback,
+        )
+        await summarizer.setup(frame_processor_setup(self.task_manager))
+        await summarizer.process_frame(LLMFullResponseStartFrame())
+        self.assertEqual(callback_calls, 0)
+        await summarizer.cleanup()
+
+        summarizer = LLMContextSummarizer(
+            context=self.context,
+            should_summarize_callback=callback,
+        )
+        await summarizer.setup(frame_processor_setup(self.task_manager))
+        summarizer._summarization_in_progress = True
+        await summarizer.process_frame(LLMFullResponseStartFrame())
+        self.assertEqual(callback_calls, 0)
+        await summarizer.cleanup()
+
+    async def test_manual_summarization_does_not_consult_callback(self):
+        """Manual summary requests bypass automatic trigger policy."""
+        callback_calls = 0
+
+        def callback(context):
+            nonlocal callback_calls
+            callback_calls += 1
+            return False
+
+        summarizer = LLMContextSummarizer(
+            context=self.context,
+            auto_trigger=False,
+            should_summarize_callback=callback,
+        )
+        await summarizer.setup(frame_processor_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
+
+        await summarizer.process_frame(LLMSummarizeContextFrame())
+
+        self.assertEqual(callback_calls, 0)
+        self.assertIsNotNone(request_frame)
+        await summarizer.cleanup()
+
+    async def test_assistant_aggregator_uses_should_summarize_callback(self):
+        """Assistant aggregator parameters control automatic summarization."""
+        callback_contexts = []
+        aggregator = LLMAssistantAggregator(
+            context=self.context,
+            params=LLMAssistantAggregatorParams(
+                enable_auto_context_summarization=True,
+                should_summarize_callback=lambda context: callback_contexts.append(context) or True,
+            ),
+        )
+
+        _, upstream_frames = await run_test(
+            aggregator,
+            frames_to_send=[LLMFullResponseStartFrame()],
+            expected_down_frames=[],
+            expected_up_frames=[LLMContextSummaryRequestFrame],
+        )
+
+        self.assertEqual(callback_contexts, [self.context])
+        self.assertEqual(upstream_frames[0].context, self.context)
+
     async def test_summarization_in_progress_prevents_duplicate(self):
         """Test that a summarization in progress prevents triggering another."""
         config = LLMAutoContextSummarizationConfig(