|
|
|
__all__ = ['UpdateMemory'] |
|
|
|
import json |
|
|
|
|
|
class UpdateMemory(): |
|
dependencies = [] |
|
|
|
inputSchema = { |
|
"name": "UpdateMemory", |
|
"description": "Updates the memory of the AI agent. This tool is used to update the memory of the AI agent with new information.", |
|
"parameters": { |
|
"type": "object", |
|
"properties":{ |
|
"memory": { |
|
"type": "string", |
|
"description": "The new memory to be added to the AI agent's memory.", |
|
} |
|
}, |
|
"required": ["memory"], |
|
}, |
|
} |
|
|
|
|
|
def run(self, **kwargs): |
|
|
|
memory = kwargs.get("memory") |
|
if not memory: |
|
return { |
|
"status": "error", |
|
"message": "Memory is required", |
|
"output": None |
|
} |
|
|
|
|
|
try: |
|
with open("src/data/memory.json", "r") as f: |
|
memory_list = json.load(f) |
|
except FileNotFoundError: |
|
memory_list = [] |
|
except json.JSONDecodeError: |
|
memory_list = [] |
|
|
|
|
|
memory_list.append(memory) |
|
|
|
with open("src/data/memory.json", "w") as f: |
|
json.dump(memory_list, f) |
|
|
|
return { |
|
"status": "success", |
|
"message": "Memory updated successfully", |
|
"output": memory_list |
|
} |