-
Notifications
You must be signed in to change notification settings - Fork 4.6k
feat: Implement memory optimization #5229
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
base: main
Are you sure you want to change the base?
Conversation
libs/agno/agno/memory/strategy.py
Outdated
| raise ValueError(f"Unsupported optimization strategy: {strategy_name}. Valid options: {[e.value for e in cls]}") | ||
|
|
||
|
|
||
| class MemoryOptimizationStrategy(ABC): |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Base class should be in strategies.
libs/agno/agno/memory/strategy.py
Outdated
| class MemoryOptimizationStrategy(ABC): | ||
| """Base class for memory optimization strategies.""" | ||
|
|
||
| def count_tokens(self, memories: List["UserMemory"]) -> int: |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This should be a general utility function. We can use it in many places.
| from agno.utils.log import log_debug | ||
|
|
||
|
|
||
| class MergeStrategy(MemoryOptimizationStrategy): |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This is probably the only strategy that makes sense for now. Replace all memories with a single memory.
|
|
||
| # Concatenate input and feedback fields | ||
| input_parts = [mem.input for mem in memories if mem.input] | ||
| merged_input = "\n---\n".join(input_parts) if input_parts else None |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think that doesn't make sense to keep. The input kinda falls away if you optimize
| merged_input = "\n---\n".join(input_parts) if input_parts else None | ||
|
|
||
| feedback_parts = [mem.feedback for mem in memories if mem.feedback] | ||
| merged_feedback = "\n---\n".join(feedback_parts) if feedback_parts else None |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Same with feedback (although we don't use it)
| feedback_parts = [mem.feedback for mem in memories if mem.feedback] | ||
| merged_feedback = "\n---\n".join(feedback_parts) if feedback_parts else None | ||
|
|
||
| # Check if agent_id and team_id are consistent |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
These as well. Since a whole new memory is created the original memory creator is lost
libs/agno/agno/memory/manager.py
Outdated
| self, | ||
| user_id: Optional[str] = None, | ||
| token_limit: int = 2000, | ||
| strategy: Union[str, MemoryOptimizationStrategyType, MemoryOptimizationStrategy] = "summarize", |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We should only support MemoryOptimizationStrategyType and MemoryOptimizationStrategy
libs/agno/agno/memory/manager.py
Outdated
| Args: | ||
| user_id: User ID to optimize memories for. Defaults to "default". | ||
| token_limit: Maximum tokens for all memories combined. |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This doesn't work for summarize, because there the token limit is per memory?
In general I don't think we should have token limit, it doesn't really make sense. The Model should just summarize.
libs/agno/agno/memory/manager.py
Outdated
| - String: "summarize", "merge" | ||
| - Enum: MemoryOptimizationStrategyType.SUMMARIZE, MemoryOptimizationStrategyType.MERGE | ||
| - Instance: Custom MemoryOptimizationStrategy instance | ||
| model: Model to use for optimization. Defaults to manager's model. |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I don't think this should be a parameter. You should just create the memory manager with a model and call optimize.
libs/agno/agno/memory/manager.py
Outdated
|
|
||
| # Get strategy instance | ||
| if isinstance(strategy, str): | ||
| strategy_type = MemoryOptimizationStrategyType.from_string(strategy) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Don't need a from_string, it is an enum so MemoryOptimizationStrategyType[strategy] should work
libs/agno/agno/memory/manager.py
Outdated
| current_tokens = strategy_instance.count_tokens(memories) | ||
| log_debug(f"Current token count: {current_tokens}, Target: {token_limit}") | ||
|
|
||
| if current_tokens <= token_limit: |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think this is a bit overengineerd. Let the user decide themselves when they want to optimize. We can give them the util for counting the tokens if needed.
libs/agno/agno/memory/manager.py
Outdated
| for opt_mem in optimized_memories: | ||
| if opt_mem.memory_id: | ||
| # Update existing memory | ||
| self.replace_user_memory( |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I believe replace is upsert, so we should be able to just call that for all cases.
libs/agno/agno/memory/strategy.py
Outdated
| raise NotImplementedError | ||
|
|
||
| @abstractmethod | ||
| def optimize( |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
And aoptimize
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think this should move to the init of the strategies directory
| from uuid import uuid4 | ||
|
|
||
| from agno.db.schemas import UserMemory | ||
| from agno.memory.strategies.base import MemoryOptimizationStrategy |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
There is no base?
| self, | ||
| memories: List[UserMemory], | ||
| model: Model, | ||
| user_id: Optional[str] = None, |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I don't think this needs to be a parameter. You can just grab it from the first memory?
libs/agno/agno/memory/manager.py
Outdated
| strategy: Union[ | ||
| MemoryOptimizationStrategyType, MemoryOptimizationStrategy | ||
| ] = MemoryOptimizationStrategyType.SUMMARIZE, | ||
| apply: bool = False, |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think this should be true by default. That seems like the default behaviour and that a "dry run" should be opt-in?
| mem for mem in memories if mem.memory_id and mem.memory_id not in optimized_memory_ids | ||
| ] | ||
|
|
||
| # Delete removed memories |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Shouldn't we rather just clear all the user's memories and add the new ones? So it is a complete replace.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This is doing the same thing. It's looping through all the user's memory and then deleting them. Do you want me to call any specific function which already pre-exists?
Summary
Describe key changes, mention related issues or motivation for the changes.
3 main classes:
MemoryOptimizationStrategyType(Enum)from_string()method for backward compatibility allows to use stratergies as simple string Eg:stratergy="merge"as well. This gives good DXMemoryOptimizationStrategy(Abstract Base Class)MemoryOptimizationStrategyFactoryChunkingStrategyFactoryUsers can also define their custom strategy.
Type of change
Checklist
./scripts/format.shand./scripts/validate.sh)Additional Notes
Add any important context (deployment instructions, screenshots, security considerations, etc.)