File size: 3,692 Bytes
b3d741d 167a572 b3d741d 167a572 b3d741d 167a572 b3d741d 167a572 b3d741d 167a572 b3d741d 167a572 b3d741d 167a572 b3d741d 167a572 b3d741d 167a572 b3d741d 167a572 b3d741d 167a572 b3d741d 167a572 b3d741d 167a572 b3d741d |
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 |
__all__ = ['MemoryManager']
import json
class MemoryManager():
dependencies = []
inputSchema = {
"name": "MemoryManager",
"description": "Updates, retrieves, or deletes the memory of an AI agent.",
"parameters": {
"type": "object",
"properties":{
"action": {
"type": "string",
"enum": ["add_memory", "get_all_memories", "delete_memory"],
"description": "The action to perform: add_memory, get_all_memories, or delete_memory.",
},
"memory": {
"type": "string",
"description": "The memory to add. Required for 'add_memory' action.",
},
"key": {
"type": "string",
"description": "The key to delete or add memory."
},
},
"required": ["action"],
},
}
def get_memories(self):
# load the memory from src/data/memory.json
try:
with open("src/data/memory.json", "r") as f:
memory_list = json.load(f)
except FileNotFoundError:
memory_list = []
except json.JSONDecodeError:
memory_list = []
return memory_list
def update_memories(self, memories):
# save the memory to src/data/memory.json
with open("src/data/memory.json", "w") as f:
json.dump(memories, f, indent=4)
def run(self, **kwargs):
# save it to src/data/memory.json
action = kwargs.get("action")
memory = kwargs.get("memory")
key = kwargs.get("key")
memories = self.get_memories()
if action == "get_all_memories":
return {
"status": "success",
"message": "Memory retrieved successfully",
"output": memories
}
if action == "add_memory":
if memory is None or key is None:
return {
"status": "error",
"message": "Memory and key are required for add_memory action",
"output": None
}
# check if the key already exists
for mem in memories:
if mem["key"] == key:
return {
"status": "error",
"message": f"Memory with key {key} already exists",
"output": None
}
memories.append({
"key": key,
"memory": memory
})
self.update_memories(memories)
return {
"status": "success",
"message": "Memory created successfully",
"output": None
}
if action == "delete_memory":
if key is None:
return {
"status": "error",
"message": "Key is required for delete_memory action",
"output": None
}
# check if the key exists
for mem in memories:
if mem["key"] == key:
memories.remove(mem)
self.update_memories(memories)
return {
"status": "success",
"message": "Memory deleted successfully",
"output": None
}
return {
"status": "error",
"message": f"Memory with key {key} not found",
"output": None
} |