1. 从“hindsight”说起为什么记忆是 Agent 落地的最后一公里“hindsight”这个词本身很有意思字面意思是“事后的洞察力”也就是我们常说的“后见之明”。把这个词放到 LLM Agent 的语境里它指向的东西非常明确Agent 在完成任务之后能不能把这次经历沉淀下来下次遇到类似场景时直接调用而不是每次都从零开始推理。这就是记忆memory问题也是当前 agent 开发里最容易被低估、但实际落地时最卡脖子的一环。我接触过不少 agent 项目从简单的工具调用到复杂的多步规划大家一开始都把精力放在 prompt 工程、工具编排、MCP 协议对接上觉得只要模型够强、工具够全agent 就能跑起来。但真正上线跑一段时间就会发现同一个用户问第二遍同样的问题agent 还是像第一次见面一样完全不知道之前发生过什么。这种“失忆”状态在 demo 阶段不明显一旦进入真实业务场景用户体验直接崩掉。hindsight 要解决的就是这个问题。它不是简单的对话历史拼接而是一套结构化的、可检索的、带时间衰减的长期记忆机制。你可以把它理解成给 agent 装了一个“经验库”每次任务执行完agent 会把关键信息写进去下次遇到新任务先从经验库里捞相关的记忆再结合当前上下文做决策。这套机制的核心价值在于降低重复推理成本、提升响应一致性、让 agent 具备跨会话的连续性。适合谁来参考如果你正在做 agent 开发尤其是涉及多轮对话、任务型助手、个人知识管理这类场景hindsight 的思路可以直接借鉴。如果你只是用现成的 LLM 框架做简单问答那可能暂时用不上但了解这套机制对你理解 agent 的演进方向有帮助。下面我会从设计思路、核心细节、实操落地、问题排查几个层面把这套东西拆开讲清楚。2. 记忆系统的整体设计与选型逻辑2.1 为什么不用简单的向量数据库堆砌很多人一提到 agent 记忆第一反应就是“上个向量数据库把历史对话 embed 一下存进去需要的时候相似度检索”。这个方案能跑但实际用下来问题很多。我最早也是这么干的用 Chroma 或者 Milvus 存对话片段检索 top-k 塞回 prompt。跑了一段时间发现几个致命问题第一检索出来的记忆是碎片化的。向量检索按相似度返回可能返回五条不相关的片段每条都只有半句话拼在一起语义不完整。第二没有时间维度。三个月前的一条记忆和昨天的记忆在向量空间里可能距离差不多但实际权重完全不一样。第三无法处理矛盾信息。用户上周说喜欢咖啡这周说改喝茶了两条记忆都在库里检索时可能同时返回agent 就懵了。hindsight 的设计思路不是推翻向量检索而是在它上面加了两层结构化抽取层和时间衰减层。结构化抽取负责把原始对话变成带实体、带关系、带事件类型的记忆单元时间衰减负责给每条记忆算一个动态权重越久远的记忆权重越低但如果是高频引用的记忆权重衰减会变慢。这样检索的时候不是单纯看语义相似度而是相似度乘以时间权重再排序。2.2 记忆分层工作记忆、短期记忆、长期记忆我把 hindsight 的记忆体系分成三层这个分层参考了认知科学里的人类记忆模型但在工程上做了简化层级存储介质生命周期容量典型用途工作记忆内存/上下文窗口单次会话受 token 限制当前对话的即时上下文短期记忆Redis/内存数据库数小时到数天较大最近几次会话的关键信息长期记忆向量库关系库永久很大用户偏好、历史事件、知识沉淀工作记忆就是当前 prompt 里的内容这个不用多说。短期记忆我一般用 Redis 存key 是用户 ID 加会话 IDvalue 是结构化的事件列表设置 TTL。长期记忆才是 hindsight 的核心它又分成两个部分向量索引用于语义检索关系图谱用于实体关联查询。比如用户问“我上次说的那个项目进展怎么样了”向量检索能找到相关对话片段关系图谱能顺着“项目”这个实体找到关联的任务、时间、参与人。2.3 写入策略什么时候该记什么时候不该记这是实操中最容易踩坑的地方。如果每轮对话都写记忆库会迅速膨胀检索质量下降成本也扛不住。我的经验是按事件触发写入而不是按轮次。具体来说以下几种情况才触发写入用户明确表达了偏好、事实、决策“我决定用 PostgreSQL 而不是 MySQL”任务执行完成产生了可复用的结果“这次部署的配置是 XXX”用户纠正了 agent 的错误“不对应该是 XXX”出现了新的实体或关系“我们团队新来了一个同事叫张三”反过来闲聊、确认性回复“好的”“嗯嗯”、中间推理过程这些都不写。写入的时候也不是原样存而是先让 LLM 做一次抽取输出结构化的 JSON包含event_type、entities、summary、timestamp、confidence这几个字段。confidence 很重要用户随口一说的事情置信度低明确确认的事情置信度高检索时权重不一样。3. 核心细节解析从抽取到检索的完整链路3.1 记忆抽取的 prompt 设计要点抽取这一步直接决定记忆质量。我试过很多版 prompt最后稳定下来的结构是这样的先给模型一个明确的角色定义“你是一个记忆抽取器”然后给几个 few-shot 示例每个示例包含原始对话和对应的 JSON 输出。关键点在于约束输出格式必须用 JSON schema 校验不合法就重试。抽取的字段我一般保留这几个{ event_type: preference | fact | decision | entity | correction, summary: 一句话概括不超过50字, entities: [实体1, 实体2], relations: [{from: 实体1, to: 实体2, type: 关系类型}], confidence: 0.0-1.0, source_text: 原始对话片段 }这里有个细节summary一定要让模型用自己的话概括不要直接截取原文。因为原文可能包含口语、指代、省略直接存进去检索时匹配效果差。概括之后语义更集中embedding 质量更高。另外entities要做归一化比如“PostgreSQL”和“postgres”要统一成同一个实体 ID不然关系图谱会散掉。3.2 时间衰减函数的参数计算时间衰减是 hindsight 区别于普通向量记忆的关键。我用的是指数衰减加引用加权的组合weight base_confidence * exp(-lambda * days_since_creation) * (1 alpha * reference_count)其中lambda是衰减系数days_since_creation是记忆创建至今的天数reference_count是这条记忆被检索命中的次数alpha是引用加权系数。参数怎么定我实测下来lambda取 0.01 到 0.03 比较合适。取 0.01 的话100 天后权重降到约 0.37取 0.03 的话100 天后降到约 0.05。具体取值看业务场景如果是个人助手记忆需要长期保留lambda取小一点如果是客服场景用户偏好变化快lambda取大一点。alpha我一般设 0.1也就是说一条记忆每被引用一次权重增加 10%。这个机制让高频使用的记忆衰减更慢相当于“用进废退”。但要注意reference_count不能无限增长我设了一个上限 50超过之后不再增加避免某条记忆权重过大压制其他记忆。3.3 检索时的混合排序策略检索不是单纯按向量相似度排而是多路召回再融合。我的做法是向量召回用 query 的 embedding 在长期记忆库里检索 top-20实体召回从 query 里抽取实体在关系图谱里找关联记忆 top-10时间召回最近 7 天的短期记忆全部纳入候选三路召回合并去重后用上面的 weight 公式算最终分数再乘以语义相似度排序取 top-5 塞回 prompt。这里有个坑语义相似度的量纲和 weight 不一样直接相乘会导致某一项主导。我的处理是先把相似度归一化到 0-1weight 也归一化到 0-1然后加权求和相似度权重 0.6时间权重 0.4。这个比例可以根据场景调任务型 agent 更看重语义个人助手更看重时间。3.4 记忆冲突的消解机制用户改主意了怎么办比如之前说“我住在北京”后来说“我搬到上海了”。两条记忆都在库里检索时可能同时返回。我的消解策略是同实体同关系类型的新记忆覆盖旧记忆。具体实现是在关系图谱里(用户, 居住地, 北京)这条边被标记为失效新边(用户, 居住地, 上海)生效。向量库里旧记忆不删除但检索时如果发现有关联的新记忆旧记忆的权重会被压制到 0.1 以下。这个机制需要实体和关系类型定义得足够细。如果关系类型太粗比如只有一个“相关信息”那就没法判断哪条新哪条旧。所以前期花时间设计 ontology 是值得的至少要把人物、地点、时间、偏好、决策这几大类分清楚。4. 实操落地从零搭一套 hindsight 记忆系统4.1 技术栈选型与依赖安装我用的技术栈比较轻量适合快速验证LLM任意支持 function calling 的模型我用的是本地部署的 7B 模型加一个 API 模型做兜底向量库Qdrant单机版够用Docker 一键起关系库Neo4j Community或者用 SQLite 加邻接表也能凑合缓存Redis存短期记忆和检索缓存编排Python 加 FastAPI不用太重型的框架安装依赖pip install qdrant-client neo4j redis openai tiktoken fastapi uvicornQdrant 和 Neo4j 用 Docker 起docker run -d -p 6333:6333 qdrant/qdrant docker run -d -p 7687:7687 -p 7474:7474 neo4j:community4.2 记忆写入的完整代码实现先定义抽取函数import json from openai import OpenAI client OpenAI(base_urlhttp://localhost:8000/v1, api_keydummy) EXTRACT_PROMPT 你是一个记忆抽取器。从以下对话中抽取值得长期记忆的信息。 输出 JSON字段包括 event_type, summary, entities, relations, confidence。 event_type 可选preference, fact, decision, entity, correction。 confidence 是 0 到 1 的浮点数。 只输出 JSON不要其他内容。 对话 {conversation} def extract_memory(conversation: str) - dict: resp client.chat.completions.create( modelqwen2.5-7b-instruct, messages[{role: user, content: EXTRACT_PROMPT.format(conversationconversation)}], temperature0.1, response_format{type: json_object} ) return json.loads(resp.choices[0].message.content)写入向量库from qdrant_client import QdrantClient from qdrant_client.models import PointStruct, VectorParams, Distance import uuid, time qdrant QdrantClient(hostlocalhost, port6333) qdrant.recreate_collection( collection_namelong_term_memory, vectors_configVectorParams(size768, distanceDistance.COSINE) ) def write_memory(user_id: str, memory: dict, embedding: list): point_id str(uuid.uuid4()) qdrant.upsert( collection_namelong_term_memory, points[PointStruct( idpoint_id, vectorembedding, payload{ user_id: user_id, summary: memory[summary], event_type: memory[event_type], entities: memory[entities], confidence: memory[confidence], created_at: time.time(), reference_count: 0 } )] ) return point_id写入关系图谱from neo4j import GraphDatabase driver GraphDatabase.driver(bolt://localhost:7687, auth(neo4j, password)) def write_relations(user_id: str, memory: dict): with driver.session() as session: for rel in memory.get(relations, []): session.run( MERGE (a:Entity {name: $from_name, user_id: $uid}) MERGE (b:Entity {name: $to_name, user_id: $uid}) MERGE (a)-[r:RELATED {type: $rel_type}]-(b) SET r.updated_at timestamp(), r.confidence $conf , from_namerel[from], to_namerel[to], rel_typerel[type], uiduser_id, confmemory[confidence] )4.3 检索与注入的实操流程检索函数import math def retrieve_memories(user_id: str, query: str, query_embedding: list, top_k: int 5): # 向量召回 hits qdrant.search( collection_namelong_term_memory, query_vectorquery_embedding, query_filter{must: [{key: user_id, match: {value: user_id}}]}, limit20 ) now time.time() scored [] for hit in hits: payload hit.payload days (now - payload[created_at]) / 86400 weight payload[confidence] * math.exp(-0.02 * days) * (1 0.1 * payload[reference_count]) final_score 0.6 * hit.score 0.4 * min(weight, 1.0) scored.append((final_score, payload)) scored.sort(keylambda x: x[0], reverseTrue) return [p for _, p in scored[:top_k]]注入 promptdef build_prompt_with_memory(user_id: str, query: str, query_embedding: list): memories retrieve_memories(user_id, query, query_embedding) memory_text \n.join([f- [{m[event_type]}] {m[summary]} for m in memories]) return f以下是与当前对话相关的历史记忆 {memory_text} 用户当前输入{query} 4.4 短期记忆的 Redis 实现短期记忆我用来存最近几次会话的原始对话TTL 设 24 小时import redis, json r redis.Redis(hostlocalhost, port6379, decode_responsesTrue) def push_short_term(user_id: str, session_id: str, role: str, content: str): key fstm:{user_id}:{session_id} r.rpush(key, json.dumps({role: role, content: content, ts: time.time()})) r.expire(key, 86400) def get_short_term(user_id: str, session_id: str, limit: int 10): key fstm:{user_id}:{session_id} items r.lrange(key, -limit, -1) return [json.loads(i) for i in items]短期记忆在构建 prompt 时直接拼在长期记忆后面作为最近上下文。注意短期记忆不要全部塞进去按 token 预算截断我一般留 2000 token 给短期记忆。5. 常见问题与排查技巧实录5.1 记忆检索不相关怎么办这是最高频的问题。排查顺序我一般是这样现象可能原因排查方法解决检索结果完全不相关embedding 模型不匹配检查写入和检索是否用同一模型统一模型重新 embed相关记忆排在后边时间权重过大打印每条记忆的 weight 和相似度调低时间权重比例检索不到任何记忆user_id 过滤错误检查写入和检索的 user_id 是否一致统一 ID 生成规则返回旧记忆冲突消解未生效检查关系图谱的 updated_at检索时过滤失效边我踩过最坑的一次是 embedding 模型换了但没重新 embed 历史数据导致新旧向量不在同一空间检索结果完全随机。所以换 embedding 模型一定要全量重建索引这个没有捷径。5.2 记忆库膨胀太快怎么控制写入触发条件太宽松是主因。我的经验是加一道去重和合并新记忆写入前先检索 top-3 相似记忆如果相似度超过 0.95就不新增而是更新旧记忆的reference_count和updated_at。如果相似度在 0.8 到 0.95 之间让 LLM 判断是合并还是新增。这样能把库的增速降低 60% 以上。另外定期做冷记忆归档超过 90 天且reference_count为 0 的记忆从向量库移到冷存储检索时不参与但保留可恢复。这个策略对个人助手场景特别有效因为很多一次性信息过了一个季度就没用了。5.3 记忆注入导致 prompt 超长top-k 不能设太大我一般 5 条每条 summary 控制在 50 字以内加上短期记忆 2000 token总共记忆部分不超过 500 token。如果还是超就做分层注入先注入高权重的 3 条如果模型表示信息不足再触发二次检索注入更多。这个可以用 function calling 实现让模型主动请求“我还需要更多记忆”。5.4 多用户记忆隔离问题user_id 必须贯穿写入和检索全链路。我见过有人用 session_id 当隔离键结果同一用户换设备就失忆了。正确做法是 user_id 和 session_id 分开user_id 标识长期身份session_id 标识单次会话。短期记忆按 session_id 存长期记忆按 user_id 存。检索时长期记忆用 user_id 过滤短期记忆用 session_id 取。5.5 记忆置信度校准confidence 是 LLM 给的不一定准。我加了一个用户反馈回路如果用户对 agent 的回答表示认可相关记忆的 confidence 加 0.1如果用户纠正相关记忆 confidence 减 0.2并触发 correction 类型的记忆写入。这样跑一段时间后confidence 会逐渐校准到合理区间。注意 confidence 要有上下限我设的是 0.1 到 1.0避免归零或溢出。6. 与 MCP 生态的衔接及扩展方向hindsight 这套记忆系统不是孤立的它可以和 MCP 协议对接把记忆读写封装成 MCP tool这样任何支持 MCP 的 agent 框架都能直接调用。具体做法是起一个 MCP server暴露write_memory、retrieve_memory、forget_memory三个 toolagent 在需要的时候主动调用。这样记忆管理就从 agent 主逻辑里解耦出来了换框架也不用重写。我实测下来MCP 方式的好处是标准化坏处是多一次网络往返延迟增加 50 到 100 毫秒。如果对延迟敏感可以本地直连不走 MCP。另外 MCP 的 tool schema 要设计得简洁参数不要太多否则模型调用时容易填错。扩展方向有几个一是记忆可视化把关系图谱画出来让用户看到 agent 记住了什么方便纠错二是记忆共享多个 agent 共享同一份记忆库比如客服 agent 和售后 agent 共享用户偏好三是记忆压缩定期用 LLM 把多条相关记忆合并成一条高层摘要降低检索复杂度。这几个方向我都在试目前记忆可视化效果最好用户看到图谱之后对 agent 的信任度明显提升。最后分享一个实操小技巧记忆系统的日志一定要打全。每次写入和检索都记录原始 query、召回结果、最终分数、注入内容。出问题的时候翻日志比猜原因快十倍。我一开始没打日志排查一个检索不相关的问题花了一下午后来加上日志同样的问题十分钟定位。这个投入绝对值得。
LLM Agent 记忆系统实战：hindsight 架构设计与落地
1. 从“hindsight”说起为什么记忆是 Agent 落地的最后一公里“hindsight”这个词本身很有意思字面意思是“事后的洞察力”也就是我们常说的“后见之明”。把这个词放到 LLM Agent 的语境里它指向的东西非常明确Agent 在完成任务之后能不能把这次经历沉淀下来下次遇到类似场景时直接调用而不是每次都从零开始推理。这就是记忆memory问题也是当前 agent 开发里最容易被低估、但实际落地时最卡脖子的一环。我接触过不少 agent 项目从简单的工具调用到复杂的多步规划大家一开始都把精力放在 prompt 工程、工具编排、MCP 协议对接上觉得只要模型够强、工具够全agent 就能跑起来。但真正上线跑一段时间就会发现同一个用户问第二遍同样的问题agent 还是像第一次见面一样完全不知道之前发生过什么。这种“失忆”状态在 demo 阶段不明显一旦进入真实业务场景用户体验直接崩掉。hindsight 要解决的就是这个问题。它不是简单的对话历史拼接而是一套结构化的、可检索的、带时间衰减的长期记忆机制。你可以把它理解成给 agent 装了一个“经验库”每次任务执行完agent 会把关键信息写进去下次遇到新任务先从经验库里捞相关的记忆再结合当前上下文做决策。这套机制的核心价值在于降低重复推理成本、提升响应一致性、让 agent 具备跨会话的连续性。适合谁来参考如果你正在做 agent 开发尤其是涉及多轮对话、任务型助手、个人知识管理这类场景hindsight 的思路可以直接借鉴。如果你只是用现成的 LLM 框架做简单问答那可能暂时用不上但了解这套机制对你理解 agent 的演进方向有帮助。下面我会从设计思路、核心细节、实操落地、问题排查几个层面把这套东西拆开讲清楚。2. 记忆系统的整体设计与选型逻辑2.1 为什么不用简单的向量数据库堆砌很多人一提到 agent 记忆第一反应就是“上个向量数据库把历史对话 embed 一下存进去需要的时候相似度检索”。这个方案能跑但实际用下来问题很多。我最早也是这么干的用 Chroma 或者 Milvus 存对话片段检索 top-k 塞回 prompt。跑了一段时间发现几个致命问题第一检索出来的记忆是碎片化的。向量检索按相似度返回可能返回五条不相关的片段每条都只有半句话拼在一起语义不完整。第二没有时间维度。三个月前的一条记忆和昨天的记忆在向量空间里可能距离差不多但实际权重完全不一样。第三无法处理矛盾信息。用户上周说喜欢咖啡这周说改喝茶了两条记忆都在库里检索时可能同时返回agent 就懵了。hindsight 的设计思路不是推翻向量检索而是在它上面加了两层结构化抽取层和时间衰减层。结构化抽取负责把原始对话变成带实体、带关系、带事件类型的记忆单元时间衰减负责给每条记忆算一个动态权重越久远的记忆权重越低但如果是高频引用的记忆权重衰减会变慢。这样检索的时候不是单纯看语义相似度而是相似度乘以时间权重再排序。2.2 记忆分层工作记忆、短期记忆、长期记忆我把 hindsight 的记忆体系分成三层这个分层参考了认知科学里的人类记忆模型但在工程上做了简化层级存储介质生命周期容量典型用途工作记忆内存/上下文窗口单次会话受 token 限制当前对话的即时上下文短期记忆Redis/内存数据库数小时到数天较大最近几次会话的关键信息长期记忆向量库关系库永久很大用户偏好、历史事件、知识沉淀工作记忆就是当前 prompt 里的内容这个不用多说。短期记忆我一般用 Redis 存key 是用户 ID 加会话 IDvalue 是结构化的事件列表设置 TTL。长期记忆才是 hindsight 的核心它又分成两个部分向量索引用于语义检索关系图谱用于实体关联查询。比如用户问“我上次说的那个项目进展怎么样了”向量检索能找到相关对话片段关系图谱能顺着“项目”这个实体找到关联的任务、时间、参与人。2.3 写入策略什么时候该记什么时候不该记这是实操中最容易踩坑的地方。如果每轮对话都写记忆库会迅速膨胀检索质量下降成本也扛不住。我的经验是按事件触发写入而不是按轮次。具体来说以下几种情况才触发写入用户明确表达了偏好、事实、决策“我决定用 PostgreSQL 而不是 MySQL”任务执行完成产生了可复用的结果“这次部署的配置是 XXX”用户纠正了 agent 的错误“不对应该是 XXX”出现了新的实体或关系“我们团队新来了一个同事叫张三”反过来闲聊、确认性回复“好的”“嗯嗯”、中间推理过程这些都不写。写入的时候也不是原样存而是先让 LLM 做一次抽取输出结构化的 JSON包含event_type、entities、summary、timestamp、confidence这几个字段。confidence 很重要用户随口一说的事情置信度低明确确认的事情置信度高检索时权重不一样。3. 核心细节解析从抽取到检索的完整链路3.1 记忆抽取的 prompt 设计要点抽取这一步直接决定记忆质量。我试过很多版 prompt最后稳定下来的结构是这样的先给模型一个明确的角色定义“你是一个记忆抽取器”然后给几个 few-shot 示例每个示例包含原始对话和对应的 JSON 输出。关键点在于约束输出格式必须用 JSON schema 校验不合法就重试。抽取的字段我一般保留这几个{ event_type: preference | fact | decision | entity | correction, summary: 一句话概括不超过50字, entities: [实体1, 实体2], relations: [{from: 实体1, to: 实体2, type: 关系类型}], confidence: 0.0-1.0, source_text: 原始对话片段 }这里有个细节summary一定要让模型用自己的话概括不要直接截取原文。因为原文可能包含口语、指代、省略直接存进去检索时匹配效果差。概括之后语义更集中embedding 质量更高。另外entities要做归一化比如“PostgreSQL”和“postgres”要统一成同一个实体 ID不然关系图谱会散掉。3.2 时间衰减函数的参数计算时间衰减是 hindsight 区别于普通向量记忆的关键。我用的是指数衰减加引用加权的组合weight base_confidence * exp(-lambda * days_since_creation) * (1 alpha * reference_count)其中lambda是衰减系数days_since_creation是记忆创建至今的天数reference_count是这条记忆被检索命中的次数alpha是引用加权系数。参数怎么定我实测下来lambda取 0.01 到 0.03 比较合适。取 0.01 的话100 天后权重降到约 0.37取 0.03 的话100 天后降到约 0.05。具体取值看业务场景如果是个人助手记忆需要长期保留lambda取小一点如果是客服场景用户偏好变化快lambda取大一点。alpha我一般设 0.1也就是说一条记忆每被引用一次权重增加 10%。这个机制让高频使用的记忆衰减更慢相当于“用进废退”。但要注意reference_count不能无限增长我设了一个上限 50超过之后不再增加避免某条记忆权重过大压制其他记忆。3.3 检索时的混合排序策略检索不是单纯按向量相似度排而是多路召回再融合。我的做法是向量召回用 query 的 embedding 在长期记忆库里检索 top-20实体召回从 query 里抽取实体在关系图谱里找关联记忆 top-10时间召回最近 7 天的短期记忆全部纳入候选三路召回合并去重后用上面的 weight 公式算最终分数再乘以语义相似度排序取 top-5 塞回 prompt。这里有个坑语义相似度的量纲和 weight 不一样直接相乘会导致某一项主导。我的处理是先把相似度归一化到 0-1weight 也归一化到 0-1然后加权求和相似度权重 0.6时间权重 0.4。这个比例可以根据场景调任务型 agent 更看重语义个人助手更看重时间。3.4 记忆冲突的消解机制用户改主意了怎么办比如之前说“我住在北京”后来说“我搬到上海了”。两条记忆都在库里检索时可能同时返回。我的消解策略是同实体同关系类型的新记忆覆盖旧记忆。具体实现是在关系图谱里(用户, 居住地, 北京)这条边被标记为失效新边(用户, 居住地, 上海)生效。向量库里旧记忆不删除但检索时如果发现有关联的新记忆旧记忆的权重会被压制到 0.1 以下。这个机制需要实体和关系类型定义得足够细。如果关系类型太粗比如只有一个“相关信息”那就没法判断哪条新哪条旧。所以前期花时间设计 ontology 是值得的至少要把人物、地点、时间、偏好、决策这几大类分清楚。4. 实操落地从零搭一套 hindsight 记忆系统4.1 技术栈选型与依赖安装我用的技术栈比较轻量适合快速验证LLM任意支持 function calling 的模型我用的是本地部署的 7B 模型加一个 API 模型做兜底向量库Qdrant单机版够用Docker 一键起关系库Neo4j Community或者用 SQLite 加邻接表也能凑合缓存Redis存短期记忆和检索缓存编排Python 加 FastAPI不用太重型的框架安装依赖pip install qdrant-client neo4j redis openai tiktoken fastapi uvicornQdrant 和 Neo4j 用 Docker 起docker run -d -p 6333:6333 qdrant/qdrant docker run -d -p 7687:7687 -p 7474:7474 neo4j:community4.2 记忆写入的完整代码实现先定义抽取函数import json from openai import OpenAI client OpenAI(base_urlhttp://localhost:8000/v1, api_keydummy) EXTRACT_PROMPT 你是一个记忆抽取器。从以下对话中抽取值得长期记忆的信息。 输出 JSON字段包括 event_type, summary, entities, relations, confidence。 event_type 可选preference, fact, decision, entity, correction。 confidence 是 0 到 1 的浮点数。 只输出 JSON不要其他内容。 对话 {conversation} def extract_memory(conversation: str) - dict: resp client.chat.completions.create( modelqwen2.5-7b-instruct, messages[{role: user, content: EXTRACT_PROMPT.format(conversationconversation)}], temperature0.1, response_format{type: json_object} ) return json.loads(resp.choices[0].message.content)写入向量库from qdrant_client import QdrantClient from qdrant_client.models import PointStruct, VectorParams, Distance import uuid, time qdrant QdrantClient(hostlocalhost, port6333) qdrant.recreate_collection( collection_namelong_term_memory, vectors_configVectorParams(size768, distanceDistance.COSINE) ) def write_memory(user_id: str, memory: dict, embedding: list): point_id str(uuid.uuid4()) qdrant.upsert( collection_namelong_term_memory, points[PointStruct( idpoint_id, vectorembedding, payload{ user_id: user_id, summary: memory[summary], event_type: memory[event_type], entities: memory[entities], confidence: memory[confidence], created_at: time.time(), reference_count: 0 } )] ) return point_id写入关系图谱from neo4j import GraphDatabase driver GraphDatabase.driver(bolt://localhost:7687, auth(neo4j, password)) def write_relations(user_id: str, memory: dict): with driver.session() as session: for rel in memory.get(relations, []): session.run( MERGE (a:Entity {name: $from_name, user_id: $uid}) MERGE (b:Entity {name: $to_name, user_id: $uid}) MERGE (a)-[r:RELATED {type: $rel_type}]-(b) SET r.updated_at timestamp(), r.confidence $conf , from_namerel[from], to_namerel[to], rel_typerel[type], uiduser_id, confmemory[confidence] )4.3 检索与注入的实操流程检索函数import math def retrieve_memories(user_id: str, query: str, query_embedding: list, top_k: int 5): # 向量召回 hits qdrant.search( collection_namelong_term_memory, query_vectorquery_embedding, query_filter{must: [{key: user_id, match: {value: user_id}}]}, limit20 ) now time.time() scored [] for hit in hits: payload hit.payload days (now - payload[created_at]) / 86400 weight payload[confidence] * math.exp(-0.02 * days) * (1 0.1 * payload[reference_count]) final_score 0.6 * hit.score 0.4 * min(weight, 1.0) scored.append((final_score, payload)) scored.sort(keylambda x: x[0], reverseTrue) return [p for _, p in scored[:top_k]]注入 promptdef build_prompt_with_memory(user_id: str, query: str, query_embedding: list): memories retrieve_memories(user_id, query, query_embedding) memory_text \n.join([f- [{m[event_type]}] {m[summary]} for m in memories]) return f以下是与当前对话相关的历史记忆 {memory_text} 用户当前输入{query} 4.4 短期记忆的 Redis 实现短期记忆我用来存最近几次会话的原始对话TTL 设 24 小时import redis, json r redis.Redis(hostlocalhost, port6379, decode_responsesTrue) def push_short_term(user_id: str, session_id: str, role: str, content: str): key fstm:{user_id}:{session_id} r.rpush(key, json.dumps({role: role, content: content, ts: time.time()})) r.expire(key, 86400) def get_short_term(user_id: str, session_id: str, limit: int 10): key fstm:{user_id}:{session_id} items r.lrange(key, -limit, -1) return [json.loads(i) for i in items]短期记忆在构建 prompt 时直接拼在长期记忆后面作为最近上下文。注意短期记忆不要全部塞进去按 token 预算截断我一般留 2000 token 给短期记忆。5. 常见问题与排查技巧实录5.1 记忆检索不相关怎么办这是最高频的问题。排查顺序我一般是这样现象可能原因排查方法解决检索结果完全不相关embedding 模型不匹配检查写入和检索是否用同一模型统一模型重新 embed相关记忆排在后边时间权重过大打印每条记忆的 weight 和相似度调低时间权重比例检索不到任何记忆user_id 过滤错误检查写入和检索的 user_id 是否一致统一 ID 生成规则返回旧记忆冲突消解未生效检查关系图谱的 updated_at检索时过滤失效边我踩过最坑的一次是 embedding 模型换了但没重新 embed 历史数据导致新旧向量不在同一空间检索结果完全随机。所以换 embedding 模型一定要全量重建索引这个没有捷径。5.2 记忆库膨胀太快怎么控制写入触发条件太宽松是主因。我的经验是加一道去重和合并新记忆写入前先检索 top-3 相似记忆如果相似度超过 0.95就不新增而是更新旧记忆的reference_count和updated_at。如果相似度在 0.8 到 0.95 之间让 LLM 判断是合并还是新增。这样能把库的增速降低 60% 以上。另外定期做冷记忆归档超过 90 天且reference_count为 0 的记忆从向量库移到冷存储检索时不参与但保留可恢复。这个策略对个人助手场景特别有效因为很多一次性信息过了一个季度就没用了。5.3 记忆注入导致 prompt 超长top-k 不能设太大我一般 5 条每条 summary 控制在 50 字以内加上短期记忆 2000 token总共记忆部分不超过 500 token。如果还是超就做分层注入先注入高权重的 3 条如果模型表示信息不足再触发二次检索注入更多。这个可以用 function calling 实现让模型主动请求“我还需要更多记忆”。5.4 多用户记忆隔离问题user_id 必须贯穿写入和检索全链路。我见过有人用 session_id 当隔离键结果同一用户换设备就失忆了。正确做法是 user_id 和 session_id 分开user_id 标识长期身份session_id 标识单次会话。短期记忆按 session_id 存长期记忆按 user_id 存。检索时长期记忆用 user_id 过滤短期记忆用 session_id 取。5.5 记忆置信度校准confidence 是 LLM 给的不一定准。我加了一个用户反馈回路如果用户对 agent 的回答表示认可相关记忆的 confidence 加 0.1如果用户纠正相关记忆 confidence 减 0.2并触发 correction 类型的记忆写入。这样跑一段时间后confidence 会逐渐校准到合理区间。注意 confidence 要有上下限我设的是 0.1 到 1.0避免归零或溢出。6. 与 MCP 生态的衔接及扩展方向hindsight 这套记忆系统不是孤立的它可以和 MCP 协议对接把记忆读写封装成 MCP tool这样任何支持 MCP 的 agent 框架都能直接调用。具体做法是起一个 MCP server暴露write_memory、retrieve_memory、forget_memory三个 toolagent 在需要的时候主动调用。这样记忆管理就从 agent 主逻辑里解耦出来了换框架也不用重写。我实测下来MCP 方式的好处是标准化坏处是多一次网络往返延迟增加 50 到 100 毫秒。如果对延迟敏感可以本地直连不走 MCP。另外 MCP 的 tool schema 要设计得简洁参数不要太多否则模型调用时容易填错。扩展方向有几个一是记忆可视化把关系图谱画出来让用户看到 agent 记住了什么方便纠错二是记忆共享多个 agent 共享同一份记忆库比如客服 agent 和售后 agent 共享用户偏好三是记忆压缩定期用 LLM 把多条相关记忆合并成一条高层摘要降低检索复杂度。这几个方向我都在试目前记忆可视化效果最好用户看到图谱之后对 agent 的信任度明显提升。最后分享一个实操小技巧记忆系统的日志一定要打全。每次写入和检索都记录原始 query、召回结果、最终分数、注入内容。出问题的时候翻日志比猜原因快十倍。我一开始没打日志排查一个检索不相关的问题花了一下午后来加上日志同样的问题十分钟定位。这个投入绝对值得。