ARTICLE DETAIL

资讯详情

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

Agent记忆系统实战:基于MCP与Docker的hindsight方案

Agent记忆系统实战:基于MCP与Docker的hindsight方案 1. 从“hindsight”说起为什么记忆是 Agent 落地的最后一公里第一次看到 “hindsight” 这个词是在一个做智能体Agent的朋友群里。有人丢了一张截图说他们的 Agent 在连续对话到第 40 轮之后开始“胡言乱语”明明前面用户已经说过自己不吃辣结果推荐餐厅的时候还是甩出一堆川菜馆。底下有人回了一句“这不就是典型的 hindsight 缺失嘛——事后看什么都明白当时就是记不住。”这个吐槽其实点到了当前 LLM Agent 落地最要命的一个环节记忆。模型本身的推理能力、工具调用能力在过去一年进步飞快MCP 协议把工具接入标准化了Docker 把环境部署标准化了但“Agent 记不住事”这个问题始终没有一个特别优雅的通用解。hindsight 这个项目标题我理解它想解决的就是这件事——让 Agent 拥有一种“回头看”的能力把历史交互、工具调用结果、用户偏好这些东西沉淀下来在需要的时候精准召回。我前后折腾过几套 Agent 记忆方案从最粗暴的“把历史对话全塞进 context”到向量库检索再到最近比较火的 working memory 分层设计踩的坑不算少。这篇文章就把我对 hindsight 这类 Agent memory 方案的理解、拆解和实操完整讲一遍。不管你是刚接触 LLM 应用开发的新手还是已经在做 Agent 产品、被记忆问题折磨过的老手应该都能从里面找到能直接抄作业的部分。先说清楚这篇文章适合谁如果你正在用 MCP 接工具、用 Docker 部署服务、想让自己的 Agent 记住用户说过的话和做过的事那这篇就是写给你的。我会从设计思路讲到具体实现包括存储结构、召回策略、和 MCP/Docker 的配合方式以及我自己踩过的那些坑。2. Agent Memory 的整体设计与思路拆解2.1 为什么“全塞进 context”这条路走不通刚开始做 Agent 的时候最直觉的做法就是把所有历史对话拼成一个超长 prompt一股脑丢给模型。我最早的一个客服 Agent 就是这么干的前 10 轮效果很好到第 30 轮开始出问题响应变慢、成本飙升、模型开始忽略中间部分的内容。这不是模型不行而是注意力机制本身的特性决定的——context 越长中间信息的有效利用率越低业内管这个叫“lost in the middle”。算一笔账你就明白了。假设每轮对话平均 200 个 token40 轮就是 8000 token再加上系统提示词、工具定义、检索到的文档轻松破万。按主流模型的定价一次调用成本可能是短对话的十几倍。更要命的是这里面 90% 的历史信息在当前这轮根本用不上纯属浪费。所以 hindsight 这类方案的核心思路本质上是把“记忆”从 context 里剥离出来做成一个独立的、可检索、可管理的存储层。context 里只放当前真正需要的那部分记忆其余的存在外面需要的时候再捞回来。2.2 三层记忆结构working memory、episodic memory、semantic memory我目前用得比较顺的一套结构是把 Agent 的记忆分成三层这个划分参考了认知科学里对人类记忆的分类落地到工程上非常自然。第一层是 working memory工作记忆。这一层就是当前对话的“活动区”存放最近几轮对话、当前任务的状态、正在调用的工具参数。它的特点是容量小、读写快、生命周期短。一般我会把最近 5 到 10 轮对话放在这里超过就往下沉。第二层是 episodic memory情景记忆。这一层记录的是“发生过什么”比如用户上周问过退款流程、昨天查询过订单 A123、刚才调用天气工具返回了暴雨预警。它带时间戳按事件组织检索的时候往往结合时间衰减——越近的事越容易被想起来。第三层是 semantic memory语义记忆。这一层是提炼出来的“事实和偏好”比如“用户是素食主义者”“用户所在城市是杭州”“用户偏好简洁回答”。它不带具体时间是长期沉淀的结论通常由 episodic memory 经过总结归纳得到。这三层的分工用一句话概括就是working memory 管“现在”episodic memory 管“刚才”semantic memory 管“一直以来”。hindsight 的价值就在于它让 Agent 在回答之前能“回头看一眼”这三层里有什么相关的而不是每次都从零开始。2.3 为什么选 MCP Docker 这套组合热词里 MCP 和 Docker 出现频率极高这不是偶然。MCPModel Context Protocol本质上是给模型和外部能力之间定的一套标准接口你可以把它类比成“AI 世界的 USB-C”——不管你是数据库、文件系统还是某个 SaaS 服务只要实现 MCP 协议模型就能用统一的方式调用。记忆系统作为一个外部能力用 MCP 暴露出去是最自然的选择Agent 通过 MCP 调用memory_store、memory_recall这些工具跟调用其他工具没有任何区别。Docker 则是解决“记忆系统跑在哪”的问题。记忆系统通常需要一个向量库比如 Milvus、Qdrant、一个关系库存结构化的事实、可能还有一个缓存Redis。这些东西本地装一遍能折腾死人用 Docker Compose 一把梭环境隔离、版本固定、迁移方便。我现在的习惯是任何需要多个服务配合的项目先写 docker-compose.yml再写业务代码。提示MCP 是软件协议层面的标准不是硬件协议。硬件协议那个概念通常叫“总线协议”或“接口标准”比如 USB、PCIe。两者容易混淆但层级完全不同。3. 核心细节解析与实操要点3.1 记忆的写入什么时候存、存什么、怎么存记忆系统最容易做错的地方不是检索而是写入。我见过太多项目把所有对话无脑往库里灌结果检索出来的全是噪音。写入这一步必须做过滤和结构化。我的做法是分三个触发点来写对话轮次触发每完成一轮用户-Agent 交互把这一轮压缩成一条 episodic 记录字段包括时间戳、用户输入摘要、Agent 回复摘要、涉及的工具调用。任务完成触发当一个多步任务结束时比如用户的问题被彻底解决触发一次总结把这次任务里值得长期记住的事实提炼成 semantic 记录。显式触发用户明确说“记住我喜欢 XX”或者 Agent 判断某条信息是长期偏好直接写 semantic。存什么内容我一般会做一层“记忆抽取”用一个轻量 prompt 让模型输出结构化 JSON而不是直接存原始文本。比如{ type: semantic, key: user_dietary_preference, value: vegetarian, no spicy food, confidence: 0.9, source_turn: 42, created_at: 2025-01-15T10:30:00Z }这里key的设计很关键。热词里提到“LLM 的 token 三个点key 我是谁、query 我在找什么、value 我能提供什么”这个类比非常精准。记忆的 key 就是“这条记忆是关于什么的”query 是“当前我需要什么”value 是“这条记忆能提供什么”。三者对齐检索才准。3.2 记忆的召回向量检索 结构化过滤的混合策略召回是 hindsight 的核心能力。纯向量检索的问题在于它对“精确匹配”不敏感——用户问“我的订单 A123 到哪了”向量检索可能召回一堆关于订单的泛泛内容但就是漏掉 A123 这条精确记录。所以我的方案是混合检索结构化过滤先行如果 query 里能抽出明确的实体订单号、日期、人名先用关系库精确查一遍。向量检索补充把 query 编码成向量在 episodic 和 semantic 库里做相似度检索取 top-k。时间衰减加权episodic 记录按时间做衰减越近的权重越高公式大概是score similarity * exp(-λ * age)λ 取 0.01 到 0.05 之间具体看业务对“新鲜度”的敏感程度。重排序把前两步的结果合并用一个小的 rerank 模型或者规则打分选出最终注入 context 的 3 到 5 条。这套流程听起来复杂但用 MCP 封装成一个memory_recall工具之后Agent 侧只需要一次调用。我在实际项目里测过混合检索相比纯向量在“精确实体查询”场景下的命中率能提升 30% 以上。3.3 存储选型向量库、关系库、缓存各司其职存储这块我的标准配置是三个组件组件选型职责备注向量库Qdrant / Milvus存 episodic 和 semantic 的向量Qdrant 部署简单单机够用关系库PostgreSQL / MySQL存结构化事实、实体索引精确查询靠它缓存Redis存 working memory、热点记忆设 TTL自动过期用 Docker Compose 编排的话一个文件就能把这三个拉起来。这里有个细节向量库和关系库之间要有一个memory_id做关联保证同一条记忆在两个库里能对上。我一般用 UUID 做主键写入时先写关系库拿到 id再写向量库。注意向量维度和距离度量cosine / euclidean一旦选定后期换 embedding 模型会非常痛苦因为要全量重建索引。建议一开始就选一个稳定、长期可用的 embedding 模型别频繁换。3.4 和 MCP 的对接把记忆能力暴露成标准工具MCP 的好处是记忆系统不需要关心上层是哪个 Agent 框架。我通常暴露这几个工具memory_store(content, type, metadata)写入一条记忆memory_recall(query, top_k, filters)召回相关记忆memory_forget(memory_id)删除某条记忆用户要求“忘掉”时用memory_summarize(session_id)把一个会话总结成 semantic 记忆Agent 侧只要在系统提示词里说明“你有记忆工具在回答前先 recall”模型就会自己决定什么时候调用。实测下来配合 few-shot 示例模型调用记忆工具的准确率能到 85% 以上。4. 实操过程与核心环节实现4.1 用 Docker Compose 拉起记忆服务栈先上编排文件这是整个系统的地基。我把它拆成三个服务网络互通数据卷持久化。version: 3.9 services: qdrant: image: qdrant/qdrant:latest ports: - 6333:6333 volumes: - ./data/qdrant:/qdrant/storage restart: unless-stopped postgres: image: postgres:16 environment: POSTGRES_USER: memory POSTGRES_PASSWORD: memory_pass POSTGRES_DB: agent_memory ports: - 5432:5432 volumes: - ./data/pg:/var/lib/postgresql/data restart: unless-stopped redis: image: redis:7-alpine ports: - 6379:6379 command: redis-server --maxmemory 512mb --maxmemory-policy allkeys-lru restart: unless-stopped启动就一句docker compose up -d。这里有几个我踩过的坑一是 Windows 上装 Docker Desktop 经常报 “virtualization support not detected”要去 BIOS 里开虚拟化或者确认没和 Hyper-V/WSL2 冲突二是数据卷一定要挂出来不然容器一删数据全没三是 Redis 一定要设maxmemory-policy不然 working memory 写多了会把内存吃满。4.2 记忆写入的代码实现下面这段是写入逻辑的核心用 Python 写依赖qdrant-client、psycopg2、redis。我把它封装成一个MemoryWriter类。import uuid import json from datetime import datetime from qdrant_client import QdrantClient from qdrant_client.models import PointStruct import psycopg2 import redis class MemoryWriter: def __init__(self, embed_fn): self.qdrant QdrantClient(hostlocalhost, port6333) self.pg psycopg2.connect( hostlocalhost, dbnameagent_memory, usermemory, passwordmemory_pass ) self.redis redis.Redis(hostlocalhost, port6379, decode_responsesTrue) self.embed embed_fn # 传入你的 embedding 函数 def store(self, content, mem_type, metadataNone): mem_id str(uuid.uuid4()) metadata metadata or {} vector self.embed(content) # 1. 写关系库 with self.pg.cursor() as cur: cur.execute( INSERT INTO memories (id, content, type, metadata, created_at) VALUES (%s, %s, %s, %s, %s), (mem_id, content, mem_type, json.dumps(metadata), datetime.utcnow()) ) self.pg.commit() # 2. 写向量库 self.qdrant.upsert( collection_nameagent_memory, points[PointStruct( idmem_id, vectorvector, payload{type: mem_type, content: content, **metadata} )] ) # 3. 如果是 working memory同步写 Redis if mem_type working: self.redis.lpush(working_memory, mem_id) self.redis.ltrim(working_memory, 0, 9) # 只留最近 10 条 return mem_id这段代码里working memory用 Redis 的 list 存ltrim保证只保留最近 10 条超出的自动淘汰。episodic 和 semantic 走 Qdrant PostgreSQL 双写。注意双写不是事务性的如果向量库写失败关系库里会留一条孤儿记录所以生产环境要加一个补偿任务定期清理。4.3 记忆召回的混合检索实现召回这块是重头戏我把结构化过滤、向量检索、时间衰减、重排序串起来。import math from datetime import datetime class MemoryRetriever: def __init__(self, embed_fn, writer): self.embed embed_fn self.writer writer self.qdrant writer.qdrant self.pg writer.pg def recall(self, query, top_k5, mem_typesNone): results [] # 1. 结构化精确匹配如果 query 里有实体 entities self._extract_entities(query) if entities: with self.pg.cursor() as cur: cur.execute( SELECT id, content, type, created_at FROM memories WHERE content ILIKE ANY(%s) LIMIT 5, ([% e % for e in entities],) ) for row in cur.fetchall(): results.append({ id: row[0], content: row[1], type: row[2], created_at: row[3], score: 1.0, source: exact }) # 2. 向量检索 qvec self.embed(query) hits self.qdrant.search( collection_nameagent_memory, query_vectorqvec, limittop_k * 2, query_filterself._build_filter(mem_types) ) now datetime.utcnow() for h in hits: created datetime.fromisoformat(h.payload.get(created_at, now.isoformat())) age_days (now - created).total_seconds() / 86400 decay math.exp(-0.03 * age_days) results.append({ id: h.id, content: h.payload[content], type: h.payload[type], created_at: created, score: h.score * decay, source: vector }) # 3. 去重 重排序 seen set() deduped [] for r in sorted(results, keylambda x: x[score], reverseTrue): if r[id] not in seen: seen.add(r[id]) deduped.append(r) return deduped[:top_k] def _extract_entities(self, query): # 简化版正则抽订单号、日期等生产环境可用 NER 模型 import re patterns [r[A-Z]\d{6,}, r\d{4}-\d{2}-\d{2}] found [] for p in patterns: found.extend(re.findall(p, query)) return found def _build_filter(self, mem_types): if not mem_types: return None from qdrant_client.models import Filter, FieldCondition, MatchValue return Filter(must[ FieldCondition(keytype, matchMatchValue(valuet)) for t in mem_types ])时间衰减系数我取 0.03意思是大约 23 天权重衰减到一半。这个值不是拍脑袋来的如果你的业务是客服用户偏好可能几个月都有效系数可以调小到 0.01如果是实时性很强的场景比如股票咨询系数可以调到 0.1 甚至更高。4.4 通过 MCP 暴露记忆工具MCP 服务端我用官方 SDK 写核心是把上面两个类包成工具。这里给一个精简版from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent app Server(hindsight-memory) writer MemoryWriter(embed_fnmy_embed) retriever MemoryRetriever(embed_fnmy_embed, writerwriter) app.list_tools() async def list_tools(): return [ Tool(namememory_store, description存储一条记忆, inputSchema{type: object, properties: { content: {type: string}, type: {type: string, enum: [working, episodic, semantic]}, metadata: {type: object} }, required: [content, type]}), Tool(namememory_recall, 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: mid writer.store(arguments[content], arguments[type], arguments.get(metadata)) return [TextContent(typetext, textfstored:{mid})] if name memory_recall: hits retriever.recall(arguments[query], arguments.get(top_k, 5)) text \n.join([f[{h[type]}] {h[content]} for h in hits]) return [TextContent(typetext, texttext or no memory found)] async def main(): async with stdio_server() as (r, w): await app.run(r, w, app.create_initialization_options())Agent 侧配置好这个 MCP server 之后模型就能像调用其他工具一样调用记忆了。我在系统提示词里会加一句“在回答用户问题前先用 memory_recall 检索相关记忆如果检索到用户偏好或历史事实优先遵循。”实测这句话能让模型主动调用记忆工具的概率大幅提升。5. 常见问题与排查技巧实录5.1 记忆检索不准先查 embedding再查过滤条件检索不准是最常见的问题。我的排查顺序是先看 embedding 模型是否适合当前语言和领域很多通用模型对中文短文本的效果一般换成多语言模型或者针对领域微调过的模型会好很多再看过滤条件是不是太严比如mem_types只传了 semantic那 episodic 里的信息就永远召不回来最后看时间衰减系数是不是设得太大导致老记忆被压得几乎为 0。有个很隐蔽的坑Qdrant 的search默认用 cosine 距离但如果你建 collection 的时候用的是 euclidean分数含义完全不同排序会乱。建 collection 时一定要显式指定距离度量并且和 embedding 模型的训练目标对齐。5.2 Docker 网络不通容器间用服务名不要用 localhost这个坑我踩过不止一次。在 docker-compose 里容器之间通信要用服务名比如qdrant、postgres不能用localhost。因为每个容器有自己的网络命名空间localhost指向的是容器自己。我一开始在代码里写hostlocalhost本地跑没问题一进容器就连不上排查了半天才反应过来。正确做法是把 host 做成环境变量本地开发用localhost容器里用服务名import os QDRANT_HOST os.getenv(QDRANT_HOST, localhost)然后在 compose 文件里给应用服务加environment: QDRANT_HOSTqdrant。5.3 记忆无限增长必须有淘汰和归档策略记忆库不会自己变小。如果不做淘汰半年后你的向量库会有几百万条记录检索变慢、成本上升。我的策略是working memory 用 Redis TTL 自动过期或者 list 长度限制。episodic memory 保留最近 90 天更早的做归档导出到冷存储或者用总结的方式压缩成 semantic。semantic memory 做去重和合并同一个 key 只保留最新、置信度最高的那条。去重这块可以用向量相似度做两条 semantic 记忆相似度超过 0.95 就合并。我写过一个定时任务每周跑一次效果不错。5.4 常见问题速查表现象可能原因排查方向检索结果全是无关内容embedding 模型不匹配换模型重建索引精确实体查不到没走结构化过滤检查实体抽取逻辑老记忆召不回时间衰减过大调小 λ 系数容器间连不上用了 localhost改用服务名记忆库膨胀快无淘汰策略加 TTL 和归档任务模型不调用记忆工具提示词没引导加 few-shot 示例写入报错但检索正常双写不一致加补偿清理任务5.5 几个独家避坑心得第一别在写入时做太重的处理。我一开始想在写入时就用大模型做深度总结结果每轮对话延迟增加好几秒用户体验很差。后来改成写入只做轻量抽取总结放到异步任务里做延迟立刻降下来。第二记忆的 key 设计要稳定。如果你用自然语言当 key同一个偏好可能被存成好几种表述检索时对不上。我现在的做法是维护一个 key 的枚举表比如user_dietary_preference、user_city、user_language写入时让模型从这个表里选保证一致性。第三给用户一个“忘掉我”的入口。这既是合规要求也是体验细节。用户说“忘掉我刚才说的”Agent 要能调用memory_forget把相关记忆删掉。我一般会按 session 或者按 key 批量删而不是只删单条。第四测试记忆系统要用“长对话”场景。短对话测不出问题一定要模拟几十轮、跨天的对话才能暴露召回不准、记忆冲突这些问题。我一般会写一个脚本自动生成 50 轮对话灌进去然后抽查召回结果。6. 记忆系统的扩展方向与个人体会hindsight 这套思路跑通之后能扩展的方向其实挺多。比如把记忆和 RAG 结合起来让 Agent 既能查外部知识库又能查自己的历史记忆再比如做多 Agent 共享记忆几个 Agent 协作时共用一个记忆层避免信息孤岛还有就是把记忆做成可解释的让用户能看到 Agent “记住了什么”增强信任感。我自己在实际项目里最大的体会是记忆系统的难点从来不在技术选型而在“什么该记、什么该忘”的判断。向量库、关系库、MCP、Docker 这些都是工具真正决定效果的是你对业务的理解——哪些信息对用户是长期有价值的哪些只是当下的噪音。这个问题没有标准答案只能靠不断观察真实对话、调整抽取和召回策略来逼近。最后分享一个小技巧在开发阶段我会给每条记忆加一个debug字段记录它是从哪轮对话、哪个 prompt 抽取出来的。这样当召回出问题时能快速回溯到源头比盲目调参高效得多。等系统稳定了再把这个字段关掉。这套方案我目前在两个项目里跑着一个客服 Agent、一个个人助理 Agent稳定运行了几个月记忆召回准确率维持在可接受的水平。如果你也在做 Agent 记忆希望这些经验能帮你少走点弯路。
返回列表