AI Agent记忆与缓存：让智能体不再重复从零开始摸数据

开发 AI Agent 时最常被低估的问题不是模型能力而是 Agent 的“失忆”。同一个 Agent同一类任务每次执行都像第一次进入系统一样从头扫描目录、探测接口、重新拉取数据、重新理解业务背景。这不是个别框架的问题而是大部分 Agent 默认按无状态请求设计的必然结果。“别再让 AI Agent 每次从零开始摸数据”这句话放到工程上其实就变成两件事给 Agent 加跨任务的持久化记忆给外部数据查询加可命中的缓存。解决了这两点Agent 才不会反复消耗 token、重复扫描同样的日志和接口。这篇文章会围绕一个最小可运行的实验项目讨论 Agent 记忆层和缓存层的实现思路。内容包括为什么 Agent 会无状态执行、记忆和缓存该如何分层、用什么技术选型、怎么用代码落地、怎么验证第二次启动不再从零开始以及实际项目中常见的排查和优化路径。适合正在做 AI Agent 开发、智能日志分析、AI Coding Agent 或想给现有 Agent 项目补上记忆能力的开发者阅读。1. 先看清楚 AI Agent 为什么会反复“摸数据”1.1 一个典型的“从零开始”场景假设你写了一个负责排查线上日志的 AI Agent。它接到的任务是分析今天订单服务报错的原因。Agent 会先调用 Elasticsearch 的 REST API 查询索引列表再尝试猜测日志索引名然后发起多条搜索请求把返回结果拼进上下文再让大模型总结。这个流程第一次跑完没有问题。问题是第二次、第三次、第 N 次。新任务进来时Agent 并不知道自己上次已经确认过日志索引叫order-service-2025.04.13也不知道错误码E1008对应的历史结论更不知道哪些查询已经跑过、结果是什么。它重新枚举索引、重新拉日志、重新做条件过滤甚至可能因为上下文窗口不够把上一次已经验证过的错误结论再次输出出来。这就是“每次从零开始摸数据”的典型表现。问题不在模型而在工程结构。Agent 的外围没有一个跨任务、跨会话保存中间结论和数据查询结果的能力。1.2 无状态设计是根因大多数 Agent 框架按“单次执行”来组织流程。一个请求进来Agent 读取 prompt 和工具列表进入循环观察 → 思考 → 行动 → 观察。这个循环内可以维护上下文但循环一旦结束状态就丢失了。从实现角度看Agent 本身是无状态的“决策器”它每次决策都依赖 prompt 里拼接的上下文。如果上下文里没有历史结论Agent 就只能重新通过工具调用去获取数据。这是“摸数据”的根本原因。从成本角度看无状态意味着每次执行都要重建上下文。同样的日志查询、同样的 API 响应、同样的中间推理会被重复计算。在真实业务里这会导致三个直接后果token 消耗成倍增长费用随任务量线性放大。任务执行时间变长因为多轮工具调用会反复等待外部接口响应。结论稳定性差同一个问题在不同时间跑可能因为上下文细节不同得到不一致的结果。所以要让 Agent 高效工作不能只指望把 prompt 写长必须把一部分外部数据和历史结论“沉淀”到 Agent 之外。1.3 记忆、缓存和技能是三件不同的事讨论 Agent 不重复摸数据时有三个概念经常被混在一起先拆开记忆跨任务保存的事实、结论和用户偏好。它回答“Agent 还记得什么”。缓存保存外部数据查询的结果避免重复调用同一个接口。它回答“同一份数据有没有必要再取一次”。技能Skills把固定操作固化成可复用的能力模块比如“解析 ES 查询结果”。它回答“Agent 会不会做这件事”。举一个具体例子。Agent 今天发现订单服务报错集中在E1008并得出“数据库连接池过小”的结论这是记忆。它把当时 ES 查询的原始结果存下来下次相同查询直接返回这是缓存。它已经把“查询 ES 并解析出错误码分布”封装成一个固定动作这是技能。三者的配合关系是技能负责执行缓存负责减少重复执行的开销记忆负责把执行后的结论保留下来。只做缓存不做记忆Agent 能快速拿到相同数据却不知道这些数据指向什么结论只做记忆不做缓存Agent 能复用结论但在需要回到原始数据时仍然要重新查询。所以工程上一般同时补上两层。2. 设计一套不重复摸数据的记忆与缓存架构2.1 分层设计短时上下文、长时记忆、工具缓存给 Agent 加记忆不是简单地把历史对话塞进数据库。更合理的做法是分层工作上下文代表当前任务进行中的临时状态比如已经读到的日志片段、正在生成的报告。它活跃时间短任务结束就可以丢弃。情景记忆代表“某一次任务发生过什么”比如“2025-04-13 排查订单服务报错时确认了索引名”。它按事件保存适合给后续任务做参考。语义记忆代表提炼后的稳定知识比如“E1008 错误在近一周的分布特征是夜间激增”。它从若干次任务中提炼不绑定具体某次执行。工具缓存代表外部接口和数据的查询结果比如 ES 的一次_search响应。它有明确的 key、TTL 和失效规则。对应到代码里工作上下文通常由 Agent 框架内部的上下文数组维护情景记忆和语义记忆落在数据库工具缓存落在 Redis、SQLite 或进程内缓存。这样设计的好处是不同生命周期的状态使用不同的存储策略不会被一个方案拖累。2.2 技术选型SQLite、Redis、向量库怎么选记忆存储和缓存存储的要求不同选型也不同。下表列出常见的组合存储类型解决什么问题适合场景典型代表SQLite持久化结构化记忆单机 Agent、学习项目、轻量服务SQLiteRedis工具结果缓存、高频读写高并发、需要 TTL 和淘汰策略Redis、KeyDB关系型数据库记忆审计、条件查询、权限控制生产环境、多用户 SaaSPostgreSQL向量数据库语义相似度检索长文本记忆、日志片段、非结构化知识pgvector、Milvus、Chroma、FAISS对象存储保存大体积原始数据日志原文、报表文件MinIO、云存储选型时先看数据量级和并发要求。个人项目用 SQLite 足够它能持久化记忆也支持结构化查询。生产环境建议把缓存交给 Redis把记忆放进 PostgreSQL再加上向量检索能力。不要一开始就上复杂中间件缓存命中率低的时候中间件本身也是维护负担。2.3 记忆生命周期要设计成什么样记忆不是写进去就完事。它有自己的生命周期至少包含五个阶段写入任务结束时从当前上下文中提炼值得记住的事实写入记忆库。检索新任务开始时根据任务描述召回相关记忆。更新当一条旧记忆与新结论冲突时进行合并或标记过期。衰减太久没有命中的记忆检索权重降低避免旧结论持续干扰新任务。清理删除确定过期的记忆控制存储体积。很多 Agent 记忆项目失败不是因为不会写代码而是没有定义生命周期。所有记忆都是新增从不更新和清理最后检索结果全被历史噪音淹没Agent 反而更不稳定。提示记忆质量比记忆数量重要。宁可从一次任务中提炼出 3 条高价值结论也不要塞入 50 条原始对话原文。3. 用最小可运行代码实现 Agent 持久化记忆3.1 项目结构与依赖这一节用一个 Python 示例说明实现思路。项目只依赖 Python 标准库和sqlite3不需要额外装第三方包便于快速跑通。agent-memory-demo/ ├── agent_memory.py # 记忆层核心实现 ├── tool_cache.py # 工具结果缓存 ├── demo_agent.py # 模拟 Agent 主流程 └── agent_memory.db # SQLite 数据库文件运行时自动生成示例中的记忆层采用非向量方案关键词匹配 最近访问计数。真实生产环境可以替换成向量检索但工程结构保持一致写入、检索、注入上下文三步。3.2 数据库表设计记忆表需要保存内容、来源、时间戳和访问次数。用 SQLite 的 DDL 表示CREATE TABLE IF NOT EXISTS memory ( id INTEGER PRIMARY KEY AUTOINCREMENT, namespace TEXT NOT NULL, content TEXT NOT NULL, metadata_json TEXT NOT NULL DEFAULT {}, source TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, last_access_at TEXT, access_count INTEGER NOT NULL DEFAULT 0 ); CREATE TABLE IF NOT EXISTS memory_tag ( memory_id INTEGER NOT NULL, tag TEXT NOT NULL, FOREIGN KEY(memory_id) REFERENCES memory(id) );namespace用于隔离记忆范围比如不同用户、不同项目、不同任务类型。metadata_json保存可扩展的结构化信息例如业务线、模型版本、任务编号。source记录记忆来源可能是日志文件、ES 查询或人工录入。这里要注意不要把原始对话直接整个塞进content。应先做摘要只保存可复用的结论。3.3 记忆写入、检索与注入的核心实现完整的记忆模块如下# agent_memory.py import json import re import sqlite3 from datetime import datetime, timezone class AgentMemory: def __init__(self, db_pathagent_memory.db): self.conn sqlite3.connect(db_path) self.conn.execute(PRAGMA journal_modeWAL;) self._init_tables() def _init_tables(self): self.conn.execute( CREATE TABLE IF NOT EXISTS memory ( id INTEGER PRIMARY KEY AUTOINCREMENT, namespace TEXT NOT NULL, content TEXT NOT NULL, metadata_json TEXT NOT NULL DEFAULT {}, source TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, last_access_at TEXT, access_count INTEGER NOT NULL DEFAULT 0 ) ) self.conn.execute( CREATE TABLE IF NOT EXISTS memory_tag ( memory_id INTEGER NOT NULL, tag TEXT NOT NULL, FOREIGN KEY(memory_id) REFERENCES memory(id) ) ) self.conn.commit() def add_memory( self, namespace, content, tagsNone, metadataNone, sourceNone, ): now datetime.now(timezone.utc).isoformat() cursor self.conn.execute( INSERT INTO memory (namespace, content, metadata_json, source, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?) , ( namespace, content, json.dumps(metadata or {}, ensure_asciiFalse), source, now, now, ), ) memory_id cursor.lastrowid for tag in tags or []: self.conn.execute( INSERT INTO memory_tag (memory_id, tag) VALUES (?, ?), (memory_id, tag), ) self.conn.commit() return memory_id def search(self, namespace, query, top_k5, min_score1.0): rows self.conn.execute( SELECT id, content, metadata_json, source, created_at, updated_at, access_count FROM memory WHERE namespace ? ORDER BY updated_at DESC LIMIT 200 , (namespace,), ).fetchall() tokens self._tokenize(query) scored [] for memory_id, content, metadata_json, source, created_at, updated_at, access_count in rows: haystack f{content} {source or }.lower() score 0.0 for token in tokens: if token in haystack: score 1.0 if score 0: continue # 最近被访问过的记忆有轻微加分模拟热度 score min(access_count, 10) * 0.1 if score min_score: scored.append( { id: memory_id, content: content, metadata: json.loads(metadata_json), source: source, created_at: created_at, updated_at: updated_at, access_count: access_count, score: round(score, 2), } ) scored.sort(keylambda x: x[score], reverseTrue) # 命中后更新访问统计 if scored: memory_ids [item[id] for item in scored[:top_k]] placeholders ,.join(? for _ in memory_ids) now datetime.now(timezone.utc).isoformat() self.conn.execute( f UPDATE memory SET access_count access_count 1, last_access_at ? WHERE id IN ({placeholders}) , [now, *memory_ids], ) self.conn.commit() return scored[:top_k] def build_context(self, namespace, query, top_k5, min_score1.0): memories self.search(namespace, query, top_ktop_k, min_scoremin_score) if not memories: return None blocks [] for item in memories: source item[source] or unknown blocks.append(f[memory:{item[id]}] from {source}: {item[content]}) return \n.join(blocks) def close(self): self.conn.close() def _tokenize(self, text): return set(re.findall(r[a-z0-9\u4e00-\u9fff], text.lower()))这段代码有三个值得留意的点namespace是第一个过滤条件先圈定记忆范围再做内容匹配。检索打分使用简单关键词命中计数教学上足够直观生产环境应替换为向量相似度或 BM25。每次检索成功后更新access_count和last_access_at让高频记忆获得少量加权这是低成本的时间热度模拟。3.4 在 Agent 循环中接入记忆层记忆层不是独立服务它要接入到 Agent 的执行循环里。接入点有两个任务开始时召回记忆注入 system prompt任务结束时提炼结论写入记忆库。# demo_agent.py import openai # 这里仅示意实际需要替换为可用的模型服务 from agent_memory import AgentMemory memory AgentMemory(agent_memory.db) NAMESPACE log-analyzer def run_analysis(task: str): # 任务开始前召回相关记忆 context memory.build_context(NAMESPACE, task, top_k3) if context: print( 命中历史记忆 ) print(context) else: print( 未命中记忆开始全量探索 ) # 将记忆拼进提示词 system_prompt 你是一个日志分析助手。下面是历史记忆中与当前任务相关的内容 if context: system_prompt \n context # 调用模型并执行工具这里省略实际请求代码 result f根据任务执行得出的结果{task} # 任务结束后沉淀值得记住的结论 memory.add_memory( namespaceNAMESPACE, content订单服务报错与 E1008 强相关原因集中在数据库连接池过小。, tags[order-service, E1008, db-pool], metadata{task: task, model: demo-model}, sourcees-search, ) return result if __name__ __main__: print(run_analysis(今天订单服务报错很多帮我分析原因)) print(---------- 第二次任务 ----------) print(run_analysis(订单服务又报 E1008 了为什么)) memory.close()注意上面的openai只是示意实际项目要换成自己可访问的模型 API 或本地推理服务。重点是记忆层接入的位置第一步先注入历史记忆最后一步再写入新结论。这样第二次执行相同任务时Agent 就不会再盲目枚举索引、反复查询了。提示记忆写入不要放在工具调用的每一步。建议只在任务级结束时提炼避免把中间过程当结论写入产生大量噪音。4. 为外部数据源加缓存避免同一批数据被重复查询4.1 缓存 key 的设计原则记忆解决“Agent 还记得结论”缓存解决“外部数据不要重复取”。缓存的核心在 key 设计。如果 key 太粗会串数据太细缓存命中率低等于没缓存。一个工具调用的 key 至少要包含下面几类信息任务或用户隔离字段防止不同业务的数据互相串。工具方法名与请求 URL。请求体的规范化内容例如 JSON 字段排序。查询语义版本当接口参数含义变化时能主动失效。常见反例是直接json.dumps(body)作为 key。当请求体字段顺序变化、空格变化或包含动态时间戳时key 会不稳定可能永远不命中或错误命中。4.2 带过期时间的工具结果缓存实现下面是一个最小缓存实现用来演示 key 生成、写入、命中和过期判断# tool_cache.py import hashlib import json import time class ToolCache: def __init__(self, default_ttl300): self.default_ttl default_ttl self._store {} def _make_key(self, namespace, method, url, bodyNone): raw json.dumps( { namespace: namespace, method: method, url: url, body: body or {}, }, sort_keysTrue, ensure_asciiFalse, ) return hashlib.sha256(raw.encode(utf-8)).hexdigest() def get(self, key): item self._store.get(key) if item is None: return None if item[expire_at] time.time(): del self._store[key] return None item[hit_count] 1 return item[value] def set(self, key, value, ttlNone): ttl ttl or self.default_ttl self._store[key] { value: value, expire_at: time.time() ttl, hit_count: 0, } def clear_expired(self): now time.time() expired [k for k, v in self._store.items() if v[expire_at] now] for key in expired: del self._store[key] return len(expired)使用方式很简单调用外部工具前先用 namespace、方法、URL、请求体生成 key查到就直接返回查不到才真正发起请求并把结果写回缓存。cache ToolCache(default_ttl60) def search_es(namespace, index, query): key cache._make_key(namespace, POST, f/{index}/_search, query) cached cache.get(key) if cached is not None: return cached # 这里替换为真正的 ES REST API 调用 result {hits: {total: 100}, errorCodes: {E1008: 80}} cache.set(key, result, ttl60) return result这个代码便于理解原理但生产环境建议直接用 Redis 的SET key value EX ttl利用它的持久化和多进程共享能力。进程内字典只适合单机演示。4.3 典型场景ES REST API 日志分析缓存结合本文开头的日志分析场景ES REST API 查询有很强的缓存价值。日志数据写入后基本不变短时间窗口内查同一段日志结果高度相似。常见优化方式如下缓存维度缓存 key 示例TTL 建议索引列表log_analyzerGET错误码分布log_analyzerPOST日志摘要log_analyzerPOST集群健康log_analyzerGET加了这一层缓存后即使 Agent 突然“忘记”历史结论重新探索时也能从缓存快速拿到原始数据而不是让 ES 重复承受相同的查询压力。记忆层负责减少分析时间缓存层负责减少数据接口的重复请求两者叠加效果才是完整的。5. 如何验证 Agent 第二次启动不再从零开始5.1 构造可复现的验证场景跑通代码后要用一个可观察的实验来验证效果。建议先清空数据库再执行以下操作运行demo_agent.py第一次执行任务确认输出中显示“未命中记忆开始全量探索”。观察数据库memory表确认写入了一条摘要记忆。再次运行任务确认输出中显示“命中历史记忆”并打印出记忆内容。检查 Agent 的输出确认它没有重新枚举索引或重复查询相同接口。在缓存场景中连续两次调用search_es第二次调用不应真正触发 ES 请求而是返回缓存值。5.2 预期日志与验证命令第二次运行成功后预期输出类似 命中历史记忆 [memory:1] from es-search: 订单服务报错与 E1008 强相关原因集中在数据库连接池过小。可以用一个命令确认记忆写入情况sqlite3 agent_memory.db select id, namespace, content, source, access_count from memory;预期结果1|log-analyzer|订单服务报错与 E1008 强相关原因集中在数据库连接池过小。|es-search|1如果添加了工具缓存可以在缓存命中分支里打一行日志cache hit。命中率是判断缓存是否有效的直接指标。5.3 学习环境与生产环境的验证差异学习环境只需要验证“第二次能命中”。生产环境还需要验证更多方面多用户隔离A 用户的记忆不能出现在 B 用户的任务里。记忆时效过期的结论不应长期生效需要有时间衰减和人工确认机制。缓存一致性外部数据更新后缓存能在预期时间内失效。故障恢复缓存节点或数据库重启后Agent 能继续工作而不是因为检索失败直接崩溃。学习环境用本地 SQLite 和进程内字典没问题但生产环境的记忆服务和缓存服务最好是独立组件让 Agent 无状态状态全部外置。这样才能水平扩容多个 Agent 实例共享同一份记忆。提示验证记忆是否生效不要只看是否打印“命中记忆”。要看最终回答是否因为记忆而变得更准确、更少调用工具。真正有效的指标是工具调用次数下降、任务耗时下降、结论一致性提升。6. 常见问题排查为什么记忆没生效6.1 现象第二次启动仍然说“我不知道”可能原因有很多按顺序排查确认namespace是否一致。写入用log-analyzer检索却传log_analysis永远查不到。确认写入是否真的提交。SQLite 中事务没有commit进程退出后数据丢失。确认检索阈值是否过高。min_score3.0但每个 token 命中只加 1 分结果被过滤。确认任务描述和记忆内容描述差异过大。关键词检索对同义词不敏感比如记的是“数据库连接池”任务里写的是“DB 连接耗尽”。解决办法是先写一条调试接口直接打印检索命中的记忆和分数。如果命中了但 Agent 不用是 prompt 注入问题如果根本没命中是写入、namespace、阈值或分词问题。6.2 检索到旧记忆或张冠李戴现象是命中了好几条无关记忆导致 Agent 被错误信息带偏。常见原因是检索打分没有区分 namespace或者没有时间衰减。建议补齐三个机制检索前强制按 namespace 过滤。给记忆加入updated_at衰减因子越久远的记忆权重越低。对冲突结论标记superseded_by新结论出现时旧结论降权而不是简单叠加。向量检索方案还要注意 embedding 模型一致性。切换模型后旧向量和新向量可能在不同语义空间必须重建索引否则相似度分数完全没有可比性。6.3 缓存命中了旧数据缓存 TTL 设置太长、接口返回结构变化但 key 没变、业务上数据被更新但缓存未主动失效都会导致旧数据被反复使用。检查步骤确认缓存的 key 是否包含接口版本和语义版本。确认 TTL 是否匹配数据变化频率。确认修改类操作是否主动删除对应缓存键。确认时间是否来自同一时钟源避免服务器时间偏移导致过期失效。# 主动失效示例 cache.delete_by_prefix(log_analyzer|POST|/order-service-*/_search)如果缓存的数据用于生产报表建议额外写入数据时间戳在命中缓存时校验数据新鲜度。6.4 并发写入与脏数据SQLite 的多进程并发写入容易出现database is locked。原因是多个 Agent 实例同时写入同一数据库。解决方向打开 WAL 模式读写并发能力会好很多。把写入操作收敛到单线程任务队列或队列服务。生产环境换 PostgreSQL配合连接池和唯一约束避免重复记忆。self.conn.execute(PRAGMA busy_timeout 5000;) self.conn.execute(PRAGMA journal_mode WAL;)6.5 排查顺序速查表问题现象优先检查看什么处理建议记忆未命中写入是否成功memory 表数据确认 commit、namespace检索结果乱打分和过滤条件命中的内容和分数加时间衰减、降阈值缓存未命中key 是否稳定key 生成参数去掉动态字段缓存命中旧数据TTL 和失效逻辑数据更新时间改短 TTL、主动失效数据库锁并发写入错误日志WAL、队列、换数据库7. 落地建议与扩展方向7.1 Agent 记忆层落地检查清单给现有 Agent 补记忆层之前先按清单过一遍[ ] 是否定义 namespace 隔离机制区分用户、项目和任务类型。[ ] 是否对记忆内容做摘要和结构化而不是直接存原始对话。[ ] 是否有记忆检索阈值和反馈机制记录命中后是否有用。[ ] 是否有更新时间衰减和冲突合并规则。[ ] 是否有工具缓存层缓存 key 是否包含隔离字段和语义版本。[ ] 缓存 TTL 是否合理是否提供主动失效入口。[ ] 是否记录命中率、写入量、检索耗时等指标。[ ] 是否过滤敏感信息日志原文不能直接入记忆库。[ ] 数据库是否有备份、恢复和回滚方案。这套清单同样适用于代码审查。评判一个 Agent 项目是否工程化不看 prompt 写得多好而看状态管理是否清晰、外部数据是否重复获取、记忆是否可审计。7.2 生产环境还需要额外补什么学习项目用 SQLite 没问题生产环境要额外考虑几个工程点记忆服务独立部署把记忆读写封装成 REST 或 gRPC 服务Agent 实例无状态便于扩容。向量检索持久化使用 pgvector 或 Milvus 一类的组件把语义检索能力和已有的业务表打通。缓存用 Redis设置 TTL、淘汰策略和命名空间前缀避免单机内存缓存导致多实例命中率低。权限和审计不同角色能看到的记忆范围要受控记忆写入要记录来源和操作者。可观测性埋点记录记忆命中率、工具调用次数、缓存命中率、单任务 token 消耗用数据验证优化效果。7.3 与 Agent Skills、MCP 配合后的架构走向记忆层解决“从零开始”还不够。在更完整的 Agent 架构里还需要与技能和工具协议配合。Skills 把固定分析动作固化成可复用模块比如“ES 错误码分布查询”就是一个 skill多个 skill 按任务调度。内存记忆、缓存、技能注册表构成 Agent 的“工作台”状态。在新一代 Agent 开发中记忆通常被抽象成服务组件通过协议暴露给多个 Agent而不是每个 Agent 内部维护一份。对 Java 开发的同学对应的思路相似可以用 Spring AI 之类的框架管理 Agent 上下文用 Redis 处理缓存用 PostgreSQL 持久化记忆。对前端同学重点是让浏览器端的 Agent 把用户偏好和历史操作保存在 IndexedDB 或本地向量索引里下次打开页面时先恢复上下文。无论什么技术栈核心判断不变Agent 的长期价值不取决于它的第一步探索能力而取决于它能不能把以前摸过的数据、踩过的坑、验证过的结论真正留下来并在下一次任务开始时直接拿出来用。现在就可以从一个小项目开始给 Agent 加一张记忆表、加一层缓存验证一下第二次调用是否真的更快、更稳、更省钱。