ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

Agent Memory实战:从向量检索到MCP结构化记忆

Agent Memory实战:从向量检索到MCP结构化记忆 1. 从“hindsight”说起为什么我们需要给Agent装上记忆第一次看到“hindsight”这个词是在一个做LLM Agent的朋友群里。有人丢了一张截图说他们的Agent在连续对话到第37轮的时候突然把用户三小时前说过的偏好设置忘得一干二净回复开始胡言乱语。底下有人回了一句“这不就是典型的hindsight缺失么——事后看它本该记住的。”这个词本身的意思是“事后聪明”但放在Agent Memory这个语境里它指的是一套让Agent能够回溯、检索、利用历史交互信息的机制。说白了就是让Agent别像金鱼一样七秒钟之后什么都不记得。我过去大半年一直在折腾LLM Agent的记忆系统从最粗暴的“把历史对话全塞进context window”到后来用向量数据库做语义检索再到最近在试的MCP协议下的结构化记忆方案。踩过的坑不少也攒了一些还算能用的经验。这篇文章不打算写成学术综述就是把我对Agent Memory这套东西的理解、实操中遇到的问题、以及目前觉得比较靠谱的方案从头到尾捋一遍。如果你正在做Agent相关的开发或者单纯对“怎么让大模型记住事儿”这个问题感兴趣那下面的内容应该能帮你省下不少试错时间。我会尽量把每个技术选择的理由讲清楚把参数怎么算、代码怎么写、坑在哪里都说明白。文章里涉及到的工具和框架都是我自己实际跑过的不是纸上谈兵。2. Agent Memory到底在解决什么问题2.1 从“金鱼记忆”到“长期记忆”的鸿沟大语言模型本身是没有记忆的。每次你调用API它都是从一个全新的状态开始你给它什么context它就只能基于这个context来生成回复。这就像你每次跟一个人说话他都会失忆一次你得把所有背景重新讲一遍。短对话还好context window塞得下。但一旦对话轮次多了或者你需要Agent跨会话记住用户的偏好、历史决策、项目状态问题就来了。最直接的做法是把所有历史对话拼成一个超长字符串塞进去但这样做有三个致命问题第一token成本线性增长聊到后面每轮请求都在烧钱第二context window有上限超了就截断截断就意味着丢失第三即使塞得下模型对长context中间部分的注意力也会衰减这就是所谓的“lost in the middle”现象。Agent Memory要解决的核心问题就是如何在有限的context预算下让Agent能够访问到最相关的历史信息。这本质上是一个检索问题而不是简单的存储问题。2.2 三种记忆类型的区分在动手之前得先把记忆分个类。我习惯把它分成三种工作记忆Working Memory当前对话轮次内的上下文包括最近几轮的用户输入和Agent回复。这部分通常直接放在context里不需要额外检索。一般保留最近5到10轮就够了具体看任务复杂度。情景记忆Episodic Memory过去发生过的具体事件比如“用户上周三说过他喜欢用Python而不是JavaScript”、“上次部署时遇到了Docker网络不通的问题”。这部分需要持久化存储并且能够按时间或语义检索。语义记忆Semantic Memory从多次交互中抽象出来的通用知识比如“这个用户偏好简洁的代码风格”、“这个项目的技术栈是ReactNode”。这部分通常需要一定的归纳和总结不是简单存储原始对话。这三种记忆的处理策略完全不同。工作记忆靠滑动窗口情景记忆靠向量检索语义记忆靠定期总结和更新。很多Agent Memory方案之所以效果不好就是因为把这三者混在一起处理了。2.3 为什么现在这个问题变得重要了两年前大家做Demo的时候Agent能完成一轮任务就行没人关心它能不能记住。但现在Agent开始往生产环境走要处理持续数天甚至数周的任务要服务同一个用户成百上千次交互记忆就成了刚需。再加上MCP协议的推广Agent可以调用的工具越来越多每次工具调用的结果都需要被记录和检索。没有一套靠谱的记忆系统Agent就像是一个每次上班都失忆的员工永远在重复问同样的问题。3. 核心方案选型从朴素到结构化3.1 方案一全量上下文Naive Approach这是最原始的做法把所有历史对话拼成一个字符串每次请求都完整带上。优点是实现简单不需要任何额外组件。缺点是token成本高而且随着对话增长会迅速触及context上限。我实测过一个客服场景的Agent平均对话轮次在15轮左右全量上下文方案下每轮请求的token消耗从最初的800涨到了最后的12000。按GPT-4的定价算单次对话成本从0.03美元涨到了0.45美元。如果每天有1000次对话一天就是450美元。这个成本在生产环境是不可接受的。所以全量上下文只适合Demo或者对话轮次极少的场景。一旦要上生产必须做检索。3.2 方案二滑动窗口摘要Sliding Window Summary这个方案的核心思路是保留最近N轮完整对话作为工作记忆更早的对话压缩成摘要。摘要可以由LLM生成也可以由规则提取关键信息。具体实现上我一般设置N6也就是保留最近6轮完整对话。当对话轮次超过6轮时把最早的两轮对话送给LLM做摘要摘要结果追加到一个“历史摘要”字段里。每次请求时context由“历史摘要最近6轮对话”组成。这个方案比全量上下文省token但摘要本身也有成本而且摘要会丢失细节。我遇到过一个问题用户在第3轮提到过一个具体的错误码“ERR_CONNECTION_REFUSED”这个信息在摘要时被压缩成了“用户遇到了连接问题”导致后面Agent无法根据具体错误码给出精准的排查建议。所以摘要方案适合那些对细节要求不高的场景比如闲聊或者通用问答。如果是技术支持、代码调试这类需要精确信息的场景摘要方案不够用。3.3 方案三向量检索增强RAG-based Memory这是目前最主流的方案。把每一轮对话或者每一个关键信息片段做embedding存到向量数据库里。每次请求时用当前用户输入做query检索最相关的K条历史记忆拼到context里。这个方案的关键在于几个参数的选择Chunk大小我试过按轮次切分每轮对话一个chunk和按语义切分用LLM判断语义边界。按轮次切分实现简单但有时候一轮对话里包含多个独立信息点检索时可能只命中其中一个。按语义切分效果更好但需要额外的LLM调用成本和延迟都会增加。我目前的折中是按轮次切分但每轮对话的embedding用整轮文本生成检索时返回完整轮次。检索数量KK太小会漏掉相关信息K太大会引入噪声。我实测下来K3到5是比较平衡的值。具体取决于你的记忆库大小和任务复杂度。如果记忆库只有几百条K3够了如果记忆库上万条可能需要K5到8。相似度阈值不是所有检索结果都值得放进context。我一般设置一个相似度阈值比如0.75低于这个值的直接丢弃。这样可以避免把不相关的记忆塞进去干扰模型。向量检索方案的效果比摘要方案好很多但也不是没有问题。最大的问题是embedding模型对“时间”不敏感。用户说“我上次说的那个方案”embedding可能检索到三个月前的一次对话而实际上用户指的是昨天说的。所以纯向量检索需要配合时间衰减因子越近的记忆权重越高。3.4 方案四MCP协议下的结构化记忆这是我最近在试的方案也是我觉得最有前景的方向。MCPModel Context Protocol本质上是一个让LLM和外部工具、数据源交互的协议。把Agent Memory做成一个MCP ServerAgent通过MCP协议来读写记忆。这个方案的好处是记忆的存储和检索逻辑被封装在MCP Server里Agent本身不需要关心具体实现。你可以随时替换底层的存储引擎从SQLite换成Postgres从内存换成Redis只要MCP接口不变Agent代码就不用改。更重要的是MCP协议支持结构化的工具调用。你可以定义这样的工具memory_store(key, value, metadata)存储一条记忆memory_retrieve(query, top_k, time_range)检索记忆memory_forget(key)删除记忆memory_summarize(time_range)对指定时间范围的记忆做摘要这种结构化的方式比纯向量检索更可控。你可以精确地指定要检索哪个时间范围、哪种类型的记忆而不是依赖embedding的模糊匹配。我目前用Docker跑了一个MCP Memory Server底层用SQLite做持久化用ChromaDB做向量索引。Agent通过MCP协议调用实测下来延迟在50ms左右完全可以接受。4. 实操从零搭建一个Agent Memory系统4.1 环境准备与依赖安装我假设你用的是Linux或者macOSWindows的话建议用WSL2。Docker是必须的因为我们要用Docker来跑MCP Server和向量数据库。先装Docker。Ubuntu下用官方脚本curl -fsSL https://get.docker.com -o get-docker.sh sudo sh get-docker.sh sudo usermod -aG docker $USER装完之后记得重新登录让用户组生效。然后验证一下docker --version docker run hello-world如果hello-world跑不起来大概率是网络问题。国内环境的话需要配置镜像加速。编辑/etc/docker/daemon.json{ registry-mirrors: [https://your-mirror-url] }然后重启Docker服务sudo systemctl restart docker接下来装Python依赖。我用的Python 3.11主要依赖这几个包pip install mcp chromadb openai tiktokenmcp是MCP协议的Python SDKchromadb是向量数据库openai用来调embedding和LLMtiktoken用来算token数。4.2 记忆存储层设计存储层我分两部分结构化存储用SQLite向量索引用ChromaDB。SQLite的表结构很简单CREATE TABLE memories ( id TEXT PRIMARY KEY, content TEXT NOT NULL, memory_type TEXT NOT NULL, -- episodic or semantic created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, last_accessed_at TIMESTAMP, access_count INTEGER DEFAULT 0, metadata TEXT -- JSON string );memory_type区分情景记忆和语义记忆。access_count和last_accessed_at用来做时间衰减和热度排序。ChromaDB的collection初始化import chromadb client chromadb.PersistentClient(path./memory_db) collection client.get_or_create_collection( nameagent_memory, metadata{hnsw:space: cosine} )这里用的是cosine距离因为embedding的语义相似度用cosine更合适。HNSW是ChromaDB默认的索引算法对于万级数据量来说性能足够。4.3 记忆写入流程每次对话结束后把这一轮的用户输入和Agent回复拼成一个文本块做embedding然后写入SQLite和ChromaDB。import uuid import json from datetime import datetime def store_memory(user_input, agent_response, memory_typeepisodic, metadataNone): memory_id str(uuid.uuid4()) content fUser: {user_input}\nAgent: {agent_response} # 写入SQLite conn.execute( INSERT INTO memories (id, content, memory_type, metadata) VALUES (?, ?, ?, ?), (memory_id, content, memory_type, json.dumps(metadata or {})) ) conn.commit() # 写入ChromaDB embedding get_embedding(content) collection.add( ids[memory_id], embeddings[embedding], documents[content], metadatas[{memory_type: memory_type, created_at: datetime.now().isoformat()}] ) return memory_id这里有个细节embedding我用的是OpenAI的text-embedding-3-small1536维。实测下来比text-embedding-ada-002效果好而且便宜。如果你要本地跑可以用sentence-transformers的all-MiniLM-L6-v2384维效果差一些但免费。4.4 记忆检索流程检索的时候用当前用户输入做query从ChromaDB里找最相似的K条然后根据时间衰减和访问频率做重排序。import numpy as np from datetime import datetime, timedelta def retrieve_memories(query, top_k5, time_decay_factor0.1): query_embedding get_embedding(query) results collection.query( query_embeddings[query_embedding], n_resultstop_k * 2 # 多取一些后面重排序 ) memories [] for i, mem_id in enumerate(results[ids][0]): distance results[distances][0][i] similarity 1 - distance # cosine距离转相似度 # 获取元数据 meta results[metadatas][0][i] created_at datetime.fromisoformat(meta[created_at]) days_old (datetime.now() - created_at).days # 时间衰减 time_weight np.exp(-time_decay_factor * days_old) # 综合得分 score similarity * 0.7 time_weight * 0.3 memories.append({ id: mem_id, content: results[documents][0][i], score: score, similarity: similarity, days_old: days_old }) # 按综合得分排序 memories.sort(keylambda x: x[score], reverseTrue) # 更新访问记录 for mem in memories[:top_k]: conn.execute( UPDATE memories SET last_accessed_at ?, access_count access_count 1 WHERE id ?, (datetime.now(), mem[id]) ) conn.commit() return memories[:top_k]这里的time_decay_factor控制时间衰减的速度。0.1意味着每天衰减约10%一周前的记忆权重降到约50%。这个值可以根据你的场景调整。如果是长期项目助手可以调小到0.05如果是短期客服可以调大到0.2。4.5 用MCP协议封装记忆服务上面是核心逻辑但要让它能被Agent调用需要封装成MCP Server。MCP的Python SDK用起来很简单from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent app Server(memory-server) app.list_tools() async def list_tools(): return [ Tool( namememory_store, description存储一条记忆, inputSchema{ type: object, properties: { content: {type: string}, memory_type: {type: string, enum: [episodic, semantic]} }, required: [content] } ), Tool( namememory_retrieve, description检索相关记忆, inputSchema{ type: object, properties: { query: {type: string}, top_k: {type: integer, default: 5} }, required: [query] } ) ] app.call_tool() async def call_tool(name, arguments): if name memory_store: mem_id store_memory( arguments[content], , arguments.get(memory_type, episodic) ) return [TextContent(typetext, textfStored memory {mem_id})] elif name memory_retrieve: memories retrieve_memories( arguments[query], arguments.get(top_k, 5) ) result \n---\n.join([m[content] for m in memories]) return [TextContent(typetext, textresult)] async def main(): async with stdio_server() as (read, write): await app.run(read, write, app.create_initialization_options()) if __name__ __main__: import asyncio asyncio.run(main())这个Server可以通过stdio方式被Agent调用也可以打包成Docker镜像通过SSE方式调用。我一般用Docker跑因为方便管理依赖和环境。Dockerfile很简单FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . CMD [python, server.py]构建和运行docker build -t memory-server . docker run -d --name memory-server -v ./data:/app/data memory-server4.6 与Agent集成Agent这边以Claude Desktop为例在配置文件里加上MCP Server的地址{ mcpServers: { memory: { command: docker, args: [run, -i, --rm, memory-server] } } }这样Agent就能通过MCP协议调用memory_store和memory_retrieve了。每次对话开始时Agent先调memory_retrieve获取相关记忆拼到system prompt里对话结束后调memory_store把本轮对话存进去。5. 踩坑记录与常见问题排查5.1 记忆检索不准确怎么办这是最常见的问题。Agent检索出来的记忆跟当前对话不相关导致回复跑偏。排查思路如下检查embedding质量先用几个已知相关的query和memory做测试看相似度分数是否合理。如果相关内容的相似度低于0.7说明embedding模型不适合你的领域。可以考虑换模型或者用领域数据做fine-tune。检查chunk切分如果一轮对话很长embedding会被平均掉导致检索时匹配不到关键信息。解决办法是把长对话切成多个chunk每个chunk单独embedding。我一般按每200个token切一刀重叠50个token。调整检索参数K值和相似度阈值需要根据实际数据调。我建议先设K10阈值0.6然后人工检查检索结果逐步收紧。加入重排序ChromaDB的检索是基于向量相似度的但向量相似度不等于语义相关性。可以加一个cross-encoder做重排序比如cross-encoder/ms-marco-MiniLM-L-6-v2。实测能提升10%到15%的准确率。5.2 Docker网络不通导致MCP Server连不上这个问题我遇到好几次了。Agent跑在宿主机上MCP Server跑在Docker里两者网络不通。排查步骤先确认Docker容器是否在运行docker ps进入容器测试网络docker exec -it memory-server ping 8.8.8.8如果容器内网络不通检查Docker的DNS配置docker run --rm alpine cat /etc/resolv.conf如果DNS有问题在daemon.json里加上dns: [8.8.8.8, 114.114.114.114]如果容器网络正常但宿主机连不上检查端口映射docker run -p 8080:8080 ...如果用stdio方式确保docker run加了-i参数否则stdin不通还有一个坑Windows下Docker Desktop有时候会报“Virtualization support not detected”。这是因为Hyper-V或者WSL2没启用。解决办法是在BIOS里开启虚拟化支持然后在Windows功能里启用“虚拟机平台”和“适用于Linux的Windows子系统”。5.3 记忆库膨胀导致检索变慢跑了一段时间后记忆库可能积累了几万条记录检索延迟从50ms涨到了500ms。解决办法定期归档超过30天的情景记忆如果访问次数少于3次移到归档表里不参与实时检索。需要的时候再手动查。分层索引最近7天的记忆用HNSW索引更早的用IVF索引。HNSW快但内存占用高IVF慢但省内存。限制检索范围检索时加上时间范围过滤比如只检索最近90天的记忆。ChromaDB支持where条件results collection.query( query_embeddings[query_embedding], n_resultstop_k, where{created_at: {$gte: (datetime.now() - timedelta(days90)).isoformat()}} )定期压缩每周跑一次任务把相似度高于0.95的记忆合并成一条减少冗余。5.4 常见问题速查表问题现象可能原因排查方法解决方案检索结果不相关embedding模型不匹配测试已知相关对的相似度换模型或fine-tune检索延迟高记忆库过大查看collection.count()归档旧记忆、加时间过滤Docker容器启动失败端口冲突或镜像问题docker logs查看日志换端口、重建镜像MCP连接超时网络配置问题容器内ping测试配DNS、检查端口映射记忆重复存储缺少去重逻辑检查content哈希存储前做相似度去重token超限检索结果太长统计拼接后的token数限制单条记忆长度、减少K值6. 进阶优化让记忆系统更聪明6.1 记忆的重要性评分不是所有记忆都同等重要。用户说“你好”和用户说“我的API key是sk-xxx”这两条记忆的价值天差地别。我加了一个重要性评分机制用LLM对每条记忆打分1到10分检索时把重要性作为权重之一。评分prompt大概是这样请对以下对话片段的重要性打分1-10分 - 10分包含关键配置、密码、重要决策 - 7-9分包含用户偏好、项目状态、技术选型 - 4-6分一般性讨论、问答 - 1-3分寒暄、闲聊、无信息量内容 对话内容{content} 只输出分数不要解释。这个评分在存储时算一次存到metadata里。检索时综合得分变成similarity * 0.5 time_weight * 0.2 importance * 0.3。6.2 语义记忆的自动归纳情景记忆是原始对话语义记忆是从中抽象出来的知识。我每周跑一次归纳任务把过去一周的情景记忆送给LLM让它提取出通用的用户偏好和项目知识。归纳prompt以下是过去一周的对话记录请提取出关于用户的长期偏好和项目状态的关键信息。 输出格式 - 用户偏好... - 项目状态... - 待办事项... 对话记录{memories}归纳结果作为语义记忆存储memory_type设为semantic。检索时语义记忆的权重比情景记忆高因为它们更精炼、更通用。6.3 记忆的遗忘机制人脑会遗忘Agent的记忆系统也应该有遗忘机制。否则记忆库会无限膨胀而且旧的不相关信息会干扰检索。我的遗忘策略是超过90天且访问次数为0的情景记忆直接删除超过180天且访问次数少于3次的情景记忆归档语义记忆不自动删除但每季度人工review一次删除之前会先做一次摘要把关键信息保留下来。这样既控制了记忆库大小又不会丢失重要信息。6.4 多Agent共享记忆如果你有多个Agent协同工作记忆共享是个问题。我的做法是每个Agent有自己的私有记忆库同时有一个共享记忆库。私有记忆存Agent自己的对话历史共享记忆存跨Agent的公共知识。共享记忆的写入需要权限控制不是所有Agent都能写。我一般只允许“协调者”角色的Agent写共享记忆其他Agent只读。MCP协议天然支持这种模式每个Agent连自己的Memory Server协调者Agent额外连一个Shared Memory Server。通过MCP的权限控制可以实现细粒度的访问管理。7. 一些实测数据与个人体会跑了大半年下来有几个数据可以分享。token节省相比全量上下文方案向量检索方案在15轮对话的场景下token消耗降低了约70%。全量方案平均每轮8000 token检索方案平均每轮2400 token。检索准确率在技术支持场景下Top-3检索的命中率约为82%。加上重排序后提升到89%。还有提升空间主要瓶颈在embedding模型对技术术语的理解。延迟MCP Server的检索延迟平均在45msembedding生成延迟在120ms用OpenAI API。如果换本地embedding模型生成延迟可以降到20ms以内但准确率会下降约5%。成本embedding成本可以忽略不计text-embedding-3-small每百万token只要0.02美元。主要成本在LLM调用上但相比全量上下文方案总体成本还是降低了不少。我个人觉得Agent Memory这个方向还有很多可以挖的。目前的结构化记忆向量检索方案已经能解决80%的问题但在记忆的自动归纳、跨模态记忆比如记住图片和音频、记忆的因果推理这些方面还有很大的探索空间。如果你也在做类似的东西建议先从最简单的向量检索方案开始跑通了再逐步加复杂度。不要一上来就搞太复杂的架构很多优化在数据量小的时候根本看不出效果。先把基础流程跑通积累一些真实数据再针对性地优化。
返回列表