storage: list[SerializeAsAny[Message]] = []index: DefaultDict[str, list[SerializeAsAny[Message]]] = Field(default_factory=lambda: defaultdict(list))ignore_id: bool = False-
storage属性是一个列表，用于存储消息对象。每个消息对象都被标记为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)@classmethoddef 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 memory-
serialize方法用于将内存对象序列化并保存到指定路径下的 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_IDif message in self.storage:returnself.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 = Nonereturn newest_msgdef delete(self, message: Message):"""Delete the specified message from storage, while updating the index"""if self.ignore_id:message.id = IGNORED_MESSAGE_IDself.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:continuenews.append(i)return newsdef 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:continuersp += self.index[action]return rsp-
find_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]：返回由指定一组动作触发的所有消息
