1. 项目缘起为什么“事后聪明”值得被工程化“Hindsight”这个词本身很有意思字面意思是“事后的洞察力”中文常翻译成“事后诸葛亮”。但在 LLM Agent 的语境里它指向一个非常具体且棘手的问题Agent 如何记住过去发生过的事情并在未来需要的时候准确地调用这些记忆。我接触过不少做 Agent 的团队大家一开始都把精力放在工具调用、提示词工程、多轮对话编排上等到系统跑了一段时间用户开始抱怨“它怎么又忘了”“上次不是说过吗”“同一个错误犯了三次”才意识到记忆层是个绕不过去的坎。Agent Memory 不是一个新话题但真正把它当作一个独立系统来设计、来运维的团队并不多。大多数项目在早期都是把对话历史往上下文里一塞靠模型的窗口长度硬扛等到上下文爆了、成本上去了、延迟不可接受了才开始想“是不是该搞个记忆系统”。Hindsight 这个项目标题我理解它要解决的核心问题是让 Agent 具备对历史交互的结构化记忆能力并且这种记忆是可检索、可更新、可遗忘的。它不是简单的“把聊天记录存数据库”而是涉及记忆的编码、存储、检索、衰减、冲突消解等一系列工程问题。结合热词里出现的 agent memory、LLM、MCP、Docker可以判断这个项目大概率是一个围绕 Agent 记忆层构建的工程实践可能包含记忆存储服务、MCP 协议对接、容器化部署等模块。这篇文章适合谁看如果你正在做 LLM Agent 相关的产品或者你已经在用 MCP 协议搭建工具链又或者你单纯对“Agent 怎么记住东西”这件事好奇那接下来的内容应该对你有用。我会从设计思路讲到实操细节把记忆系统的关键决策点一个个拆开说尽量让你看完能直接动手搭一个能跑的东西出来。2. 记忆系统的整体设计从“存什么”到“怎么取”2.1 记忆不是一种东西至少分三层很多人一上来就说“我要给 Agent 加记忆”但记忆这个词太笼统了。我在实际项目里会把 Agent 的记忆至少分成三层来设计每一层的生命周期、存储介质、检索方式都不一样。第一层是工作记忆Working Memory也就是当前对话轮次里模型需要立刻用到的信息。这部分通常就是上下文窗口里的内容生命周期以“轮”为单位对话结束就可以丢弃。它的特点是容量小、访问快、不需要持久化。热词里提到的“agent 存储 working memory”说的就是这个层面的事情。第二层是情景记忆Episodic Memory记录的是“什么时候发生了什么”。比如用户在某次对话里说“我下周三要去上海出差”这就是一条情景记忆。它需要持久化需要带时间戳需要能被后续对话检索到。这一层是 Hindsight 这类项目的主战场。第三层是语义记忆Semantic Memory是从多次交互中抽象出来的稳定知识。比如“这个用户偏好简洁的回答风格”“这个项目的代码规范要求用 TypeScript 严格模式”。语义记忆不依赖于某一次具体对话而是从情景记忆里提炼出来的。为什么要分三层因为如果不分你要么把所有东西都塞进上下文成本爆炸要么把所有东西都存数据库然后每次全量检索延迟爆炸。分层之后每一层可以用不同的策略来管理。工作记忆用滑动窗口加摘要情景记忆用向量检索加时间衰减语义记忆用定期归纳加人工审核。2.2 记忆的编码Key-Value 结构为什么好用热词里有一条很有意思“llm 的 token 三个点 key 我是谁、query 我在找什么、value 我能提供什么”。这其实是在用类比的方式解释注意力机制里的 QKV但放到记忆系统里同样适用。我在设计记忆存储结构的时候基本都会采用类似 Key-Value 的模型但这里的 Key 不是简单的字符串而是一个复合结构。一条记忆记录通常包含以下字段字段含义示例memory_id唯一标识mem_20240513_001content记忆正文用户下周三去上海出差embedding向量表示[0.023, -0.114, ...]timestamp创建时间2024-05-13T10:30:00Zlast_access最后访问时间2024-05-14T09:00:00Zaccess_count访问次数7importance重要性评分0.85source来源conversation_42tags标签[出行, 日程]这个结构的好处是检索的时候可以综合多个维度来排序。纯向量检索的问题在于它只看语义相似度不看时间、不看重要性、不看访问频率。但实际场景里一条三天前的记忆和一条三个月前的记忆即使语义相似度一样优先级也应该不同。一条被访问过十次的记忆和一条从没被访问过的记忆价值也不一样。所以我在检索排序的时候会用这样的公式final_score w1 * cosine_similarity w2 * time_decay w3 * importance w4 * access_frequency其中 time_decay 可以用指数衰减exp(-lambda * hours_since_creation)lambda 根据业务场景调整。如果是日程类应用lambda 可以大一点让旧记忆快速衰减如果是知识库类应用lambda 可以小一点让旧记忆保持更久。2.3 为什么选 MCP 作为对接层热词里 MCP 出现的频率很高还有“mcp 协议”“mcp 是软件协议 硬件协议那个概念叫什么来着”这样的搜索词。MCP 全称是 Model Context Protocol是一个让 LLM 应用与外部工具、数据源对接的协议标准。它的核心价值在于解耦记忆服务不需要知道上层是哪个 Agent 框架Agent 框架也不需要知道记忆服务底层用什么数据库。我选择用 MCP 来暴露记忆服务主要考虑三点。第一复用性。同一个记忆服务可以同时给多个 Agent 用只要它们都支持 MCP 客户端。第二可测试性。MCP 有标准的请求-响应格式我可以单独测试记忆服务的每个接口不用把整个 Agent 跑起来。第三生态兼容。现在越来越多的工具链开始支持 MCP比如热词里提到的 Playwright MCP、Chrome DevTools MCP、Unity MCP 等用 MCP 意味着你的记忆服务可以跟这些工具在同一个编排层里协作。具体到接口设计我会暴露这么几个 MCP 工具memory_store存入一条记忆参数包括 content、tags、importancememory_search检索记忆参数包括 query、top_k、time_rangememory_update更新一条记忆的内容或元数据memory_forget删除或标记一条记忆为失效memory_summarize对一段时间内的记忆做归纳生成语义记忆每个工具都有明确的输入输出 schema这样 Agent 在调用的时候不会因为参数格式问题反复试错。2.4 Docker 化部署为什么不是可选项热词里 Docker 相关的搜索词非常多从“docker 安装”到“docker desktop 安装教程”到“windows 安装 docker”到“docker 网络不通”说明很多人在这一步踩过坑。我的观点很明确记忆服务必须容器化原因有三个。第一依赖隔离。记忆服务通常要连向量数据库、关系数据库、缓存这些依赖的版本冲突是家常便饭。容器化之后每个服务在自己的环境里跑互不干扰。第二可复现。我在本地跑通的配置打成镜像之后在服务器上也能跑通不会出现“在我机器上是好的”这种情况。第三弹性伸缩。记忆检索是 IO 密集型操作流量大的时候可以单独扩容记忆服务不用动 Agent 主进程。我一般会用 Docker Compose 来编排至少包含三个服务memory-service记忆服务本体、vector-db向量存储、cache热点记忆缓存。如果要做持久化再加一个 postgres 或者 mysql 来存元数据。3. 核心细节解析记忆的写入、检索与遗忘3.1 写入策略不是所有对话都值得记新手最容易犯的错误是“什么都记”。用户说了一句“今天天气不错”也存一条记忆用户发了个“嗯”也存一条。结果就是记忆库迅速膨胀检索的时候噪声比信号还多。我的做法是在写入之前加一层过滤和提炼。具体来说分三步走。第一步规则过滤。太短的少于 10 个 token、纯表情的、纯停用词的直接丢弃。这一步能过滤掉大概 30% 的无意义内容。第二步重要性评分。用一个轻量级的 LLM 调用或者本地小模型给每条候选记忆打一个 0 到 1 的分数。评分维度包括是否包含具体事实时间、地点、人物、数字、是否是用户的明确偏好或指令、是否与已有记忆冲突。分数低于 0.3 的直接丢弃0.3 到 0.6 的存入但标记为低优先级0.6 以上的正常存入。第三步去重和合并。如果新记忆和已有记忆的语义相似度超过 0.95就不新建而是更新已有记忆的 last_access 和 access_count。如果相似度在 0.8 到 0.95 之间可以考虑合并成一条更完整的记忆。这里有个实操心得重要性评分不要用太大的模型。我试过用 GPT-4 级别的模型来打分效果确实好一点但成本和延迟都上去了。后来换成一个 7B 级别的本地模型配合精心设计的提示词准确率能到 85% 左右完全够用。毕竟记忆写入是高频操作每次省 200ms一天下来就是几万秒。3.2 检索策略多路召回加重排序检索是记忆系统里最影响体验的环节。用户问一个问题Agent 能不能找到相关的历史记忆直接决定了回答的质量。我采用的是多路召回 重排序的架构。多路召回包括向量召回用 embedding 做语义相似度检索召回 top 50关键词召回用 BM25 或者全文索引做关键词匹配召回 top 30时间召回取最近 N 条记忆召回 top 20标签召回如果 query 里能提取出标签按标签过滤召回 top 20四路召回的结果合并去重之后大概有 80 到 100 条候选。然后用一个重排序模型可以用 cross-encoder也可以用 LLM 做 listwise 排序对这 100 条做精排取 top 5 到 top 10 注入到上下文里。为什么要这么麻烦因为单一召回策略都有盲区。向量召回对语义相似但用词不同的情况好但对精确匹配比如订单号、人名不行关键词召回反过来时间召回能保证新鲜度但可能召回不相关的内容。多路召回的本质是用不同的视角看同一个问题然后把最相关的挑出来。重排序这一步我强烈建议不要省。我做过对比实验不加重排序的检索准确率大概在 60% 左右加了之后能到 85% 以上。代价是每次检索多 100 到 200ms但这个延迟在大多数场景下是可以接受的。3.3 遗忘机制记忆系统也需要“断舍离”热词里有一条“a-memguard: a proactive defense framework for llm-based agent memory”这提醒我一个很重要但常被忽视的问题记忆系统需要主动防御和清理。遗忘机制我一般分三种。第一种是时间衰减前面提过用指数衰减函数降低旧记忆的权重。第二种是容量淘汰给每个用户或每个 Agent 设定一个记忆上限比如 10000 条超了就按 final_score 从低到高淘汰。第三种是冲突消解当新记忆和旧记忆矛盾时比如用户先说“我喜欢咖啡”后说“我戒咖啡了”把旧记忆标记为失效而不是直接删除保留审计线索。这里有个坑不要物理删除记忆。我早期版本为了省空间直接 delete 掉低分记忆结果后来做数据分析的时候发现很多有价值的统计信息没了。后来改成软删除加一个is_active字段检索的时候过滤掉但数据还在。这样既保证了检索质量又保留了回溯能力。另外遗忘策略要跟业务场景匹配。如果是客服 Agent记忆保留周期可以长一点因为用户可能几个月后回来问同样的问题。如果是实时助手记忆保留一周就够了太久远的记忆反而会干扰。4. 实操过程从零搭一个可用的记忆服务4.1 环境准备与 Docker Compose 编排先说环境。我假设你用的是 Ubuntu 或者 macOSWindows 的话建议用 WSL2因为 Docker Desktop 在 Windows 上的网络配置有时候会出问题热词里“docker 网络不通”和“virtualization support not detected”就是典型症状。第一步安装 Docker 和 Docker Compose。Ubuntu 上可以用官方脚本curl -fsSL https://get.docker.com -o get-docker.sh sudo sh get-docker.sh sudo usermod -aG docker $USER装完之后记得重新登录让用户组生效。验证一下docker --version docker compose version第二步创建项目目录结构hindsight/ ├── docker-compose.yml ├── memory-service/ │ ├── Dockerfile │ ├── requirements.txt │ └── app/ │ ├── main.py │ ├── models.py │ ├── store.py │ └── search.py ├── vector-db/ │ └── data/ └── cache/ └── data/第三步写 docker-compose.yml。我用的向量数据库是 Qdrant缓存用 Redis元数据存 PostgreSQLversion: 3.9 services: memory-service: build: ./memory-service ports: - 8080:8080 environment: - QDRANT_HOSTvector-db - REDIS_HOSTcache - POSTGRES_HOSTmetadata-db depends_on: - vector-db - cache - metadata-db networks: - hindsight-net vector-db: image: qdrant/qdrant:latest ports: - 6333:6333 volumes: - ./vector-db/data:/qdrant/storage networks: - hindsight-net cache: image: redis:7-alpine ports: - 6379:6379 volumes: - ./cache/data:/data networks: - hindsight-net metadata-db: image: postgres:16-alpine environment: - POSTGRES_USERhindsight - POSTGRES_PASSWORDhindsight_dev - POSTGRES_DBhindsight ports: - 5432:5432 volumes: - ./metadata-db/data:/var/lib/postgresql/data networks: - hindsight-net networks: hindsight-net: driver: bridge这里有个细节一定要自定义网络。默认的 bridge 网络里容器之间只能用 IP 互访不能用服务名。自定义网络之后memory-service 可以直接用vector-db这个主机名连 Qdrant配置起来清爽很多。4.2 记忆服务的核心代码实现memory-service 我用 FastAPI 来写因为它的异步支持好而且自动生成 OpenAPI 文档方便调试。先看数据模型from pydantic import BaseModel, Field from datetime import datetime from typing import Optional, List class MemoryItem(BaseModel): memory_id: str content: str embedding: Optional[List[float]] None timestamp: datetime Field(default_factorydatetime.utcnow) last_access: datetime Field(default_factorydatetime.utcnow) access_count: int 0 importance: float 0.5 source: str unknown tags: List[str] [] is_active: bool True class MemoryCreate(BaseModel): content: str importance: float 0.5 source: str conversation tags: List[str] [] class MemorySearchQuery(BaseModel): query: str top_k: int 5 time_range_hours: Optional[int] None tags: Optional[List[str]] None然后是存储层。我用 Qdrant 存向量PostgreSQL 存元数据Redis 做热点缓存import redis from qdrant_client import QdrantClient from qdrant_client.models import PointStruct, Distance, VectorParams import psycopg2 import hashlib class MemoryStore: def __init__(self, qdrant_host, redis_host, pg_host): self.qdrant QdrantClient(hostqdrant_host, port6333) self.redis redis.Redis(hostredis_host, port6379, decode_responsesTrue) self.pg psycopg2.connect( hostpg_host, dbnamehindsight, userhindsight, passwordhindsight_dev ) self._init_collection() def _init_collection(self): collections self.qdrant.get_collections().collections if not any(c.name memories for c in collections): self.qdrant.create_collection( collection_namememories, vectors_configVectorParams(size1536, distanceDistance.COSINE) )写入逻辑里我加了一个简单的去重检查def store(self, item: MemoryCreate, embedding: List[float]): # 先查有没有高度相似的记忆 similar self.qdrant.search( collection_namememories, query_vectorembedding, limit1, score_threshold0.95 ) if similar: # 更新已有记忆的访问信息 existing_id similar[0].id self._update_access(existing_id) return existing_id memory_id hashlib.md5( f{item.content}{datetime.utcnow().isoformat()}.encode() ).hexdigest()[:16] self.qdrant.upsert( collection_namememories, points[PointStruct( idmemory_id, vectorembedding, payload{ content: item.content, importance: item.importance, source: item.source, tags: item.tags, timestamp: datetime.utcnow().isoformat() } )] ) return memory_id检索部分是多路召回的简化版def search(self, query: str, query_embedding: List[float], top_k: int 5): # 向量召回 vector_results self.qdrant.search( collection_namememories, query_vectorquery_embedding, limit50 ) # 时间召回取最近 20 条 recent self.qdrant.scroll( collection_namememories, limit20, with_payloadTrue )[0] # 合并去重 candidates {} for r in vector_results: candidates[r.id] {score: r.score, payload: r.payload} for r in recent: if r.id not in candidates: candidates[r.id] {score: 0.5, payload: r.payload} # 重排序综合向量分、时间衰减、重要性 now datetime.utcnow() scored [] for mid, data in candidates.items(): ts datetime.fromisoformat(data[payload][timestamp]) hours_old (now - ts).total_seconds() / 3600 time_decay math.exp(-0.01 * hours_old) importance data[payload].get(importance, 0.5) final 0.6 * data[score] 0.2 * time_decay 0.2 * importance scored.append((final, mid, data[payload])) scored.sort(reverseTrue) return scored[:top_k]这段代码里时间衰减的 lambda 我设的是 0.01意味着大约 70 小时后权重降到一半。这个值可以根据业务调整日程类应用可以设 0.05知识库类可以设 0.001。4.3 MCP 接口暴露与 Agent 对接记忆服务跑起来之后下一步是把它暴露成 MCP 工具。我用的是 Python 的 mcp 库from mcp.server import Server from mcp.types import Tool, TextContent app Server(hindsight-memory) app.list_tools() async def list_tools(): return [ Tool( namememory_store, description存入一条记忆, inputSchema{ type: object, properties: { content: {type: string}, importance: {type: number, default: 0.5}, tags: {type: array, items: {type: string}} }, required: [content] } ), Tool( namememory_search, description检索相关记忆, inputSchema{ type: object, properties: { query: {type: string}, top_k: {type: integer, default: 5} }, required: [query] } ) ] app.call_tool() async def call_tool(name: str, arguments: dict): if name memory_store: embedding get_embedding(arguments[content]) mid store.store(MemoryCreate(**arguments), embedding) return [TextContent(typetext, textf已存储ID: {mid})] elif name memory_search: embedding get_embedding(arguments[query]) results store.search(arguments[query], embedding, arguments.get(top_k, 5)) text \n.join([f- {r[2][content]} for r in results]) return [TextContent(typetext, texttext)]Agent 那边配置 MCP 客户端的时候只需要填上记忆服务的地址和端口。我用 Claude Desktop 测试的时候配置文件大概长这样{ mcpServers: { hindsight: { command: python, args: [-m, memory_service.mcp_server], env: { QDRANT_HOST: localhost, REDIS_HOST: localhost } } } }这里有个实操心得MCP 工具的 description 要写清楚。Agent 决定调不调这个工具、怎么调很大程度上依赖 description。我一开始写得太简略Agent 经常把 search 和 store 搞混。后来把 description 改成“检索历史记忆用于回答需要回忆过去信息的问题”和“存储新的记忆仅在用户明确要求记住某事时调用”准确率明显提升。4.4 启动与验证所有代码写完之后启动整个栈docker compose up -d --build等所有容器 healthy 之后验证一下# 检查记忆服务 curl http://localhost:8080/health # 存入一条记忆 curl -X POST http://localhost:8080/memory \ -H Content-Type: application/json \ -d {content: 用户偏好简洁的回答, importance: 0.8, tags: [偏好]} # 检索 curl -X POST http://localhost:8080/search \ -H Content-Type: application/json \ -d {query: 用户喜欢什么样的回答, top_k: 3}如果一切正常检索结果里应该能看到刚才存的那条记忆。这时候再通过 MCP 客户端连上去让 Agent 试着存一条、查一条整个链路就通了。5. 常见问题与排查技巧实录5.1 记忆检索不准的排查思路这是最高频的问题。用户反馈“Agent 明明之前知道现在又不知道了”排查顺序我一般是这样第一确认记忆有没有存进去。直接查 Qdrant 的 collection看 point 数量对不对。如果没存进去检查写入过滤逻辑是不是太严格把有效记忆误杀了。第二确认 embedding 模型一致。写入和检索必须用同一个 embedding 模型否则向量空间不对齐相似度计算全是错的。我踩过这个坑换了模型之后忘了重新索引结果检索全乱。第三检查重排序权重。如果时间衰减的 lambda 设得太大旧记忆会被压得很低。可以临时把 lambda 调小看检索结果有没有改善。第四看 top_k 是不是太小。有时候相关记忆排在第六第七位但 top_k 设的是 5就被截断了。可以先把 top_k 调到 20看目标记忆在不在里面。症状可能原因排查方法完全检索不到记忆未写入查 Qdrant point 数量检索到但不相关embedding 不一致检查模型版本旧记忆检索不到时间衰减过强调小 lambda相关记忆排后面重排序权重不合理调整权重系数检索延迟高候选集太大减少召回数量5.2 Docker 网络问题的典型场景热词里“docker 网络不通”出现多次我总结几个常见原因。场景一容器之间用 localhost 互连。这是新手最容易犯的错。在容器里localhost 指的是容器自己不是宿主机。要连宿主机上的服务得用host.docker.internalDocker Desktop或者宿主机的实际 IP。场景二端口映射写错。ports: - 8080:8080前面是宿主机端口后面是容器端口。如果写反了外面就连不上。场景三自定义网络没配。前面提过默认 bridge 网络不支持服务名解析。要么用自定义网络要么用 links已废弃要么直接用 IP。场景四防火墙拦截。Ubuntu 上 ufw 默认可能拦截 Docker 的流量。可以临时关掉 ufw 测试确认是防火墙问题之后再配规则。5.3 记忆冲突的处理经验用户先说“我喜欢咖啡”过两天说“我戒咖啡了”。如果两条记忆都存着检索的时候可能同时召回Agent 就懵了。我的处理方式是加一个冲突检测步骤。写入新记忆之前先检索语义相似度在 0.85 以上的旧记忆然后用一个 LLM 判断是否矛盾。如果矛盾把旧记忆标记为is_activeFalse新记忆正常写入并在新记忆的 payload 里加一个supersedes字段指向旧记忆 ID。这样做的好处是检索的时候只返回 active 的记忆不会出现矛盾信息。同时保留了完整的变更历史方便回溯。5.4 性能优化的几个实用技巧记忆服务跑久了性能会下降。我一般从这几个地方优化缓存热点记忆把 access_count 最高的 100 条记忆缓存在 Redis 里检索的时候先查缓存命中就直接返回不用走向量数据库。批量写入如果短时间内有大量记忆写入攒一批再批量 upsert比一条条写快很多。索引优化Qdrant 的 HNSW 索引参数可以调m和ef_construct越大越准但越慢根据数据量权衡。异步化embedding 计算是 CPU 密集型的放到单独的 worker 里异步做不要阻塞主请求。我在实际项目里优化前检索 P99 延迟是 800ms优化后降到 200ms 左右。主要贡献来自缓存和批量写入。5.5 安全与合规的注意事项记忆系统存的是用户的历史交互涉及隐私。几点必须注意加密存储敏感字段如用户 ID、对话内容在数据库里要加密不要明文存。访问控制MCP 接口要有鉴权不能裸奔。至少加一个 token 校验。数据隔离不同用户的记忆要逻辑隔离检索的时候必须带 user_id 过滤防止串数据。审计日志谁在什么时候存了什么、查了什么要有日志方便追溯。遗忘权用户要求删除记忆的时候要能彻底删除包括向量库、元数据库、缓存里的所有副本。这些不是可选项是底线。我见过因为记忆串数据导致的事故修复成本远高于前期做好隔离。6. 记忆系统的扩展方向与个人体会这套东西搭起来之后能跑但离“好用”还有距离。我后续会往几个方向扩展。一个是记忆的自动归纳。现在情景记忆是散点式的时间久了数量很大。可以定期跑一个归纳任务把同一主题的多条情景记忆合并成一条语义记忆。比如用户过去一个月提了五次“不喜欢冗长的回答”可以归纳成一条“用户偏好简洁风格”的语义记忆权重更高。另一个是跨 Agent 的记忆共享。如果多个 Agent 服务同一个用户它们的记忆应该能互通。这需要在记忆服务里加一个 namespace 概念不同 Agent 可以读写同一个 namespace 下的记忆但要有权限控制。还有一个是记忆的可解释性。现在检索出来一条记忆Agent 直接用但用户不知道它为什么用这条。可以在返回结果里带上检索理由比如“因为与当前问题语义相似度 0.92且是三天内的高重要性记忆”。这样调试和排查都方便很多。我个人在实际操作中的体会是记忆系统的难点不在技术选型而在策略设计。用什么数据库、什么 embedding 模型这些都有成熟方案。但“什么该记、什么该忘、怎么排序、怎么消解冲突”这些策略问题没有标准答案必须结合具体业务场景反复调。我建议一开始不要追求大而全先把写入过滤和向量检索跑通让 Agent 能记住最近的重要信息然后再逐步加时间衰减、重排序、冲突消解。每加一个策略都要有对应的评估指标不然很容易越调越乱。最后分享一个小技巧给记忆系统加一个“调试模式”。在这个模式下每次检索都返回完整的候选列表和打分明细而不是只返回 top_k。这样排查问题的时候你能看到每条记忆的向量分、时间分、重要性分分别是多少很快就能定位是哪个环节出了问题。这个功能我一开始没做后来被逼着加上现在成了最常用的调试工具。