feat: implement memory optimization

cookbook/memory/09_memory_summarize_strategy.py

@@ -0,0 +1,107 @@
+"""
+Memory optimization using the "summarize" strategy.
+
+The "summarize" strategy combines all memories into a single comprehensive summary,
+achieving maximum compression. This is useful when you need aggressive token
+reduction and can accept losing the individual memory structure.
+
+Run: python cookbook/memory/09_memory_summarize_strategy.py
+"""
+
+from agno.agent import Agent
+from agno.db.postgres import PostgresDb
+from agno.memory import MemoryManager, SummarizeStrategy
+from agno.memory.strategy import MemoryOptimizationStrategyType
+from agno.models.openai import OpenAIChat
+
+db_url = "postgresql+psycopg://ai:ai@localhost:5532/ai"
+db = PostgresDb(db_url=db_url)
+
+user_id = "user2"
+
+# Create agent with memory enabled
+agent = Agent(
+    model=OpenAIChat(id="gpt-4o-mini"),
+    db=db,
+    enable_user_memories=True,
+)
+
+# Create some memories for a user
+print("Creating memories...")
+agent.print_response(
+    "I have a wonderful pet dog named Max who is 3 years old. He's a golden retriever and he's such a friendly and energetic dog. "
+    "We got him as a puppy when he was just 8 weeks old. He loves playing fetch in the park and going on long walks. "
+    "Max is really smart too - he knows about 15 different commands and tricks. Taking care of him has been one of the most "
+    "rewarding experiences of my life. He's basically part of the family now.",
+    user_id=user_id,
+)
+agent.print_response(
+    "I currently live in San Francisco, which is an amazing city despite all its challenges. I've been here for about 5 years now. "
+    "I work in the tech industry as a product manager at a mid-sized software company. The tech scene here is incredible - "
+    "there are so many smart people working on interesting problems. The cost of living is definitely high, but the opportunities "
+    "and the community make it worthwhile. I live in the Mission district which has great food and a vibrant culture.",
+    user_id=user_id,
+)
+agent.print_response(
+    "On weekends, I really enjoy hiking in the beautiful areas around the Bay Area. There are so many amazing trails - "
+    "from Mount Tamalpais to Big Basin Redwoods. I usually go hiking with a group of friends and we try to explore new trails every month. "
+    "I also love trying new restaurants. San Francisco has such an incredible food scene with cuisines from all over the world. "
+    "I'm always on the lookout for hidden gems and new places to try. My favorite types of cuisine are Japanese, Thai, and Mexican.",
+    user_id=user_id,
+)
+agent.print_response(
+    "I've been learning to play the piano for about a year and a half now. It's something I always wanted to do but never had time for. "
+    "I finally decided to commit to it and I practice almost every day, usually for 30-45 minutes. "
+    "I'm working through classical pieces right now - I can play some simple Bach and Mozart compositions. "
+    "My goal is to eventually be able to play some jazz piano as well. Having a creative hobby like this has been great for my mental health "
+    "and it's nice to have something completely different from my day job.",
+    user_id=user_id,
+)
+
+# Check current memories
+print("\nBefore optimization:")
+memories_before = agent.get_user_memories(user_id=user_id)
+print(f"  Memory count: {len(memories_before)}")
+
+# Count tokens before optimization
+strategy = SummarizeStrategy()
+tokens_before = strategy.count_tokens(memories_before)
+print(f"  Token count: {tokens_before} tokens")
+
+print("\nIndividual memories:")
+for i, memory in enumerate(memories_before, 1):
+    print(f"  {i}. {memory.memory}")
+
+# Create memory manager and optimize memories
+memory_manager = MemoryManager(
+    model=OpenAIChat(id="gpt-4o-mini"),
+    db=db,
+)
+
+print("\nOptimizing memories with 'summarize' strategy...")
+memory_manager.optimize_memories(
+    user_id=user_id,
+    strategy=MemoryOptimizationStrategyType.SUMMARIZE,  # Combine all memories into one
+    apply=True,  # Apply changes to database
+)
+
+# Check optimized memories
+print("\nAfter optimization:")
+memories_after = agent.get_user_memories(user_id=user_id)
+print(f"  Memory count: {len(memories_after)}")
+
+# Count tokens after optimization
+tokens_after = strategy.count_tokens(memories_after)
+print(f"  Token count: {tokens_after} tokens")
+
+# Calculate reduction
+if tokens_before > 0:
+    reduction_pct = ((tokens_before - tokens_after) / tokens_before) * 100
+    tokens_saved = tokens_before - tokens_after
+    print(f"  Reduction: {reduction_pct:.1f}% ({tokens_saved} tokens saved)")
+
+if memories_after:
+    print("\nSummarized memory:")
+    print(f"  {memories_after[0].memory}")
+else:
+    print("\n No memories found after optimization")

cookbook/memory/10_custom_memory_strategy.py

@@ -0,0 +1,155 @@
+"""
+Creating a custom memory optimization strategy.
+
+This cookbook shows how to create your own memory optimization strategy by
+subclassing MemoryOptimizationStrategy. This is useful when you need custom
+logic that the built-in "summarize" and "merge" strategies don't provide.
+
+Run: python cookbook/memory/10_custom_memory_strategy.py
+"""
+
+from datetime import datetime
+from typing import List, Optional
+
+from agno.agent import Agent
+from agno.db.postgres import PostgresDb
+from agno.db.schemas import UserMemory
+from agno.memory import MemoryManager, MemoryOptimizationStrategy
+from agno.models.base import Model
+from agno.models.openai import OpenAIChat
+
+
+# Define a custom strategy that keeps only the N most recent memories
+class RecentOnlyStrategy(MemoryOptimizationStrategy):
+    """Keep only the N most recent memories."""
+
+    def __init__(self, keep_count: int = 2):
+        self.keep_count = keep_count
+
+    def get_system_prompt(self) -> str:
+        """Not used by this strategy."""
+        return "RecentOnlyStrategy doesn't use LLM"
+
+    def optimize(
+        self,
+        memories: List[UserMemory],
+        model: Model,
+        user_id: Optional[str] = None,
+    ) -> List[UserMemory]:
+        """Keep only the most recent N memories."""
+        # Sort by updated_at or created_at, most recent first
+        sorted_memories = sorted(
+            memories,
+            key=lambda m: m.updated_at or m.created_at or datetime.min,
+            reverse=True,
+        )
+        # Keep only the specified number
+        return sorted_memories[: self.keep_count]
+
+    async def aoptimize(
+        self,
+        memories: List[UserMemory],
+        model: Model,
+        user_id: Optional[str] = None,
+    ) -> List[UserMemory]:
+        """Async version: Keep only the most recent N memories."""
+        sorted_memories = sorted(
+            memories,
+            key=lambda m: m.updated_at or m.created_at or datetime.min,
+            reverse=True,
+        )
+        # Keep only the specified number
+        return sorted_memories[: self.keep_count]
+
+
+# Example usage
+db_url = "postgresql+psycopg://ai:ai@localhost:5532/ai"
+db = PostgresDb(db_url=db_url)
+
+user_id = "user3"
+
+agent = Agent(
+    model=OpenAIChat(id="gpt-4o-mini"),
+    db=db,
+    enable_user_memories=True,
+)
+
+# Create some memories
+print("Creating memories...")
+agent.print_response(
+    "I'm currently learning machine learning and it's been an incredible journey so far. I started about 6 months ago with the basics - "
+    "linear regression, decision trees, and simple classification algorithms. Now I'm diving into more advanced topics like deep learning "
+    "and neural networks. I'm using Python with libraries like scikit-learn, TensorFlow, and PyTorch. "
+    "The math can be challenging sometimes, especially the calculus and linear algebra, but I'm working through it step by step.",
+    user_id=user_id,
+)
+agent.print_response(
+    "I recently completed an excellent online course on neural networks from Coursera. The course covered everything from basic perceptrons "
+    "to complex architectures like CNNs and RNNs. The instructor did a great job explaining backpropagation and gradient descent. "
+    "I completed all the programming assignments where we built neural networks from scratch and also used TensorFlow. "
+    "The final project was building an image classifier that achieved 92% accuracy on the test set. I'm really proud of that accomplishment.",
+    user_id=user_id,
+)
+agent.print_response(
+    "My ultimate goal is to build my own AI projects that solve real-world problems. I have several ideas I want to explore - "
+    "maybe a recommendation system, a chatbot for customer service, or perhaps something in computer vision. "
+    "I'm trying to identify problems where AI can make a real difference and where I have the skills to build something meaningful. "
+    "I know I need more experience and practice, but I'm committed to working on personal projects to build my portfolio.",
+    user_id=user_id,
+)
+agent.print_response(
+    "I'm particularly interested in natural language processing applications. The recent advances in large language models are fascinating. "
+    "I've been experimenting with transformer architectures and trying to understand how attention mechanisms work. "
+    "I'd love to work on projects involving text classification, sentiment analysis, or maybe even building conversational AI. "
+    "NLP feels like it's at the cutting edge right now and there are so many interesting problems to solve in this space.",
+    user_id=user_id,
+)
+
+# Check current memories
+print("\nBefore optimization:")
+memories_before = agent.get_user_memories(user_id=user_id)
+print(f"  Memory count: {len(memories_before)}")
+
+# Count tokens before optimization using custom strategy instance
+custom_strategy = RecentOnlyStrategy(keep_count=2)
+tokens_before = custom_strategy.count_tokens(memories_before)
+print(f"  Token count: {tokens_before} tokens")
+
+print("\nAll memories:")
+for i, memory in enumerate(memories_before, 1):
+    print(f"  {i}. {memory.memory}")
+
+# Use custom strategy to keep only 2 most recent memories
+print("\nOptimizing with custom RecentOnlyStrategy (keep_count=2)...")
+
+memory_manager = MemoryManager(
+    model=OpenAIChat(id="gpt-4o-mini"),
+    db=db,
+)
+
+memory_manager.optimize_memories(
+    user_id=user_id,
+    strategy=custom_strategy,  # Pass custom strategy instance
+    apply=True,  # Apply changes to database
+)
+
+# Check optimized memories
+print("\nAfter optimization:")
+memories_after = agent.get_user_memories(user_id=user_id)
+print(f"  Memory count: {len(memories_after)}")
+
+# Count tokens after optimization
+tokens_after = custom_strategy.count_tokens(memories_after)
+print(f"  Token count: {tokens_after} tokens")
+
+# Show what was kept/removed
+if tokens_before > 0:
+    reduction_pct = ((tokens_before - tokens_after) / tokens_before) * 100
+    tokens_saved = tokens_before - tokens_after
+    print(
+        f"  Reduction: {reduction_pct:.1f}% ({tokens_saved} tokens saved by keeping 2 most recent)"
+    )
+
+print("\nRemaining memories (2 most recent):")
+for i, memory in enumerate(memories_after, 1):
+    print(f"  {i}. {memory.memory}")

libs/agno/agno/memory/__init__.py

@@ -1,3 +1,16 @@
 from agno.memory.manager import MemoryManager, UserMemory
+from agno.memory.strategies import SummarizeStrategy
+from agno.memory.strategy import (
+    MemoryOptimizationStrategy,
+    MemoryOptimizationStrategyFactory,
+    MemoryOptimizationStrategyType,
+)
 
-__all__ = ["MemoryManager", "UserMemory"]
+__all__ = [
+    "MemoryManager",
+    "UserMemory",
+    "MemoryOptimizationStrategy",
+    "MemoryOptimizationStrategyType",
+    "MemoryOptimizationStrategyFactory",
+    "SummarizeStrategy",
+]