LLM Agent记忆系统实战：从遗忘到精准召回

1. 从“hindsight”说起：为什么我们需要给Agent装上记忆

“hindsight”这个词本身很有意思，字面意思是“事后的洞察力”，也就是我们常说的“后见之明”。放在LLM Agent的语境里，它指向一个非常具体且棘手的问题：Agent如何记住过去发生过的事情，并在后续决策中真正用上这些经验。

我接触过不少做Agent项目的团队，大家一开始都兴致勃勃地搭ReAct循环、接工具调用，跑几个demo觉得效果惊艳。但一旦进入真实业务场景，问题就暴露了——Agent像个失忆症患者，每次对话都从零开始，用户上周提过的偏好、上个月处理过的类似工单、昨天刚纠正过的错误，它统统不记得。你可能会说，不是有上下文窗口吗？但上下文窗口再大也有上限，而且把几十轮历史全塞进去，token成本飙升不说，模型注意力还会被稀释，关键信息反而被淹没。

这就是“hindsight”要解决的核心矛盾：Agent需要一种机制，把过去的交互经验沉淀下来，在需要的时候精准召回，而不是每次都把全部历史重新读一遍。它本质上是一个记忆管理系统，涉及存储结构设计、检索策略、遗忘机制、与LLM推理循环的集成等多个层面。

适合谁来参考这篇内容？如果你正在做LLM Agent开发，或者对Agent Memory这个方向感兴趣，又或者你只是好奇“为什么我的Agent总是记不住事”，那接下来的内容应该能给你一些可直接落地的思路。我会从整体设计讲到具体实现，包括存储选型、检索算法、与MCP协议的配合、Docker环境搭建，以及我在实际项目中踩过的坑。

2. Agent Memory的整体设计与核心思路拆解

2.1 为什么传统上下文管理不够用

先说说大家最常用的做法：把对话历史拼成一个长字符串，每次请求都带上。这个方案在早期确实能用，但它有三个致命问题。

第一是成本问题。假设每轮对话平均500 token，50轮就是25000 token。按GPT-4级别的价格算，每次请求光历史部分就要花掉不少钱。如果Agent每天处理上千次请求，这个成本会迅速失控。

第二是注意力稀释。Transformer的注意力机制虽然理论上能处理长序列，但实际表现是“中间遗忘”——序列中间部分的信息容易被忽略。你把100轮历史塞进去，模型真正能有效利用的可能只有开头和结尾的几轮。

第三是信息冗余与冲突。用户可能在不同时间说了矛盾的话，比如“我喜欢简洁的回答”和“能不能详细解释一下”。如果没有一个机制去管理这些信息的优先级和时效性，模型就会困惑。

所以我们需要一个独立的记忆层，它负责选择性地存储、组织、检索和遗忘信息，让LLM每次只看到最相关的那部分。

2.2 记忆的分层结构：Working Memory与Long-term Memory

参考认知科学的模型，我把Agent记忆分成两层：

Working Memory（工作记忆）：当前会话的短期上下文，容量有限，通常就是最近N轮对话或当前任务相关的信息。它的特点是访问速度快、生命周期短、随会话结束而清空或归档。

Long-term Memory（长期记忆）：跨会话持久化的知识，包括用户偏好、历史事件、学到的经验等。它的特点是容量大、生命周期长、需要检索机制来访问。

这两层之间的交互是关键。Working Memory中的信息在会话结束时，经过筛选和压缩后写入Long-term Memory；Long-term Memory中的相关内容在需要时被召回，注入到Working Memory中供LLM使用。

2.3 记忆的三种类型：Key-Query-Value视角

热词里提到一个很有意思的说法：“LLM的token三个点：key我是谁、query我在找什么、value我能提供什么”。这其实借用了注意力机制的隐喻，用来理解记忆检索非常贴切。

• Key（我是谁）：这条记忆是关于什么的？比如“用户偏好”、“项目配置”、“错误解决方案”。它是记忆的标签和索引。
• Query（我在找什么）：当前情境需要什么信息？比如用户问“上次那个bug怎么修的”，Query就是“bug修复方案”。
• Value（我能提供什么）：记忆的实际内容。比如“bug是因为Docker网络配置冲突，解决方案是自定义bridge网络”。

一个好的记忆系统，就是让Key和Query的匹配尽可能精准，从而召回最相关的Value。

2.4 为什么选择向量数据库+结构化存储的混合方案

纯向量检索的问题是它只能做语义相似度匹配，无法处理精确条件过滤。比如你想找“上周关于Docker的所有记忆”，向量检索可能返回一堆语义相关但时间不对的内容。

纯结构化存储（比如SQL）的问题是它无法处理模糊语义查询。用户说“我上次说的那个部署问题”，你没法用SQL精确匹配。

所以我的方案是混合存储：用向量数据库做语义检索，用关系型数据库或文档数据库做结构化过滤，两者结合。具体来说，每条记忆同时存储向量嵌入和结构化元数据（时间戳、类型、标签、重要性评分等），检索时先做结构化过滤缩小范围，再做向量相似度排序。

3. 核心细节解析与实操要点

3.1 记忆的写入策略：什么时候该记，什么时候不该记

不是所有对话都值得写入长期记忆。如果每句话都存，记忆库会迅速膨胀，检索质量也会下降。我的策略是基于重要性评分的选择性写入。

重要性评分可以从几个维度计算：

• 显式标记：用户说“记住这个”、“以后都这样”时，直接高分。
• 信息密度：包含具体事实、偏好、决策的内容比寒暄高分。
• 重复频率：同一信息多次出现，说明它重要。
• 任务相关性：与当前Agent核心任务相关的信息优先。

我通常设一个阈值，比如0.6分以上才写入长期记忆。低于阈值的只在Working Memory中保留，会话结束就丢弃。

注意：写入时一定要做去重和冲突检测。如果新记忆和已有记忆矛盾，要么更新旧记忆，要么标记冲突让Agent在检索时知道存在不同版本。

3.2 记忆的检索策略：多路召回+重排序

检索是记忆系统最核心的环节。我的做法是多路召回：

1. 向量语义检索：用embedding模型把Query和所有记忆的Key做相似度计算，取Top-K。
2. 关键词检索：用BM25或类似算法做精确词匹配，补充向量检索可能漏掉的精确匹配。
3. 时间衰减加权：越近的记忆权重越高，但也不能完全忽略旧记忆。我通常用指数衰减，半衰期设7天左右。
4. 重要性加权：写入时的重要性评分在检索时也参与排序。

最后用一个重排序模型（可以是cross-encoder，也可以简单的加权求和）把多路结果融合，取最终Top-N注入到LLM上下文。

3.3 记忆的遗忘机制：不是所有记忆都值得永远保留

遗忘不是bug，是feature。没有遗忘机制，记忆库会变成垃圾场。我的遗忘策略分三种：

• 时间衰减：超过一定时间未被访问的记忆，重要性评分逐渐降低，低于阈值后归档或删除。
• 容量限制：每个用户或每个Agent的记忆总量设上限，超出时淘汰最低分的。
• 主动遗忘：用户明确要求删除某条记忆时，立即执行。

实操心得：我建议保留一个“归档”层，而不是直接删除。归档的记忆不参与常规检索，但在特定情况下（比如用户问“很久以前那个事”）可以被召回。这样既控制了活跃记忆的规模，又不会真正丢失信息。

3.4 与MCP协议的集成：让记忆成为可插拔的能力

MCP（Model Context Protocol）是当前Agent工具集成的主流协议。把记忆系统做成一个MCP Server，好处是解耦——Agent不需要知道记忆的具体实现，只需要通过标准协议调用即可。

具体来说，我会暴露几个MCP工具：

• memory_write：写入一条记忆，参数包括内容、类型、标签、重要性。
• memory_search：检索记忆，参数包括Query、时间范围、类型过滤、返回数量。
• memory_forget：删除或归档记忆。
• memory_summarize：对一段对话做摘要后写入。

这样任何支持MCP的Agent框架都能直接接入，不需要改代码。

3.5 Docker环境下的部署要点

用Docker部署记忆系统有几个关键点：

网络配置：如果向量数据库和Agent在不同容器，要确保它们在同一个Docker网络中。我习惯创建一个自定义bridge网络，而不是用默认的。

数据持久化：向量数据库和关系型数据库的数据目录必须挂载到宿主机，否则容器重启数据就没了。

资源限制：向量检索是内存密集型操作，给容器设合理的内存上限，避免OOM。

健康检查：配置healthcheck，确保依赖服务（数据库、embedding服务）就绪后再启动记忆服务。

4. 实操过程与核心环节实现

4.1 环境准备：Docker与依赖服务

先确保Docker环境正常。Windows用户如果遇到“Virtualization support not detected”错误，需要在BIOS里开启虚拟化支持，然后在Windows功能里启用WSL2或Hyper-V。

# 检查Docker是否正常运行 docker --version docker compose version # 创建自定义网络 docker network create agent-memory-net

接下来用Docker Compose编排依赖服务。我选Qdrant做向量数据库（轻量、性能好、API简洁），PostgreSQL做结构化存储，Redis做Working Memory的缓存。

version: '3.8' services: qdrant: image: qdrant/qdrant:latest ports: - "6333:6333" volumes: - ./data/qdrant:/qdrant/storage networks: - agent-memory-net deploy: resources: limits: memory: 2G postgres: image: postgres:16 environment: POSTGRES_DB: agent_memory POSTGRES_USER: memory_user POSTGRES_PASSWORD: memory_pass ports: - "5432:5432" volumes: - ./data/postgres:/var/lib/postgresql/data networks: - agent-memory-net redis: image: redis:7-alpine ports: - "6379:6379" volumes: - ./data/redis:/data networks: - agent-memory-net networks: agent-memory-net: external: true

启动后验证各服务：

docker compose up -d docker compose ps curl http://localhost:6333/healthz # Qdrant健康检查

4.2 记忆数据模型设计

在PostgreSQL中建表存储记忆的元数据：

CREATE TABLE memories ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), agent_id VARCHAR(64) NOT NULL, user_id VARCHAR(64), content TEXT NOT NULL, memory_type VARCHAR(32) NOT NULL, -- preference, fact, event, skill tags TEXT[], importance FLOAT DEFAULT 0.5, access_count INT DEFAULT 0, created_at TIMESTAMP DEFAULT NOW(), last_accessed_at TIMESTAMP DEFAULT NOW(), archived BOOLEAN DEFAULT FALSE ); CREATE INDEX idx_memories_agent_user ON memories(agent_id, user_id); CREATE INDEX idx_memories_type ON memories(memory_type); CREATE INDEX idx_memories_importance ON memories(importance DESC);

Qdrant中存储向量：

from qdrant_client import QdrantClient from qdrant_client.models import Distance, VectorParams, PointStruct client = QdrantClient(host="localhost", port=6333) client.create_collection( collection_name="agent_memories", vectors_config=VectorParams(size=1536, distance=Distance.COSINE) )

4.3 记忆写入的完整流程

写入一条记忆的步骤：

1. 内容预处理：清洗文本，提取关键信息。
2. 重要性评分：用规则+小模型打分。
3. 生成嵌入：调用embedding模型（如text-embedding-3-small）生成向量。
4. 写入PostgreSQL：存储元数据和原始内容。
5. 写入Qdrant：存储向量和ID映射。
6. 更新Redis：如果是当前会话的Working Memory，同步更新缓存。

import uuid from datetime import datetime def write_memory(agent_id, user_id, content, memory_type, tags, importance): memory_id = str(uuid.uuid4()) # 1. 写入PostgreSQL cursor.execute(""" INSERT INTO memories (id, agent_id, user_id, content, memory_type, tags, importance) VALUES (%s, %s, %s, %s, %s, %s, %s) """, (memory_id, agent_id, user_id, content, memory_type, tags, importance)) # 2. 生成嵌入并写入Qdrant embedding = get_embedding(content) client.upsert( collection_name="agent_memories", points=[PointStruct( id=memory_id, vector=embedding, payload={ "agent_id": agent_id, "user_id": user_id, "memory_type": memory_type, "importance": importance, "created_at": datetime.now().isoformat() } )] ) # 3. 更新Redis Working Memory redis_client.lpush(f"wm:{agent_id}:{user_id}", memory_id) redis_client.ltrim(f"wm:{agent_id}:{user_id}", 0, 49) # 保留最近50条 return memory_id

4.4 记忆检索的完整流程

检索时，我采用多路召回+融合排序：

def search_memories(agent_id, user_id, query, top_k=5, time_range=None, memory_type=None): # 1. 向量检索 query_embedding = get_embedding(query) vector_results = client.search( collection_name="agent_memories", query_vector=query_embedding, query_filter={ "must": [ {"key": "agent_id", "match": {"value": agent_id}}, {"key": "user_id", "match": {"value": user_id}} ] }, limit=top_k * 3 # 多召回一些，后续重排序 ) # 2. 结构化过滤（时间范围、类型） memory_ids = [r.id for r in vector_results] sql = "SELECT * FROM memories WHERE id = ANY(%s) AND archived = FALSE" params = [memory_ids] if time_range: sql += " AND created_at >= %s" params.append(time_range) if memory_type: sql += " AND memory_type = %s" params.append(memory_type) cursor.execute(sql, params) candidates = cursor.fetchall() # 3. 重排序：向量相似度 * 时间衰减 * 重要性 now = datetime.now() scored = [] for mem in candidates: vector_score = next(r.score for r in vector_results if r.id == mem['id']) days_old = (now - mem['created_at']).days time_decay = 0.5 ** (days_old / 7) # 半衰期7天 final_score = vector_score * 0.6 + time_decay * 0.2 + mem['importance'] * 0.2 scored.append((final_score, mem)) scored.sort(key=lambda x: x[0], reverse=True) return [mem for _, mem in scored[:top_k]]

4.5 与LLM推理循环的集成

在Agent的推理循环中，每次调用LLM之前，先用当前Query检索相关记忆，把结果注入到system prompt或context中：

def build_context(agent_id, user_id, current_query): # 检索长期记忆 long_term = search_memories(agent_id, user_id, current_query, top_k=5) # 获取Working Memory wm_ids = redis_client.lrange(f"wm:{agent_id}:{user_id}", 0, 9) working_memories = fetch_memories_by_ids(wm_ids) # 构建上下文 context_parts = [] if long_term: context_parts.append("## 相关历史记忆\n" + "\n".join( f"- [{m['memory_type']}] {m['content']}" for m in long_term )) if working_memories: context_parts.append("## 当前会话上下文\n" + "\n".join( f"- {m['content']}" for m in working_memories )) return "\n\n".join(context_parts)

4.6 MCP Server的实现

把记忆系统封装成MCP Server，暴露标准工具接口：

from mcp.server import Server, NotificationOptions from mcp.server.models import InitializationOptions import mcp.server.stdio server = Server("agent-memory") @server.list_tools() async def handle_list_tools(): return [ Tool(name="memory_write", description="写入一条记忆", inputSchema={...}), Tool(name="memory_search", description="检索记忆", inputSchema={...}), Tool(name="memory_forget", description="遗忘记忆", inputSchema={...}), ] @server.call_tool() async def handle_call_tool(name, arguments): if name == "memory_write": memory_id = write_memory(**arguments) return [TextContent(type="text", text=f"已写入记忆 {memory_id}")] elif name == "memory_search": results = search_memories(**arguments) return [TextContent(type="text", text=format_results(results))] # ...

启动MCP Server：

python -m mcp_server --port 8080

在Agent配置中注册这个MCP Server，Agent就能通过标准协议调用记忆能力了。

5. 常见问题与排查技巧实录

5.1 记忆检索不准确怎么办

这是最常见的问题。排查思路：

先看嵌入模型。不同embedding模型对中文、专业术语的表现差异很大。我试过用通用模型处理医疗领域术语，效果很差，换成领域微调过的模型后明显改善。

再看分块策略。如果一条记忆太长，嵌入会丢失细节。我通常把长记忆拆成200-500字的块，每块单独嵌入，检索时返回块再拼接。

最后看重排序权重。向量相似度、时间衰减、重要性的权重需要根据场景调。客服场景时间权重高，知识库场景重要性权重高。

5.2 Docker网络不通的排查

容器间网络不通是高频问题。我的排查顺序：

1. docker network inspect agent-memory-net确认容器都在同一网络。
2. 在容器内ping另一个容器名，确认DNS解析正常。
3. 检查防火墙规则，确保端口没有被宿主机防火墙拦截。
4. 如果用的是Docker Desktop，确认WSL2网络配置正常。

实操心得：我习惯在compose文件里给每个服务加container_name，这样容器间可以直接用名字通信，不用记IP。

5.3 记忆库膨胀太快怎么控制

如果发现记忆数量增长过快，检查几个点：

• 重要性阈值是不是设太低了？提高到0.7试试。
• 是不是把每轮对话都写入了？应该只写筛选后的。
• 有没有做去重？相似度超过0.95的记忆应该合并。
• 归档策略有没有生效？定期跑归档任务。

5.4 与Agent框架集成的兼容性问题

不同Agent框架对MCP的支持程度不同。如果遇到集成问题：

• 确认框架版本支持MCP协议。
• 检查MCP Server的传输方式（stdio还是SSE）是否匹配。
• 看日志确认工具调用是否成功，参数格式是否正确。

5.5 性能优化：检索延迟太高怎么办

向量检索延迟主要来自嵌入计算和向量搜索。优化手段：

• 嵌入计算用缓存，相同Query不重复计算。
• 向量索引用HNSW，比暴力搜索快几个数量级。
• 结构化过滤先执行，缩小向量搜索范围。
• 批量检索时用异步并行。

6. 记忆系统的扩展方向与个人经验

6.1 从被动记忆到主动记忆

目前的记忆系统是被动的——用户问什么，检索什么。下一步可以做主动记忆：Agent在空闲时主动整理记忆，发现关联，生成更高层的洞察。比如把多条关于“Docker问题”的记忆归纳成一条“Docker常见问题及解决方案”。

6.2 记忆的图结构

当前是扁平的记忆列表，未来可以引入图结构，把记忆之间的关联显式建模。比如“用户偏好”和“项目配置”之间可能有关联，图结构能让检索沿着关联路径扩散，召回更全面。

6.3 多Agent共享记忆

多个Agent协作时，共享记忆层可以让它们互相学习。一个Agent踩过的坑，另一个Agent可以直接避开。这需要解决权限控制和冲突合并的问题。

6.4 我在实际项目中的几点体会

第一，不要追求一步到位。我一开始想设计一个完美的记忆系统，结果拖了很久没上线。后来先用最简单的向量检索跑起来，再逐步加时间衰减、重要性评分、归档机制，迭代速度快很多。

第二，监控比功能更重要。记忆系统的效果很难直观判断，必须加监控：检索命中率、用户反馈、记忆增长率、检索延迟。没有数据就没法优化。

第三，遗忘和记忆同样重要。我早期版本没有遗忘机制，跑了三个月后记忆库有几十万条，检索质量断崖式下降。加上归档和淘汰后，效果明显回升。

第四，测试要覆盖边界情况。比如空记忆库、超长记忆、矛盾记忆、并发写入。这些在demo阶段不会遇到，但生产环境一定会碰到。

最后分享一个小技巧：在记忆的元数据里加一个source字段，记录这条记忆来自哪次对话、哪个任务。排查问题时可以追溯，也方便做数据清理。这个字段我一开始没加，后来补上时发现很多历史记忆已经没法追溯来源了，只能全部归档重来。