storage: list[SerializeAsAny[Message]] = []
index: DefaultDict[str, list[SerializeAsAny[Message]]] = Field(default_factory=lambda: defaultdict(list))
ignore_id: bool = Falsestorage属性是一个列表，用于存储消息对象。每个消息对象都被标记为SerializeAsAny[Message]类型，这表示它们可以序列化为任何类型的消息。默认情况下，它被初始化为空列表[]。index属性是一个字典，用于创建消息的索引。键是字符串类型，对应着消息的某个属性（可能是消息的原因）。值是一个列表，存储与该属性相匹配的消息对象。它使用了DefaultDict来提供一个默认工厂函数，以便在索引中添加新键时自动创建一个空列表。默认情况下，它被初始化为一个空字典。ignore_id属性是一个布尔值，用于控制是否忽略消息的 ID。默认情况下，它被设置为False，表示不忽略消息的 ID。
def serialize(self, stg_path: Path):
"""stg_path = ./storage/team/environment/ or ./storage/team/environment/roles/{role_class}_{role_name}/"""
memory_path = stg_path.joinpath("memory.json")
storage = self.model_dump()
write_json_file(memory_path, storage)
@classmethod
def deserialize(cls, stg_path: Path) -> "Memory":
"""stg_path = ./storage/team/environment/ or ./storage/team/environment/roles/{role_class}_{role_name}/"""
memory_path = stg_path.joinpath("memory.json")
memory_dict = read_json_file(memory_path)
memory = Memory(**memory_dict)
return memoryserialize方法用于将内存对象序列化并保存到指定路径下的 JSON 文件中。参数stg_path是保存序列化文件的路径，可以是一个固定路径，也可以是具有角色类和角色名称的路径。首先，构建存储文件的路径memory_path。然后，调用self.model_dump()方法来获取要序列化的数据，并将其写入 JSON 文件中。deserialize方法是一个类方法，用于从指定路径下的 JSON 文件中反序列化内存对象。参数stg_path是指定路径，类似于serialize方法。首先，构建存储文件的路径memory_path。然后，使用read_json_file函数从 JSON 文件中读取数据，并将其作为关键字参数传递给Memory类的构造函数，从而创建并返回内存对象。
add
def add(self, message: Message):
"""Add a new message to storage, while updating the index"""
if self.ignore_id:
message.id = IGNORED_MESSAGE_ID
if message in self.storage:
return
self.storage.append(message)
if message.cause_by:
self.index[message.cause_by].append(message)
def add_batch(self, messages: Iterable[Message]):
for message in messages:
self.add(message)向存储中添加新的消息，并更新索引。
如果 ignore_id 设置为 True，则将消息的 id 设置为预定义的常量 IGNORED_MESSAGE_ID。
将消息添加到存储中，并根据消息的 cause_by 属性更新索引
def get_by_role(self, role: str) -> list[Message]:
"""Return all messages of a specified role"""
return [message for message in self.storage if message.role == role]
def get_by_content(self, content: str) -> list[Message]:
"""Return all messages containing a specified content"""
return [message for message in self.storage if content in message.content] get_by_role(self, role: str) -> list[Message]: 返回指定角色的所有消息
get_by_content(self, content: str) -> list[Message]: 返回包含指定内容的所有消息
def delete_newest(self) -> "Message":
"""delete the newest message from the storage"""
if len(self.storage) > 0:
newest_msg = self.storage.pop()
if newest_msg.cause_by and newest_msg in self.index[newest_msg.cause_by]:
self.index[newest_msg.cause_by].remove(newest_msg)
else:
newest_msg = None
return newest_msg
def delete(self, message: Message):
"""Delete the specified message from storage, while updating the index"""
if self.ignore_id:
message.id = IGNORED_MESSAGE_ID
self.storage.remove(message)
if message.cause_by and message in self.index[message.cause_by]:
self.index[message.cause_by].remove(message) delete_newest(self) -> "Message"：从存储中删除最新的消息。
- 返回值为被删除的最新消息对象，如果存储为空，则返回
None。 - 如果存储中最新消息包含在索引中，并且有引起其产生的消息，那么也会从索引中删除。
delete(self, message: Message)：从存储中删除指定的消息，并更新索引。
message：要删除的消息对象。- 如果
ignore_id为真，则会将消息的 ID 设置为IGNORED_MESSAGE_ID。 - 从存储中删除指定的消息对象。
- 如果消息有引起其产生的消息，并且存在于索引中，则也会从索引中删除该消息
def clear(self):
"""Clear storage and index"""
self.storage = []
self.index = defaultdict(list)
def count(self) -> int:
"""Return the number of messages in storage"""
return len(self.storage)
def try_remember(self, keyword: str) -> list[Message]:
"""Try to recall all messages containing a specified keyword"""
return [message for message in self.storage if keyword in message.content]clear(self)：将存储和索引都重置为空列表或字典。count(self) -> int：返回存储中消息的数量try_remember(self, keyword: str) -> list[Message]：尝试回忆所有包含指定关键字的消息。keyword：要搜索的关键字。- 返回存储中所有包含指定关键字的消息列表。
get
def get(self, k=0) -> list[Message]:
"""Return the most recent k memories, return all when k=0"""
return self.storage[-k:]从storage里面拿最近k个
def find_news(self, observed: list[Message], k=0) -> list[Message]:
"""find news (previously unseen messages) from the the most recent k memories, from all memories when k=0"""
already_observed = self.get(k)
news: list[Message] = []
for i in observed:
if i in already_observed:
continue
news.append(i)
return news
def get_by_action(self, action) -> list[Message]:
"""Return all messages triggered by a specified Action"""
index = any_to_str(action)
return self.index[index]
def get_by_actions(self, actions: Set) -> list[Message]:
"""Return all messages triggered by specified Actions"""
rsp = []
indices = any_to_str_set(actions)
for action in indices:
if action not in self.index:
continue
rsp += self.index[action]
return rspfind_news(self, observed: list[Message], k=0) -> list[Message]：从最近的 k 个存储的消息中查找新消息（之前未见过的消息），当 k=0 时从所有消息中查找。返回未见过的新消息列表。get_by_action(self, action) -> list[Message]：返回由指定动作触发的所有消息get_by_actions(self, actions: Set) -> list[Message]：返回由指定一组动作触发的所有消息
