​
ContextSummarizerToolkit

class ContextSummarizerToolkit(BaseToolkit):
A toolkit that provides intelligent context summarization and management for agents. This toolkit enables agents to compress conversation context through intelligent summarization, save conversation history to markdown files, and search through past conversations. It handles all context management needs in a single toolkit. Key features:
  • Intelligent context compression with over-compression prevention
  • Markdown file storage with session management
  • Simple text-based search through conversation history
  • Configurable summarization prompts
  • Context loading and saving capabilities

​
init

def __init__(
    self,
    agent: 'ChatAgent',
    working_directory: Optional[str] = None,
    timeout: Optional[float] = None,
    summary_prompt_template: Optional[str] = None
):
Initialize the ContextSummarizerToolkit. Parameters:
  • agent (ChatAgent): The agent that is using the toolkit. This is required to access the agent’s memory.
  • working_directory (str, optional): The directory path where notes will be stored. If not provided, a default directory will be used.
  • timeout (Optional[float]): The timeout for the toolkit.
  • summary_prompt_template (Optional[str]): Custom prompt template for summarization. If None, a default task-focused template is used. Users can customize this for different use cases.

​
_setup_storage

def _setup_storage(self, working_directory: Optional[str]):
Initialize storage paths and create session-specific directories using ContextUtility for file management.

​
_summarize_messages

def _summarize_messages(self, memory_records: List['MemoryRecord']):
Generate a summary of the conversation context. Parameters:
  • memory_records (List["MemoryRecord"]): A list of memory records to summarize.
Returns: str: The summary of the conversation context.

​
_save_summary

def _save_summary(self, summary: str):
Persist conversation summary to markdown file with metadata including timestamp and session information. Parameters:
  • summary (str): The summary text to save.
Returns: str: Success message or error message.

​
_save_history

def _save_history(self, memory_records: List['MemoryRecord']):
Export complete conversation transcript as formatted markdown with message roles, agent IDs, and content structure preserved. Parameters:
  • memory_records (List["MemoryRecord"]): The list of memory records to save.
Returns: str: Success message or error message.

​
_compress_and_save

def _compress_and_save(self, memory_records: List['MemoryRecord']):
Complete compression pipeline: summarize and save both history and summary. Parameters:
  • memory_records (List["MemoryRecord"]): The memory records to compress and save.
Returns: str: The generated summary text.

​
_load_summary

def _load_summary(self):
Returns: str: The summary content, or empty string if not found.

​
_load_history

def _load_history(self):
Returns: str: The history content, or empty string if not found.

​
_format_conversation

def _format_conversation(self, memory_records: List['MemoryRecord']):
Convert memory records into human-readable conversation format with role names and message content for summarization processing. Parameters:
  • memory_records (List["MemoryRecord"]): A list of memory records to format.
Returns: str: The formatted conversation.

​
_create_summary_prompt

def _create_summary_prompt(self, conversation_text: str):
Construct detailed summarization prompt with instructions for extracting key information, goals, and progress from conversation. Parameters:
  • conversation_text (str): The formatted conversation to summarize.
Returns: str: The complete prompt for summarization.

​
summarize_full_conversation_history

def summarize_full_conversation_history(self):
Returns: str: Success message with brief summary, or error message.

​
_refresh_context_with_summary

def _refresh_context_with_summary(self, summary: str):
Empty the agent’s memory and replace it with a summary of the conversation history. Parameters:
  • summary (str): The summary of the conversation history.
Returns: bool: True if the context was refreshed successfully, False otherwise.

​
get_conversation_memory_info

def get_conversation_memory_info(self):
Returns: str: Information about current memory and saved files.

​
search_full_conversation_history

def search_full_conversation_history(self, keywords: List[str], top_k: int = 4):
Search the conversation history using keyword matching. This is used when information is missing from the summary and the current conversation, and can potentially be found in the full conversation history before it was summarized. Searches through the current session’s history.md file to find the top messages that contain the most keywords. Parameters:
  • keywords (List[str]): List of keywords to search for. The keywords must be explicitly related to the information the user is looking for, and not general terms that might be found about any topic. For example, if the user is searching for the price of the flight to “Paris” which was discussed previously, the keywords should be [“Paris”, “price”, “flight”, ”$”, “costs”].
  • top_k (int): The number of results to return (default 4).
Returns: str: The search results or error message.

​
should_compress_context

def should_compress_context(self, message_limit: int = 40, token_limit: Optional[int] = None):
Check if context should be compressed based on limits. Parameters:
  • message_limit (int): Maximum number of messages before compression.
  • token_limit (Optional[int]): Maximum number of tokens before compression.
Returns: bool: True if context should be compressed.

​
reset

def reset(self):
Clear all compression state including stored summaries, compressed message tracking, and compression counters.

​
get_current_summary

def get_current_summary(self):
Returns: Optional[str]: The current summary, or None if no summary exists.

​
set_summary

def set_summary(self, summary: str):
Override the current in-memory summary with provided content without affecting saved files or compression tracking. Parameters:
  • summary (str): The summary to store.

​
get_tools

def get_tools(self):
Returns: List[FunctionTool]: The list of tools.