当前位置： 首页 > news >正文

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

news 来源：原创 2025/5/23 10:34:28

概述

Memory 是一个用于存储和管理对话消息的数据结构，它是 Agent 的核心组件，负责维护对话历史和上下文信息。Memory 基于 Pydantic 的 BaseModel 实现，提供了类型安全和数据验证功能。

Memory 数据结构是 Agent 系统的核心组件，它提供了存储和管理对话历史的功能，为代理提供了必要的上下文信息，使代理能够进行连贯的对话和决策。

数据结构

Memory 类

class Memory(BaseModel):
    messages: List[Message] = Field(default_factory=list)
    max_messages: int = Field(default=100)
参数名称类型默认值描述
messagesList[Message][]存储消息的列表
max_messagesint100内存中保存的最大消息数量

依赖的数据结构

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)
参数名称类型默认值描述
roleROLE_TYPE必填消息发送者的角色
contentOptional[str]None消息内容
tool_callsOptional[List[ToolCall]]None工具调用列表
nameOptional[str]None工具名称（用于工具消息）
tool_call_idOptional[str]None工具调用ID（用于工具消息）
Role 枚举
class Role(str, Enum):
    """Message role options"""
    SYSTEM = "system"
    USER = "user"
    ASSISTANT = "assistant"
    TOOL = "tool"
值描述
SYSTEM系统消息，通常用于设置指令或上下文
USER用户消息，表示用户输入
ASSISTANT助手消息，表示AI助手的响应，一般是询问 LLM 调用什么 tool call
TOOL工具消息，表示工具执行的结果
ToolCall 类
class ToolCall(BaseModel):
    """Represents a tool/function call in a message"""
    id: str
    type: str = "function"
    function: Function
属性名类型默认值描述
idstr必填工具调用的唯一标识符
typestr“function”工具调用类型
functionFunction必填函数信息
Function 类
class Function(BaseModel):
    name: str
    arguments: str
属性名类型默认值描述
namestr必填函数名称
argumentsstr必填函数参数（JSON格式字符串）

方法分析

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. 模块化：

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

相关文章：

  • 安装配置Anaconda，配置VSCode
  • 数据分析项目：基于LSTM的微博评论情感分析
  • 2.5[frontEnd]
  • Java 集合框架中 `List` 接口及其子类的详细介绍，并用 UML 图表展示层次结构关系，用表格对比各个类的差异。
  • Notepad++插件：快捷选择成对括号之间的内容
  • 代码随想录算法训练营第三十五天（20250303） |01背包问题 二维，01背包问题 一维，416. 分割等和子集 -[补卡20250316]
  • vue-treeselect 【单选/多选】的时候只选择最后一层（绑定的值只绑定最后一层）
  • 热key探测技术架构设计与实践
  • AI战略家：AI政务应用思考——AI与区块链融合对政府权力结构的重构：从“技术赋能”到“制度革命”
  • 音视频入门基础：RTP专题（20）——通过FFprobe显示RTP流每个packet的信息
  • RabbitMQ 基本原理详解
  • mysql数据库中多张表导出成excel方式
  • 【蓝桥杯速成】| 4.递归
  • CTP开发爬坑指北（九）
  • spring声明式事务原理01-调用第1层@Transactional方法（事务访问入口）
  • [蓝桥杯]花束搭配【算法赛】
  • Ubuntu从源码安装Webots
  • 网络编程、URI和URL的区别、TCP/IP协议、IP和端口、URLConnection
  • MySQL相关参数
  • 【C++多线程】thread
  • 安装好的字体怎么用wordpress/杭州seo俱乐部
  • wordpress url参数/seo 是什么
  • 网站模板免费下载php/优化游戏性能的软件
  • 一个服务器做两个网站吗/百度指数网页版
  • 新乡网站建设.com/天津百度推广代理商
  • 重庆网站快速排名优化/专业培训机构