ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

Day 16 · 长任务与记忆:让 Agent 记住上下文

Day 16 · 长任务与记忆:让 Agent 记住上下文 「AI Python 系列」第 01 栏 · AI 时代的 Python 办公自动化全栏 18 篇 · 零成本跟完品牌梅雅达编程笔记摘要前面两篇搭好了 ReAct Agent但它有个致命问题——每次对话都是新生上一轮说过的话下一轮就忘。本篇给 Agent 加装双层记忆系统短时记忆用对话列表维持上下文连贯长时记忆用 JSON 文件做跨会话持久化。本篇使用 GLM-4.7-Flash永久免费4 轮对话约消耗 5000 token零成本跑完。关键词Agent 记忆、上下文管理、JSON 持久化、长任务、System Prompt 注入环境与前置条件项目版本 / 说明操作系统Windows 10/11、macOS、Linux 均可Python3.9 及以上本文用 3.12 验证依赖库json标准库、llm_client本专栏封装的 LLM 调用模块见 Day 04LLM 提供商智谱 GLM-4.7-Flash免费代码中可切换 DeepSeek / Qwen前置知识需先完成 Day 14-15理解 ReAct Agent 和 Function Calling 的基本原理预计耗时阅读 15 分钟 动手验证 10 分钟开篇 · 一个让我翻车的真实场景Day 14-15 我们做了能调用工具的 ReAct Agent我兴冲冲拿给学生用。结果第一个学生就反馈“老师它记性怎么这么差”事情是这样的学生先跟 Agent 说我叫小明今年 10 岁Agent 回复你好小明。然后学生紧接着问我叫什么Agent 直接回了句抱歉我不知道你的名字。学生当场就愣住了然后跑来找我告状。我一开始还以为是模型出问题了排查了半天才发现——根本不是 AI 不够聪明是代码层面压根没把上一轮对话传过去。每次调 API 都是一次独立请求模型本身无状态你不把历史消息塞进去它当然什么都不记得。这就像你跟一个人打电话每说一句话就挂断再重拨对方怎么可能记得你刚才说了什么找到原因后我给 Agent 加了一套双层记忆系统。今天这篇就讲这个事情——怎么让 Agent 真的记住你。一、为什么要分两层一层 JSON 不行吗我第一版其实是只用了一个 JSON 文件把所有东西都往里塞。跑了两天就发现不对——每次对话都要读写磁盘速度肉眼可见地变慢。打开 JSON 文件一看好家伙光对话历史就涨了好几 KB。后来我改成只存在内存里速度是快了但程序一关全没了。用户第二天打开又得重新自我介绍体验极差。折腾了两版之后我才意识到这两种记忆的性质完全不同必须分开处理。维度短时记忆内存长时记忆磁盘生命周期程序关了就没了写进文件重启还在访问频率每轮对话都要读写只在记住/回忆时才动丢了代价低——当前对话断就断了高——用户数据永久丢失适合存什么当前聊天的上下文用户名、偏好、长期事实热数据放内存冷数据落磁盘——这个思路不是我发明的数据库早就这么干了。但在 Agent 里实现起来其实很朴素。二、短时记忆就是个列表别想复杂了短时记忆的实现特别简单——一个 Python 列表每轮对话把用户消息和 AI 回复都append进去就这样。classMemoryAgent:def__init__(self,memory_fileagent_memory.json,providerglm):self.clientLLMClient(providerprovider)self.memory_filememory_file self.history[]# 短时记忆就是个列表self.long_termself._load_memory()# 长时记忆从文件加载对话的时候把最近的历史拼成文本传给模型defchat(self,message):self.history.append({role:user,content:message})recentself.history[-10:]# 只取最近10条context\n.join([f{用户ifm[role]userelse助手}:{m[content]}forminrecent])responseself.client.chat(context,system_promptsystem_prompt,temperature0.5,max_tokens500)self.history.append({role:assistant,content:response})returnresponse你可能会问为什么不把全部历史都传给模型我一开始还真是这么干的——结果聊了 20 轮之后API 直接报 token 超限的错误。这才想起来模型的上下文窗口是有限的。后来我改成[-10:]只取最近 10 条问题就解决了。10 条这个数字也不是精确计算出来的就是试了几次发现够用用户说那个文件刚才那个之类的指代模型基本都能回溯到。如果你的场景需要更长的回溯可以调大到 15-20 条但别贪多token 消耗是实打实的。三、长时记忆一个 JSON 文件搞定长时记忆要解决的是关掉程序再打开之前记住的东西还在。我用 JSON 文件来做——够简单够直观不需要装额外的库。3.1 加载与保存def_load_memory(self):ifos.path.exists(self.memory_file):try:withopen(self.memory_file,r,encodingutf-8)asf:returnjson.load(f)except(json.JSONDecodeError,IOError):passreturn{facts:{},notes:[]}def_save_memory(self):withopen(self.memory_file,w,encodingutf-8)asf:json.dump(self.long_term,f,ensure_asciiFalse,indent2)这里有个我踩过的坑有一次我调试的时候手动 CtrlC 强杀了程序结果 JSON 文件写到一半就断了变成了非法格式。下次启动 Agent 的时候直接报错崩溃。后来我加了try/except捕获JSONDecodeError文件坏了就给个空结构重新来至少不会让整个程序挂掉。不过说实话这么做有个隐患——用户的记忆文件坏了你静默丢弃用户可能根本不知道。生产环境最好加个日志记录或者把损坏的文件备份成.bak再新建。3.2 记住与回忆defremember(self,key,value):self.long_term[facts][key]value self._save_memory()returnf已记住{key}{value}defrecall(self,key):valueself.long_term[facts].get(key)ifvalue:returnf回忆起来{key}{value}returnf没有关于「{key}」的记忆每次remember都立即写盘。这样做最安全不会丢数据。但有个代价——如果你频繁调用 remember比如让它批量记住一堆东西磁盘 IO 会很频繁。我个人使用场景下感觉不到延迟但如果是服务端多人并发用建议改成攒一批再写的模式并且加文件锁。3.3 笔记给没有明确 key 的信息留个位置除了键值对还有种信息不太好塞进 facts——比如用户说下周要交报告。这不是一个keyvalue能表达的东西更像一条备忘录。所以我加了一个 notes 列表每条带时间戳defadd_note(self,note):timestampdatetime.now().strftime(%Y-%m-%d %H:%M)self.long_term[notes].append({time:timestamp,note:note})self._save_memory()returnf已记录笔记{note}简单总结下分工有明确名字的东西用户名、所在城市存 facts事件、提醒、随手记存 notes。四、最关键的一步记忆怎么喂给模型记忆存好了还不够。模型看不到你的 JSON 文件它只能看到你传给它的文本。所以必须在每轮对话时把记忆格式化后塞进 system prompt。这个环节折腾了我一阵。一开始我把记忆拼成一个长字符串直接丢进去结果发现模型有时候会无视我注入的信息。后来我学聪明了在 system prompt 里加了明确的规则告诉模型这些是你记住的信息要在回答中用到。格式化函数长这样def_format_memory_for_prompt(self):parts[]ifself.long_term[facts]:facts\n.join([f -{k}:{v}fork,vinself.long_term[facts].items()])parts.append(f已知信息\n{facts})ifself.long_term[notes]:notes\n.join([f - [{n[time]}]{n[note]}forninself.long_term[notes][-5:]])parts.append(f近期笔记\n{notes})return\n\n.join(parts)ifpartselse暂无记忆然后在chat里把它拼进 system promptmemory_strself._format_memory_for_prompt()system_promptf你是一个有记忆能力的智能助手。 你的长时记忆{memory_str}规则 - 对话中自然地使用你记住的信息 - 如果用户说记住xxx是yyy主动调用记忆功能 - 如果用户问你还记得xxx吗从记忆中查找 - 保持简洁友好的风格有两件事需要注意第一notes 我取了[-5:]只给最近 5 条。原因跟对话历史截断一样——记忆越多 token 消耗越大而且一周前的笔记对今天的对话通常没什么参考价值。第二这个方案的 token 成本会随记忆量线性增长。100 条 facts 大概占 500 token每轮对话都重复发送。如果你的 Agent 需要记住很多东西后面可以考虑改成向量检索——只注入跟当前问题相关的记忆。这个进阶方案在文末有提到。五、记忆在长任务里还有另一层作用前面说的都是记住用户是谁这类基础功能。但记忆系统在更复杂的场景里作用更大——它可以充当任务的检查点。举个实际例子假设让 Agent 帮你做一份月度报告要读 Excel、统计品类、分析趋势、生成文字、输出 Markdown5 个步骤。如果没有记忆第 3 步的时候第 1 步读的数据已经丢了。有了记忆系统每完成一步就把关键中间结果存下来# 第1步完成后agent.remember(销售数据_总行数,12453)agent.remember(销售数据_时间范围,2026-07-01 至 2026-07-31)agent.add_note(第1步完成数据读取成功共12453条)更实际的价值在于中断恢复。我有一次让学生跑一个多步骤的数据处理任务跑到第 4 步程序崩了好像是内存不够。如果没有记忆得从头来。但有了记忆重启后 Agent 看到 notes 里写着前 3 步已完成可以直接从第 4 步接着干。这种检查点模式在很多工程系统里都有游戏里的存档也是同一个思路。本篇的代码实现了基础记忆机制检查点功能可以在此基础上扩展。六、我踩过的三个坑写到这里说几个我在实际开发中真实遇到的问题不是理论上的可能出问题是真真切切踩过的。6.1 JSON 文件写到一半程序崩了前面提到过一次这里展开说。原因是_save_memory在写入过程中如果程序被强杀CtrlC、断电、OOM文件就只写了一半JSON 格式不完整。下次启动时json.load()直接抛JSONDecodeError。我的处理方式是加了异常捕获文件坏了就给空结构。但更好的做法是先写临时文件再原子重命名def_save_memory(self):tmp_fileself.memory_file.tmpwithopen(tmp_file,w,encodingutf-8)asf:json.dump(self.long_term,f,ensure_asciiFalse,indent2)os.replace(tmp_file,self.memory_file)# 原子替换os.replace在大多数操作系统上是原子操作即使写入过程中断电旧文件也不会被破坏。这个改动虽然小但能避免很多莫名其妙的数据丢失。6.2 两个人同时用记忆会互相覆盖这个坑是我把 Agent 部署成 Web 服务之后发现的。两个用户同时调用remember都触发了_save_memory后写的直接覆盖先写的——用户 A 的记忆把用户 B 的盖掉了。如果只是个人本地用这个问题不存在。但如果多人共用必须加锁。Linux 上用fcntlWindows 上用msvcrt.locking或者干脆用filelock这个跨平台的库。6.3 记忆越攒越多JSON 越来越大用了一个月之后打开 JSON 文件里面存了几百条 facts。每次启动都要全量加载到内存每轮对话都要把几百条 facts 格式化后塞进 system prompt——token 消耗肉眼可见地在涨。目前我的临时方案是facts 超过 50 条就手动清理一下。但这显然不是长久之计。后续准备加一个淘汰策略比如超过 30 天没被 recall 过的 facts 自动归档notes 超过 3 个月的自动清理。更彻底的方案是做记忆总结——让 AI 把零散的记忆压缩成几条核心摘要。七、完整代码与运行验证核心代码都在下面了。llm_client.py是 Day 04 封装好的 LLM 调用模块这里直接复用。importosimportjsonfromllm_clientimportLLMClientclassMemoryAgent:def__init__(self,memory_fileagent_memory.json,providerglm):self.clientLLMClient(providerprovider)self.memory_filememory_file self.history[]# 短时记忆self.long_termself._load_memory()# 长时记忆def_load_memory(self):ifos.path.exists(self.memory_file):try:withopen(self.memory_file,r,encodingutf-8)asf:returnjson.load(f)except(json.JSONDecodeError,IOError):passreturn{facts:{},notes:[]}def_save_memory(self):# 先写临时文件再原子替换防止写到一半崩溃tmp_fileself.memory_file.tmpwithopen(tmp_file,w,encodingutf-8)asf:json.dump(self.long_term,f,ensure_asciiFalse,indent2)os.replace(tmp_file,self.memory_file)defremember(self,key,value):self.long_term[facts][key]value self._save_memory()returnf已记住{key}{value}defrecall(self,key):valueself.long_term[facts].get(key)returnf回忆起来{key}{value}ifvalueelsef没有关于「{key}」的记忆defadd_note(self,note):fromdatetimeimportdatetime timestampdatetime.now().strftime(%Y-%m-%d %H:%M)ifnotesnotinself.long_term:self.long_term[notes][]self.long_term[notes].append({time:timestamp,note:note})self._save_memory()returnf已记录笔记{note}defchat(self,message):# 把长时记忆注入 system promptmemory_strself._format_memory_for_prompt()system_promptf你是一个有记忆的助手。已知信息\n{memory_str}\n自然地使用这些信息回答。# 短时记忆记录对话只取最近10条self.history.append({role:user,content:message})recentself.history[-10:]context\n.join([f{用户ifm[role]userelse助手}:{m[content]}forminrecent])responseself.client.chat(context,system_promptsystem_prompt,temperature0.5,max_tokens500)self.history.append({role:assistant,content:responseor(API异常)})returnresponsedef_format_memory_for_prompt(self):parts[]ifself.long_term.get(facts):facts\n.join([f -{k}:{v}fork,vinself.long_term[facts].items()])parts.append(f已知信息\n{facts})ifself.long_term.get(notes):notes\n.join([f - [{n[time]}]{n[note]}forninself.long_term[notes][-5:]])parts.append(f近期笔记\n{notes})return\n\n.join(parts)ifpartselse暂无记忆怎么跑目录结构day16_code/ ├── memory_agent.py # 本篇代码 ├── llm_client.py # Day 04 的封装直接复用 └── requirements.txtcdday16_code pipinstall-rrequirements.txt python memory_agent.py跑起来之后试试这几步感受一下记忆是怎么工作的你: /remember 用户名梅雅达 已记住用户名 梅雅达 你: /remember 所在城市北京 已记住所在城市 北京 你: 你好我是梅雅达 助手: 你好梅雅达你在北京工作有什么需要帮忙的吗 你: 你还记得我叫什么吗 助手: 你叫梅雅达在北京工作。然后退出程序。你会发现当前目录多了一个agent_memory.json{facts:{用户名:梅雅达,所在城市:北京},notes:[]}重新运行python memory_agent.py直接输入我叫什么——如果它还能回答出来说明记忆确实落盘了。如果重启后记忆丢了检查两点①memory_file路径是不是相对路径导致写到了别的目录建议用绝对路径② 程序有没有当前目录的写权限。八、写在最后Agent 的记忆系统说到底就是状态管理——模型本身是无状态的每一次 API 调用都是一个独立的请求。我们要做的是在外部维护状态然后在对话时把相关状态喂给模型。短时记忆管当前对话的连贯性长时记忆管跨会话的数据持久化两者通过 system prompt 注入汇合。这个分层思路不只适用于聊天机器人——你以后做更复杂的 Agent不管是自动化报表、数据分析还是工具编排都会用到类似的状态管理机制。JSON 方案够轻量、够直观适合个人使用和小规模场景。等记忆量涨到几百条以上或者需要多人并发就该考虑 SQLite 或向量数据库了。但在那之前先用 JSON 把功能跑通再说。你可以接着折腾的方向加个遗忘功能。用户改名了、搬城市了旧记忆就该能删掉。实现起来就一行del self.long_term[facts][key]但别忘了处理 key 不存在的情况。用向量检索替代全量注入。每轮对话把所有 facts 都塞进 prompt记忆多了 token 扛不住。试试给每条记忆生成 embedding查找时按相似度排序只注入 Top-3。sentence-transformers可以本地跑也可以调 GLM 的 embedding API。做一个记忆总结。对话超过 20 轮时让 AI 把前面的内容总结成一段摘要存进 notes然后清空短时记忆。核心逻辑大致是if len(history) 20→ 调模型总结 →add_note(summary)→history.clear()。这个思路和 MemGPT 论文里的做法很像感兴趣的可以去翻翻。下期预告Day 17 · 综合实战 A日报/周报/月报自动化 AgentAgent 有工具、有记忆了该来干正事了。明天做一个报告自动化 Agent——读取经营数据AI 分析趋势和异常自动生成 Markdown 格式的日报/周报/月报。整合前面学的所有技能。资源与工具Python json 模块文档https://docs.python.org/3/library/json.htmlLLM 记忆机制综述MemGPThttps://arxiv.org/abs/2304.11477智谱 GLM 模型文档https://docs.bigmodel.cn/本专栏配套代码CSDN 下载区往期回顾Day 15 · 多工具 Agent让 AI 自主选择和组合工具专栏「AI Python 系列」第 01 栏 · AI办公自动化©️ 梅雅达编程笔记原创 · 首发 CSDN · 转载请注明出处
返回列表