-
Notifications
You must be signed in to change notification settings - Fork 80
feat(memory): update session manager for Strands integration #170
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?
Changes from all commits
7f76f2c
58ed5f7
40f319c
ec7eea3
dcba595
7a8906b
b1115da
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,8 +1,27 @@ | ||
| """Configuration for AgentCore Memory Session Manager.""" | ||
|
|
||
| from typing import Dict, Optional | ||
| from typing import Dict, List, Optional | ||
|
|
||
| from pydantic import BaseModel, Field | ||
| from pydantic import BaseModel, Field, field_validator | ||
|
|
||
| from bedrock_agentcore.memory.constants import MessageRole | ||
| from bedrock_agentcore.memory.models import StringValue | ||
|
|
||
|
|
||
| class BranchConfig(BaseModel): | ||
| """Configuration for AgentCore Memory branching. | ||
|
|
||
| Attributes: | ||
| name: Descriptive name for the branch | ||
| root_event_id: ID of the event from which this branch originates | ||
| """ | ||
|
|
||
| name: str = Field(min_length=1) | ||
| root_event_id: Optional[str] = "" | ||
|
|
||
| def to_agentcore_format(self) -> dict: | ||
| """Convert to AgentCore Memory API format.""" | ||
| return {"name": self.name, "rootEventId": self.root_event_id} | ||
|
|
||
|
|
||
| class RetrievalConfig(BaseModel): | ||
|
|
@@ -21,6 +40,13 @@ class RetrievalConfig(BaseModel): | |
| initialization_query: Optional[str] = None | ||
|
|
||
|
|
||
| class ShortTermRetrievalConfig(BaseModel): | ||
| """Configuration for Short term memory retrieval operations""" | ||
|
|
||
| branch_filter: Optional[bool] = True | ||
| metadata: Optional[Dict[str, StringValue]] = None | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This is how you would filter by agent_id. Use to get a agent's memory
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. this is currently supportly supported.
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Customers don't really need metadata filtering just keep agent_id. |
||
|
|
||
|
|
||
| class AgentCoreMemoryConfig(BaseModel): | ||
| """Configuration for AgentCore Memory Session Manager. | ||
|
|
||
|
|
@@ -29,9 +55,29 @@ class AgentCoreMemoryConfig(BaseModel): | |
| session_id: Required unique ID for the session | ||
| actor_id: Required unique ID for the agent instance/user | ||
| retrieval_config: Optional dictionary mapping namespaces to retrieval configurations | ||
| default_branch: Optional default branch configuration for the session | ||
| message_types: Optional list of message types to filter | ||
| metadata: Optional dictionary of metadata to include with events | ||
| """ | ||
|
|
||
| memory_id: str = Field(min_length=1) | ||
| session_id: str = Field(min_length=1) | ||
| actor_id: str = Field(min_length=1) | ||
| retrieval_config: Optional[Dict[str, RetrievalConfig]] = None | ||
| default_branch: Optional[BranchConfig] = Field( | ||
| default_factory=lambda: BranchConfig(name="main", root_event_id="") | ||
| ) | ||
| short_term_retrieval_config: Optional[ShortTermRetrievalConfig] = ( | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. does this needs it's own variable? Can we allow for this to be set to the |
||
| ShortTermRetrievalConfig() | ||
| ) | ||
| message_types: Optional[List[str]] = Field(default=["user", "assistant"]) | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. we to emphasize its agentcore message_types ("user", "assistant", "toolResult", "other"). *check if this filter issues. add to default "other", "tool_use". |
||
| metadata: Optional[Dict[str, StringValue]] = ( | ||
| None # Currently only supports agent_id. Will be extended further. | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. the comment is lie. it allow customer's metada. |
||
| ) | ||
|
|
||
| @field_validator("memory_id", "session_id", "actor_id") | ||
| @classmethod | ||
| def validate_non_empty_strings(cls, v: str) -> str: | ||
| if not v or not v.strip(): | ||
| raise ValueError("must be a non-empty string") | ||
| return v | ||
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.
Is there a plan to have branching on LTM? We can refactor later but then we should consider lifting the
branch_filterinto a base classThere 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.
yes there is.