Agent记忆系统实战：基于MCP协议与pgvector实现hindsight记忆架构-北京尧图网络

1. 从“hindsight”说起为什么我们需要给Agent装上“后视镜”第一次看到“hindsight”这个词是在做一个多轮对话Agent的复盘工具时。当时团队里有个争论Agent到底需不需要“记住”上一次任务失败的原因有人觉得每次请求都是独立的上下文窗口塞进去就够了有人坚持认为没有跨会话记忆的Agent永远只能当个“金鱼脑”助手。后来我们做了一组对比测试同一个代码修复任务带记忆的Agent第二次执行时平均少走了3.2步token消耗降低约41%。这个数字让我彻底站到了“记忆派”这边。hindsight这个词本身很有意思字面意思是“事后的洞察力”说白了就是“回头看才明白”。放在Agent记忆这个语境里它指向的是一类非常具体的能力让LLM驱动的Agent能够回溯、检索、利用过去发生过的事情来指导当前的决策。它不是简单的“把历史对话存下来”而是涉及存储结构、检索策略、遗忘机制、上下文注入方式等一整套工程问题。你可能会问这不就是RAG吗不完全是。RAG更多是面向静态知识库的检索增强而hindsight关注的是Agent自身经历的记忆——它做过什么、失败了什么、什么策略有效、什么路径被验证过。这两者有交集但hindsight更强调“自我参照”和“时间维度”。一个典型的hindsight系统需要回答三个问题存什么、怎么存、什么时候取出来用。这篇文章适合谁看如果你正在做Agent相关的产品或者你在用MCP协议搭建工具链又或者你只是好奇“Agent记忆”到底怎么落地那接下来的内容应该能给你一些可以直接抄作业的思路。我会从架构设计讲到具体实现包括Docker环境搭建、MCP协议对接、存储层选型以及我在实际项目中踩过的坑。2. Agent记忆的核心架构存什么、怎么存、怎么取2.1 记忆的三种类型与hindsight的定位在动手写代码之前得先把“记忆”这个概念拆清楚。业界通常把Agent记忆分为三类工作记忆Working Memory当前会话的上下文就是塞在prompt里的那些token。生命周期短容量受限于上下文窗口。情景记忆Episodic Memory具体发生过的事件比如“2024年3月15日用户让我修复了一个Python的ImportError我用了重装依赖的方案成功了”。语义记忆Semantic Memory从多次经历中抽象出来的规律比如“在这个项目里ImportError通常是因为虚拟环境没激活”。hindsight主要覆盖的是情景记忆和语义记忆这两层。工作记忆是LLM框架自己管的而hindsight要做的是跨会话、跨任务的长期记忆管理。为什么这么分因为不同类型的记忆存储和检索策略完全不同。工作记忆追求低延迟直接放内存情景记忆需要时间索引和事件关联适合用文档数据库或向量库语义记忆需要抽象和归纳往往要配合知识图谱或本体ontology来做。我见过不少项目一上来就把所有对话历史往向量库里塞结果检索出来的东西又碎又杂Agent反而被干扰。问题就出在没有区分记忆类型把工作记忆和情景记忆混在一起了。2.2 存储层选型向量库、文档库还是图数据库存储层的选择直接决定了hindsight系统的上限。我整理了一个对比表基于实际项目中的使用体验存储方案适合的记忆类型检索方式优势坑点向量数据库如Chroma、Qdrant情景记忆、语义记忆语义相似度模糊匹配强适合自然语言查询时间维度弱精确过滤差文档数据库如MongoDB情景记忆结构化查询全文索引时间范围查询方便字段灵活语义检索需要额外集成图数据库如Neo4j语义记忆、本体关系图遍历关系推理强适合知识图谱写入复杂运维成本高关系型数据库如PostgreSQLpgvector全部SQL向量混合一套搞定事务安全向量性能不如专用库我的建议是起步阶段用PostgreSQLpgvector。原因很简单你不需要同时维护两套存储系统SQL的过滤能力加上向量检索能覆盖80%的场景。等记忆量上到百万级再考虑拆分成专用向量库。这里有个关键设计每条记忆记录应该包含哪些字段我的方案是这样的# 记忆记录的数据结构以Python dict为例 memory_record { id: uuid, agent_id: agent_001, # 哪个Agent的记忆 session_id: sess_20240315, # 属于哪个会话 timestamp: 2024-03-15T10:30:00Z, memory_type: episodic, # episodic / semantic content: 用户要求修复ImportError方案是重装依赖, embedding: [0.12, -0.34, ...], # 向量表示 metadata: { task_type: code_fix, outcome: success, tools_used: [pip, python], token_cost: 1250 }, access_count: 3, # 被检索次数 last_accessed: 2024-03-16T09:00:00Z, decay_score: 0.85 # 衰减分数 }decay_score这个字段是我后来加的非常有用。记忆不是存得越多越好有些过时的、低价值的记忆应该逐渐“遗忘”。我用的衰减公式是decay_score base_score * exp(-λ * days_since_last_access) access_bonus其中λ取0.05access_bonus是每次被检索时加0.1。这样经常被用到的记忆衰减慢长期不用的自然沉底。检索时只取decay_score高于阈值的记录能有效控制噪声。2.3 检索策略不只是向量相似度很多人做记忆检索就是“把query转成向量然后top-k”。这在简单场景下能用但在hindsight场景里不够。因为Agent需要的记忆往往和当前任务有时间关联、因果关联、工具关联。我实际用的检索策略是多路召回重排序第一路是向量相似度召回取top-20。第二路是时间窗口召回取最近N条同类型任务的记忆。第三路是工具关联召回如果当前任务要用到某个工具把历史上用过这个工具的记忆捞出来。三路合并去重后用一个轻量级的重排序模型比如bge-reranker打分取top-5注入上下文。这个策略听起来复杂但实现起来并不难。关键是每一路的召回数量要控制好太多会拖慢响应太少会漏掉关键记忆。我的经验值是向量路20条时间路10条工具路10条合并后重排取5条。整体检索延迟控制在200ms以内。3. 用MCP协议打通Agent与记忆服务3.1 MCP是什么为什么它适合做记忆层MCPModel Context Protocol这两年在Agent圈子里热度很高。简单说它是一套标准化的协议让LLM能够以统一的方式调用外部工具和数据源。你可以把它理解成“AI世界的USB接口”——不管背后是什么服务只要实现了MCP协议LLM就能即插即用。为什么hindsight适合用MCP来做因为记忆服务本质上就是一个“工具”Agent需要的时候去查查完了拿回来用。用MCP封装记忆服务有几个好处解耦记忆服务的实现和Agent框架无关换LLM、换框架都不用改记忆层。可组合记忆服务可以和别的MCP工具比如Playwright MCP、BurpSuite MCP一起挂载Agent统一调度。标准化MCP定义了工具描述、参数schema、返回格式省去了自己设计接口的麻烦。我现在的项目里记忆服务就是一个独立的MCP ServerAgent通过MCP Client调用它。下面讲讲具体怎么搭。3.2 搭建记忆MCP Server的完整步骤先说一下环境。我用的是Docker Desktop来跑各种服务因为记忆服务需要数据库用Docker管理依赖最省心。如果你在Windows上装Docker Desktop遇到“Virtualization support not detected”的报错大概率是BIOS里的虚拟化没开进BIOS把Intel VT-x或AMD-V打开就行。这个坑我踩过折腾了半天才发现是BIOS设置问题。第一步起一个PostgreSQLpgvector的容器docker run -d \ --name hindsight-db \ -e POSTGRES_PASSWORDyourpassword \ -e POSTGRES_DBhindsight \ -p 5432:5432 \ -v hindsight_data:/var/lib/postgresql/data \ pgvector/pgvector:pg16这里用pgvector/pgvector:pg16这个镜像它已经预装了pgvector扩展省得自己编译。数据卷挂载到hindsight_data容器删了数据还在。第二步初始化数据库表结构CREATE EXTENSION IF NOT EXISTS vector; CREATE TABLE memories ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), agent_id VARCHAR(64) NOT NULL, session_id VARCHAR(64), memory_type VARCHAR(16) NOT NULL, content TEXT NOT NULL, embedding vector(768), metadata JSONB DEFAULT {}, access_count INT DEFAULT 0, decay_score FLOAT DEFAULT 1.0, created_at TIMESTAMPTZ DEFAULT NOW(), last_accessed TIMESTAMPTZ DEFAULT NOW() ); CREATE INDEX idx_memories_agent ON memories(agent_id); CREATE INDEX idx_memories_embedding ON memories USING ivfflat (embedding vector_cosine_ops) WITH (lists 100); CREATE INDEX idx_memories_metadata ON memories USING gin(metadata);向量维度768是因为我用的embedding模型输出是768维。如果你用OpenAI的text-embedding-3-small维度是1536记得改。ivfflat索引的lists参数一般取数据量的平方根初期数据少取100就够了。第三步写MCP Server。我用Python的mcp库来实现from mcp.server import Server from mcp.types import Tool, TextContent import asyncpg import json app Server(hindsight-memory) app.list_tools() async def list_tools(): return [ Tool( namestore_memory, description存储一条Agent记忆, inputSchema{ type: object, properties: { agent_id: {type: string}, content: {type: string}, memory_type: {type: string, enum: [episodic, semantic]}, metadata: {type: object} }, required: [agent_id, content] } ), Tool( nameretrieve_memory, description检索相关记忆, inputSchema{ type: object, properties: { agent_id: {type: string}, query: {type: string}, top_k: {type: integer, default: 5} }, required: [agent_id, query] } ) ] app.call_tool() async def call_tool(name: str, arguments: dict): conn await asyncpg.connect(postgresql://postgres:yourpasswordlocalhost:5432/hindsight) try: if name store_memory: embedding await get_embedding(arguments[content]) await conn.execute( INSERT INTO memories (agent_id, content, memory_type, embedding, metadata) VALUES ($1, $2, $3, $4, $5), arguments[agent_id], arguments[content], arguments.get(memory_type, episodic), embedding, json.dumps(arguments.get(metadata, {})) ) return [TextContent(typetext, text记忆已存储)] elif name retrieve_memory: query_embedding await get_embedding(arguments[query]) rows await conn.fetch( SELECT content, metadata, decay_score, 1 - (embedding $1) AS similarity FROM memories WHERE agent_id $2 AND decay_score 0.3 ORDER BY embedding $1 LIMIT $3, query_embedding, arguments[agent_id], arguments.get(top_k, 5) ) results [dict(r) for r in rows] return [TextContent(typetext, textjson.dumps(results, ensure_asciiFalse))] finally: await conn.close()这段代码的核心逻辑很直白store_memory把内容向量化后存库retrieve_memory把查询向量化后做相似度检索。是pgvector的余弦距离操作符1 - distance就是相似度。第四步把MCP Server注册到Agent框架。如果你用的是支持MCP的客户端比如某些IDE或Agent平台在配置里加上这个Server的启动命令就行。配置格式大概是{ mcpServers: { hindsight: { command: python, args: [path/to/hindsight_server.py], env: { DATABASE_URL: postgresql://postgres:yourpasswordlocalhost:5432/hindsight } } } }3.3 记忆注入的时机与格式MCP Server搭好了下一个问题是什么时候调用检索检索回来的记忆怎么塞进prompt我的做法是在Agent的system prompt里预留一个记忆槽位每次任务开始前先调一次retrieve_memory把结果格式化后填进去。格式大概是这样## 相关历史经验以下是你过去处理类似任务时的记录供参考 1. [2024-03-10] 任务修复Python依赖冲突方案使用pip-compile锁定版本结果成功耗时3轮对话 2. [2024-03-12] 任务修复ImportError 方案检查虚拟环境激活状态结果成功耗时1轮对话注意记忆注入不是越多越好。我试过塞10条进去结果Agent反而犹豫不决因为它看到了太多“可能相关”的方案。5条是个比较舒服的数量而且要按照相似度排序最相关的放最前面。还有一个细节记忆的表述要用“任务-方案-结果”的结构化格式不要直接塞原始对话。原始对话噪声太大LLM提取关键信息的成本很高。我在存储阶段就做了一层摘要把对话压缩成结构化记录检索出来直接能用。4. 记忆的写入、衰减与冲突处理4.1 什么值得记写入策略的设计不是所有交互都值得存成记忆。如果Agent每说一句话就存一条数据库很快就爆了检索质量也会被稀释。我用的写入策略是事件驱动价值过滤。事件驱动的意思是只在特定事件发生时触发写入任务完成时成功或失败用户给出明确反馈时“对了”/“不对”Agent使用了工具并得到结果时遇到错误并解决时价值过滤则是在写入前做一个快速判断这条记忆是否包含可复用的信息判断逻辑可以用一个轻量级的LLM调用也可以用规则。我的规则是这样的def should_store(interaction): # 包含工具调用结果的存 if interaction.get(tool_results): return True # 用户明确反馈的存 if interaction.get(user_feedback) in [positive, negative]: return True # 任务状态变化的存 if interaction.get(task_status) in [completed, failed]: return True # 纯闲聊不存 if interaction.get(task_type) chitchat: return False # 其他情况看内容长度和是否包含关键实体 return len(interaction.get(content, )) 200这套规则跑下来写入量大概只有原始交互的15%左右但检索命中率反而更高了。少即是多在记忆系统里体现得特别明显。4.2 记忆衰减让系统学会“忘记”前面提到了decay_score这里展开讲讲衰减机制的设计。为什么需要衰减因为Agent的记忆如果只增不减会出现两个问题一是检索噪声越来越大二是过时的信息会误导当前决策。我的衰减策略是时间衰减访问增强的组合import math from datetime import datetime, timedelta def update_decay_scores(conn): # 每天跑一次更新所有记忆的衰减分数 rows conn.fetch(SELECT id, decay_score, last_accessed, access_count FROM memories) for row in rows: days_since (datetime.now() - row[last_accessed]).days # 基础衰减半衰期约14天 base row[decay_score] * math.exp(-0.05 * days_since) # 访问增强每次访问加0.1上限1.0 bonus min(row[access_count] * 0.1, 0.5) new_score min(base bonus, 1.0) conn.execute(UPDATE memories SET decay_score $1 WHERE id $2, new_score, row[id])半衰期14天的意思是一条记忆如果两周没被访问它的分数会降到原来的约50%。这个参数可以根据业务调整如果是长期项目助手半衰期可以设长一点比如30天如果是实时客服可以短一点7天。检索时加一个decay_score 0.3的过滤条件低于这个阈值的记忆就不会被召回。但它们还在数据库里只是“沉底”了。如果某天突然被访问到分数会回升。这种设计比直接删除更优雅保留了“想起来”的可能性。4.3 记忆冲突当新旧信息矛盾时怎么办这是hindsight系统里最棘手的问题之一。假设Agent上周记住“这个项目的测试命令是pytest tests/”这周项目重构了测试命令变成了pytest test/。两条记忆矛盾检索时都召回了Agent该信哪个我的处理方案是时间优先显式覆盖时间优先很简单检索结果按时间倒序排列新的在前。但光靠时间不够因为有些旧记忆是“长期有效”的比如“这个项目的代码风格用black格式化”。显式覆盖是指当Agent明确知道某条旧记忆失效时主动调用一个invalidate_memory工具把旧记忆的decay_score直接设为0并标记metadata.invalidated_by指向新记忆的ID。这样旧记忆就不会再被召回。app.call_tool() async def invalidate_memory(memory_id: str, reason: str): await conn.execute( UPDATE memories SET decay_score 0, metadata metadata || {invalidated: true, reason: ...} WHERE id $1, memory_id )在实际运行中我会在system prompt里加一条指令“如果你发现历史记忆与当前情况矛盾请调用invalidate_memory标记旧记忆。”这样Agent自己就能维护记忆的一致性。5. 实战中的问题排查与性能调优5.1 常见问题速查表问题现象可能原因排查步骤解决方案检索结果不相关embedding模型不匹配检查存储和查询是否用同一模型统一embedding模型重建索引检索延迟高向量索引未生效EXPLAIN ANALYZE看是否走索引重建ivfflat索引调整lists参数记忆重复写入事件触发条件太宽查日志看哪些事件触发了写入收紧should_store规则加去重逻辑Agent忽略记忆注入位置不对检查记忆是否在system prompt靠前位置把记忆放在system prompt开头Docker容器连不上数据库网络配置问题docker exec进容器ping数据库用host.docker.internal或自定义网络MCP工具调用失败schema不匹配看MCP Client的报错日志检查inputSchema的required字段5.2 性能调优从500ms到80ms的优化过程最初版本的检索延迟在500ms左右对于交互式Agent来说太慢了。我做了几轮优化第一轮加向量索引。之前是全表扫描加了ivfflat索引后降到200ms。注意ivfflat索引需要在数据量达到一定规模后重建否则效果不好。我的经验是数据量超过1000条后重建一次。第二轮连接池。之前每次请求都新建数据库连接开销很大。换成asyncpg的连接池后降到120ms。pool await asyncpg.create_pool( postgresql://..., min_size5, max_size20 )第三轮embedding缓存。相同的query不重复计算embedding用LRU缓存存最近1000条query的向量。这一下降到80ms。第四轮预取。在Agent开始任务时根据任务类型预取一批记忆到内存后续检索先查内存。这个优化对多轮任务效果明显但实现复杂度高我目前还没上。5.3 几个容易踩的坑坑一embedding维度不一致。我一开始用某个模型存了768维的向量后来换了个模型输出1024维结果检索直接报错。教训是embedding模型一旦确定不要轻易换。要换就得全量重建。坑二metadata字段的JSONB索引。PostgreSQL的JSONB很好用但如果不建GIN索引按metadata过滤会很慢。我一开始没建查metadata-task_type code_fix的时候全表扫描加了CREATE INDEX ... USING gin(metadata)之后快了一个数量级。坑三Docker网络。如果你把数据库和MCP Server分别放在不同容器里它们默认不在同一网络互相访问不了。解决办法是创建一个自定义网络docker network create hindsight-net docker run --network hindsight-net --name hindsight-db ... docker run --network hindsight-net --name hindsight-mcp ...然后在MCP Server里用容器名hindsight-db作为主机名连接数据库。坑四MCP工具描述太模糊。MCP协议里每个工具都有description这个描述是给LLM看的。如果写得太模糊LLM不知道该什么时候调用。我的做法是在description里写清楚使用场景“当需要回忆过去处理类似任务的经验时调用此工具”。加上这句话后Agent主动调用检索的频率明显提高。6. 记忆系统的扩展方向从hindsight到proactivehindsight解决的是“回头看”的问题但一个成熟的Agent记忆系统还应该具备“向前看”的能力。最近有个研究方向叫a-memguard思路是在记忆写入前做一层主动防御过滤掉可能有害或误导性的记忆。这个思路我觉得很有价值因为Agent的记忆一旦被污染后续所有决策都会受影响。我在自己的项目里做了一个简化版的防御层写入前用一个分类模型判断这条记忆是否包含“绝对化表述”比如“永远不要用X”、“X一定是错的”如果有就降低它的初始decay_score或者要求人工确认。这个机制拦住了不少因为单次失败就产生的过度泛化记忆。另一个扩展方向是记忆的本体化。现在我的记忆都是扁平的文本向量检索靠相似度。但如果把记忆组织成本体ontology结构比如“任务类型-工具-结果”的三元组图谱就能做更精确的推理。比如Agent可以问“有哪些任务用了pip工具但失败了”这种查询用向量检索很难做好但用图查询就很自然。不过本体化也有代价就是写入复杂度高需要预先定义schema。我的建议是先用扁平结构跑起来等记忆量上到万级、检索质量遇到瓶颈时再考虑引入本体层。过早优化是万恶之源在记忆系统里同样适用。最后分享一个我在实际使用中的小技巧定期导出记忆库用LLM做一次“记忆整理”把碎片化的情景记忆归纳成语义记忆。比如把10条“修复ImportError”的记录归纳成一条“ImportError的常见原因和解决方案”。这样既能压缩存储又能提升检索质量。我一般每两周跑一次整理任务效果不错。

Agent记忆系统实战：基于MCP协议与pgvector实现hindsight记忆架构

相关推荐

Agent Memory实战：基于MCP与Docker构建LLM长期记忆系统

hindsight视角下的Agent Memory：三层记忆架构与检索优化实践

Vue3 + Nginx 在 Windows 部署的三大核心冲突与配置原理

更多资讯

视觉SLAM全流程实战：从算法原理到嵌入式部署与调优

从零手搓AI工程：数据管线、模型推理优化与FastAPI服务部署实战

护网行动红蓝对抗实战：从攻击链路到防御体系的完整指南

Model-Optimizer：大模型GPU推理的工程方法论与实战调优

大模型推理优化实战：TensorRT、vLLM与Model-Optimizer工程方法论

TensorFlow 2024实操指南：从安装到部署的完整避坑手册

今日更新

MATLAB+Yalmip+CPLEX实战：综合能源系统优化调度全流程解析

I3C比I2C快10倍？RK3576实战：速率、DTS配置与混合总线避坑指南

字符串转对象：JSON.parse、new Function与URLSearchParams

本周精选

如何彻底离线使用 Laya-CoreML：Hugging Face 下载、缓存机制与 local_files_only 详解

基于1Panel 部署MySQL MCP 服务：TaoToken 统一 Key 接入与 config.toml 配置实战

崇州园区营销网站建设避坑指南：5大注意事项保备案安全

本月热门

基于YOLO的焊缝缺陷检测：从数据标注到工业部署全流程实践

分布式服务的决策记录

创业产品的首版范围

读完文章，想聊聊您的网站？