Skip to content

feat: add enhanced memory management foundation #626

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 2 commits into
base: main
Choose a base branch
from

Conversation

yike5460
Copy link

@yike5460 yike5460 commented Aug 7, 2025

Description

The current agent state management in Strands lacks sophisticated memory organization capabilities. As multi-agent workflows become more complex, agents need intelligent memory management to handle context efficiently, prevent memory bloat, and maintain performance at scale. This PR introduces a foundational memory management system that categorizes agent state by usage patterns, automatically manages, memory lifecycle, and provides comprehensive monitoring - all while maintaining complete backward compatibility with existing AgentState usage.

Related Issues

Documentation PR

Architecture Overview

The implementation follows a layered approach with 4 core components:

  • Configuration Layer (config.py) - Defines memory categories, thresholds, and behavior policies
  • Enhanced State Layer (enhanced_state.py) - Extends AgentState with categorized memory management
  • Lifecycle Management (lifecycle.py) - Handles automatic promotion, demotion, and cleanup
  • Metrics Layer (metrics.py) - Tracks performance and provides analytics

Key Design

  • Backward Compatibility First: The EnhancedAgentState inherits from AgentState and preserves all existing behavior. Existing code works unchanged.
  • Tiered Memory Model: Memory items are categorized into 4 tiers based on access patterns:
    -- ACTIVE: Currently being worked with, fastest access
    -- CACHED: Recently used but not currently active
    -- ARCHIVED: Historical data kept for potential retrieval
    -- METADATA: System information and statistics
  • Automatic Lifecycle Management: Items automatically transition between tiers based on access patterns and configurable time thresholds, reducing manual memory management overhead.

Type of Change

New feature

Testing

How have you tested the change? Verify that the changes do not break functionality or introduce warnings in consuming repositories: agents-docs, agents-tools, agents-cli

test coverage with new 65 unit tests and 100% pass rate (python -m pytest tests/strands/agent/memory/ -v) across all components, all tests achieve 100% pass rate with thorough error handling and boundary condition coverage.

  • [Y] I ran hatch run prepare

Checklist

  • [Y] I have read the CONTRIBUTING document
  • [Y] I have added any necessary tests that prove my fix is effective or my feature works
  • [Y] I have updated the documentation accordingly
  • [Y] I have added an appropriate example to the documentation to outline the feature, or no new docs are needed
  • [Y] My changes generate no new warnings
  • [Y] Any dependent changes have been merged and published

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

* Add tiered memory categorization system (ACTIVE, CACHED, ARCHIVED, METADATA)
* Implement EnhancedAgentState with backward compatibility to existing AgentState
* Add automatic memory lifecycle management with promotion, demotion, and archival
* Include comprehensive metrics tracking with trend analysis and performance monitoring
* Provide configurable memory thresholds and cleanup strategies
* Add emergency cleanup for memory limit enforcement
* Maintain full JSON validation and key validation from parent classes

Breaking: None - fully backward compatible with existing AgentState interface
Testing: 65 comprehensive unit tests with 100% pass rate
* Add missing docstrings for __init__ methods
* Fix variable name collision with unused loop variable
* Add Optional typing import and proper type annotations
* Create thresholds property to handle Optional MemoryThresholds
* Fix return type annotation for optimize_memory method
* Ensure proper type safety throughout memory management system
@yike5460
Copy link
Author

yike5460 commented Aug 7, 2025

Sample code for your reference:

Configurable Memory Policies

from strands.agent.memory import EnhancedAgentState, MemoryConfig, MemoryThresholds

# Use preset configurations
conservative_state = EnhancedAgentState(MemoryConfig.conservative())  # Lower memory usage
aggressive_state = EnhancedAgentState(MemoryConfig.aggressive())      # Higher performance
minimal_state = EnhancedAgentState(MemoryConfig.minimal())            # Bare minimum features

# Custom configuration
custom_config = MemoryConfig(
    enable_categorization=True,
    enable_lifecycle=True,
    thresholds=MemoryThresholds(
        active_memory_limit=8192,
        cleanup_threshold=0.8
    )
)
state = EnhancedAgentState(memory_config=custom_config)

Memory Analytics and Optimization

# Get memory usage statistics
stats = state.get_memory_stats()
print(f"Total items: {stats['total_items']}")
print(f"Hit rate: {stats['current_stats']['hit_rate']:.2%}")

# Automatic cleanup when needed
removed_count = state.cleanup_memory()
print(f"Cleaned up {removed_count} items")

# Memory optimization with detailed results
results = state.optimize_memory()
print(f"Reduced memory by {results['size_reduction_pct']:.1f}%")
print(f"Removed {results['items_removed']} items")

Advanced Memory Management

# Manual memory lifecycle control
state.configure_memory(MemoryConfig.aggressive())

# Category-based retrieval for performance-critical paths
critical_data = state.get_by_category(MemoryCategory.ACTIVE)
historical_data = state.get_by_category(MemoryCategory.ARCHIVED)

# Monitor memory trends over time
stats = state.get_memory_stats()
trends = stats.get('trends', {})
if trends.get('trend', 0) > 100:  # Growing by 100 units per measurement
    print("Memory usage trending upward - consider cleanup")

# Get comprehensive memory report
report = state.get_memory_stats()
print(f"Memory utilization: {report['current_stats']['distribution']}")
print(f"Performance metrics: {report['performance']}")

Integration with Existing Agents

import time
from strands.agent.memory import EnhancedAgentState, MemoryCategory

# Gradual migration approach
class MyAgent:
    def __init__(self):
        # Replace AgentState with EnhancedAgentState
        self.state = EnhancedAgentState()
        # All existing code continues to work unchanged
    
    def process_task(self, task):
        # Use new categorization for better memory management
        self.state.set("current_task", task, MemoryCategory.ACTIVE)
        self.state.set_metadata("task_start_time", time.time())
        
        # Existing logic works as before
        result = self.state.get("current_task")
        return result
    
    def cleanup_session(self):
        # Leverage new cleanup capabilities
        removed = self.state.cleanup_memory(force=True)
        stats = self.state.get_memory_stats()
        return {"removed_items": removed, "final_stats": stats}

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

Successfully merging this pull request may close these issues.

1 participant