【Agent】OpenManus-Agent-Memory详细设计

概述

Memory 是一个用于存储和管理对话消息的数据结构，它是 Agent 的核心组件，负责维护对话历史和上下文信息。Memory 基于 Pydantic 的 BaseModel 实现，提供了类型安全和数据验证功能。 Memory 数据结构是 Agent 系统的核心组件，它提供了存储和管理对话历史的功能，为代理提供了必要的上下文信息，使代理能够进行连贯的对话和决策。

数据结构

Memory 类

class Memory(BaseModel): messages: List[Message] = Field(default_factory=list) max_messages: int = Field(default=100)

依赖的数据结构

Message 类

class Message(BaseModel): role: ROLE_TYPE = Field(…) content: Optional[str] = Field(default=None) tool_calls: Optional[List[ToolCall]] = Field(default=None) name: Optional[str] = Field(default=None) tool_call_id: Optional[str] = Field(default=None)

Role 枚举

class Role(str, Enum): “““Message role options””” SYSTEM = “system” USER = “user” ASSISTANT = “assistant” TOOL = “tool”

ToolCall 类

class ToolCall(BaseModel): “““Represents a tool/function call in a message””” id: str type: str = “function” function: Function

Function 类

class Function(BaseModel): name: str arguments: str

方法分析

Memory 类方法

add_message

def add_message(self, message: Message) -> None: “““Add a message to memory””” self.messages.append(message)

Optional: Implement message limit

if len(self.messages) > self.max_messages: self.messages = self.messages[-self.max_messages :] 功能 ：向 Memory 中添加单个消息。 参数 ：

• message: Message - 要添加的消息对象 实现细节 ：
• 将消息追加到消息列表末尾
• 检查消息数量是否超过最大限制
• 如果超过限制，保留最近的 max_messages 条消息 设计理念 ：
• 实现滑动窗口机制，防止内存无限增长
• 保留最近的消息，确保最相关的上下文信息被保留

add_messages

在 planning agent 里面使用的 def add_messages(self, messages: List[Message]) -> None: “““Add multiple messages to memory””” self.messages.extend(messages) 功能 ：向内存中批量添加多个消息。 参数 ：

• messages: List[Message] - 要添加的消息列表 实现细节 ：
• 使用 extend 方法将消息列表追加到现有消息列表
• 注意：此方法不检查最大消息限制 设计理念 ：
• 提供批量添加消息的便捷方法
• 优化性能，避免多次调用 add_message

clear

def clear(self) -> None: “““Clear all messages””” self.messages.clear() 功能 ：清空内存中的所有消息。 实现细节 ：

• 使用 Python 列表的 clear 方法清空消息列表 设计理念 ：
• 提供重置对话历史的能力
• 在需要开始新对话时使用

get_recent_messages

def get_recent_messages(self, n: int) -> List[Message]: “““Get n most recent messages””” return self.messages[-n:] 功能 ：获取最近的 n 条消息。 参数 ：

• n: int - 要获取的消息数量 返回值 ：
• List[Message] - 最近的 n 条消息列表 实现细节 ：
• 使用 Python 列表切片获取最后 n 个元素 设计理念 ：
• 允许获取有限数量的最近消息
• 用于构建有限上下文窗口

to_dict_list

def to_dict_list(self) -> List[dict]: “““Convert messages to list of dicts””” return [msg.to_dict() for msg in self.messages] 功能 ：将消息列表转换为字典列表。 返回值 ：

• List[dict] - 消息字典列表 实现细节 ：
• 使用列表推导式对每个消息调用 to_dict 方法
• 依赖 Message 类的 to_dict 方法 设计理念 ：
• 提供序列化功能，便于与外部系统交互
• 将复杂对象转换为简单数据结构

Message 类相关方法

to_dict

def to_dict(self) -> dict: “““Convert message to dictionary format””” message = {“role”: self.role} if self.content is not None: message[“content”] = self.content if self.tool_calls is not None: message[“tool_calls”] = [tool_call.dict() for tool_call in self.tool_calls] if self.name is not None: message[“name”] = self.name if self.tool_call_id is not None: message[“tool_call_id”] = self.tool_call_id return message 功能 ：将消息对象转换为字典格式。 返回值 ：

• dict - 消息的字典表示

工厂方法

Message 类提供了多个工厂方法，用于创建不同类型的消息： @classmethod def user_message(cls, content: str) -> “Message”: “““Create a user message””” return cls(role=Role.USER, content=content) @classmethod def system_message(cls, content: str) -> “Message”: “““Create a system message””” return cls(role=Role.SYSTEM, content=content) @classmethod def assistant_message(cls, content: Optional[str] = None) -> “Message”: “““Create an assistant message””” return cls(role=Role.ASSISTANT, content=content) @classmethod def tool_message(cls, content: str, name, tool_call_id: str) -> “Message”: “““Create a tool message””” return cls( role=Role.TOOL, content=content, name=name, tool_call_id=tool_call_id ) @classmethod def from_tool_calls( cls, tool_calls: List[Any], content: Union[str, List[str]] = “”, **kwargs ): “““Create ToolCallsMessage from raw tool calls.””” formatted_calls = [ {“id”: call.id, “function”: call.function.model_dump(), “type”: “function”} for call in tool_calls ] return cls( role=Role.ASSISTANT, content=content, tool_calls=formatted_calls, **kwargs ) 功能 ：提供创建各种类型消息的便捷方法。

运算符重载

Message 类重载了 + 运算符，支持消息与列表或其他消息的组合： 注意：这个重载是不校验 memory中messages的长度的，也就是不受 max_messages 参数限制。一般用在添加 user msg 的时候，可见 ToolAgent 中 think 方法 如果是为了保证 user msg 一定被加到 memory messages 里面，感觉用一个特别的方法名称来标识会更好一点。 def add(self, other) -> List[“Message”]: “““支持 Message + list 或 Message + Message 的操作””” if isinstance(other, list): return [self] + other elif isinstance(other, Message): return [self, other] else: raise TypeError( f"unsupported operand type(s) for +: ‘{type(self).name}’ and ‘{type(other).name}’" ) def radd(self, other) -> List[“Message”]: “““支持 list + Message 的操作””” if isinstance(other, list): return other + [self] else: raise TypeError( f"unsupported operand type(s) for +: ‘{type(other).name}’ and ‘{type(self).name}’" ) 功能 ：允许使用 + 运算符组合消息和消息列表。 设计理念 ：

• 提供直观的语法糖，简化消息列表操作
• 支持多种组合方式，增强灵活性

使用示例

创建内存实例

memory = Memory()

添加不同类型的消息

memory.add_message(Message.system_message(“你是一个有用的助手”)) memory.add_message(Message.user_message(“你好，请帮我写一首诗”)) memory.add_message(Message.assistant_message(“好的，这是一首关于春天的诗：…”))

获取最近的消息

recent_msgs = memory.get_recent_messages(2)

转换为字典列表（用于API调用）

dict_msgs = memory.to_dict_list()

清空内存

memory.clear()

设计理念总结

1. 类型安全 ：

• 使用 Pydantic 模型确保数据类型正确
• 使用枚举类型限制可能的值（如角色）

2. 内存管理 ：

• 实现最大消息限制，防止内存无限增长
• 保留最近的消息，确保相关上下文

3. 灵活性 ：

• 支持多种消息类型（用户、系统、助手、工具）
• 提供工厂方法简化消息创建

4. 序列化 ：

• 提供转换为字典的方法，便于与API交互
• 只包含非空字段，减少数据大小

5. 便捷操作 ：

• 重载运算符，简化消息组合
• 提供批量操作方法，提高效率

6. 模块化 ：

• 将消息和内存分离为独立类
• 清晰的职责分离，遵循单一职责原则