住房和城乡建设岗位证书查询官网东莞seo关键词
概述
Memory 是一个用于存储和管理对话消息的数据结构，它是 Agent 的核心组件，负责维护对话历史和上下文信息。Memory 基于 Pydantic 的 BaseModel 实现，提供了类型安全和数据验证功能。
Memory 数据结构是 Agent 系统的核心组件，它提供了存储和管理对话历史的功能，为代理提供了必要的上下文信息，使代理能够进行连贯的对话和决策。
数据结构
Memory 类
class Memory(BaseModel):messages: List[Message] = Field(default_factory=list)max_messages: int = Field(default=100)
| 参数名称 | 类型 | 默认值 | 描述 |
|---|---|---|---|
| messages | List[Message] | [] | 存储消息的列表 |
| max_messages | int | 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 | ROLE_TYPE | 必填 | 消息发送者的角色 |
| content | Optional[str] | None | 消息内容 |
| tool_calls | Optional[List[ToolCall]] | None | 工具调用列表 |
| name | Optional[str] | None | 工具名称（用于工具消息） |
| tool_call_id | Optional[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: strtype: str = "function"function: Function
| 属性名 | 类型 | 默认值 | 描述 |
|---|---|---|---|
| id | str | 必填 | 工具调用的唯一标识符 |
| type | str | “function” | 工具调用类型 |
| function | Function | 必填 | 函数信息 |
Function 类
class Function(BaseModel):name: strarguments: str
| 属性名 | 类型 | 默认值 | 描述 |
|---|---|---|---|
| name | str | 必填 | 函数名称 |
| arguments | str | 必填 | 函数参数（JSON格式字符串） |
方法分析
Memory 类方法
add_message
def add_message(self, message: Message) -> None:"""Add a message to memory"""self.messages.append(message)# Optional: Implement message limitif 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.contentif 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.nameif self.tool_call_id is not None:message["tool_call_id"] = self.tool_call_idreturn 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] + otherelif 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()
设计理念总结
-
类型安全：
- 使用 Pydantic 模型确保数据类型正确
- 使用枚举类型限制可能的值（如角色）
-
内存管理：
- 实现最大消息限制，防止内存无限增长
- 保留最近的消息，确保相关上下文
-
灵活性：
- 支持多种消息类型（用户、系统、助手、工具）
- 提供工厂方法简化消息创建
-
序列化：
- 提供转换为字典的方法，便于与API交互
- 只包含非空字段，减少数据大小
-
便捷操作：
- 重载运算符，简化消息组合
- 提供批量操作方法，提高效率
-
模块化：
- 将消息和内存分离为独立类
- 清晰的职责分离，遵循单一职责原则