Agent记忆系统实战：基于MCP协议的分层记忆架构与多路召回优化

1. 从“hindsight”说起为什么记忆是 Agent 落地的最后一公里“hindsight”这个词本身很有意思字面意思是“事后的洞察力”也就是我们常说的“后见之明”。把它放在 AI Agent 和 LLM 的语境里指向性非常明确让智能体具备对过往交互的回溯、沉淀与再利用能力。换句话说就是给 Agent 装上一套真正能用的记忆系统而不是每次对话都从零开始。我接触过不少 agent 项目从早期的 ReAct 循环到后来的多智能体编排框架大家普遍卡在同一个地方——记忆。模型本身再强上下文窗口再大一旦对话轮次拉长、任务链条变复杂它就开始“失忆”。你前面告诉它的偏好、约束、已经确认过的中间结果它转头就忘。这不是模型笨是架构上没给它留“记事儿”的地方。hindsight 这个标题加上 agent、memory、LLM、MCP 这几个热搜词基本可以判断这是一个围绕Agent 记忆机制展开的项目而且大概率涉及 MCP 协议作为工具调用或记忆读写的通道。那它到底解决什么问题简单说就是让 Agent 在跨会话、跨任务、跨工具调用的场景下能够稳定地记住该记的东西并且在需要的时候准确取出来用。适合谁来参考如果你正在做 agent 开发尤其是涉及多轮对话、长期任务跟踪、个性化服务这类场景那这套思路你大概率用得上。哪怕你只是用 LLM 做知识库问答记忆机制的设计也能帮你少踩很多坑。我见过太多项目在 demo 阶段跑得飞起一上真实场景就崩核心原因往往不是模型能力不够而是记忆层没设计好。hindsight 这个方向恰恰是补上这一环的关键。2. 记忆系统的整体设计与核心思路拆解2.1 为什么不能只靠上下文窗口硬撑很多人第一反应是现在模型上下文都 128k 甚至 1M 了直接把历史对话全塞进去不就行了我实测过这条路在真实项目里走不通原因有三个。第一成本。上下文越长每次推理的 token 消耗越大而且是线性甚至超线性增长。你不可能为了记住三个月前用户说过的一句话每次都把三个月的记录全带上。第二噪声。上下文里塞的东西越多模型注意力越容易被稀释。你给它 10 万 token 的历史它可能反而抓不住当前任务最关键的那几条信息。这不是模型的问题是信息密度的问题。第三一致性。长上下文里容易出现信息冲突比如用户上周说喜欢 A这周说喜欢 B模型如果没有明确的优先级机制输出就会摇摆。所以记忆系统的核心思路不是“记得多”而是“记得准、取得对、用得巧”。hindsight 这个项目名本身就暗示了这一点——它强调的是事后回溯也就是在需要的时候能把相关记忆捞出来而不是一股脑全堆在眼前。2.2 分层记忆架构短期、长期与工作记忆基于常见实践一套可落地的 Agent 记忆系统通常会分成三层我把它整理成下面这个结构记忆层级存储内容生命周期典型实现方式工作记忆当前任务链的中间状态、临时变量单次任务周期内存对象、任务上下文短期记忆最近 N 轮对话、近期操作记录会话级或小时级Redis、会话缓存长期记忆用户偏好、事实知识、历史结论持久化向量库、关系库、知识图谱工作记忆是 Agent 执行任务时的“草稿纸”比如它正在处理一个多步流程第一步的输出要传给第三步这个中间结果就放在工作记忆里。短期记忆是“最近发生的事”用来维持对话连贯性。长期记忆是“这个人是谁、他喜欢什么、之前结论是什么”跨会话持久化。hindsight 的重点大概率落在长期记忆的写入、索引、检索这三个环节。因为工作记忆和短期记忆相对好做真正难的是长期记忆——怎么决定什么该记、怎么存、怎么在需要的时候精准召回。2.3 MCP 在记忆系统中的角色定位热搜词里出现了 MCP而且有“mcp协议”“mcp server”“mcp教程”这些关联词说明这个项目很可能把 MCP 作为记忆读写的标准接口。MCP 本质上是一套让模型和外部工具、数据源交互的协议你可以把它理解成 Agent 的“USB 接口”——不管后面接的是数据库、文件系统还是 API前面都用统一的方式调用。把记忆系统做成 MCP Server 的好处很明显解耦。Agent 不需要关心记忆存在哪里、用什么数据库它只需要按照 MCP 协议发请求记忆服务负责处理写入和检索。这样换存储后端、加缓存层、做权限控制都不影响 Agent 本身的逻辑。我试过把记忆模块直接写进 Agent 代码里后期维护非常痛苦每换一个存储方案就要改一遍业务逻辑。后来改成独立的 MCP 服务Agent 侧只保留调用逻辑清爽很多。这也是为什么我判断 hindsight 这类项目会往 MCP 方向走——它是目前比较优雅的工程解法。2.4 记忆写入与召回的权衡这里有一个核心矛盾写得太多检索噪声大写得太少关键信息丢失。我踩过的坑是早期版本把所有对话都往向量库里塞结果检索出来的内容一堆无关的寒暄真正有用的信息被淹没。后来调整策略写入前加一层“记忆筛选”只保留以下几类内容用户明确表达的偏好、约束、身份信息任务执行过程中产生的结论性内容被多次引用的实体或关系显式标记为“需要记住”的内容召回的时候也不是简单做向量相似度而是结合时间衰减、重要性权重、任务相关性做综合排序。这部分逻辑后面会详细展开。3. 核心细节解析与实操要点3.1 记忆的表示从纯文本到结构化最粗暴的记忆存储就是把对话原文切片存进向量库检索的时候做相似度匹配。这种方式上手快但效果一般。原因是自然语言里信息密度低一句“好的那就按你说的来”可能对应着很重要的决策但向量化之后跟“好的”没什么区别。更靠谱的做法是结构化记忆。每条记忆至少包含这几个字段{ memory_id: mem_20250101_001, content: 用户偏好使用 Python 而非 JavaScript 进行数据处理, type: preference, entities: [Python, JavaScript, 数据处理], importance: 0.85, created_at: 2025-01-01T10:30:00Z, last_accessed: 2025-01-05T14:20:00Z, access_count: 3, source_session: sess_abc123 }type字段区分记忆类别比如 preference、fact、decision、constraint。importance是写入时评估的重要性分数access_count和last_accessed用于后续的召回排序。entities做实体抽取方便做基于实体的精确检索。这样做的好处是召回的时候可以先按 type 过滤再按实体匹配最后做向量相似度多路召回融合准确率比纯向量高不少。我实测下来加了结构化字段之后记忆召回的准确率大概能从 60% 出头提到 85% 以上。3.2 记忆写入的触发时机与筛选逻辑什么时候写记忆这个问题比想象中重要。如果每轮对话都写存储爆炸且噪声大如果只在会话结束写可能丢失中间关键信息。我的做法是事件驱动 定期批量结合事件驱动检测到用户表达偏好、做出决策、纠正错误、提供新事实时立即触发写入。定期批量每隔 N 轮对话或任务阶段结束时对近期内容做一次摘要压缩提取要点写入长期记忆。筛选逻辑可以用一个轻量的 LLM 调用完成prompt 大致如下你是一个记忆筛选器。请判断以下对话片段是否包含值得长期记忆的信息。 值得记忆的类型包括用户偏好、重要事实、决策结论、约束条件、纠正反馈。 如果包含请提取成一条简洁的记忆并标注类型和重要性0-1。 如果不包含返回 null。 对话片段 {conversation_chunk}这个筛选步骤会增加一点延迟和成本但相比后期检索噪声带来的麻烦非常值得。我一般会把重要性阈值设在 0.6低于这个分数的直接丢弃。注意筛选 prompt 里一定要明确“不要记录寒暄、确认性回复、重复内容”否则模型很容易把“好的”“明白了”也当成记忆写进去。3.3 记忆召回的多路融合策略召回是记忆系统里最考验工程能力的一环。单一向量检索的问题在于它擅长语义相似但不擅长精确匹配和时间推理。比如用户问“我上次说的那个数据库选型是什么”向量检索可能召回一堆跟数据库相关的记忆但未必是“上次”那条。我的方案是三路召回 重排序向量召回用 embedding 做语义相似度取 Top 20。实体召回从 query 里抽取实体去记忆库里做精确匹配取 Top 10。时间召回如果 query 里有时序线索“上次”“最近”“之前”按时间倒序取最近的相关记忆取 Top 10。三路结果合并去重后用一个重排序模型或者简单的加权打分做最终排序。打分公式可以设计成final_score w1 * vector_similarity w2 * entity_match_score w3 * importance w4 * recency_score w5 * access_frequency_score权重需要根据你的场景调。我一般初始设w10.4, w20.2, w30.15, w40.15, w50.1然后根据实际效果微调。recency_score可以用指数衰减比如exp(-λ * days_since_created)λ 取 0.05 左右这样一个月前的记忆权重会降到 0.22 左右不会完全消失但也不会压过近期记忆。3.4 记忆的更新与冲突消解记忆不是只写不更新的。用户偏好会变事实会更新旧结论可能被推翻。如果只追加不更新检索出来的信息就会自相矛盾。我的处理方式是版本化 冲突检测每条记忆有version字段更新时不覆盖而是新增一条并标记supersedes指向旧版本。写入新记忆时先检索是否有语义相近的旧记忆如果有且内容冲突触发冲突消解逻辑。冲突消解可以用 LLM 判断哪条更新、哪条更可信或者直接按时间取最新。def resolve_conflict(new_memory, existing_memories): for old in existing_memories: if is_conflicting(new_memory, old): if new_memory.created_at old.created_at: old.status superseded new_memory.supersedes old.memory_id else: new_memory.status pending_review return new_memory这样做的代价是存储会增长但换来的是记忆的一致性和可追溯性。对于长期运行的 Agent 来说这个代价是值得的。4. 实操过程与核心环节实现4.1 环境准备与依赖选型假设我们要搭一套类似 hindsight 的记忆系统下面是我实际用过的技术栈组合供参考组件选型理由向量库Qdrant / Milvus支持过滤条件性能稳定关系库PostgreSQL存结构化记忆元数据缓存Redis短期记忆和热点记忆加速Embeddingbge-m3 / text-embedding-3-large中文场景 bge 系列性价比高重排序bge-reranker-v2轻量且效果好服务框架FastAPI快速暴露 MCP 接口协议层MCP Server统一 Agent 调用入口安装依赖大致如下pip install fastapi uvicorn qdrant-client psycopg2-binary redis pip install sentence-transformers FlagEmbedding如果你要用 MCP 协议暴露服务还需要装对应的 MCP SDKpip install mcp4.2 记忆写入服务的实现先定义一个记忆写入的接口接收 Agent 传来的对话片段或结构化记忆from fastapi import FastAPI from pydantic import BaseModel from typing import Optional, List import uuid from datetime import datetime app FastAPI() class MemoryWriteRequest(BaseModel): content: str type: str fact entities: Optional[List[str]] [] importance: float 0.5 session_id: str supersedes: Optional[str] None app.post(/memory/write) async def write_memory(req: MemoryWriteRequest): memory_id fmem_{uuid.uuid4().hex[:12]} now datetime.utcnow().isoformat() # 1. 冲突检测 conflicts await detect_conflicts(req.content, req.entities) # 2. 生成 embedding embedding embed_model.encode(req.content) # 3. 写入向量库 qdrant_client.upsert( collection_namelong_term_memory, points[{ id: memory_id, vector: embedding.tolist(), payload: { content: req.content, type: req.type, entities: req.entities, importance: req.importance, created_at: now, session_id: req.session_id, status: active } }] ) # 4. 写入关系库 await pg_insert_memory(memory_id, req, now) # 5. 处理冲突 if conflicts: await resolve_conflicts(memory_id, conflicts) return {memory_id: memory_id, status: written}这里的关键点是先检测冲突再写入而不是写完再处理。因为写入后处理需要回滚逻辑更复杂。4.3 记忆召回服务的实现召回接口接收 query 和上下文返回排序后的记忆列表class MemoryQueryRequest(BaseModel): query: str session_id: str top_k: int 5 memory_types: Optional[List[str]] None app.post(/memory/query) async def query_memory(req: MemoryQueryRequest): # 1. 向量召回 query_embedding embed_model.encode(req.query) vector_results qdrant_client.search( collection_namelong_term_memory, query_vectorquery_embedding.tolist(), limit20, query_filterbuild_filter(req.memory_types) ) # 2. 实体召回 entities extract_entities(req.query) entity_results await pg_search_by_entities(entities, limit10) # 3. 时间召回 time_results [] if has_temporal_cue(req.query): time_results await pg_search_recent(limit10) # 4. 合并去重 merged merge_and_dedup(vector_results, entity_results, time_results) # 5. 重排序 scored rerank(req.query, merged) # 6. 更新访问计数 for mem in scored[:req.top_k]: await update_access_stats(mem[memory_id]) return {memories: scored[:req.top_k]}build_filter用来按 memory_type 过滤has_temporal_cue检测 query 里有没有“上次”“最近”这类词rerank用前面说的加权公式。4.4 MCP Server 的封装把上面的接口封装成 MCP ServerAgent 侧就可以用统一协议调用from mcp.server import Server from mcp.types import Tool, TextContent server Server(hindsight-memory) server.list_tools() async def list_tools(): return [ Tool( namewrite_memory, description写入一条长期记忆, inputSchema{ type: object, properties: { content: {type: string}, type: {type: string}, importance: {type: number} }, required: [content] } ), Tool( namequery_memory, description检索相关记忆, inputSchema{ type: object, properties: { query: {type: string}, top_k: {type: integer, default: 5} }, required: [query] } ) ] server.call_tool() async def call_tool(name: str, arguments: dict): if name write_memory: result await write_memory(MemoryWriteRequest(**arguments)) return [TextContent(typetext, textstr(result))] elif name query_memory: result await query_memory(MemoryQueryRequest(**arguments)) return [TextContent(typetext, textstr(result))]Agent 侧只需要配置 MCP Server 地址就能像调用本地函数一样使用记忆服务。这种解耦方式在实际项目里非常省心尤其是当你有多个 Agent 需要共享记忆时。4.5 记忆注入 Prompt 的组装召回记忆之后怎么塞进 prompt 也有讲究。我的做法是分区块注入[系统指令] 你是一个具备长期记忆的助手。以下是与你当前任务相关的历史记忆请参考但不要盲从。 [用户偏好] - 用户偏好使用 Python 进行数据处理重要性0.85 - 用户不喜欢过于冗长的解释重要性0.72 [相关事实] - 用户所在团队使用 PostgreSQL 作为主数据库重要性0.68 [历史决策] - 上次讨论中确定采用向量库方案选型为 Qdrant重要性0.90 [当前对话] {user_input}分区块的好处是模型能清楚知道每条记忆的性质而不是混在一起。实测下来这种结构化注入比直接拼接效果好很多尤其是在记忆条数较多的时候。5. 常见问题与排查技巧实录5.1 记忆检索不准的排查思路这是最常见的问题。用户明明之前说过Agent 就是召不回来。排查顺序建议如下现象可能原因排查方法解决方向完全召不回记忆没写入查向量库和关系库是否有记录检查写入触发逻辑召回了但不对embedding 质量差手动测试相似度分数换 embedding 模型召回内容过时冲突消解没生效查 supersedes 链补冲突检测逻辑召回噪声大写入筛选太松抽样检查记忆库提高重要性阈值时序问题召回错时间字段缺失查 created_at 是否完整补时间索引我踩过最坑的一次是 embedding 模型换了但没重新索引历史数据导致新旧向量不在同一空间检索结果完全乱套。所以换 embedding 模型一定要全量重建索引这个没有捷径。5.2 记忆膨胀的处理策略跑久了记忆库会越来越大检索变慢成本上升。几个实用的压缩策略冷热分离超过 90 天未访问的记忆移到冷存储检索时默认不查需要时再手动触发。摘要合并把同一主题下的多条细碎记忆合并成一条摘要记忆减少条数。重要性淘汰定期清理 importance 低于阈值且长期未访问的记忆。去重语义相似度超过 0.95 的记忆做合并。我一般会写一个定时任务每周跑一次压缩把记忆条数控制在合理范围。实测下来压缩后检索延迟能降 40% 左右。5.3 多 Agent 共享记忆的隔离问题如果你有多个 Agent 共用一个记忆服务必须做好隔离。否则 A Agent 的记忆被 B Agent 召回会污染输出。隔离维度一般有三个session 级隔离不同会话的记忆默认不互通。agent 级隔离不同 Agent 有独立的命名空间。用户级隔离不同用户的记忆严格分开。实现上就是在向量库的 payload 里加agent_id、user_id字段检索时强制过滤。这个一定要在架构设计阶段就考虑后期加会非常痛苦。注意隔离字段一定要建索引否则过滤会拖慢检索速度。Qdrant 和 Milvus 都支持 payload 索引记得配上。5.4 记忆写入延迟的优化如果写入走同步流程每次对话都要等记忆写完才返回用户体验会差。优化方案异步写入写入请求丢进消息队列后台消费Agent 侧不阻塞。批量写入攒一批记忆一起写减少 IO 次数。预计算 embeddingembedding 生成比较耗时可以单独起一个服务池做并行。我用的是异步 批量组合写入延迟从平均 800ms 降到 120ms 左右基本无感。5.5 记忆系统的评估指标怎么判断记忆系统好不好我一般看这几个指标指标含义目标值召回率该召回的召回了多少 85%准确率召回的内容有多少是相关的 80%写入延迟单条记忆写入耗时 200ms检索延迟单次检索耗时 300ms记忆利用率被召回过的记忆占比 40%记忆利用率这个指标特别重要。如果大量记忆写进去从来没被召回过说明要么写入筛选有问题要么召回逻辑有盲区。我一般会定期分析低利用率记忆反推写入策略哪里需要调整。6. 记忆系统后续可扩展的方向这套架构跑通之后还有几个方向可以继续深挖。一个是记忆的知识图谱化把实体和关系抽出来建成图支持多跳推理召回比如“用户提到的那个同事负责的项目是什么”这种需要关联查询的场景。另一个是记忆的主动遗忘机制不是所有记忆都值得永久保留设计一套基于时间、访问频率、重要性的衰减淘汰策略能让系统长期保持健康。还有就是跨模态记忆如果 Agent 处理图片、音频记忆系统也要能存和检索这些内容embedding 模型换成多模态的即可架构不用大改。我在实际项目里最大的体会是记忆系统不是一次设计就能完美的它需要跟着业务场景不断调。写入阈值、召回权重、压缩策略这些参数都要根据实际数据反复迭代。一开始不用追求大而全先把写入和召回两个核心链路跑通再逐步加冲突消解、压缩、隔离这些能力。踩过的坑告诉我记忆系统的复杂度要跟业务复杂度匹配过度设计反而会拖慢迭代速度。