feat(summarizing_conversation_manager): implement summarization of older messages with model integration

Author: stefanoamorelli

src/strands/agent/conversation_manager/__init__.py

@@ -6,6 +6,8 @@
 - NullConversationManager: A no-op implementation that does not modify conversation history
 - SlidingWindowConversationManager: An implementation that maintains a sliding window of messages to control context
   size while preserving conversation coherence
+- SummarizingConversationManager: An extension of sliding window that can optionally summarize older context instead
+  of simply trimming it
 
 Conversation managers help control memory usage and context length while maintaining relevant conversation state, which
 is critical for effective agent interactions.
@@ -14,5 +16,11 @@
 from .conversation_manager import ConversationManager
 from .null_conversation_manager import NullConversationManager
 from .sliding_window_conversation_manager import SlidingWindowConversationManager
+from .summarizing_conversation_manager import SummarizingConversationManager
 
-__all__ = ["ConversationManager", "NullConversationManager", "SlidingWindowConversationManager"]
+__all__ = [
+    "ConversationManager",
+    "NullConversationManager",
+    "SlidingWindowConversationManager",
+    "SummarizingConversationManager",
+]

src/strands/agent/conversation_manager/summarizing_conversation_manager.py

@@ -0,0 +1,192 @@
+"""Summarizing conversation history management with configurable options."""
+
+import logging
+from typing import List, Optional, cast
+
+from ...types.content import ContentBlock, Message, Messages
+from ...types.exceptions import ContextWindowOverflowException
+from ...types.models.model import Model
+from ...types.tools import ToolResult
+from .sliding_window_conversation_manager import SlidingWindowConversationManager
+
+logger = logging.getLogger(__name__)
+
+
+class SummarizingConversationManager(SlidingWindowConversationManager):
+    """Extends sliding window manager with optional summarization of older messages.
+
+    This manager provides a configurable option to summarize older context instead of
+    simply trimming it, helping preserve important information while staying within
+    context limits.
+    """
+
+    def __init__(
+        self,
+        window_size: int = 40,
+        enable_summarization: bool = False,
+        summarization_model: Optional[Model] = None,
+        summary_ratio: float = 0.3,
+        preserve_recent_messages: int = 10,
+    ):
+        """Initialize the summarizing conversation manager.
+
+        Args:
+            window_size: Maximum number of messages to keep in history.
+                Defaults to 40 messages.
+            enable_summarization: Whether to enable summarization of older context.
+                Defaults to False (falls back to sliding window behavior).
+            summarization_model: Model to use for generating summaries.
+                Required if enable_summarization is True.
+            summary_ratio: Ratio of messages to summarize vs keep when window is exceeded.
+                Value between 0.1 and 0.8. Defaults to 0.3 (summarize 30% of oldest messages).
+            preserve_recent_messages: Minimum number of recent messages to always keep.
+                Defaults to 10 messages.
+        """
+        super().__init__(window_size)
+        self.enable_summarization = enable_summarization
+        self.summarization_model = summarization_model
+        self.summary_ratio = max(0.1, min(0.8, summary_ratio))
+        self.preserve_recent_messages = preserve_recent_messages
+
+        if enable_summarization and summarization_model is None:
+            raise ValueError("summarization_model is required when enable_summarization is True")
+
+    def reduce_context(self, messages: Messages, e: Optional[Exception] = None) -> None:
+        """Reduce context using summarization if enabled, otherwise fall back to sliding window.
+
+        Args:
+            messages: The messages to reduce.
+                This list is modified in-place.
+            e: The exception that triggered the context reduction, if any.
+
+        Raises:
+            ContextWindowOverflowException: If the context cannot be reduced further.
+        """
+        if not self.enable_summarization or self.summarization_model is None:
+            # Fall back to standard sliding window behavior
+            super().reduce_context(messages, e)
+            return
+
+        try:
+            self._reduce_context_with_summarization(messages, e)
+        except Exception as summarization_error:
+            logger.warning("Summarization failed, falling back to sliding window: %s", summarization_error)
+            # Fall back to sliding window if summarization fails
+            super().reduce_context(messages, e)
+
+    def _reduce_context_with_summarization(self, messages: Messages, e: Optional[Exception] = None) -> None:
+        """Reduce context by summarizing older messages.
+
+        Args:
+            messages: The messages to reduce.
+                This list is modified in-place.
+            e: The exception that triggered the context reduction, if any.
+
+        Raises:
+            ContextWindowOverflowException: If the context cannot be reduced further.
+        """
+        if len(messages) <= self.preserve_recent_messages:
+            raise ContextWindowOverflowException("Cannot summarize: too few messages to preserve context") from e
+
+        # Calculate how many messages to summarize
+        messages_to_summarize_count = max(1, int(len(messages) * self.summary_ratio))
+
+        # Ensure we don't summarize recent messages
+        messages_to_summarize_count = min(messages_to_summarize_count, len(messages) - self.preserve_recent_messages)
+
+        if messages_to_summarize_count <= 0:
+            raise ContextWindowOverflowException("Cannot summarize: insufficient messages for summarization") from e
+
+        # Extract messages to summarize
+        messages_to_summarize = messages[:messages_to_summarize_count]
+        remaining_messages = messages[messages_to_summarize_count:]
+
+        # Generate summary
+        summary_content = self._generate_summary(messages_to_summarize)
+
+        # Create a summary message
+        summary_message: Message = {"role": "assistant", "content": [ContentBlock(text=summary_content)]}
+
+        # Replace the summarized messages with the summary
+        messages[:] = [summary_message] + remaining_messages
+
+    def _generate_summary(self, messages: List[Message]) -> str:
+        """Generate a summary of the provided messages.
+
+        Args:
+            messages: The messages to summarize.
+
+        Returns:
+            A text summary of the conversation history.
+
+        Raises:
+            Exception: If summary generation fails.
+        """
+        if self.summarization_model is None:
+            raise ValueError("Summarization model is required but not provided")
+
+        # Convert messages to a readable format for summarization
+        conversation_text = self._format_messages_for_summarization(messages)
+
+        # Create system prompt for summarization task
+        system_prompt = (
+            "You are a conversation summarizer. Provide a concise summary of the conversation history. "
+            "Focus on key decisions, important context, and any ongoing tasks or topics. "
+            "Keep the summary under 500 words and be factual and objective."
+        )
+
+        # Create user message with conversation to summarize
+        user_prompt = f"Please summarize this conversation:\n\n{conversation_text}"
+        summary_messages: Messages = [{"role": "user", "content": [ContentBlock(text=user_prompt)]}]
+
+        summary_response = ""
+        for chunk in self.summarization_model.converse(summary_messages, system_prompt=system_prompt):
+            # Extract text from streaming events
+            if "contentBlockDelta" in chunk and "delta" in chunk["contentBlockDelta"]:
+                delta = chunk["contentBlockDelta"]["delta"]
+                if "text" in delta:
+                    summary_response += delta["text"]
+
+        return f"[CONVERSATION SUMMARY]: {summary_response.strip()}"
+
+    def _format_messages_for_summarization(self, messages: List[Message]) -> str:
+        """Format messages into readable text for summarization.
+
+        Args:
+            messages: The messages to format.
+
+        Returns:
+            A formatted string representation of the messages.
+        """
+        formatted_parts = []
+
+        for _, message in enumerate(messages):
+            role = message["role"].capitalize()
+            content_parts = []
+
+            for content in message["content"]:
+                if "text" in content:
+                    content_parts.append(content["text"])
+                elif "toolUse" in content:
+                    tool_use = content["toolUse"]
+                    content_parts.append(f"[Used tool: {tool_use.get('name', 'unknown')}]")
+                elif "toolResult" in content:
+                    tool_result = cast(ToolResult, content["toolResult"])
+                    status = tool_result.get("status", "")
+                    if tool_result.get("content"):
+                        for tr_content in tool_result["content"]:
+                            if "text" in tr_content:
+                                content_parts.append(f"[Tool result: {tr_content['text'][:100]}...]")
+                            elif "json" in tr_content:
+                                content_parts.append("[Tool result: JSON data]")
+                    else:
+                        content_parts.append(f"[Tool result: {status}]")
+                elif "image" in content:
+                    content_parts.append("[Image content]")
+                elif "document" in content:
+                    content_parts.append("[Document content]")
+
+            if content_parts:
+                formatted_parts.append(f"{role}: {' '.join(content_parts)}")
+
+        return "\n".join(formatted_parts)