blob: b85ea8e673403b44a8fb2e4f7f906f381fbc70a5 (
plain)
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
|
from __future__ import annotations
from pathlib import Path
from typing import Iterator
import orjson
from autogpt.config import Config
from autogpt.logs import logger
from ..memory_item import MemoryItem
from .base import VectorMemoryProvider
class JSONFileMemory(VectorMemoryProvider):
"""Memory backend that stores memories in a JSON file"""
SAVE_OPTIONS = orjson.OPT_SERIALIZE_NUMPY | orjson.OPT_SERIALIZE_DATACLASS
file_path: Path
memories: list[MemoryItem]
def __init__(self, config: Config) -> None:
"""Initialize a class instance
Args:
config: Config object
Returns:
None
"""
workspace_path = Path(config.workspace_path)
self.file_path = workspace_path / f"{config.memory_index}.json"
self.file_path.touch()
logger.debug(
f"Initialized {__class__.__name__} with index path {self.file_path}"
)
self.memories = []
try:
self.load_index()
logger.debug(f"Loaded {len(self.memories)} MemoryItems from file")
except Exception as e:
logger.warn(f"Could not load MemoryItems from file: {e}")
self.save_index()
def __iter__(self) -> Iterator[MemoryItem]:
return iter(self.memories)
def __contains__(self, x: MemoryItem) -> bool:
return x in self.memories
def __len__(self) -> int:
return len(self.memories)
def add(self, item: MemoryItem):
self.memories.append(item)
logger.debug(f"Adding item to memory: {item.dump()}")
self.save_index()
return len(self.memories)
def discard(self, item: MemoryItem):
try:
self.remove(item)
except:
pass
def clear(self):
"""Clears the data in memory."""
self.memories.clear()
self.save_index()
def load_index(self):
"""Loads all memories from the index file"""
if not self.file_path.is_file():
logger.debug(f"Index file '{self.file_path}' does not exist")
return
with self.file_path.open("r") as f:
logger.debug(f"Loading memories from index file '{self.file_path}'")
json_index = orjson.loads(f.read())
for memory_item_dict in json_index:
self.memories.append(MemoryItem(**memory_item_dict))
def save_index(self):
logger.debug(f"Saving memory index to file {self.file_path}")
with self.file_path.open("wb") as f:
return f.write(orjson.dumps(self.memories, option=self.SAVE_OPTIONS))
|
