Context summarization feature implementation.

src/pipecat/frames/frames.py

@@ -1991,6 +1991,56 @@ class LLMFullResponseEndFrame(ControlFrame):
         self.skip_tts = None
 
 
+@dataclass
+class LLMContextSummaryRequestFrame(ControlFrame):
+    """Frame requesting context summarization from an LLM service.
+
+    Sent by aggregators to LLM services when conversation context needs to be
+    compressed. The LLM service generates a summary of older messages while
+    preserving recent conversation history.
+
+    Parameters:
+        request_id: Unique identifier to match this request with its response.
+            Used to handle async responses and avoid race conditions.
+        context: The full LLM context containing all messages to analyze and summarize.
+        min_messages_to_keep: Number of recent messages to preserve uncompressed.
+            These messages will not be included in the summary.
+        target_context_tokens: Maximum token size for the generated summary. This value
+            is passed directly to the LLM as the max_tokens parameter when generating
+            the summary text.
+        summarization_prompt: System prompt instructing the LLM how to generate
+            the summary.
+    """
+
+    request_id: str
+    context: "LLMContext"
+    min_messages_to_keep: int
+    target_context_tokens: int
+    summarization_prompt: str
+
+
+@dataclass
+class LLMContextSummaryResultFrame(ControlFrame, UninterruptibleFrame):
+    """Frame containing the result of context summarization.
+
+    Sent by LLM services back to aggregators after generating a summary.
+    Contains the formatted summary message and metadata about what was summarized.
+
+    Parameters:
+        request_id: Identifier matching the original request. Used to correlate
+            async responses.
+        summary: The formatted summary message ready to be inserted into context.
+        last_summarized_index: Index (0-based) of the last message that was
+            included in the summary. Messages after this index are preserved.
+        error: Error message if summarization failed, None on success.
+    """
+
+    request_id: str
+    summary: str
+    last_summarized_index: int
+    error: Optional[str] = None
+
+
 @dataclass
 class FunctionCallInProgressFrame(ControlFrame, UninterruptibleFrame):
     """Frame signaling that a function call is currently executing.

src/pipecat/processors/aggregators/llm_context_summarizer.py

@@ -0,0 +1,315 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""This module defines a summarizer for managing LLM context summarization."""
+
+import uuid
+from typing import Optional
+
+from loguru import logger
+
+from pipecat.frames.frames import (
+    Frame,
+    InterruptionFrame,
+    LLMContextSummaryRequestFrame,
+    LLMContextSummaryResultFrame,
+    LLMFullResponseStartFrame,
+)
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.utils.asyncio.task_manager import BaseTaskManager
+from pipecat.utils.base_object import BaseObject
+from pipecat.utils.context.llm_context_summarization import (
+    LLMContextSummarizationConfig,
+    LLMContextSummarizationUtil,
+)
+
+
+class LLMContextSummarizer(BaseObject):
+    """Summarizer for managing LLM context summarization.
+
+    This class manages automatic context summarization when token or message
+    limits are reached. It monitors the LLM context size, triggers
+    summarization requests, and applies the results to compress conversation history.
+
+    Event handlers available:
+
+    - on_request_summarization: Emitted when summarization should be triggered.
+        The aggregator should broadcast this frame to the LLM service.
+
+    Example::
+
+        @summarizer.event_handler("on_request_summarization")
+        async def on_request_summarization(summarizer, frame: LLMContextSummaryRequestFrame):
+            await aggregator.broadcast_frame(
+                LLMContextSummaryRequestFrame,
+                request_id=frame.request_id,
+                context=frame.context,
+                ...
+            )
+    """
+
+    def __init__(
+        self,
+        *,
+        context: LLMContext,
+        config: Optional[LLMContextSummarizationConfig] = None,
+    ):
+        """Initialize the context summarizer.
+
+        Args:
+            context: The LLM context to monitor and summarize.
+            config: Configuration for summarization behavior. If None, uses default config.
+        """
+        super().__init__()
+
+        self._context = context
+        self._config = config or LLMContextSummarizationConfig()
+
+        self._task_manager: Optional[BaseTaskManager] = None
+
+        self._summarization_in_progress = False
+        self._pending_summary_request_id: Optional[str] = None
+
+        self._register_event_handler("on_request_summarization", sync=True)
+
+    @property
+    def task_manager(self) -> BaseTaskManager:
+        """Returns the configured task manager."""
+        if not self._task_manager:
+            raise RuntimeError(f"{self} context summarizer was not properly setup")
+        return self._task_manager
+
+    async def setup(self, task_manager: BaseTaskManager):
+        """Initialize the summarizer with the given task manager.
+
+        Args:
+            task_manager: The task manager to be associated with this instance.
+        """
+        self._task_manager = task_manager
+
+    async def cleanup(self):
+        """Cleanup the summarizer."""
+        await super().cleanup()
+        await self._clear_summarization_state()
+
+    async def process_frame(self, frame: Frame):
+        """Process an incoming frame to detect when summarization is needed.
+
+        Args:
+            frame: The frame to be processed.
+        """
+        if isinstance(frame, LLMFullResponseStartFrame):
+            await self._handle_llm_response_start(frame)
+        elif isinstance(frame, LLMContextSummaryResultFrame):
+            await self._handle_summary_result(frame)
+        elif isinstance(frame, InterruptionFrame):
+            await self._handle_interruption()
+
+    async def _handle_llm_response_start(self, frame: LLMFullResponseStartFrame):
+        """Handle LLM response start to check if summarization is needed.
+
+        Args:
+            frame: The LLM response start frame.
+        """
+        if self._should_summarize():
+            await self._request_summarization()
+
+    async def _handle_interruption(self):
+        """Handle interruption by canceling summarization in progress.
+
+        Args:
+            frame: The interruption frame.
+        """
+        # Reset summarization state to allow new requests. This is necessary because
+        # the request frame (LLMContextSummaryRequestFrame) may have been cancelled
+        # during interruption. We preserve _pending_summary_request_id to handle the
+        # response frame (LLMContextSummaryResultFrame), which is uninterruptible and
+        # will still be delivered.
+        self._summarization_in_progress = False
+
+    async def _clear_summarization_state(self):
+        """Cancel pending summarization."""
+        if self._summarization_in_progress:
+            logger.debug(f"{self}: Clearing pending summarization")
+            self._summarization_in_progress = False
+            self._pending_summary_request_id = None
+
+    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.
+
+        Returns:
+            True if all conditions are met:
+            - No summarization currently in progress
+            - AND either:
+              - Token count exceeds max_context_tokens
+              - OR message count exceeds max_unsummarized_messages since last summary
+        """
+        logger.trace(f"{self}: Checking if context summarization is needed")
+
+        if self._summarization_in_progress:
+            logger.debug(f"{self}: Summarization already in progress")
+            return False
+
+        # Estimate tokens in context
+        total_tokens = LLMContextSummarizationUtil.estimate_context_tokens(self._context)
+        num_messages = len(self._context.messages)
+
+        # Check if we've reached the token limit
+        token_limit = self._config.max_context_tokens
+        token_limit_exceeded = total_tokens >= token_limit
+
+        # Check if we've exceeded max unsummarized messages
+        messages_since_summary = len(self._context.messages) - 1
+        message_threshold_exceeded = (
+            messages_since_summary >= self._config.max_unsummarized_messages
+        )
+
+        logger.trace(
+            f"{self}: Context has {num_messages} messages, "
+            f"~{total_tokens} tokens (limit: {token_limit}), "
+            f"{messages_since_summary} messages since last summary "
+            f"(message threshold: {self._config.max_unsummarized_messages})"
+        )
+
+        # Trigger if either limit is exceeded
+        if not token_limit_exceeded and not message_threshold_exceeded:
+            logger.trace(
+                f"{self}: Neither token limit nor message threshold exceeded, skipping summarization"
+            )
+            return False
+
+        reason = []
+        if token_limit_exceeded:
+            reason.append(f"~{total_tokens} tokens (>={token_limit} limit)")
+        if message_threshold_exceeded:
+            reason.append(
+                f"{messages_since_summary} messages (>={self._config.max_unsummarized_messages} threshold)"
+            )
+
+        logger.debug(f"{self}: ✓ Summarization needed - {', '.join(reason)}")
+        return True
+
+    async def _request_summarization(self):
+        """Request context summarization from LLM service.
+
+        Creates a summarization request frame and emits it via event handler.
+        Tracks the request ID to match async responses and prevent race conditions.
+        """
+        # Generate unique request ID
+        request_id = str(uuid.uuid4())
+        min_keep = self._config.min_messages_after_summary
+
+        # Mark summarization in progress
+        self._summarization_in_progress = True
+        self._pending_summary_request_id = request_id
+
+        logger.debug(f"{self}: Sending summarization request (request_id={request_id})")
+
+        # Create the request frame
+        request_frame = LLMContextSummaryRequestFrame(
+            request_id=request_id,
+            context=self._context,
+            min_messages_to_keep=min_keep,
+            target_context_tokens=self._config.target_context_tokens,
+            summarization_prompt=self._config.summary_prompt,
+        )
+
+        # Emit event for aggregator to broadcast
+        await self._call_event_handler("on_request_summarization", request_frame)
+
+    async def _handle_summary_result(self, frame: LLMContextSummaryResultFrame):
+        """Handle context summarization result from LLM service.
+
+        Processes the summary result by validating the request ID, checking for
+        errors, validating context state, and applying the summary.
+
+        Args:
+            frame: The summary result frame containing the generated summary.
+        """
+        logger.debug(f"{self}: Received summary result (request_id={frame.request_id})")
+
+        # Check if this is the result we're waiting for
+        if frame.request_id != self._pending_summary_request_id:
+            logger.debug(f"{self}: Ignoring stale summary result (request_id={frame.request_id})")
+            return
+
+        # Clear pending state
+        await self._clear_summarization_state()
+
+        # Check for errors
+        if frame.error:
+            logger.error(f"{self}: Context summarization failed: {frame.error}")
+            return
+
+        # Validate context state
+        if not self._validate_summary_context(frame.last_summarized_index):
+            logger.warning(f"{self}: Context state changed, skipping summary application")
+            return
+
+        # Apply summary
+        await self._apply_summary(frame.summary, frame.last_summarized_index)
+
+    def _validate_summary_context(self, last_summarized_index: int) -> bool:
+        """Validate that context state is still valid for applying summary.
+
+        Args:
+            last_summarized_index: The index of the last summarized message.
+
+        Returns:
+            True if the context state is still consistent with the summary.
+        """
+        if last_summarized_index < 0:
+            return False
+
+        # Check if we still have enough messages
+        if last_summarized_index >= len(self._context.messages):
+            return False
+
+        min_keep = self._config.min_messages_after_summary
+        remaining = len(self._context.messages) - 1 - last_summarized_index
+        if remaining < min_keep:
+            return False
+
+        return True
+
+    async def _apply_summary(self, summary: str, last_summarized_index: int):
+        """Apply summary to compress the conversation context.
+
+        Reconstructs the context with:
+        [first_system_message] + [summary_message] + [recent_messages]
+
+        Args:
+            summary: The generated summary text.
+            last_summarized_index: Index of the last message that was summarized.
+        """
+        messages = self._context.messages
+
+        # Find the first system message to preserve
+        first_system_msg = next((msg for msg in messages if msg.get("role") == "system"), None)
+
+        # Get recent messages to keep
+        recent_messages = messages[last_summarized_index + 1 :]
+
+        # Create summary message as an assistant message
+        summary_message = {"role": "assistant", "content": f"Conversation summary: {summary}"}
+
+        # Reconstruct context
+        new_messages = []
+        if first_system_msg:
+            new_messages.append(first_system_msg)
+        new_messages.append(summary_message)
+        new_messages.extend(recent_messages)
+
+        # Update context
+        self._context.set_messages(new_messages)
+
+        logger.info(
+            f"{self}: Applied context summary, compressed {last_summarized_index + 1} messages "
+            f"into summary. Context now has {len(new_messages)} messages (was {len(messages)})"
+        )

src/pipecat/processors/aggregators/llm_response_universal.py

@@ -37,6 +37,7 @@ from pipecat.frames.frames import (
     InterruptionFrame,
     LLMContextAssistantTimestampFrame,
     LLMContextFrame,
+    LLMContextSummaryRequestFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     LLMMessagesAppendFrame,
@@ -68,6 +69,7 @@ from pipecat.processors.aggregators.llm_context import (
     LLMSpecificMessage,
     NotGiven,
 )
+from pipecat.processors.aggregators.llm_context_summarizer import LLMContextSummarizer
 from pipecat.processors.frame_processor import FrameCallback, FrameDirection, FrameProcessor
 from pipecat.turns.user_idle_controller import UserIdleController
 from pipecat.turns.user_mute import BaseUserMuteStrategy
@@ -76,6 +78,7 @@ from pipecat.turns.user_stop import BaseUserTurnStopStrategy, UserTurnStoppedPar
 from pipecat.turns.user_turn_completion_mixin import UserTurnCompletionConfig
 from pipecat.turns.user_turn_controller import UserTurnController
 from pipecat.turns.user_turn_strategies import ExternalUserTurnStrategies, UserTurnStrategies
+from pipecat.utils.context.llm_context_summarization import LLMContextSummarizationConfig
 from pipecat.utils.string import TextPartForConcatenation, concatenate_aggregated_text
 from pipecat.utils.time import time_now_iso8601
 
@@ -121,9 +124,17 @@ class LLMAssistantAggregatorParams:
             in text frames by adding spaces between tokens. This parameter is
             ignored when used with the newer LLMAssistantAggregator, which
             handles word spacing automatically.
+        enable_context_summarization: Enable automatic context summarization when token
+            limits are reached (disabled by default). When enabled, older conversation
+            messages are automatically compressed into summaries to manage context size.
+        context_summarization_config: Configuration for context summarization behavior.
+            Controls thresholds, message preservation, and summarization prompts. If None
+            and summarization is enabled, uses default configuration values.
     """
 
     expect_stripped_words: bool = True
+    enable_context_summarization: bool = False
+    context_summarization_config: Optional[LLMContextSummarizationConfig] = None
 
 
 @dataclass
@@ -807,6 +818,17 @@ class LLMAssistantAggregator(LLMContextAggregator):
         self._thought_aggregation: List[TextPartForConcatenation] = []
         self._thought_start_time: str = ""
 
+        # Context summarization
+        self._summarizer: Optional[LLMContextSummarizer] = None
+        if self._params.enable_context_summarization:
+            self._summarizer = LLMContextSummarizer(
+                context=self._context,
+                config=self._params.context_summarization_config,
+            )
+            self._summarizer.add_event_handler(
+                "on_request_summarization", self._on_request_summarization
+            )
+
         self._register_event_handler("on_assistant_turn_started")
         self._register_event_handler("on_assistant_turn_stopped")
         self._register_event_handler("on_assistant_thought")
@@ -840,7 +862,12 @@ class LLMAssistantAggregator(LLMContextAggregator):
         """
         await super().process_frame(frame, direction)
 
-        if isinstance(frame, InterruptionFrame):
+        if isinstance(frame, StartFrame):
+            # Push StartFrame before start(), because we want StartFrame to be
+            # processed by every processor before any other frame is processed.
+            await self.push_frame(frame, direction)
+            await self._start(frame)
+        elif isinstance(frame, InterruptionFrame):
             await self._handle_interruptions(frame)
             await self.push_frame(frame, direction)
         elif isinstance(frame, (EndFrame, CancelFrame)):
@@ -883,6 +910,14 @@ class LLMAssistantAggregator(LLMContextAggregator):
         else:
             await self.push_frame(frame, direction)
 
+        # Pass frames to summarizer for monitoring
+        if self._summarizer:
+            await self._summarizer.process_frame(frame)
+
+    async def _start(self, frame: StartFrame):
+        if self._summarizer:
+            await self._summarizer.setup(self.task_manager)
+
     async def push_aggregation(self) -> str:
         """Push the current assistant aggregation with timestamp."""
         if not self._aggregation:
@@ -921,6 +956,8 @@ class LLMAssistantAggregator(LLMContextAggregator):
 
     async def _handle_end_or_cancel(self, frame: Frame):
         await self._trigger_assistant_turn_stopped()
+        if self._summarizer:
+            await self._summarizer.cleanup()
 
     async def _handle_function_calls_started(self, frame: FunctionCallsStartedFrame):
         function_names = [f"{f.function_name}:{f.tool_call_id}" for f in frame.function_calls]
@@ -1197,6 +1234,19 @@ class LLMAssistantAggregator(LLMContextAggregator):
         # Only strip whitespace if we removed a marker
         return text.strip() if marker_found else text
 
+    async def _on_request_summarization(
+        self, summarizer: LLMContextSummarizer, frame: LLMContextSummaryRequestFrame
+    ):
+        """Handle summarization request from the summarizer.
+
+        Push the request frame UPSTREAM to the LLM service for processing.
+
+        Args:
+            summarizer: The summarizer that generated the request.
+            frame: The summarization request frame to broadcast.
+        """
+        await self.push_frame(frame, FrameDirection.UPSTREAM)
+
 
 class LLMContextAggregatorPair:
     """Pair of LLM context aggregators for updating context with user and assistant messages."""

src/pipecat/services/anthropic/llm.py

@@ -261,11 +261,15 @@ class AnthropicLLMService(LLMService):
             response = await api_call(**params)
             return response
 
-    async def run_inference(self, context: LLMContext | OpenAILLMContext) -> Optional[str]:
+    async def run_inference(
+        self, context: LLMContext | OpenAILLMContext, max_tokens: Optional[int] = None
+    ) -> Optional[str]:
         """Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context.
 
         Args:
             context: The LLM context containing conversation history.
+            max_tokens: Optional maximum number of tokens to generate. If provided,
+                overrides the service's default max_tokens setting.
 
         Returns:
             The LLM's response as a string, or None if no response is generated.
@@ -290,7 +294,7 @@ class AnthropicLLMService(LLMService):
         # Build params using the same method as streaming completions
         params = {
             "model": self.model_name,
-            "max_tokens": self._settings["max_tokens"],
+            "max_tokens": max_tokens if max_tokens is not None else self._settings["max_tokens"],
             "stream": False,
             "temperature": self._settings["temperature"],
             "top_k": self._settings["top_k"],

src/pipecat/services/aws/llm.py

@@ -844,11 +844,15 @@ class AWSBedrockLLMService(LLMService):
             inference_config["topP"] = self._settings["top_p"]
         return inference_config
 
-    async def run_inference(self, context: LLMContext | OpenAILLMContext) -> Optional[str]:
+    async def run_inference(
+        self, context: LLMContext | OpenAILLMContext, max_tokens: Optional[int] = None
+    ) -> Optional[str]:
         """Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context.
 
         Args:
             context: The LLM context containing conversation history.
+            max_tokens: Optional maximum number of tokens to generate. If provided,
+                overrides the service's default max_tokens setting.
 
         Returns:
             The LLM's response as a string, or None if no response is generated.
@@ -868,6 +872,10 @@ class AWSBedrockLLMService(LLMService):
         # Prepare request parameters using the same method as streaming
         inference_config = self._build_inference_config()
 
+        # Override maxTokens if provided
+        if max_tokens is not None:
+            inference_config["maxTokens"] = max_tokens
+
         request_params = {
             "modelId": self.model_name,
             "messages": messages,

src/pipecat/services/google/llm.py

@@ -799,11 +799,15 @@ class GoogleLLMService(LLMService):
         """Create the Gemini client instance. Subclasses can override this."""
         self._client = genai.Client(api_key=self._api_key, http_options=self._http_options)
 
-    async def run_inference(self, context: LLMContext | OpenAILLMContext) -> Optional[str]:
+    async def run_inference(
+        self, context: LLMContext | OpenAILLMContext, max_tokens: Optional[int] = None
+    ) -> Optional[str]:
         """Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context.
 
         Args:
             context: The LLM context containing conversation history.
+            max_tokens: Optional maximum number of tokens to generate. If provided,
+                overrides the service's default max_tokens setting.
 
         Returns:
             The LLM's response as a string, or None if no response is generated.
@@ -828,6 +832,10 @@ class GoogleLLMService(LLMService):
             system_instruction=system, tools=tools if tools else None
         )
 
+        # Override max_output_tokens if provided
+        if max_tokens is not None:
+            generation_params["max_output_tokens"] = max_tokens
+
         generation_config = GenerateContentConfig(**generation_params)
 
         # Use the new google-genai client's async method

src/pipecat/services/llm_service.py

@@ -39,6 +39,8 @@ from pipecat.frames.frames import (
     FunctionCallsStartedFrame,
     InterruptionFrame,
     LLMConfigureOutputFrame,
+    LLMContextSummaryRequestFrame,
+    LLMContextSummaryResultFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     LLMTextFrame,
@@ -57,6 +59,9 @@ from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.ai_service import AIService
 from pipecat.turns.user_turn_completion_mixin import UserTurnCompletionLLMServiceMixin
+from pipecat.utils.context.llm_context_summarization import (
+    LLMContextSummarizationUtil,
+)
 
 # Type alias for a callable that handles LLM function calls.
 FunctionCallHandler = Callable[["FunctionCallParams"], Awaitable[None]]
@@ -195,6 +200,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
         self._sequential_runner_task: Optional[asyncio.Task] = None
         self._tracing_enabled: bool = False
         self._skip_tts: Optional[bool] = None
+        self._summary_task: Optional[asyncio.Task] = None
 
         self._register_event_handler("on_function_calls_started")
         self._register_event_handler("on_completion_timeout")
@@ -218,13 +224,17 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
         """
         return self.get_llm_adapter().create_llm_specific_message(message)
 
-    async def run_inference(self, context: LLMContext | OpenAILLMContext) -> Optional[str]:
+    async def run_inference(
+        self, context: LLMContext | OpenAILLMContext, max_tokens: Optional[int] = None
+    ) -> Optional[str]:
         """Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context.
 
         Must be implemented by subclasses.
 
         Args:
             context: The LLM context containing conversation history.
+            max_tokens: Optional maximum number of tokens to generate. If provided,
+                overrides the service's default max_tokens/max_completion_tokens setting.
 
         Returns:
             The LLM's response as a string, or None if no response is generated.
@@ -286,6 +296,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
         await super().stop(frame)
         if not self._run_in_parallel:
             await self._cancel_sequential_runner_task()
+        await self._cancel_summary_task()
 
     async def cancel(self, frame: CancelFrame):
         """Cancel the LLM service.
@@ -296,6 +307,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
         await super().cancel(frame)
         if not self._run_in_parallel:
             await self._cancel_sequential_runner_task()
+        await self._cancel_summary_task()
 
     async def _update_settings(self, settings: Mapping[str, Any]):
         """Update LLM service settings.
@@ -339,6 +351,8 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
             await self._handle_interruptions(frame)
         elif isinstance(frame, LLMConfigureOutputFrame):
             self._skip_tts = frame.skip_tts
+        elif isinstance(frame, LLMContextSummaryRequestFrame):
+            await self._handle_summary_request(frame)
 
     async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
         """Pushes a frame.
@@ -372,6 +386,121 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
             if entry.cancel_on_interruption:
                 await self._cancel_function_call(function_name)
 
+    async def _handle_summary_request(self, frame: LLMContextSummaryRequestFrame):
+        """Handle context summarization request from aggregator.
+
+        Processes a summarization request by generating a compressed summary
+        of conversation history. Uses the adapter to format the summary
+        according to the provider's requirements. Broadcasts the result back
+        to the aggregator for context reconstruction.
+
+        Args:
+            frame: The summary request frame containing context and parameters.
+        """
+        logger.debug(f"{self}: Processing summarization request {frame.request_id}")
+
+        # Create a background task to generate the summary without blocking
+        self._summary_task = self.create_task(self._generate_summary_task(frame))
+
+    async def _generate_summary_task(self, frame: LLMContextSummaryRequestFrame):
+        """Background task to generate summary without blocking the pipeline.
+
+        Args:
+            frame: The summary request frame containing context and parameters.
+        """
+        summary = ""
+        last_index = -1
+        error = None
+
+        try:
+            summary, last_index = await self._generate_summary(frame)
+        except Exception as e:
+            error = f"Error generating context summary: {e}"
+            await self.push_error(error, exception=e)
+
+        await self.broadcast_frame(
+            LLMContextSummaryResultFrame,
+            request_id=frame.request_id,
+            summary=summary,
+            last_summarized_index=last_index,
+            error=error,
+        )
+
+        self._summary_task = None
+
+    async def _generate_summary(self, frame: LLMContextSummaryRequestFrame) -> tuple[str, int]:
+        """Generate a compressed summary of conversation context.
+
+        Uses the message selection logic to identify which messages
+        to summarize, formats them as a transcript, and invokes the LLM to
+        generate a concise summary. The summary is formatted according to the
+        LLM provider's requirements using the adapter.
+
+        Args:
+            frame: The summary request frame containing context and configuration.
+
+        Returns:
+            Tuple of (formatted summary message, last_summarized_index).
+
+        Raises:
+            RuntimeError: If there are no messages to summarize, the service doesn't
+                support run_inference(), or the LLM returns an empty summary.
+
+        Note:
+            Requires the service to implement run_inference() method for
+            synchronous LLM calls.
+        """
+        # Get messages to summarize using utility method
+        result = LLMContextSummarizationUtil.get_messages_to_summarize(
+            frame.context, frame.min_messages_to_keep
+        )
+
+        if not result.messages:
+            logger.debug(f"{self}: No messages to summarize")
+            raise RuntimeError("No messages to summarize")
+
+        logger.debug(
+            f"{self}: Generating summary for {len(result.messages)} messages "
+            f"(index 0 to {result.last_summarized_index}), "
+            f"target_context_tokens={frame.target_context_tokens}"
+        )
+
+        # Create summary context
+        transcript = LLMContextSummarizationUtil.format_messages_for_summary(result.messages)
+        prompt_messages = [
+            {
+                "role": "system",
+                "content": frame.summarization_prompt,
+            },
+            {
+                "role": "user",
+                "content": f"Conversation history:\n{transcript}",
+            },
+        ]
+        summary_context = LLMContext(messages=prompt_messages)
+
+        # Generate summary using run_inference
+        # This will be overridden by each LLM service implementation
+        try:
+            summary_text = await self.run_inference(
+                summary_context, max_tokens=frame.target_context_tokens
+            )
+        except NotImplementedError:
+            raise RuntimeError(
+                f"LLM service {self.__class__.__name__} does not implement run_inference"
+            )
+
+        if not summary_text:
+            raise RuntimeError("LLM returned empty summary")
+
+        summary_text = summary_text.strip()
+        logger.info(
+            f"{self}: Generated summary of {len(summary_text)} characters "
+            f"for {len(result.messages)} messages"
+        )
+
+        return summary_text, result.last_summarized_index
+
     def register_function(
         self,
         function_name: Optional[str],
@@ -588,6 +717,11 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
             await self.cancel_task(self._sequential_runner_task)
             self._sequential_runner_task = None
 
+    async def _cancel_summary_task(self):
+        if self._summary_task:
+            await self.cancel_task(self._summary_task)
+            self._summary_task = None
+
     async def _sequential_runner_handler(self):
         while True:
             runner_item = await self._sequential_runner_queue.get()

src/pipecat/services/openai/base_llm.py

@@ -265,11 +265,15 @@ class BaseOpenAILLMService(LLMService):
         params.update(self._settings["extra"])
         return params
 
-    async def run_inference(self, context: LLMContext | OpenAILLMContext) -> Optional[str]:
+    async def run_inference(
+        self, context: LLMContext | OpenAILLMContext, max_tokens: Optional[int] = None
+    ) -> Optional[str]:
         """Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context.
 
         Args:
             context: The LLM context containing conversation history.
+            max_tokens: Optional maximum number of tokens to generate. If provided,
+                overrides the service's default max_tokens/max_completion_tokens setting.
 
         Returns:
             The LLM's response as a string, or None if no response is generated.
@@ -291,6 +295,14 @@ class BaseOpenAILLMService(LLMService):
         params["stream"] = False
         params.pop("stream_options", None)
 
+        # Override max_tokens if provided
+        if max_tokens is not None:
+            # Use max_completion_tokens for newer models, fallback to max_tokens
+            if "max_completion_tokens" in params:
+                params["max_completion_tokens"] = max_tokens
+            else:
+                params["max_tokens"] = max_tokens
+
         # LLM completion
         response = await self._client.chat.completions.create(**params)
 

src/pipecat/utils/context/llm_context_summarization.py

@@ -0,0 +1,396 @@
+#
+# Copyright (c) 2024–2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Utility for context summarization in LLM services.
+
+This module provides reusable functionality for automatically compressing conversation
+context when token limits are reached, enabling efficient long-running conversations.
+"""
+
+from dataclasses import dataclass
+from typing import List, Optional
+
+from loguru import logger
+
+from pipecat.processors.aggregators.llm_context import LLMContext
+
+# Token estimation constants
+CHARS_PER_TOKEN = 4  # Industry-standard heuristic: 1 token ≈ 4 characters
+TOKEN_OVERHEAD_PER_MESSAGE = 10  # Estimated structural overhead per message
+IMAGE_TOKEN_ESTIMATE = 500  # Rough estimate for image content
+SUMMARY_TOKEN_BUFFER = 0.8  # Keep summary at 80% of available space for safety
+MIN_SUMMARY_TOKENS = 100  # Minimum tokens to allocate for summary
+
+DEFAULT_SUMMARIZATION_PROMPT = """You are summarizing a conversation between a user and an AI assistant.
+
+Your task:
+1. Create a concise summary that preserves:
+   - Key facts, decisions, and agreements
+   - Important context needed to continue the conversation
+   - User preferences and requirements mentioned
+   - Any unresolved questions or action items
+
+2. Format:
+   - Use clear, factual statements
+   - Group related information
+   - Prioritize information likely to be referenced later
+   - Keep the summary concise to fit within the specified token budget
+
+3. Omit:
+   - Greetings and small talk
+   - Redundant information
+   - Tangential discussions that were resolved
+
+The conversation transcript follows. Generate only the summary, no other text."""
+
+
+@dataclass
+class LLMContextSummarizationConfig:
+    """Configuration for context summarization behavior.
+
+    Controls when and how conversation context is automatically compressed
+    to manage token limits in long-running conversations.
+
+    Parameters:
+        max_context_tokens: Maximum allowed context size in tokens. When this
+            limit is reached, summarization is triggered to compress the context.
+            The tokens are calculated using the industry-standard approximation
+            of 1 token ≈ 4 characters.
+        target_context_tokens: Maximum token size for the generated summary.
+            This value is passed directly to the LLM as the max_tokens parameter
+            when generating the summary. Should be sized appropriately to allow
+            the summary plus recent preserved messages to fit within reasonable
+            context limits.
+        max_unsummarized_messages: Maximum number of new messages that can
+            accumulate since the last summary before triggering a new
+            summarization. This ensures regular compression even if token
+            limits are not reached.
+        min_messages_after_summary: Number of recent messages to preserve
+            uncompressed after each summarization. These messages maintain
+            immediate conversational context.
+        summarization_prompt: Custom prompt for the LLM to use when generating
+            summaries. If None, uses DEFAULT_SUMMARIZATION_PROMPT.
+    """
+
+    max_context_tokens: int = 8000
+    target_context_tokens: int = 6000
+    max_unsummarized_messages: int = 20
+    min_messages_after_summary: int = 4
+    summarization_prompt: Optional[str] = None
+
+    def __post_init__(self):
+        """Validate configuration parameters."""
+        if self.max_context_tokens <= 0:
+            raise ValueError("max_context_tokens must be positive")
+        if self.target_context_tokens <= 0:
+            raise ValueError("target_context_tokens must be positive")
+
+        # Auto-adjust target_context_tokens if it exceeds max_context_tokens
+        if self.target_context_tokens > self.max_context_tokens:
+            # Use 80% of max_context_tokens as a reasonable default
+            self.target_context_tokens = int(self.max_context_tokens * 0.8)
+
+        if self.max_unsummarized_messages < 1:
+            raise ValueError("max_unsummarized_messages must be at least 1")
+        if self.min_messages_after_summary < 0:
+            raise ValueError("min_messages_after_summary must be positive")
+
+    @property
+    def summary_prompt(self) -> str:
+        """Get the summarization prompt to use.
+
+        Returns:
+            The custom prompt if set, otherwise the default summarization prompt.
+        """
+        return self.summarization_prompt or DEFAULT_SUMMARIZATION_PROMPT
+
+
+@dataclass
+class LLMMessagesToSummarize:
+    """Result of get_messages_to_summarize operation.
+
+    Parameters:
+        messages: Messages to include in the summary
+        last_summarized_index: Index of the last message being summarized
+    """
+
+    messages: List[dict]
+    last_summarized_index: int
+
+
+class LLMContextSummarizationUtil:
+    """Utility providing context summarization capabilities for LLM processing.
+
+    This utility enables automatic conversation context compression when token
+    limits are reached. It provides functionality for both aggregators
+    (which decide when to summarize) and LLM services (which generate summaries).
+
+    Key features:
+    - Token estimation using character-count heuristics (chars // 4)
+    - Smart message selection (preserves system messages and recent context)
+    - Function call awareness (avoids summarizing incomplete tool interactions)
+    - Flexible transcript formatting for summarization
+    - Maximum summary token calculation with safety buffers
+
+    Usage:
+        Use the static methods directly on the class:
+
+        tokens = LLMContextSummarizationUtil.estimate_context_tokens(context)
+        result = LLMContextSummarizationUtil.get_messages_to_summarize(context, 4)
+        transcript = LLMContextSummarizationUtil.format_messages_for_summary(messages)
+
+    Note:
+        Token estimation uses the industry-standard heuristic of 1 token ≈ 4 characters.
+    """
+
+    @staticmethod
+    def estimate_tokens(text: str) -> int:
+        """Estimate token count for text using character count heuristic.
+
+        Uses the industry-standard approximation of 1 token ≈ 4 characters.
+        This works well across different content types (prose, code, etc.)
+        and languages.
+
+        Note:
+            For more accurate token counts, use the model's official tokenizer.
+            This is a rough estimate suitable for threshold checks and budgeting.
+
+        Args:
+            text: Text to estimate tokens for
+
+        Returns:
+            Estimated token count (characters // 4)
+        """
+        if not text:
+            return 0
+        return len(text) // CHARS_PER_TOKEN
+
+    @staticmethod
+    def estimate_context_tokens(context: LLMContext) -> int:
+        """Estimate total token count for a context.
+
+        Calculates an approximate token count by analyzing all messages,
+        including text content, tool calls, and structural overhead.
+
+        Args:
+            context: LLM context to estimate.
+
+        Returns:
+            Estimated total token count including:
+            - Message content (text, images)
+            - Tool calls and their arguments
+            - Tool results
+            - Structural overhead (TOKEN_OVERHEAD_PER_MESSAGE per message)
+        """
+        total = 0
+
+        for message in context.messages:
+            # Role and structure overhead
+            total += TOKEN_OVERHEAD_PER_MESSAGE
+
+            # Message content
+            content = message.get("content", "")
+            if isinstance(content, str):
+                total += LLMContextSummarizationUtil.estimate_tokens(content)
+            elif isinstance(content, list):
+                for item in content:
+                    if isinstance(item, dict):
+                        item_type = item.get("type", "")
+                        # Text content
+                        if item_type == "text":
+                            total += LLMContextSummarizationUtil.estimate_tokens(
+                                item.get("text", "")
+                            )
+                        # Image content
+                        elif item_type in ("image_url", "image"):
+                            # Images are expensive, rough estimate
+                            total += IMAGE_TOKEN_ESTIMATE
+
+            # Tool calls
+            if "tool_calls" in message:
+                tool_calls = message["tool_calls"]
+                if isinstance(tool_calls, list):
+                    for tool_call in tool_calls:
+                        if isinstance(tool_call, dict):
+                            func = tool_call.get("function", {})
+                            if isinstance(func, dict):
+                                total += LLMContextSummarizationUtil.estimate_tokens(
+                                    func.get("name", "") + func.get("arguments", "")
+                                )
+
+            # Tool call ID
+            if "tool_call_id" in message:
+                total += TOKEN_OVERHEAD_PER_MESSAGE
+
+        return total
+
+    @staticmethod
+    def _get_function_calls_in_progress_index(messages: List[dict], start_idx: int) -> int:
+        """Find the earliest message index with incomplete function calls.
+
+        Scans messages to identify function/tool calls that haven't received
+        their results yet. This prevents summarizing incomplete tool interactions
+        which would break the request-response pairing.
+
+        Args:
+            messages: List of messages to check.
+            start_idx: Index to start checking from.
+
+        Returns:
+            Index of first message with function call in progress, or -1 if all
+            function calls are complete.
+        """
+        # Track tool call IDs mapped to their message index
+        pending_tool_calls: dict[str, int] = {}
+
+        for i in range(start_idx, len(messages)):
+            msg = messages[i]
+            role = msg.get("role")
+
+            # Check for tool calls in assistant messages
+            if role == "assistant" and "tool_calls" in msg:
+                tool_calls = msg.get("tool_calls", [])
+                if isinstance(tool_calls, list):
+                    for tool_call in tool_calls:
+                        if isinstance(tool_call, dict):
+                            tool_call_id = tool_call.get("id")
+                            if tool_call_id:
+                                pending_tool_calls[tool_call_id] = i
+
+            # Check for tool results
+            if role == "tool":
+                tool_call_id = msg.get("tool_call_id")
+                if tool_call_id and tool_call_id in pending_tool_calls:
+                    pending_tool_calls.pop(tool_call_id)
+
+        # If we have pending tool calls, return the earliest index
+        if pending_tool_calls:
+            return min(pending_tool_calls.values())
+
+        return -1
+
+    @staticmethod
+    def get_messages_to_summarize(
+        context: LLMContext, min_messages_to_keep: int
+    ) -> LLMMessagesToSummarize:
+        """Determine which messages should be included in summarization.
+
+        Intelligently selects messages for summarization while preserving:
+        - The first system message (defines assistant behavior)
+        - The last N messages (maintains immediate conversation context)
+        - Incomplete function call sequences (preserves tool interaction integrity)
+
+        Args:
+            context: The LLM context containing all messages.
+            min_messages_to_keep: Number of recent messages to exclude from
+                summarization.
+
+        Returns:
+            LLMMessagesToSummarize containing the messages to summarize and the
+            index of the last message included.
+        """
+        messages = context.messages
+        if len(messages) <= min_messages_to_keep:
+            return LLMMessagesToSummarize(messages=[], last_summarized_index=-1)
+
+        # Find first system message index
+        first_system_index = next(
+            (i for i, msg in enumerate(messages) if msg.get("role") == "system"), -1
+        )
+
+        # Messages to summarize are between first system and recent messages
+        # We exclude the first system message itself
+        if first_system_index >= 0:
+            summary_start = first_system_index + 1
+        else:
+            summary_start = 0
+
+        # Get messages to keep (last N messages)
+        summary_end = len(messages) - min_messages_to_keep
+
+        if summary_start >= summary_end:
+            return LLMMessagesToSummarize(messages=[], last_summarized_index=-1)
+
+        # Check for function calls in progress in the range we want to summarize
+        function_call_start = LLMContextSummarizationUtil._get_function_calls_in_progress_index(
+            messages, summary_start
+        )
+        if function_call_start >= 0 and function_call_start < summary_end:
+            # Stop summarization before the function call
+            logger.debug(
+                f"ContextSummarization: Found function call in progress at index {function_call_start}, "
+                f"stopping summary before it (was going to summarize up to {summary_end})"
+            )
+            # Count how many messages we're skipping
+            skipped_messages = summary_end - function_call_start
+            summary_end = function_call_start
+            if skipped_messages > 0:
+                logger.info(
+                    f"ContextSummarization: Skipping {skipped_messages} messages with "
+                    f"function calls in progress (will summarize after results are available)"
+                )
+
+        if summary_start >= summary_end:
+            return LLMMessagesToSummarize(messages=[], last_summarized_index=-1)
+
+        messages_to_summarize = messages[summary_start:summary_end]
+        last_summarized_index = summary_end - 1
+
+        return LLMMessagesToSummarize(
+            messages=messages_to_summarize, last_summarized_index=last_summarized_index
+        )
+
+    @staticmethod
+    def format_messages_for_summary(messages: List[dict]) -> str:
+        """Format messages as a transcript for summarization.
+
+        Args:
+            messages: Messages to format
+
+        Returns:
+            Formatted transcript string
+        """
+        transcript_parts = []
+
+        for msg in messages:
+            role = msg.get("role", "unknown")
+            content = msg.get("content", "")
+
+            # Handle different content types
+            if isinstance(content, str):
+                text = content
+            elif isinstance(content, list):
+                text_parts = []
+                for item in content:
+                    if isinstance(item, dict) and item.get("type") == "text":
+                        text_parts.append(item.get("text", ""))
+                text = " ".join(text_parts)
+            else:
+                text = str(content)
+
+            if text:
+                # Capitalize role for readability
+                formatted_role = role.upper()
+                transcript_parts.append(f"{formatted_role}: {text}")
+
+            # Include tool calls if present
+            if "tool_calls" in msg:
+                tool_calls = msg.get("tool_calls", [])
+                if isinstance(tool_calls, list):
+                    for tool_call in tool_calls:
+                        if isinstance(tool_call, dict):
+                            func = tool_call.get("function", {})
+                            if isinstance(func, dict):
+                                name = func.get("name", "unknown")
+                                args = func.get("arguments", "")
+                                transcript_parts.append(f"TOOL_CALL: {name}({args})")
+
+            # Include tool results
+            if role == "tool":
+                tool_call_id = msg.get("tool_call_id", "unknown")
+                transcript_parts.append(f"TOOL_RESULT[{tool_call_id}]: {text}")
+
+        return "\n\n".join(transcript_parts)