Skip to content

strands.agent.conversation_manager.compression.context_compression

Shared helpers for context compression strategies.

These functions are used by both the conversation managers (reactive/proactive compression) and the agentic context-management tools (model-driven compression). They cover the low-level mechanics: finding safe split/trim boundaries that don’t break tool-use/tool-result pairs, generating a summary via the model, and filtering messages by type.

Filter selecting which messages a compression operation targets.

  • "tools": only messages containing a toolUse or toolResult block.
  • "messages": only messages without any toolUse or toolResult block.
  • "all": every message.
def adjust_split_point_for_tool_pairs(messages: list[Message],
split_point: int) -> int

Defined in: src/strands/agent/conversation_manager/compression/context_compression.py:62

Adjust a split point forward to avoid breaking toolUse/toolResult pairs.

Walks the split point forward until the message at that position is neither an orphaned toolResult nor a toolUse without an immediately following toolResult.

Arguments:

  • messages - The full list of messages.
  • split_point - The initially calculated split point.

Returns:

The adjusted split point that doesn’t break a toolUse/toolResult pair.

Raises:

  • ContextWindowOverflowException - If the split point exceeds the message array length, or if no valid split point can be found (walked past all messages).
def find_valid_trim_point(messages: list[Message], start_index: int) -> int

Defined in: src/strands/agent/conversation_manager/compression/context_compression.py:107

Find a valid trim point for truncation starting at start_index.

A valid trim point must:

  1. Be a user message (required by most model providers)
  2. Not be an orphaned toolResult
  3. Not be a toolUse unless its toolResult immediately follows

Arguments:

  • messages - The full list of messages.
  • start_index - The index to begin searching from.

Returns:

The valid trim index, or len(messages) if none is found.

async def generate_summary(messages_to_summarize: list[Message],
model: "Model",
system_prompt: str | None = None) -> Message

Defined in: src/strands/agent/conversation_manager/compression/context_compression.py:149

Generate a summary of the provided messages by calling the model directly.

This bypasses the full agent pipeline (lock, metrics, traces, tool loop) and simply asks the underlying model to summarize the conversation.

Arguments:

  • messages_to_summarize - The messages to summarize.
  • model - The model used to generate the summary.
  • system_prompt - Optional system prompt override. Defaults to :data:DEFAULT_SUMMARIZATION_PROMPT.

Returns:

A user-role message containing the model-generated summary.

Raises:

  • RuntimeError - If the model fails to produce a response.
def matches_message_type(message: Message, filter: MessageType) -> bool

Defined in: src/strands/agent/conversation_manager/compression/context_compression.py:195

Return True if the message matches the given type filter.

Arguments:

  • message - The message to test.
  • filter - The message-type filter ("tools", "messages", or "all").

Returns:

True if the message matches the filter.