From dbb52675fd9f1ed99ca351f407692395ddc700c6 Mon Sep 17 00:00:00 2001 From: Kihyeon Myung Date: Mon, 23 Feb 2026 20:09:21 -0700 Subject: [PATCH 1/3] feat(conversation-manager): improve tool result truncation strategy Partially truncate large tool results (preserve first/last 200 chars) instead of replacing with error. Replace image blocks with placeholders, and target oldest tool results first. --- .../sliding_window_conversation_manager.py | 139 +++++++--- tests/strands/agent/test_agent.py | 10 +- .../agent/test_conversation_manager.py | 246 ++++++++++++++++-- 3 files changed, 330 insertions(+), 65 deletions(-) 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..d42590b86 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. @@ -23,7 +26,13 @@ class SlidingWindowConversationManager(ConversationManager): 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 +53,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 +169,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 +209,15 @@ 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 (both at the top level of the message content and nested inside tool + result content) are replaced with a short descriptive placeholder. Args: messages: The conversation message history. @@ -214,50 +231,90 @@ def _truncate_tool_results(self, messages: Messages, msg_idx: int) -> bool: 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}] + new_content: list[ContentBlock] = [] + + for content in message.get("content", []): + # Replace top-level image blocks with a descriptive placeholder + if "image" in content: + image: Any = content["image"] + media_type = image.get("format", "unknown") + source: Any = image.get("source", {}) + data = source.get("bytes", b"") + size = len(data) if data else 0 + new_content.append({"text": f"[image: {media_type}, {size} bytes]"}) changes_made = True + continue + + 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: + image = item["image"] + media_type = image.get("format", "unknown") + source = image.get("source", {}) + data = source.get("bytes", b"") + size = len(data) if data else 0 + new_items.append({"text": f"[image: {media_type}, {size} bytes]"}) + 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..b88042910 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,211 @@ 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_in_message_content_replaced_with_placeholder(): + """Top-level image blocks in a message are replaced with a text placeholder.""" + manager = SlidingWindowConversationManager(window_size=10) + image_data = b"abc123" + messages = [ + { + "role": "user", + "content": [ + { + "toolResult": { + "toolUseId": "1", + "content": [{"text": "result"}], + "status": "success", + } + }, + { + "image": { + "format": "png", + "source": {"bytes": image_data}, + } + }, + ], + } + ] + + changed = manager._truncate_tool_results(messages, 0) + + assert changed + # The image block should be replaced + content = messages[0]["content"] + assert any(isinstance(c, dict) and c.get("text") == f"[image: png, {len(image_data)} bytes]" for c in content) + # No raw image blocks should remain + assert not any(isinstance(c, dict) and "image" in c for c in content) + + +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 From a0e19251368397e3e75bf95ef4454914bdd6b74d Mon Sep 17 00:00:00 2001 From: Kihyeon Myung Date: Thu, 26 Feb 2026 18:13:18 -0700 Subject: [PATCH 2/3] docs: update class docstring to describe truncation behavior --- .../sliding_window_conversation_manager.py | 5 +++++ 1 file changed, 5 insertions(+) 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 d42590b86..56acfc512 100644 --- a/src/strands/agent/conversation_manager/sliding_window_conversation_manager.py +++ b/src/strands/agent/conversation_manager/sliding_window_conversation_manager.py @@ -23,6 +23,11 @@ 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 rather than completely + replaced — preserving the first and last 200 characters — and image blocks 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. """ From fb640291f287b6e79c1ab579599414e2c4541ea4 Mon Sep 17 00:00:00 2001 From: Kihyeon Myung Date: Fri, 27 Feb 2026 11:35:02 -0700 Subject: [PATCH 3/3] refactor: extract image placeholder helper --- .../sliding_window_conversation_manager.py | 35 +++++++------------ .../agent/test_conversation_manager.py | 35 ------------------- 2 files changed, 12 insertions(+), 58 deletions(-) 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 56acfc512..b97de0b06 100644 --- a/src/strands/agent/conversation_manager/sliding_window_conversation_manager.py +++ b/src/strands/agent/conversation_manager/sliding_window_conversation_manager.py @@ -23,10 +23,10 @@ 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 rather than completely - replaced — preserving the first and last 200 characters — and image blocks 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. + 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. """ @@ -221,8 +221,7 @@ def _truncate_tool_results(self, messages: Messages, msg_idx: int) -> bool: and the removed middle is replaced with a notice indicating how many characters were removed. The tool result status is not changed. - Image blocks (both at the top level of the message content and nested inside tool - result content) are replaced with a short descriptive placeholder. + Image blocks nested inside tool result content are replaced with a short descriptive placeholder. Args: messages: The conversation message history. @@ -234,22 +233,17 @@ 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 new_content: list[ContentBlock] = [] for content in message.get("content", []): - # Replace top-level image blocks with a descriptive placeholder - if "image" in content: - image: Any = content["image"] - media_type = image.get("format", "unknown") - source: Any = image.get("source", {}) - data = source.get("bytes", b"") - size = len(data) if data else 0 - new_content.append({"text": f"[image: {media_type}, {size} bytes]"}) - changes_made = True - continue - if "toolResult" in content: tool_result: Any = content["toolResult"] tool_result_items = tool_result.get("content", []) @@ -259,12 +253,7 @@ def _truncate_tool_results(self, messages: Messages, msg_idx: int) -> bool: for item in tool_result_items: # Replace image items nested inside toolResult content if "image" in item: - image = item["image"] - media_type = image.get("format", "unknown") - source = image.get("source", {}) - data = source.get("bytes", b"") - size = len(data) if data else 0 - new_items.append({"text": f"[image: {media_type}, {size} bytes]"}) + new_items.append({"text": _image_placeholder(item["image"])}) item_changed = True continue diff --git a/tests/strands/agent/test_conversation_manager.py b/tests/strands/agent/test_conversation_manager.py index b88042910..fd88954e8 100644 --- a/tests/strands/agent/test_conversation_manager.py +++ b/tests/strands/agent/test_conversation_manager.py @@ -498,41 +498,6 @@ def test_truncation_does_not_change_status_to_error(): assert messages[0]["content"][0]["toolResult"]["status"] == "success" -def test_image_blocks_in_message_content_replaced_with_placeholder(): - """Top-level image blocks in a message are replaced with a text placeholder.""" - manager = SlidingWindowConversationManager(window_size=10) - image_data = b"abc123" - messages = [ - { - "role": "user", - "content": [ - { - "toolResult": { - "toolUseId": "1", - "content": [{"text": "result"}], - "status": "success", - } - }, - { - "image": { - "format": "png", - "source": {"bytes": image_data}, - } - }, - ], - } - ] - - changed = manager._truncate_tool_results(messages, 0) - - assert changed - # The image block should be replaced - content = messages[0]["content"] - assert any(isinstance(c, dict) and c.get("text") == f"[image: png, {len(image_data)} bytes]" for c in content) - # No raw image blocks should remain - assert not any(isinstance(c, dict) and "image" in c for c in content) - - 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)