第 4 章 · 01 AgentMemory 与步骤类型


第 4 章 · 01 AgentMemory 与步骤类型

本节摘要:本节精读 memory.py(316 行)。AgentMemory 把 agent 的完整运行轨迹结构化为「系统提示 + 步骤列表」,五种步骤类型各司其职:SystemPromptStep 存系统提示、TaskStep 存用户任务、ActionStep 存每一步动作(含 observations/error/token_usage)、PlanningStep 存规划、FinalAnswerStep 存最终答案。ToolCall 记录工具调用,CallbackRegistry 提供观察者回调,write_memory_to_messages 把记忆转成 LLM 可消费的消息列表(含图片多模态处理)。这是理解 agent「怎么记住自己做过什么」的关键一节。

内容来源:src/smolagents/memory.pysrc/smolagents/agents.py(write_memory_to_messages 部分)

⚠️ 注意:memory.py 里没有任何「持久化」逻辑——它是纯内存对象,agent 重启即失。真正的保存/加载在第 4.2 节的 serialization 与 save/load 中。

学习目标

  1. 读懂 AgentMemory 的两个成员:system_prompt 与 steps。
  2. 记住五种步骤类型各自的字段与用途。
  3. 理解 ActionStep.to_messages 如何把一步动作拆成多条 LLM 消息。
  4. 掌握 CallbackRegistry 的观察者模式。
  5. 会用 write_memory_to_messages 把记忆转成消息列表。

一、AgentMemory:系统提示 + 步骤列表

AgentMemory 是 agent 的大脑海马体。它只有两个成员,极简:

class AgentMemory: def __init__(self, system_prompt: str): self.system_prompt: SystemPromptStep = SystemPromptStep(system_prompt=system_prompt) self.steps: list[TaskStep | ActionStep | PlanningStep] = [] def reset(self): """Reset the agent's memory, clearing all steps and keeping the system prompt.""" self.steps = []

解读三点:

  1. system_prompt 不是裸字符串,而是一个 SystemPromptStep——这样系统提示与其他步骤享有统一的 to_messages 接口。
  2. steps 是异构列表,按时间顺序追加 TaskStep、ActionStep、PlanningStep(运行循环里还可能混入 FinalAnswerStep)。
  3. reset() 只清 steps 不清系统提示——agent 换任务时,人设还在。

辅助方法三个,都基于 steps 派生:get_succinct_steps 剔除 model_input_messages 再返回(该字段每步都完整复制当时的输入消息,体积会指数膨胀,适合打印日志);get_full_steps 返回完整字典;return_full_code 把 CodeAgent 各步的 code_action 拼成一个脚本——调试时「看看 agent 到底写了什么代码」一行搞定。replay(logger, detailed) 则用富文本回放整个轨迹,detailed 模式连每步输入消息都打出来,官方注释警告「日志长度指数级增长,仅用于调试」。

二、五种步骤类型:一条时间线上的五种事件

所有步骤继承自 MemoryStep,约定两个方法:dict()(用 dataclasses 的 asdict 转字典)和 to_messages()(转 LLM 消息)。

@dataclass class MemoryStep: def dict(self): return asdict(self) def to_messages(self, summary_mode: bool = False) -> list[ChatMessage]: raise NotImplementedError

2.1 SystemPromptStep 与 TaskStep:一头一尾的静态信息

@dataclass class SystemPromptStep(MemoryStep): system_prompt: str # to_messages: summary_mode=True 时返回空;否则一条 SYSTEM 角色消息 @dataclass class TaskStep(MemoryStep): task: str task_images: list["PIL.Image.Image"] | None = None def to_messages(self, summary_mode: bool = False) -> list[ChatMessage]: content = [{"type": "text", "text": f"New task:\n{self.task}"}] if self.task_images: content.extend([{"type": "image", "image": image} for image in self.task_images]) return [ChatMessage(role=MessageRole.USER, content=content)]

注意 summary_mode=True 时 SystemPromptStep 返回空——摘要模式就是「去掉系统提示和规划输出,只看动作史」,第 4.2 节规划步生成时会用到。TaskStep 支持 task_images 字段:任务本身可以是多模态的,to_messages 里图片以 {"type": "image", "image": image} 追加进 content 列表,与任务文本一起发给视觉模型。

2.2 ActionStep:信息量最大的一步

ActionStep 是五种类型里字段最多的,完整贴出:

@dataclass class ActionStep(MemoryStep): step_number: int timing: Timing model_input_messages: list[ChatMessage] | None = None tool_calls: list[ToolCall] | None = None error: AgentError | None = None model_output_message: ChatMessage | None = None model_output: str | list[dict[str, Any]] | None = None code_action: str | None = None observations: str | None = None observations_images: list["PIL.Image.Image"] | None = None action_output: Any = None token_usage: TokenUsage | None = None is_final_answer: bool = False

逐个看:timing/token_usage 记录耗时与 token 消耗(监控与成本核算的原始数据);model_input_messages 保存这一步发给 LLM 的完整输入(记忆快照);model_output 保存 LLM 原始输出,CodeAgent 里另有 code_action 单存代码;observations 是工具执行观察结果,observations_images 存观察中的图片(如生成的图像);error 存这一步的异常——出错的信息也会进记忆,让 LLM 下一步看到「上次为什么失败」;is_final_answer 标记这一步是否给出了最终答案。

to_messages 是本节的精华——一步动作如何拆成多条消息(节选):

def to_messages(self, summary_mode: bool = False) -> list[ChatMessage]: messages = [] if self.model_output is not None and not summary_mode: messages.append(ChatMessage(role=MessageRole.ASSISTANT, content=[{"type": "text", "text": self.model_output.strip()}])) if self.tool_calls is not None: messages.append(ChatMessage(role=MessageRole.TOOL_CALL, content=[{"type": "text", "text": "Calling tools:\n" + str([tc.dict() for tc in self.tool_calls])}])) if self.observations_images: messages.append(ChatMessage(role=MessageRole.USER, content=[{"type": "image", "image": image} for image in self.observations_images])) if self.observations is not None: messages.append(ChatMessage(role=MessageRole.TOOL_RESPONSE, content=[{"type": "text", "text": f"Observation:\n{self.observations}"}])) if self.error is not None: error_message = ( "Error:\n" + str(self.error) + "\nNow let's retry: take care not to repeat previous errors! If you have retried several times, try a completely different approach.\n" ) ...

三个细节值得咀嚼:

  1. 角色轮转符合 ReAct 对话惯例:assistant(我想了什么)→ tool_call(我调了什么)→ tool_response(观察结果),错误也伪装成 TOOL_RESPONSE 喂回去,并附上「别重复犯错,试过几次就换思路」的引导语——这是写给 LLM 的错误恢复话术。
  2. 图片消息单独成条,以 USER 角色发送——这就是记忆层的多模态处理:上一轮生成的图,下一轮作为输入再喂给视觉模型。
  3. dict() 里对 observations_images 特殊处理为 [image.tobytes() for image in ...]——PIL 图像没法进 JSON,转成原始字节。

2.3 PlanningStep 与 FinalAnswerStep

@dataclass class PlanningStep(MemoryStep): model_input_messages: list[ChatMessage] model_output_message: ChatMessage plan: str timing: Timing token_usage: TokenUsage | None = None def to_messages(self, summary_mode: bool = False) -> list[ChatMessage]: if summary_mode: return [] return [ ChatMessage(role=MessageRole.ASSISTANT, content=[{"type": "text", "text": self.plan.strip()}]), ChatMessage(role=MessageRole.USER, content=[{"type": "text", "text": "Now proceed and carry out this plan."}]), ]

plan 是一段完整的规划文本(事实清单 + 行动计划,见第 4.2 节)。to_messages 在计划后面补了一条 USER 消息「Now proceed and carry out this plan.」——源码注释点明用意:强制切换角色,防止模型顺着计划文本继续「写计划」而不开始干活。summary_mode 下 PlanningStep 也返回空:重新规划时不需要旧计划干扰。

@dataclass class FinalAnswerStep(MemoryStep): output: Any

FinalAnswerStep 最简单,只有一个 output——它是给外部调用者与 Gradio UI 用的终点标记,不参与 to_messages(任务已经结束,不需要再喂回 LLM)。

三、ToolCall 与 CallbackRegistry

ToolCall 只有三个字段:name/arguments/id,dict() 方法把 arguments 经 make_json_serializable 处理后,拼成:

{"id": self.id, "type": "function", "function": {"name": self.name, "arguments": make_json_serializable(self.arguments)}}

dict() 的返回结构与 OpenAI 风格的 function call 格式完全对齐——smolagents 在 CodeAgent(工具即 Python 函数)与 ToolCallingAgent(工具即 JSON 调用)两种范式间共用这一份数据结构。

CallbackRegistry 是典型的观察者模式,register(step_cls, callback) 把回调按步骤类型收进 dict[Type[MemoryStep], list[Callable]],触发时遍历 MRO:

def callback(self, memory_step, **kwargs): for cls in memory_step.__class__.__mro__: for cb in self._callbacks.get(cls, []): cb(memory_step) if len(inspect.signature(cb).parameters) == 1 else cb(memory_step, **kwargs)

每产生一个新步骤(如 PlanningStep),注册在该类型上的回调就被触发;__mro__ 遍历让注册在父类上的回调对子类也生效;inspect.signature 数参数个数是为了兼容——单参旧回调只收 memory_step,多参新回调还能拿到 agent=agent 这样的额外上下文。它是第 4.2 节 human-in-the-loop(人工审批计划)的底层机制。

四、write_memory_to_messages:记忆 → LLM 消息

AgentMemory 本身没有这个方法,它在 agents.py 的 MultiStepAgent 上,实现只有三行:

def write_memory_to_messages(self, summary_mode: bool = False) -> list[ChatMessage]: messages = self.memory.system_prompt.to_messages(summary_mode=summary_mode) for memory_step in self.memory.steps: messages.extend(memory_step.to_messages(summary_mode=summary_mode)) return messages

就是「系统提示转消息 + 逐个步骤转消息再拼接」。妙处在于多态:它不需要知道每个步骤是什么类型,to_messages 的统一接口让五种步骤各自决定如何呈现自己。下一轮 LLM 调用的输入消息列表,就是这一函数的返回值——记忆与提示工程的衔接点全在这。summary_mode 一开,系统提示和规划自动隐身,只剩「任务 + 动作 + 观察」的动作史,专门服务于「重新规划」场景。

💡 阶梯要点:本阶登上了「记忆」这一装备。核心心智模型:agent 的记忆不是聊天记录,而是结构化的步骤流水账;LLM 看到的「历史消息」是这份流水账经 to_messages 实时渲染出来的视图,渲染方式(summary_mode、图片处理、角色分配)集中封装在各步骤类型里,AgentMemory 只管收集。

本节要点回顾

  • AgentMemory = SystemPromptStep + steps 列表,reset 只清 steps。
  • 五种步骤:SystemPromptStep(系统提示)、TaskStep(用户任务,支持图片)、ActionStep(动作,含 observations/error/token_usage/code_action 等 13 个字段)、PlanningStep(规划文本 plan)、FinalAnswerStep(最终答案)。
  • ActionStep.to_messages 把一步拆成 assistant→tool_call→图片→observation→错误多条消息,错误里带「换思路」引导语。
  • PlanningStep 在计划后补一条 USER 消息强制角色切换,防止模型只规划不执行。
  • ToolCall.dict 对齐 OpenAI function call 格式;CallbackRegistry 按步骤类型注册观察者回调。
  • write_memory_to_messages 是记忆到 LLM 消息的唯一入口,summary_mode 控制精简视图。

下一节:02 planning_interval 规划与序列化——让 agent 每 N 步停下来重新规划,以及把 agent 序列化保存、分享到 HuggingFace Hub。


作者与出处
原作者: 灏天文库
整理: 灏天文库整理
本站整理收录,版权归原作者/开源协议所有;欢迎通过原文链接访问源仓库。
发布者: 作者: 灏天文库 转发
评论区 (0)
U