没有记忆的 Agent,每次对话都是陌生人。有了记忆,它才是真正的助手。


今天的目标

让 Agent 能记住:

  1. 对话历史:刚才聊了什么
  2. 关键信息:用户提到的重要事实
  3. 工具使用记录:之前用了什么工具,得到了什么结果

为什么记忆很重要

想象这个场景:

你:帮我创建一个叫 notes.txt 的文件,内容写"今天天气真好"
Agent:好的,文件已创建。

你:把刚才那个文件名改成 weather.txt
Agent:哪个文件?我没有找到叫"那个文件"的文件。

没有记忆,Agent 不知道"那个文件"是什么。


三层记忆架构

我们设计三层记忆,就像人类的记忆系统:

┌─────────────────────────────────────┐
│  长期记忆(Summary Memory)           │
│  - 关键事实的摘要                     │
│  - 持久化到文件                       │
├─────────────────────────────────────┤
│  工作记忆(Working Memory)           │
│  - 最近几轮对话的完整内容              │
│  - 当前任务的上下文                    │
├─────────────────────────────────────┤
│  即时记忆(Immediate Memory)         │
│  - 当前这一轮的输入输出                │
└─────────────────────────────────────┘
  • 即时记忆:每一轮对话的内容,不需要特殊处理
  • 工作记忆:保留最近 N 轮对话,用于理解上下文
  • 长期记忆:把重要信息压缩成摘要,永久保存

完整代码

工具实现部分参考 Day 2。这里只展示记忆系统和核心逻辑。

# day4_memory.py

import json
import os
from openai import OpenAI


# ========== 记忆系统 ==========

class Memory:
    """Agent 的记忆系统"""

    def __init__(self, max_working=10, summary_threshold=8):
        self.working_memory = []  # 工作记忆:最近的对话
        self.long_term_memory = []  # 长期记忆:重要信息摘要
        self.max_working = max_working
        self.summary_threshold = summary_threshold
        self.context_file = "agent_memory.json"

    def add_conversation(self, role: str, content: str):
        """添加一条对话到工作记忆"""
        self.working_memory.append({"role": role, "content": content})

        # 如果工作记忆太长,触发摘要
        if len(self.working_memory) > self.summary_threshold:
            self._summarize()

    def _summarize(self):
        """把工作记忆中较早的内容压缩成摘要"""
        if len(self.working_memory) <= 4:
            return  # 太短不需要摘要

        # 保留最近 4 轮,其余压缩
        to_summarize = self.working_memory[:-4]
        self.working_memory = self.working_memory[-4:]

        # 生成摘要
        conversation_text = "\n".join(
            f"{msg['role']}: {msg['content']}" for msg in to_summarize
        )

        client = OpenAI(api_key="你的API Key", base_url="https://api.mimo.ai/v1")
        response = client.chat.completions.create(
            model="mimo-v2-flash",
            messages=[
                {"role": "system", "content": "用中文简洁地总结以下对话的关键信息,包括用户的需求、工具的使用结果、以及任何重要事实。"},
                {"role": "user", "content": conversation_text},
            ],
            temperature=0,
        )
        summary = response.choices[0].message.content.strip()

        # 存入长期记忆
        self.long_term_memory.append(summary)

    def get_context(self) -> list:
        """获取发送给 AI 的上下文"""
        messages = []

        # 加入长期记忆摘要
        if self.long_term_memory:
            summary_text = "\n".join(self.long_term_memory)
            messages.append({
                "role": "system",
                "content": f"之前对话的关键信息摘要:\n{summary_text}",
            })

        # 加入工作记忆
        messages.extend(self.working_memory)

        return messages

    def save(self):
        """持久化到文件"""
        data = {
            "working": self.working_memory,
            "long_term": self.long_term_memory,
        }
        with open(self.context_file, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)

    def load(self):
        """从文件加载"""
        if os.path.exists(self.context_file):
            with open(self.context_file, "r", encoding="utf-8") as f:
                data = json.load(f)
                self.working_memory = data.get("working", [])
                self.long_term_memory = data.get("long_term", [])

    def clear(self):
        """清空记忆"""
        self.working_memory = []
        self.long_term_memory = []
        if os.path.exists(self.context_file):
            os.remove(self.context_file)

    def get_stats(self) -> dict:
        return {
            "工作记忆": len(self.working_memory),
            "长期记忆": len(self.long_term_memory),
        }


# ========== AI 决策(带记忆)==========

def think(user_input: str, memory: Memory) -> dict:
    """AI 思考并决策——这次带上了记忆"""
    client = OpenAI(api_key="你的API Key", base_url="https://api.mimo.ai/v1")
    tools_prompt = build_tools_prompt()  # 从 Day 2 导入

    # 从记忆中获取上下文
    context = memory.get_context()

    system_prompt = f"""你是一个智能助手,有记忆能力,可以使用工具。

{tools_prompt}

重要:
1. 你可以记住之前的对话内容
2. 如果用户提到"刚才"、"之前"、"那个文件"等,参考记忆中的信息
3. 用 JSON 格式回复:
   - 工具调用:{{"thought": "思考", "tool": "工具名", "params": {{}}}}
   - 聊天回复:{{"thought": "思考", "tool": "chat", "params": {{"response": "回复"}}}}
4. 只输出 JSON"""

    messages = [{"role": "system", "content": system_prompt}]
    messages.extend(context)
    messages.append({"role": "user", "content": user_input})

    response = client.chat.completions.create(
        model="mimo-v2-flash",
        messages=messages,
        temperature=0,
    )

    reply = response.choices[0].message.content.strip()
    try:
        if reply.startswith("```"):
            reply = reply.split("\n", 1)[1]
            reply = reply.rsplit("```", 1)[0]
        return json.loads(reply)
    except json.JSONDecodeError:
        return {"thought": "解析失败", "tool": "chat", "params": {"response": reply}}


def run_agent(user_input: str, memory: Memory):
    """运行 Agent——这次会记住对话"""
    print(f"\n你:{user_input}")

    decision = think(user_input, memory)
    tool_name = decision.get("tool", "chat")
    params = decision.get("params", {})
    thought = decision.get("thought", "")

    if thought:
        print(f"[思考]: {thought}")

    if tool_name != "chat":
        print(f"[调用工具]: {tool_name}")

    result = execute_tool(tool_name, params)  # 从 Day 2 导入
    print(f"Agent:{result}")

    # 记住这轮对话
    memory.add_conversation("user", user_input)
    memory.add_conversation("assistant", result)

    stats = memory.get_stats()
    print(f"(记忆:工作 {stats['工作记忆']} 条,长期 {stats['长期记忆']} 条)")

运行测试

你:帮我创建一个叫 notes.txt 的文件,内容写"今天天气真好"
Agent:成功写入文件 'notes.txt',共 7 字符
(记忆:工作 2 条,长期 0 条)

你:把刚才那个文件名改成 weather.txt
Agent:好的,我先读取 notes.txt 的内容,然后写入 weather.txt。
[调用工具]: read_file
[调用工具]: write_file
Agent:已完成!文件已从 notes.txt 改名为 weather.txt。
(记忆:工作 4 条,长期 0 条)

你:那个文件里写了什么?
Agent:weather.txt 的内容是:"今天天气真好"
(记忆:工作 6 条,长期 0 条)

Agent 记住了"那个文件"是 weather.txt。


核心概念

工作记忆 vs 长期记忆

# 工作记忆:完整的对话历史(保留细节)
self.working_memory = [
    {"role": "user", "content": "帮我创建 notes.txt"},
    {"role": "assistant", "content": "文件已创建"},
]

# 长期记忆:压缩后的摘要(保留关键信息)
self.long_term_memory = [
    "用户创建了一个 notes.txt 文件,内容是'今天天气真好',后来改名为 weather.txt"
]

工作记忆保留细节,长期记忆保留关键信息。

摘要触发机制

if len(self.working_memory) > self.summary_threshold:
    self._summarize()

当工作记忆超过阈值时,自动压缩较早的内容为摘要。这解决了 LLM 上下文窗口限制的问题。

记忆持久化

# 保存到文件
memory.save()

# 下次启动时加载
memory.load()

这样 Agent 跨多次运行也能记住之前聊过的内容。


常见错误

错误 1:摘要触发太频繁

# 错误:每轮对话都触发摘要,浪费 API 调用
self.summary_threshold = 2  # 太小了

# 正确:留够工作记忆的空间
self.summary_threshold = 8  # 工作记忆超过 8 条才触发

错误 2:长期记忆无限增长

长期记忆条数会越来越多,最终也会超出上下文窗口。解决方案:

def _summarize(self):
    # 如果长期记忆也太多,把它们也压缩
    if len(self.long_term_memory) > 5:
        combined = "\n".join(self.long_term_memory)
        # 用 AI 把所有摘要再压缩成一个
        # ...
        self.long_term_memory = [compressed_summary]

错误 3:忘记保存记忆

用户退出时必须调用 memory.save(),否则记忆丢失。

if user_input == "/quit":
    memory.save()  # 退出前保存
    print("记忆已保存。再见!")
    break

动手实验

  1. 调整 max_workingsummary_threshold,观察记忆行为变化
  2. 在多次运行之间保持记忆:运行 → 退出 → 再运行,看 Agent 是否记得之前的内容
  3. 输入一个很长的对话,触发摘要压缩,观察长期记忆的内容

明天预告

今天我们有了记忆,但 Agent 的执行还是"一步一动"的。用户说"帮我规划一个项目",Agent 会直接开始执行,而不是先想清楚再做。

明天是整个系列最关键的一天:Agent Loop。你会理解为什么 Agent 需要"循环",以及这个循环如何让 Agent 变得真正强大。

Day 5:Agent Loop——最关键的一天

Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐