用纯 Python 实现三类 Agent 记忆
教程区分上下文中的短期记忆、跨会话的长期事实和多 Agent 共享记忆,并提供 Python 3.9+ 标准库示例。代码可离线运行,模型调用使用占位实现,共享记忆写入本地数据库。
AI 深度解读
模型不会自动记住上次调用发生了什么。Agent 的「记忆」,是应用代码决定保存哪些信息、什么时候检索,以及把什么重新放进上下文。原文用 Python 和 SQLite 拆出了短期、长期、共享三种记忆,核心结论很直接:先管好信息的进入、召回和删除,再考虑框架。
它值得程序员读的地方,是把记忆问题落到了接口和数据结构上。不过,这是一套可离线运行的教学实现,没有生产效果对照数据。压缩策略、检索算法和安全边界,都需要分开判断。
短期记忆要保住任务,也要保住工具调用关系
短期记忆服务于当前任务。作者保留三部分:system prompt、滚动摘要、最近几轮对话。历史超过预算后,把较早的内容合并进摘要,尽量让剩余上下文回到预算的一半左右。
分批压缩有两个理由:减少摘要模型的调用次数;让两次压缩之间的提示词前缀保持稳定,便于利用提示词缓存。摘要单独存储,每次结合旧摘要和新移出的对话更新,避免重复累积。
示例使用 budget=400,摘要最多保留 400 个字符。按英文约 4 个字符一个 token 粗算,摘要约占预算的四分之一。循环加入 8 组问答后,示例输出是 5 messages, 330 tokens,首条保留内容从 Question 7: 开始。
这些数字说明了代码行为,不是压缩质量评测。摘要函数只是截取用户消息的前 30 个字符;生产环境需要真正的摘要调用,保留决策、待解决问题、名称和数字。
压缩历史时,消息之间的依赖关系比截掉多少 token 更重要。
工具结果不能脱离对应调用单独出现。示例会向前回退,让保留窗口从普通 user 消息开始,避免留下没有调用记录的 tool_result。作者指出,Anthropic 和 OpenAI API 都会拒绝缺失调用的工具结果。
这里也没有硬性的预算保证:最新消息始终保留,为保持消息完整还可能多留几轮。生产环境应使用提供商的 token 计数器;接 Anthropic API 时,messages()[0]["content"] 要放到独立的 system 参数,其余才作为 messages。
长期记忆先存事实,再决定怎么找回来
长期记忆保存跨会话仍有价值的信息,例如偏好、项目约束和已经确定的决策。作者建议一条记录只放一个简短事实,并记录 subject、source 和创建时间。
subject 负责隔离归属。例如 user:42 的事实,只能在这个用户的范围内读写。缺少这层隔离,记忆就可能把一个用户的信息带进另一个用户的提示词。
示例 FactMemory 使用 SQLite,通过 UNIQUE(subject, fact) 去重,只接受长度为 10–300 个字符的事实。默认路径 :memory: 不会持久保存,传入文件路径才会落盘。
检索也很克制:只扫描该主体最新的 500 条事实,按单词重合数排序,时间权重为 0.1 / (1 + age_days),默认 k=3。因此,记忆中的 Python 3.12、AWS Lambda、Postgres 16,只有匹配当前问题时才会被召回。
限制很明确。words() 使用 [a-z0-9]+,无法直接承担中文检索;询问「which database」,也未必能找到只有「Postgres 16」的记录。作者建议需要时换成 FTS5 或 embeddings,同时保持 remember/recall/forget 接口稳定。
向量检索先将事实向量和查询向量归一化,再通过点积计算余弦相似度:
def vector_recall(rows, query: str, k: int = 3) -> list[str]:
"""rows: (fact, unit_vector, created) for one subject. Both sides are unit vectors, so dot = cosine."""
q, now = normalise(embed(query)), time.time()
scored = []
for fact, vec, created in rows:
cosine = sum(a * b for a, b in zip(q, vec))
recency = 1 / (1 + (now - created) / 86400)
scored.append((cosine + 0.1 * recency, fact))
return [fact for _, fact in sorted(scored, reverse=True)[:k]]
要看清 cosine + 0.1 * recency 这一行:在连续的相似度分数里,时间权重可能改变接近结果的顺序,并非严格只打破平局。示例的 embed() 还是 256 维哈希词袋占位实现,不能拿它证明语义检索有效。
作者建议先把向量存在事实旁边的 BLOB 列中,按主体扫描,必要时用 NumPy 加速。单个主体的数据多到无法接受逐次扫描时,再考虑向量数据库。
写入和删除,才是记忆最容易出问题的地方
作者没有让每条消息自动进入长期记忆,而是提供 remember 工具,要求模型只保存用户明确说过、未来仍有用的单条事实,排除秘密、临时细节、整段消息和网页或工具结果。
下面的处理函数检查疑似秘密,并将写入结果返回给模型:
def handle_remember(mem, user_id, tool_input, turn_id):
fact = tool_input.get("fact", "")
if SECRET.search(fact):
return "not saved: looks like a secret"
ok = mem.remember(f"user:{user_id}", fact, source=turn_id)
return "saved" if ok else "not saved: duplicate, too short or too long"
关键在于 SECRET.search(fact) 的能力有限:它依赖前文正则,既可能误拦,也无法判断事实究竟来自用户还是网页。工具描述是一层约束,代码没有提供来源验证。因此,作者建议只在人与 Agent 对话时开放 remember,不要给浏览不可信网页的 Agent 同样的写入入口。
召回内容放进 <memory> 标签前,会转义 < 和 >,并明确声明这些内容是用户数据,不是指令。转义能防止事实文本闭合标签,但不能保证消除提示词注入。
工具协议也要对应:原文是 Anthropic Messages API 的 input_schema;OpenAI Chat Completions 使用 function.parameters,Responses API 使用 parameters。
删除同样属于必要功能。forget(subject, "python") 会删除所有包含 Python 的事实,执行前应先展示匹配项。PRAGMA secure_delete = ON 用于覆盖 SQLite 已删除内容;WAL 模式下,作者还要求执行 PRAGMA wal_checkpoint(TRUNCATE)。完整删除请求还要覆盖摘要、日志、运行追踪和备份。
共享记忆适合同机协作,扩展能力有边界
多个 Agent 处理同一任务时,作者建议共享一份事实库,以任务作为 subject,例如 job:report-42;source 则记录写入 Agent 和资料来源。
示例创建 research.db,设置 PRAGMA journal_mode = WAL,连接使用 timeout=10,让同机多个进程可以协作读写。任务结束后通过 forget_all(job) 清理。
WAL 不适用于 NFS、SMB 这类网络文件系统。写入者超过少数几个后,作者建议迁移到 Postgres,对应使用 INSERT ... ON CONFLICT DO NOTHING、strpos 和 %s 占位符,但没有给出具体并发阈值或压测数据。
共享也不代表可信。研究笔记需要规范化 URL、稳定来源 ID,并验证引用确实出现在原页面;Agent 名称则帮助追查错误笔记。数据库解决共同访问,来源核验解决内容是否有依据。
我的判断
我赞成先把记忆做成独立事实、限制召回数量,并保留来源和删除接口。相比不断回灌完整聊天记录,这种结构更容易检查「为什么模型拿到了这条信息」。
但原文没有提供摘要保真度、召回准确率、成本或延迟的对照实验。半预算压缩、默认召回三条、何时迁移数据库,都是起步配置或经验之谈。
实现里也缺少事实更新机制:UNIQUE(subject, fact) 只能去掉完全相同的记录。用户从 Python 3.12 切到其他版本,旧事实仍可能存在;偏好发生冲突时该信哪条,原文没有解决。向量召回还缺少最低相关性门槛,候选都不相关时也可能返回前 k 条。
框架选择同样要看已有工程。LangGraph 提供 checkpointers 和长期 store;Mem0、Letta(原名 MemGPT)提供专门的记忆层;Anthropic memory tool、OpenAI stored conversations 也提供存储相关能力。这些组件可以省工程时间,保存什么、何时失效、怎样删除,仍需要应用自己定义。
能拿来做什么
- 给日常 AI 编程助手添加跨会话记忆时,可以先记录运行时版本、部署环境和代码偏好,并按用户或项目隔离,再限制每次注入的条数与字符数。
- 做中文或中英混合的 Node 服务时,可以沿用
remember/recall/forget接口,但应替换英文分词规则,用真实问题检查检索结果,再决定是否接 embeddings。 - 给会调用工具的 Agent 加历史压缩时,应检查工具调用与结果是否完整保留,并验证摘要有没有丢失任务约束和未完成事项。
- 搭建多个 Agent 协作的研究任务时,可以先用同机 SQLite 共享事实,同时记录来源、写入者和任务归属,再根据写入竞争与扫描开销决定是否迁移。