File size: 3,892 Bytes
b3d741d 941e057 b3d741d 0ab24d1 b3d741d 167a572 b3d741d 167a572 b3d741d 941e057 b3d741d 167a572 b3d741d 941e057 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 110 111 112 |
__all__ = ['MemoryManager']
import json
import os
class MemoryManager():
dependencies = []
inputSchema = {
"name": "MemoryManager",
"description": "Updates, retrieves, or deletes the memory for a user. Only store important information such as user preferences, error solutions, and other information that can help you improve your performance.",
"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):
os.makedirs("src/data", exist_ok=True)
# 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
} |