Spaces:
Paused
Paused
File size: 4,126 Bytes
b9f94e1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 | """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}
|