diff --git a/src/strands/agent/conversation_manager/sliding_window_conversation_manager.py b/src/strands/agent/conversation_manager/sliding_window_conversation_manager.py index 709c876e7..b97de0b06 100644 --- a/src/strands/agent/conversation_manager/sliding_window_conversation_manager.py +++ b/src/strands/agent/conversation_manager/sliding_window_conversation_manager.py @@ -7,12 +7,15 @@ from ...agent.agent import Agent from ...hooks import BeforeModelCallEvent, HookRegistry -from ...types.content import Messages +from ...types.content import ContentBlock, Messages from ...types.exceptions import ContextWindowOverflowException +from ...types.tools import ToolResultContent from .conversation_manager import ConversationManager logger = logging.getLogger(__name__) +_PRESERVE_CHARS = 200 + class SlidingWindowConversationManager(ConversationManager): """Implements a sliding window strategy for managing conversation history. @@ -20,10 +23,21 @@ class SlidingWindowConversationManager(ConversationManager): This class handles the logic of maintaining a conversation window that preserves tool usage pairs and avoids invalid window states. + When truncation is enabled (the default), large tool results are partially truncated, preserving the first + and last 200 characters, and image blocks inside tool results are replaced with descriptive text placeholders. + Truncation targets the oldest tool results first so the most relevant recent context is preserved as long + as possible. + Supports proactive management during agent loop execution via the per_turn parameter. """ - def __init__(self, window_size: int = 40, should_truncate_results: bool = True, *, per_turn: bool | int = False): + def __init__( + self, + window_size: int = 40, + should_truncate_results: bool = True, + *, + per_turn: bool | int = False, + ): """Initialize the sliding window conversation manager. Args: @@ -44,6 +58,9 @@ def __init__(self, window_size: int = 40, should_truncate_results: bool = True, Raises: ValueError: If per_turn is 0 or a negative integer. """ + if isinstance(per_turn, int) and not isinstance(per_turn, bool) and per_turn <= 0: + raise ValueError(f"per_turn must be a positive integer, True, or False, got {per_turn}") + super().__init__() self.window_size = window_size @@ -157,14 +174,14 @@ def reduce_context(self, agent: "Agent", e: Exception | None = None, **kwargs: A messages = agent.messages # Try to truncate the tool result first - last_message_idx_with_tool_results = self._find_last_message_with_tool_results(messages) - if last_message_idx_with_tool_results is not None and self.should_truncate_results: + oldest_message_idx_with_tool_results = self._find_oldest_message_with_tool_results(messages) + if oldest_message_idx_with_tool_results is not None and self.should_truncate_results: logger.debug( - "message_index=<%s> | found message with tool results at index", last_message_idx_with_tool_results + "message_index=<%s> | found message with tool results at index", oldest_message_idx_with_tool_results ) - results_truncated = self._truncate_tool_results(messages, last_message_idx_with_tool_results) + results_truncated = self._truncate_tool_results(messages, oldest_message_idx_with_tool_results) if results_truncated: - logger.debug("message_index=<%s> | tool results truncated", last_message_idx_with_tool_results) + logger.debug("message_index=<%s> | tool results truncated", oldest_message_idx_with_tool_results) return # Try to trim index id when tool result cannot be truncated anymore @@ -197,10 +214,14 @@ def reduce_context(self, agent: "Agent", e: Exception | None = None, **kwargs: A messages[:] = messages[trim_index:] def _truncate_tool_results(self, messages: Messages, msg_idx: int) -> bool: - """Truncate tool results in a message to reduce context size. + """Truncate tool results and replace image blocks in a message to reduce context size. + + For text blocks within tool results, all blocks are partially truncated unless they + have already been truncated. The first and last _PRESERVE_CHARS characters are kept, + and the removed middle is replaced with a notice indicating how many characters were + removed. The tool result status is not changed. - When a message contains tool results that are too large for the model's context window, this function - replaces the content of those tool results with a simple error message. + Image blocks nested inside tool result content are replaced with a short descriptive placeholder. Args: messages: The conversation message history. @@ -212,52 +233,82 @@ def _truncate_tool_results(self, messages: Messages, msg_idx: int) -> bool: if msg_idx >= len(messages) or msg_idx < 0: return False + def _image_placeholder(image_block: Any) -> str: + source: Any = image_block.get("source", {}) + media_type = image_block.get("format", "unknown") + data = source.get("bytes", b"") + return f"[image: {media_type}, {len(data) if data else 0} bytes]" + message = messages[msg_idx] changes_made = False - tool_result_too_large_message = "The tool result was too large!" - for i, content in enumerate(message.get("content", [])): - if isinstance(content, dict) and "toolResult" in content: - tool_result_content_text = next( - (item["text"] for item in content["toolResult"]["content"] if "text" in item), - "", - ) - # make the overwriting logic togglable - if ( - message["content"][i]["toolResult"]["status"] == "error" - and tool_result_content_text == tool_result_too_large_message - ): - logger.info("ToolResult has already been updated, skipping overwrite") - return False - # Update status to error with informative message - message["content"][i]["toolResult"]["status"] = "error" - message["content"][i]["toolResult"]["content"] = [{"text": tool_result_too_large_message}] - changes_made = True + new_content: list[ContentBlock] = [] + + for content in message.get("content", []): + if "toolResult" in content: + tool_result: Any = content["toolResult"] + tool_result_items = tool_result.get("content", []) + new_items: list[ToolResultContent] = [] + item_changed = False + + for item in tool_result_items: + # Replace image items nested inside toolResult content + if "image" in item: + new_items.append({"text": _image_placeholder(item["image"])}) + item_changed = True + continue + + # Partially truncate text items that have not already been truncated + if "text" in item: + text = item["text"] + truncation_marker = "... [truncated:" + if truncation_marker not in text and len(text) > 2 * _PRESERVE_CHARS: + prefix = text[:_PRESERVE_CHARS] + suffix = text[-_PRESERVE_CHARS:] + removed = len(text) - 2 * _PRESERVE_CHARS + truncated_text = ( + f"{prefix}...\n\n... [truncated: {removed} chars removed] ...\n\n...{suffix}" + ) + new_items.append({"text": truncated_text}) + item_changed = True + continue + + new_items.append(item) + + if item_changed: + updated_tool_result: Any = { + **{k: v for k, v in tool_result.items() if k != "content"}, + "content": new_items, + } + new_content.append({"toolResult": updated_tool_result}) + changes_made = True + else: + new_content.append(content) + continue + + new_content.append(content) + + if changes_made: + message["content"] = new_content return changes_made - def _find_last_message_with_tool_results(self, messages: Messages) -> int | None: - """Find the index of the last message containing tool results. + def _find_oldest_message_with_tool_results(self, messages: Messages) -> int | None: + """Find the index of the oldest message containing tool results. - This is useful for identifying messages that might need to be truncated to reduce context size. + Iterates from oldest to newest so that truncation targets the least-recent + (and therefore least relevant) tool results first. Args: messages: The conversation message history. Returns: - Index of the last message with tool results, or None if no such message exists. + Index of the oldest message with tool results, or None if no such message exists. """ - # Iterate backwards through all messages (from newest to oldest) - for idx in range(len(messages) - 1, -1, -1): - # Check if this message has any content with toolResult + # Iterate from oldest to newest + for idx in range(len(messages)): current_message = messages[idx] - has_tool_result = False - for content in current_message.get("content", []): if isinstance(content, dict) and "toolResult" in content: - has_tool_result = True - break - - if has_tool_result: - return idx + return idx return None diff --git a/tests/strands/agent/test_agent.py b/tests/strands/agent/test_agent.py index 55de68ff1..6d73fc177 100644 --- a/tests/strands/agent/test_agent.py +++ b/tests/strands/agent/test_agent.py @@ -621,7 +621,7 @@ def test_agent__call__retry_with_overwritten_tool(mock_model, agent, tool, agene }, }, }, - {"contentBlockDelta": {"delta": {"toolUse": {"input": '{"random_string": "abcdEfghI123"}'}}}}, + {"contentBlockDelta": {"delta": {"toolUse": {"input": '{"random_string": "' + "X" * 500 + '"}'}}}}, {"contentBlockStop": {}}, {"messageStop": {"stopReason": "tool_use"}}, ] @@ -635,12 +635,14 @@ def test_agent__call__retry_with_overwritten_tool(mock_model, agent, tool, agene agent("test message") + large_input = "X" * 500 + truncated_text = large_input[:200] + "...\n\n... [truncated: 100 chars removed] ...\n\n..." + large_input[-200:] expected_messages = [ {"role": "user", "content": [{"text": "test message"}]}, { "role": "assistant", "content": [ - {"toolUse": {"toolUseId": "t1", "name": "tool_decorated", "input": {"random_string": "abcdEfghI123"}}} + {"toolUse": {"toolUseId": "t1", "name": "tool_decorated", "input": {"random_string": large_input}}} ], }, { @@ -649,8 +651,8 @@ def test_agent__call__retry_with_overwritten_tool(mock_model, agent, tool, agene { "toolResult": { "toolUseId": "t1", - "status": "error", - "content": [{"text": "The tool result was too large!"}], + "status": "success", + "content": [{"text": truncated_text}], } } ], diff --git a/tests/strands/agent/test_conversation_manager.py b/tests/strands/agent/test_conversation_manager.py index 46876d8e5..fd88954e8 100644 --- a/tests/strands/agent/test_conversation_manager.py +++ b/tests/strands/agent/test_conversation_manager.py @@ -177,37 +177,25 @@ def test_sliding_window_conversation_manager_with_untrimmable_history_raises_con def test_sliding_window_conversation_manager_with_tool_results_truncated(): + large_text = "A" * 300 + "B" * 300 + "C" * 300 manager = SlidingWindowConversationManager(1) messages = [ {"role": "assistant", "content": [{"toolUse": {"toolUseId": "456", "name": "tool1", "input": {}}}]}, { "role": "user", - "content": [ - {"toolResult": {"toolUseId": "789", "content": [{"text": "large input"}], "status": "success"}} - ], + "content": [{"toolResult": {"toolUseId": "789", "content": [{"text": large_text}], "status": "success"}}], }, ] test_agent = Agent(messages=messages) manager.reduce_context(test_agent) - expected_messages = [ - {"role": "assistant", "content": [{"toolUse": {"toolUseId": "456", "name": "tool1", "input": {}}}]}, - { - "role": "user", - "content": [ - { - "toolResult": { - "toolUseId": "789", - "content": [{"text": "The tool result was too large!"}], - "status": "error", - } - } - ], - }, - ] - - assert messages == expected_messages + result_text = messages[1]["content"][0]["toolResult"]["content"][0]["text"] + assert result_text.startswith("A" * 200) + assert result_text.endswith("C" * 200) + assert "... [truncated:" in result_text + # Status must NOT be changed to error + assert messages[1]["content"][0]["toolResult"]["status"] == "success" def test_null_conversation_manager_reduce_context_raises_context_window_overflow_exception(): @@ -267,6 +255,16 @@ def test_per_turn_parameter_validation(): assert SlidingWindowConversationManager(per_turn=3).per_turn == 3 +def test_per_turn_zero_raises_value_error(): + with pytest.raises(ValueError, match="per_turn"): + SlidingWindowConversationManager(per_turn=0) + + +def test_per_turn_negative_raises_value_error(): + with pytest.raises(ValueError, match="per_turn"): + SlidingWindowConversationManager(per_turn=-5) + + def test_conversation_manager_is_hook_provider(): """Test that ConversationManager implements HookProvider protocol.""" manager = NullConversationManager() @@ -420,3 +418,176 @@ def test_per_turn_backward_compatibility(): agent = Agent(model=model, conversation_manager=manager) result = agent("Hello") assert result is not None + + +# ============================================================================== +# Improved Truncation Strategy Tests +# ============================================================================== + + +def test_truncation_targets_oldest_message_first(): + """Oldest message with tool results is truncated before newer ones.""" + large_text = "X" * 20000 + manager = SlidingWindowConversationManager(window_size=10) + messages = [ + {"role": "assistant", "content": [{"toolUse": {"toolUseId": "1", "name": "tool1", "input": {}}}]}, + { + "role": "user", + "content": [{"toolResult": {"toolUseId": "1", "content": [{"text": large_text}], "status": "success"}}], + }, + {"role": "assistant", "content": [{"toolUse": {"toolUseId": "2", "name": "tool2", "input": {}}}]}, + { + "role": "user", + "content": [{"toolResult": {"toolUseId": "2", "content": [{"text": large_text}], "status": "success"}}], + }, + ] + test_agent = Agent(messages=messages) + + manager.reduce_context(test_agent) + + # The oldest tool result (index 1) must be truncated + oldest_text = messages[1]["content"][0]["toolResult"]["content"][0]["text"] + assert "... [truncated:" in oldest_text + + # The newest tool result (index 3) must remain untouched after the first reduce_context call + newest_text = messages[3]["content"][0]["toolResult"]["content"][0]["text"] + assert "... [truncated:" not in newest_text + + +def test_large_tool_result_partially_truncated_with_context_preserved(): + """Large tool results are truncated in the middle while the beginning and end are preserved.""" + preserve = 200 # matches _PRESERVE_CHARS + # Build text with distinct prefix, middle, and suffix + prefix_text = "P" * preserve + middle_text = "M" * 500 + suffix_text = "S" * preserve + large_text = prefix_text + middle_text + suffix_text + + manager = SlidingWindowConversationManager(window_size=10) + messages = [ + { + "role": "user", + "content": [{"toolResult": {"toolUseId": "1", "content": [{"text": large_text}], "status": "success"}}], + } + ] + + truncated = manager._truncate_tool_results(messages, 0) + + assert truncated + result_text = messages[0]["content"][0]["toolResult"]["content"][0]["text"] + assert result_text.startswith(prefix_text) + assert result_text.endswith(suffix_text) + assert "... [truncated:" in result_text + removed = len(large_text) - 2 * preserve + assert f"... [truncated: {removed} chars removed] ..." in result_text + + +def test_truncation_does_not_change_status_to_error(): + """Partial truncation must not change the tool result status.""" + large_text = "Z" * 15000 + manager = SlidingWindowConversationManager(window_size=10) + messages = [ + { + "role": "user", + "content": [{"toolResult": {"toolUseId": "1", "content": [{"text": large_text}], "status": "success"}}], + } + ] + + manager._truncate_tool_results(messages, 0) + + assert messages[0]["content"][0]["toolResult"]["status"] == "success" + + +def test_image_blocks_inside_tool_result_replaced_with_placeholder(): + """Image blocks nested inside toolResult content are replaced with a text placeholder.""" + manager = SlidingWindowConversationManager(window_size=10) + image_data = b"base64encodeddata" + messages = [ + { + "role": "user", + "content": [ + { + "toolResult": { + "toolUseId": "1", + "content": [ + {"text": "some text"}, + { + "image": { + "format": "jpeg", + "source": {"bytes": image_data}, + } + }, + ], + "status": "success", + } + } + ], + } + ] + + changed = manager._truncate_tool_results(messages, 0) + + assert changed + tool_result_items = messages[0]["content"][0]["toolResult"]["content"] + assert not any(isinstance(item, dict) and "image" in item for item in tool_result_items) + expected_placeholder = f"[image: jpeg, {len(image_data)} bytes]" + assert any(isinstance(item, dict) and item.get("text") == expected_placeholder for item in tool_result_items) + + +def test_already_truncated_text_not_truncated_again(): + """A text block that already contains the truncation marker is not truncated a second time.""" + manager = SlidingWindowConversationManager(window_size=10) + already_truncated = "A" * 200 + "...\n\n... [truncated: 990 chars removed] ...\n\n..." + "Z" * 200 + messages = [ + { + "role": "user", + "content": [ + { + "toolResult": { + "toolUseId": "1", + "content": [{"text": already_truncated}], + "status": "success", + } + } + ], + } + ] + + changed = manager._truncate_tool_results(messages, 0) + + assert not changed + assert messages[0]["content"][0]["toolResult"]["content"][0]["text"] == already_truncated + + +def test_short_text_in_tool_result_not_truncated(): + """Text no longer than 2 * _PRESERVE_CHARS must not be modified.""" + manager = SlidingWindowConversationManager(window_size=10) + short_text = "X" * 100 # 100 < 2 * 200 + messages = [ + { + "role": "user", + "content": [{"toolResult": {"toolUseId": "1", "content": [{"text": short_text}], "status": "success"}}], + } + ] + + changed = manager._truncate_tool_results(messages, 0) + + assert not changed + assert messages[0]["content"][0]["toolResult"]["content"][0]["text"] == short_text + + +def test_boundary_text_in_tool_result_not_truncated(): + """Text of exactly 2 * _PRESERVE_CHARS must not be truncated.""" + manager = SlidingWindowConversationManager(window_size=10) + boundary_text = "X" * 400 # exactly 2 * 200 + messages = [ + { + "role": "user", + "content": [{"toolResult": {"toolUseId": "1", "content": [{"text": boundary_text}], "status": "success"}}], + } + ] + + changed = manager._truncate_tool_results(messages, 0) + + assert not changed + assert messages[0]["content"][0]["toolResult"]["content"][0]["text"] == boundary_text