diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 7b821a9bc..5634d79ee 100644 --- a/src/pipecat/frames/frames.py +++ b/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. diff --git a/src/pipecat/processors/aggregators/llm_context_summarizer.py b/src/pipecat/processors/aggregators/llm_context_summarizer.py new file mode 100644 index 000000000..a1a613ccc --- /dev/null +++ b/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)})" + ) diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index 574af7bbf..6b28b5bf2 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/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.""" diff --git a/src/pipecat/services/anthropic/llm.py b/src/pipecat/services/anthropic/llm.py index b184ea29d..a21296fe3 100644 --- a/src/pipecat/services/anthropic/llm.py +++ b/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"], diff --git a/src/pipecat/services/aws/llm.py b/src/pipecat/services/aws/llm.py index 32562109a..1778ae74e 100644 --- a/src/pipecat/services/aws/llm.py +++ b/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, diff --git a/src/pipecat/services/google/llm.py b/src/pipecat/services/google/llm.py index 0e7556f83..563acadb3 100644 --- a/src/pipecat/services/google/llm.py +++ b/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 diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index c59b102b4..af7e691b0 100644 --- a/src/pipecat/services/llm_service.py +++ b/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() diff --git a/src/pipecat/services/openai/base_llm.py b/src/pipecat/services/openai/base_llm.py index 54e514508..d8669f622 100644 --- a/src/pipecat/services/openai/base_llm.py +++ b/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) diff --git a/src/pipecat/utils/context/__init__.py b/src/pipecat/utils/context/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/pipecat/utils/context/llm_context_summarization.py b/src/pipecat/utils/context/llm_context_summarization.py new file mode 100644 index 000000000..6865a00d9 --- /dev/null +++ b/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)