diff options
Diffstat (limited to 'personal_agent/tools/memory.py')
| -rw-r--r-- | personal_agent/tools/memory.py | 194 |
1 files changed, 193 insertions, 1 deletions
diff --git a/personal_agent/tools/memory.py b/personal_agent/tools/memory.py index 2287d4e..851a69d 100644 --- a/personal_agent/tools/memory.py +++ b/personal_agent/tools/memory.py @@ -1,4 +1,7 @@ -from personal_agent.memory.models import WorkingMemory +from uuid import UUID + +from personal_agent.memory.models import LongTermMemory, MemoryCategory, WorkingMemory +from personal_agent.memory.store import LongTermMemoryStore from personal_agent.tools.base import Tool, ToolError, ToolResult @@ -195,3 +198,192 @@ class MemoryRemoveNoteTool(Tool): self.memory.remove_note(note) return ToolResult(content=f"Removed note: {note}") + + +class LongTermMemorySaveTool(Tool): + def __init__(self, store: LongTermMemoryStore) -> None: + self.store = store + + @property + def name(self) -> str: + return "memory_save" + + @property + def description(self) -> str: + return ( + "Save durable information for use across future conversations. " + "Use semantic for stable facts or preferences, and episodic for " + "meaningful past events." + ) + + @property + def schema(self) -> dict[str, object]: + return { + "type": "object", + "properties": { + "content": { + "type": "string", + "description": "The durable information to remember.", + }, + "category": { + "type": "string", + "enum": ["semantic", "episodic"], + "description": ( + "semantic for stable facts/preferences; " + "episodic for meaningful past events." + ), + }, + }, + "required": ["content", "category"], + } + + def execute(self, arguments: dict[str, object]) -> ToolResult: + content = arguments.get("content") + category = arguments.get("category") + + if not isinstance(content, str): + raise ToolError("content must be a string") + + if not content.strip(): + raise ToolError("content must not be empty") + + if not isinstance(category, str): + raise ToolError("category must be a string") + + try: + memory_category = MemoryCategory(category) + except ValueError: + raise ToolError("category must be 'semantic' or 'episodic'") + + memory = LongTermMemory.create(content, memory_category) + self.store.add(memory) + + return ToolResult(content=f"Saved memory: {memory.id}") + + +class LongTermMemorySearchTool(Tool): + def __init__(self, store: LongTermMemoryStore) -> None: + self.store = store + + @property + def name(self) -> str: + return "memory_search" + + @property + def description(self) -> str: + return ( + "Search long-term memory for information relevant to the current " + "conversation. Use semantic for stable facts or preferences, and " + "episodic for meaningful past events." + ) + + @property + def schema(self) -> dict[str, object]: + return { + "type": "object", + "properties": { + "query": { + "type": "string", + "description": "The information to search for.", + }, + "category": { + "type": "string", + "enum": ["semantic", "episodic"], + "description": "Optionally restrict results to one category.", + }, + "limit": { + "type": "integer", + "minimum": 1, + "description": "Maximum number of memories to return.", + }, + }, + "required": ["query"], + } + + def execute(self, arguments: dict[str, object]) -> ToolResult: + query = arguments.get("query") + category = arguments.get("category") + limit = arguments.get("limit", 10) + + if not isinstance(query, str): + raise ToolError("query must be a string") + + if not query.strip(): + raise ToolError("query must not be empty") + + if category is not None: + if not isinstance(category, str): + raise ToolError("category must be a string") + + try: + memory_category = MemoryCategory(category) + except ValueError: + raise ToolError("category must be 'semantic' or 'episodic'") + else: + memory_category = None + + if not isinstance(limit, int) or isinstance(limit, bool): + raise ToolError("limit must be an integer") + + if limit <= 0: + raise ToolError("limit must be positive") + + memories = self.store.search( + query=query, + category=memory_category, + limit=limit, + ) + + if not memories: + return ToolResult(content="No matching memories found.") + + lines = [ + f"- [{memory.category.value}] {memory.id}: {memory.content}" + for memory in memories + ] + + return ToolResult(content="\n".join(lines)) + + +class LongTermMemoryDeleteTool(Tool): + def __init__(self, store: LongTermMemoryStore) -> None: + self.store = store + + @property + def name(self) -> str: + return "memory_delete" + + @property + def description(self) -> str: + return ( + "Delete a long-term memory by its exact ID. Only delete a memory " + "when its ID is known from a previous memory search or save result." + ) + + @property + def schema(self) -> dict[str, object]: + return { + "type": "object", + "properties": { + "id": { + "type": "string", + "description": "The exact UUID of the memory to delete.", + }, + }, + "required": ["id"], + } + + def execute(self, arguments: dict[str, object]) -> ToolResult: + memory_id = arguments.get("id") + + if not isinstance(memory_id, str): + raise ToolError("id must be a string") + + try: + parsed_id = UUID(memory_id) + except ValueError: + raise ToolError("id must be a valid UUID") + + self.store.delete(parsed_id) + + return ToolResult(content=f"Deleted memory: {parsed_id}") |
