Agent之记忆memory-消息链就是大脑

上篇你学会了 MCP——用标准协议接入全世界的工具。但工具多了、对话长了，一个新问题冒出来：Agent 记不住刚才说过什么。这篇讲 Agent 的记忆机制——为什么消息列表就是 Agent 的大脑，以及 Function Calling 之后记忆怎么维护。

小明发现 AI 失忆了

小明的 Agent 跑得很好。查天气、搜新闻、算数学，样样都行。

但今天测试时遇到了一个奇怪的现象——

你：我叫小明
Agent：你好小明，有什么可以帮你的？

你：帮我查一下成都天气
Agent：成都多云，16°C，微风。

你：我刚才说我叫什么名字？
Agent：你没有告诉我你的名字。

小明愣住了。”我刚才第一句话就说了啊！它怎么就忘了？”

他去找老张。老张说：”因为大模型没有记忆。”

“什么意思？”

“每一次 API 调用都是独立的。你第一次发’我叫小明’和第二次发’我刚才说我叫什么名字’——这两次调用之间没有任何关联。模型不记得上一轮的对话。”

小明说：”可我之前的 Agent 明明能记住上下文？”

老张说：”那你要检查一下——你每次调用 API 时，传的 messages 列表里到底有几种消息。如果只传了当前用户输入这一条，模型当然只能根据这一条回答。”

核心认知：LLM 无状态，记忆是你传的

老张在白板上写了四个字：

模型无状态

“大模型本身不存储任何对话历史。每次 API 调用都是一次’初次见面’。模型能’记住’之前说过什么，是因为你把完整的历史消息都放进了 messages 列表里，每次都重新传一遍。“

他画了张图：

第 1 轮：
  你发给模型 → [{"role":"user","content":"我叫小明"}]
  模型返回   → "你好小明"
  ✅ 模型知道用户叫小明（因为消息里有）

第 2 轮（没传历史）：
  你发给模型 → [{"role":"user","content":"我刚才说我叫什么名字？"}]
  模型返回   → "你没有告诉我你的名字。"
  ❌ 模型不知道（因为上轮的消息没传过去）

第 2 轮（传了历史）：
  你发给模型 → [
  {"role":"user","content":"我叫小明"},
  {"role":"assistant","content":"你好小明"},
  {"role":"user","content":"我刚才说我叫什么名字？"}
  ]
  模型返回   → "你刚才说你叫小明。"
  ✅ 模型知道（因为完整历史都传了）

“看明白没？模型能不能’记住’，不取决于模型本身，取决于你的程序有没有把历史消息带上。”

小明恍然大悟：”所以 Agent 的记忆本质上就是一个列表——每次请求把整个列表传给模型。”

“对。消息列表就是 Agent 的短期记忆。 你写不写数据库、用不用向量存储都不重要——只要你每次 API 调用都传完整的历史消息，Agent 就能实现最基本的’多轮对话记忆’。”

普通对话的记忆很简单

老张说：”在没有 Function Calling 时，消息列表的结构很简单——三种角色轮转：”

messages = [
  {"role": "system", "content": "你是一个智能助手..."},
  {"role": "user", "content": "我叫小明"},
  {"role": "assistant", "content": "你好小明！"},
  {"role": "user", "content": "帮我查天气"},
  {"role": "assistant", "content": "好的，请告诉我要查哪个城市？"},
  {"role": "user", "content": "成都"},
  {"role": "assistant", "content": "成都今天多云..."},
]

“流程很简单——每次对话追加两条消息（user + assistant），调用 API 时把整个列表传过去。”

Function Calling 让记忆变复杂了

小明说：”但这和普通的聊天记录差不多啊。为什么说 Function Calling 之后记忆变复杂了？”

老张说：”因为 Function Calling 在一轮对话里，消息结构不再是简单的 user → assistant。中间多了一层工具调用链。“

他在白板上画出完整的一轮 Function Calling 消息流：

普通对话（无工具）：
  user → assistant(回答)

Function Calling（有工具）：
  user
  → assistant(tool_calls: get_weather, args: {city:"成都"})  ← 模型决定调工具
  → tool(tool_call_id: "xxx", name: "get_weather", content: "多云 16°C")  ← 你的代码执行结果
  → assistant(回答: "成都今天多云，16°C，微风")  ← 模型基于结果生成回答

“一轮 Function Calling 可能产生 4 条消息——比你想象的多了两倍。”

“你以为问一句话就加两条消息，实际上加了四条。如果你的记忆实现得不好，这些消息的顺序和配对关系就可能出问题。”

协议硬规则：assistant(tool_calls) 和 tool 消息必须成对

老张的表情严肃起来：”官方的 Function Calling 协议有一条铁律——”

assistant 消息里的 tool_calls 必须在 messages 列表中紧跟对应的 tool（结果）消息。
不能拆散，不能漏掉，不能乱序。

# ✅ 正确：assistant(tool_calls) 紧接 tool 结果
[
  {"role": "user", "content": "成都天气？"},
  {"role": "assistant", "content": None, "tool_calls": [...]},  # 1. 调用请求
  {"role": "tool", "tool_call_id": "xxx", "content": "多云"},  # 2. 工具结果（紧接）
]

# ❌ 错误：assistant 和 tool 之间插了别的消息
[
  {"role": "user", "content": "成都天气？"},
  {"role": "assistant", "content": None, "tool_calls": [...]},
  {"role": "user", "content": "上海呢？"},       # ← 破坏了配对！
  {"role": "tool", "tool_call_id": "xxx", "content": "多云"},
]
# → API 报错：An assistant message with 'tool_calls'
#       must be followed by tool messages

小明问：”如果第二轮的 user 消息不小心插在 assistant(tool_calls) 和 tool 之间，就炸了？”

“对。API 直接报错——协议规定 tool_calls 之后只能接 tool 消息。这就是为什么记忆管理不只是简单的 append——你得保证消息顺序合法。“

实现短期记忆：ShortTermMemory 类

老张说：”搞清了规则，看代码实现。核心就是一个 ShortTermMemory 类，底层是一个 list[dict]。”

@dataclass
class ShortTermMemory:
  max_messages: int = 40      # 最多保留多少条消息
  _messages: list[dict] = field(default_factory=list)

  def add(self, message: dict) -> None:
    """追加一条消息，超过上限时自动裁剪"""
    self._messages.append(message)
    self._trim()

  def to_api_format(self) -> list[dict]:
    """返回完整消息列表，传给 OpenAI API"""
    return list(self._messages)

  def clear_non_system(self) -> None:
    """清空除 system 外的所有消息"""
    self._messages = [m for m in self._messages if m.get("role") == "system"]

“三个核心方法：”

“记忆的核心就是 _messages 这个列表。每次对话产生的新消息通过 add() 追加进去，调用 API 时通过 to_api_format() 把整个列表传过去。模型因为看到了全部历史，就能’记住’之前的对话。”

裁剪策略：删一条就够？不够

小明问：”_trim() 是做什么的？”

“消息列表不能无限增长。”老张说，”假设你设置了 max_messages=40，超过 40 条后，_trim() 自动删除最旧的消息。但这里有个坑——不能只删一条。“

假设 messages 列表超大了，最前面是这几条：

[0] {"role":"system","content":"你是智能助手"}     ← system，不能删
[1] {"role":"user","content":"成都天气？"}      ← 最旧的非 system
[2] {"role":"assistant","content":None,"tool_calls":[...]}  ← 要删
[3] {"role":"tool","tool_call_id":"xxx","content":"多云"}  ← 也要删！
[4] {"role":"assistant","content":"成都多云..."}       ← 也要删！删完这组
[5] {"role":"user","content":"上海呢？"}

“如果只删 [1]（user 消息），[2] 的 tool_calls 孤零零留在前面，[3] 的 tool 结果就没有前置的 tool_calls 了——协议直接报错。”

正确的裁剪策略：

def _trim(self) -> None:
  while self.message_count() > self.max_messages:
    # 1. 找到最旧的非 system 消息
    idx = next(
      (i for i, m in enumerate(self._messages)
       if m.get("role") != "system"),
      None,
    )
    if idx is None:
      break

    # 2. 删除它
    self._messages.pop(idx)

    # 3. 删除之后，把紧跟在后面的 tool 消息一并删掉
    #   （避免留下"孤儿"tool 消息）
    while idx < len(self._messages) and \
        self._messages[idx].get("role") == "tool":
      self._messages.pop(idx)

老张逐行解释：

1. 跳过 system — system 消息是 Agent 的角色设定，不能删
2. 找最旧的非 system — 从列表头部开始，找到第一条非 system 的消息
3. 弹出它 — 删掉这条消息
4. 吞掉紧跟的 tool 消息 — 如果被删的消息后面紧跟着 role=tool，一并删除

“这个策略叫’删当前 + 向后吞 tool 组‘。核心目的只有一个——保证不会留下 orphan tool 消息破坏协议。“

完整的一条对话流

老张把 Agent 的一轮对话完整展开：

class Agent:
  def __init__(self):
    self.memory = ShortTermMemory(max_messages=40)
    self.memory.add_system("你是一个智能助手，可以使用工具...")
    self.tools = get_tools_schema()

  def chat(self, user_input: str) -> str:
    # 第 0 步：用户输入写入记忆
    self.memory.add({"role": "user", "content": user_input})

    # 第 1 步：第一轮 API 调用（带 tools，让模型决定要不要调工具）
    message = chat(self.memory.to_api_format(), tools=self.tools)

    # 如果模型不需要工具，消息链到此结束
    if not message.tool_calls:
      self.memory.add({"role": "assistant", "content": message.content})
      return message.content

    # 第 2 步：把模型的 tool_calls 请求写入记忆
    self.memory.add(assistant_to_dict(message))

    # 第 3 步：执行工具，每条结果写入记忆
    for tool_call in message.tool_calls:
      result = execute_tool(tool_call.function.name,
                  tool_call.function.arguments)
      self.memory.add({
        "role": "tool",
        "tool_call_id": tool_call.id,
        "content": result,
      })

    # 第 4 步：第二轮 API 调用（不带 tools）
    final = chat(self.memory.to_api_format(), tools=None)
    self.memory.add({"role": "assistant", "content": final.content})
    return final.content

“注意每次对话回合都调 self.memory.add()，把新消息追加进去。关键操作都是对 self.memory 的 读（to_api_format()）和 写（add()）——Agent 本身不再直接操作 messages 列表。”

小明对照着时序图看：

第1轮对话："我叫小明"
  memory.add(user: "我叫小明")
  API 调用 → 模型直接回答（不需要工具）
  memory.add(assistant: "你好小明")

  memory 状态：[system, user("我叫小明"), assistant("你好小明")]

第2轮对话："成都天气？"
  memory.add(user: "成都天气？")
  API 调用 → 模型决定调 get_weather("成都")
  memory.add(assistant with tool_calls)
  执行 get_weather("成都") → "多云 16°C"
  memory.add(tool: "多云 16°C")
  API 调用 → 模型基于结果回答
  memory.add(assistant: "成都多云 16°C")

  memory 状态：[system,
          user("我叫小明"), assistant("你好小明"),
          user("成都天气？"), assistant(tool_calls),
          tool("多云"), assistant("成都多云")]

第3轮对话："我刚才说我叫什么名字"
  memory.add(user: "我刚才说我叫什么名字")
  API 调用 → 模型看到完整历史，回答"小明"
  memory.add(assistant: "你刚才说你叫小明")

“第三轮调用时，模型看到整个 message 列表里有’我叫小明’和’你好小明’，所以知道用户叫小明。记忆就是靠这个列表——不是模型记住了，是你传过去的历史让它看到了。“

/clear：可控的遗忘

老张说：”记忆除了会’记’，还要会’忘’。Agent 需要提供 重置记忆 的功能。”

“为什么？”

“比如用户想开始一段全新的对话，不想被之前的话题干扰。再比如测试的时候，你想验证新功能而不受旧对话影响。”

“实现很简单——保留 system 消息，清空其余：”

def clear_non_system(self) -> None:
  self._messages = [m for m in self._messages
            if m.get("role") == "system"]

“在 CLI 里绑一个命令：”

if user_input.lower() == "/clear":
  agent.clear()
  print("记忆已清除，开始新对话。")
  continue

“清空前：”

[system, user("我叫小明"), assistant("你好小明"),
 user("成都天气？"), assistant(tool_calls), tool("多云"), assistant("多云16°C")]

“清空后：”

[system]

“相当于 Agent 重启——只记得自己的角色设定，对话历史全忘。”

小明试了一下：

你：我叫小明
Agent：你好小明！
（当前记忆：2 条消息）

/clear
记忆已清除，开始新对话。

你：我刚才说我叫什么名字？
Agent：你没有告诉我你的名字。

assistant_to_dict：把 SDK 对象转成可存储的 dict

老张指着一处细节说：”assistant_to_dict() 这个函数值得单独讲。”

“为什么？直接用 message.to_dict() 不行吗？”

“OpenAI SDK 返回的 message 对象可以直接 .to_dict()，但不同厂商的返回格式有差异。而且 content 可能是 None（当消息是 tool_calls 时）。为了兼容更多厂商，我们手写转换：”

def assistant_to_dict(message) -> dict:
  item = {
    "role": "assistant",
    "content": message.content if message.content is not None else "",
  }
  if message.tool_calls:
    item["tool_calls"] = [
      {
        "id": tc.id,
        "type": "function",
        "function": {
          "name": tc.function.name,
          "arguments": tc.function.arguments,
        },
      }
      for tc in message.tool_calls
    ]
  return item

“两个要点：”

LLM 封装：第二轮传 tools=None

老张又指着 llm.py 说：”还有一个容易被忽略的参数——第二轮 API 调用时 tools 必须传 None。“

def chat(messages, tools=None, tool_choice=None):
  kwargs = {
    "model": MODEL,
    "messages": messages,
    "temperature": 0,
  }
  if tools is not None:
    kwargs["tools"] = tools
  if tool_choice is not None:
    kwargs["tool_choice"] = tool_choice
  return client.chat.completions.create(**kwargs)

“第一轮调用时 tools=self.tools，模型可以决定调不调工具。第二轮调用时 tools=None——告诉模型’工具结果已经给你了，不许再调工具，直接生成回答’。“

“如果第二轮还传 tools，模型可能觉得’这个工具结果不够，我再调一个’——陷入无限循环。Day4 的规定是每轮用户输入最多一次工具回合，所以第二轮必须禁止工具调用。”

不同层的记忆

老张说：”今天讲的是短期记忆——存在 Agent 进程的 list 里，进程重启就没了。这是最基本的记忆层。但完整的 Agent 记忆体系有三层：”

“今天你学的是第一层。这是基础——消息链记不住，后面两层无从谈起。但只要消息链机制搞清楚了，后面的记忆层都是在它的基础上扩展。”

这篇文章的核心

小明把笔记整理成了一张速查表：

总结

老张最后在白板上写了一句话：

Agent 的记忆不是模型的功能，是你的程序写出来的。消息列表怎么管理，记忆就多好。

“上两篇你学会了工程化管理多个工具、用 MCP 接入外部服务。这篇你学会了给 Agent 加记忆——让它记住整个会话的历史。”

“核心三件事：

1. 消息列表就是记忆 — 每次 API 调用把完整历史传过去
2. Function Calling 让记忆变复杂 — assistant(tool_calls) 和 tool 必须配对，裁剪时不能留孤儿
3. 用 ShortTermMemory 封装 — 统一管理消息的读、写、裁剪、清空，Agent 代码不直接碰 messages 列表”

小明说：”所以之前我们写的 Function Calling 代码，本质上就是在操作 Agent 的记忆？”

“对。messages.append() 每次都是在往记忆里写东西。只不过我们之前没有把这部分逻辑封装出来——记忆逻辑和对话逻辑混在一起。现在把记忆抽成独立的 ShortTermMemory 类，职责清晰了。”

“下一步呢？”

老张说：”下一步是 Agent 循环。你注意到了，今天的一轮对话最多调一次工具——模型说’需要天气’，你调了，喂结果，模型就回答了。但有时候，模型的第一次工具结果不够，它需要再调一个工具才能回答——这就需要 ‘多步工具链’。”

“怎么实现？”

“很简单——把今天的 if tool_calls → 执行 → 回答 换成 while tool_calls → 执行 → 再问模型。让模型在判断’我还需要更多信息’时继续调工具，直到它觉得够了、给出最终回答。”

“这就是下一篇文章的内容——Agent Loop 多步工具调用。”