Spaces:
Paused
Paused
| """Memory tool implementation — wired to LongTermMemory for persistence.""" | |
| from __future__ import annotations | |
| import logging | |
| from typing import Any | |
| from hermes.tools.base.tool import BaseTool, ToolSchema | |
| logger = logging.getLogger(__name__) | |
| _long_term_memory: Any = None | |
| def _get_memory() -> Any: | |
| """Get or initialize the long-term memory backend.""" | |
| global _long_term_memory | |
| if _long_term_memory is None: | |
| from hermes.memory.long_term.memory import LongTermMemory | |
| _long_term_memory = LongTermMemory() | |
| return _long_term_memory | |
| class MemoryTool(BaseTool): | |
| """Tool for storing and retrieving memories with persistent backend.""" | |
| def __init__(self, memory_store: Any | None = None) -> None: | |
| super().__init__() | |
| self._memory_store = memory_store | |
| def _define_schema(self) -> ToolSchema: | |
| return ToolSchema( | |
| name="memory_store", | |
| description="Store and retrieve information in persistent memory", | |
| parameters={ | |
| "action": { | |
| "type": "string", | |
| "description": "Action to perform: store, retrieve, list, delete", | |
| }, | |
| "key": { | |
| "type": "string", | |
| "description": "Memory key", | |
| }, | |
| "value": { | |
| "type": "string", | |
| "description": "Value to store (for store action)", | |
| }, | |
| "category": { | |
| "type": "string", | |
| "description": "Memory category", | |
| "default": "general", | |
| }, | |
| "query": { | |
| "type": "string", | |
| "description": "Search query (for retrieve action)", | |
| }, | |
| }, | |
| required=["action"], | |
| category="memory", | |
| tags=["memory", "store", "retrieve"], | |
| ) | |
| async def execute(self, **kwargs: Any) -> dict[str, Any]: | |
| """Execute memory operation.""" | |
| action = kwargs["action"] | |
| try: | |
| if action == "store": | |
| return await self._store(kwargs["key"], kwargs["value"], kwargs.get("category", "general")) | |
| elif action == "retrieve": | |
| return await self._retrieve(kwargs.get("key"), kwargs.get("query"), kwargs.get("category")) | |
| elif action == "list": | |
| return await self._list(kwargs.get("category")) | |
| elif action == "delete": | |
| return await self._delete(kwargs["key"]) | |
| else: | |
| return {"error": f"Unknown action: {action}"} | |
| except Exception as e: | |
| logger.error(f"Memory operation error: {e}") | |
| return {"error": str(e)} | |
| async def _store(self, key: str, value: str, category: str) -> dict[str, Any]: | |
| """Store a memory in persistent backend.""" | |
| memory = _get_memory() | |
| await memory.store(key=key, value=value, category=category) | |
| return {"success": True, "key": key, "category": category, "persistent": True} | |
| async def _retrieve( | |
| self, key: str | None, query: str | None, category: str | None | |
| ) -> dict[str, Any]: | |
| """Retrieve memories from persistent backend.""" | |
| memory = _get_memory() | |
| if key: | |
| results = await memory.retrieve(key=key) | |
| if results: | |
| return {"found": True, "key": key, **results[0]} | |
| return {"found": False, "key": key} | |
| results = await memory.retrieve(query=query, category=category) | |
| return {"results": results, "count": len(results)} | |
| async def _list(self, category: str | None) -> dict[str, Any]: | |
| """List all memories.""" | |
| memory = _get_memory() | |
| results = await memory.list_all(category=category) | |
| return {"results": results, "count": len(results)} | |
| async def _delete(self, key: str) -> dict[str, Any]: | |
| """Delete a memory.""" | |
| memory = _get_memory() | |
| deleted = await memory.delete(key) | |
| return {"success": deleted, "key": key} | |