1. 从“hindsight”说起为什么我们需要给Agent装上“后视镜”“hindsight”这个词本身的意思就是“事后之明”——事情发生之后才明白当初应该怎么做。把这个词用在Agent Memory智能体记忆这个领域指向性非常明确我们要解决的是LLM驱动的Agent在长期运行中“记不住、想不起、用不对”的核心痛点。你如果最近在折腾LLM Agent相关的项目大概率已经踩过这样的坑——Agent在单轮对话里表现惊艳一旦对话轮次拉长到几十轮它就开始胡言乱语前面明确说过的约束条件被抛到九霄云外用户纠正过的错误反复再犯。这不是模型能力不够而是记忆架构没设计好。我最初接触这个方向是因为一个实际需求做一个能持续跟踪项目进展的辅助Agent它需要记住过去几周内用户提过的所有需求变更、技术决策和踩坑记录。用最朴素的方式——把全部历史对话塞进context window——很快就撞墙了。Token消耗爆炸不说模型对超长上下文的注意力衰减非常明显关键信息淹没在大量无关对话里。后来试过简单的摘要压缩又发现摘要会丢失细节等到真正需要某个具体参数的时候摘要里根本没保留。这就是“hindsight”要解决的核心问题如何让Agent像人一样在需要的时候能准确回忆起过去发生的关键事件并且理解这些事件之间的因果关系。这篇文章适合几类人看一是正在做LLM Agent应用开发、被记忆问题困扰的工程师二是对Agent架构设计感兴趣、想了解记忆模块怎么落地的人三是已经在用Docker部署各种服务、想把手头的LLM工具链整合起来的实践者。我会从整体设计思路讲起然后拆解核心细节接着给出完整的实操流程最后分享我在排查问题时积累的经验。文中涉及的技术栈包括LLM、MCP协议、Docker容器化部署以及Agent Memory的分层存储设计。2. 整体设计与思路拆解Agent Memory到底该怎么分层2.1 为什么不能只靠Context Window硬撑很多人刚开始做Agent的时候第一反应就是把所有历史对话都拼到prompt里。短对话没问题一旦超过模型的有效上下文长度要么截断丢信息要么就得换更大上下文的模型成本直线上升。更关键的是即使模型支持128K甚至更长的上下文中间部分的信息检索准确率也会显著下降——这是注意力机制的固有特性不是换个模型就能解决的。我做过一个简单的对比测试让Agent记住20轮对话前用户提到的一个特定数字然后在一段无关对话之后询问这个数字。把全部历史塞进context的方案准确率大概在60%左右而采用分层记忆架构、把关键信息单独存储并在需要时检索注入的方案准确率能到95%以上。这个差距在真实应用里就是“能用”和“不能用”的区别。所以核心思路很明确Context Window只放当前最相关的信息长期记忆放到外部存储通过检索机制按需注入。这就是hindsight这类Agent Memory系统的设计起点。2.2 三层记忆架构Working Memory、Episodic Memory、Semantic Memory参考认知科学里对人类记忆的分类我在实际项目中把Agent Memory分成三层Working Memory工作记忆当前对话轮次直接相关的信息放在context window里。这部分容量有限需要严格控制。通常只保留最近几轮对话的原文加上从长期记忆中检索出来的相关片段。Episodic Memory情景记忆具体发生过的事件记录。比如“用户在第三轮对话中要求把数据库从MySQL换成PostgreSQL”、“第七轮对话中用户反馈某个API返回格式不对”。这些记录带时间戳、带上下文按事件粒度存储。Semantic Memory语义记忆从多个事件中抽象出来的通用知识。比如“这个用户偏好使用ORM而不是手写SQL”、“这个项目的API设计遵循RESTful风格”。这部分是对情景记忆的进一步提炼存储的是结论而非原始事件。三层之间的流转关系是这样的对话进行时原始信息先进入Working Memory对话结束后关键事件被抽取出来存入Episodic Memory当同类事件积累到一定数量系统自动或手动触发提炼生成Semantic Memory。检索的时候优先查Semantic Memory因为信息密度高不够再查Episodic Memory最后才回退到Working Memory。注意不要一上来就追求全自动的语义提炼。我踩过的坑是自动提炼的语义记忆经常出现过度概括或错误归纳反而污染了记忆库。建议初期以手动或半自动方式管理Semantic Memory等积累足够多的样本后再考虑自动化。2.3 为什么选择MCP协议做记忆服务的接口层MCPModel Context Protocol是Anthropic推出的一个开放协议用来标准化LLM与外部工具、数据源之间的交互方式。我选择用MCP来封装记忆服务主要考虑几点第一解耦。记忆服务作为一个独立的MCP Server运行LLM Agent通过标准协议调用不依赖具体的Agent框架。今天用LangChain明天换AutoGen记忆服务不用改。第二可组合。MCP Server可以同时暴露多个工具比如store_memory、retrieve_memory、summarize_episodesAgent根据需要调用。而且多个MCP Server可以并行运行记忆服务可以和文件系统服务、数据库服务共存。第三生态兼容。现在越来越多的工具支持MCP协议包括各种IDE插件、浏览器扩展、命令行工具。把记忆服务做成MCP Server意味着任何支持MCP的客户端都能直接接入。具体实现上我用Python写了一个MCP Server底层存储用SQLite轻量、零配置加向量数据库用于语义检索。SQLite存结构化的事件记录向量库存embedding用于相似度搜索。两者通过事件ID关联。2.4 Docker化部署为什么不用裸机跑把记忆服务跑在Docker容器里好处是环境隔离和可移植性。我试过直接在宿主机上装依赖结果Python版本冲突、SQLite扩展加载失败、向量数据库的C库版本不匹配折腾了一整天。换成Docker之后所有依赖打包在镜像里换台机器docker compose up就能跑起来。另外Docker的网络管理让MCP Server和Agent之间的通信更可控。我可以把记忆服务放在一个内部网络里只暴露必要的端口Agent通过服务名访问不用关心IP地址。如果后续要加认证和限流在Docker网络层面做也比在应用层做更干净。3. 核心细节解析与实操要点从存储格式到检索策略3.1 记忆条目的数据结构设计每条记忆记录我设计了这些字段字段名类型说明idUUID全局唯一标识timestampISO8601事件发生时间session_idString所属会话标识memory_typeEnumworking/episodic/semanticcontentText记忆正文embeddingVector语义向量维度768metadataJSON扩展字段如来源、置信度、标签ttlInteger过期时间秒0表示永不过期access_countInteger被检索次数用于热度排序ttl字段的设计很关键。Working Memory的记录通常设置较短的TTL比如1小时过期自动清理。Episodic Memory的TTL可以设长一些比如30天Semantic Memory则永不过期。这样能自动控制存储增长避免记忆库无限膨胀。access_count用来做热度加权。检索的时候除了向量相似度还会考虑这条记忆被访问的频率。经常被用到的记忆说明价值高排序时应该靠前。3.2 向量化模型的选择与权衡Embedding模型我试过好几个最后选的是text-embedding-3-smallOpenAI和bge-m3本地部署两个方案并行。原因如下OpenAI的方案优点是稳定、无需维护适合快速验证。缺点是API调用有延迟和成本而且数据要出境如果在意的话。本地部署bge-m3的优点是数据不出本地、无API成本缺点是需要GPU资源推理速度取决于硬件。实际使用中我做了个路由策略对延迟敏感的检索走本地模型对精度要求高的走API模型。两者生成的向量存在不同的表里检索时分别查询再合并结果。提示embedding维度不一致会导致无法直接比较相似度。如果混用不同模型务必在metadata里标记embedding来源检索时按来源分组查询。3.3 检索策略向量相似度关键词时间衰减单纯的向量相似度检索有个问题它擅长语义匹配但对精确的关键词匹配不够敏感。比如用户问“上次说的那个端口号是多少”向量检索可能返回一堆关于“端口”的讨论但真正包含具体端口号的那条记录可能排在后面。我的做法是混合检索向量检索用query的embedding去向量库搜Top-KK取20。关键词检索用BM25算法在content字段上做全文搜索取Top-20。合并去重两路结果按id合并计算综合得分。时间衰减综合得分乘以时间衰减因子越久远的记忆得分越低。衰减公式我用的是score * exp(-lambda * days_ago)lambda取0.01意味着大约70天后权重降到一半。热度加权再乘以log(1 access_count)让高频访问的记忆获得加成。最终排序取Top-5返回给Agent。这套组合拳实测下来检索准确率比单用向量检索提升了大概20个百分点。尤其是在需要精确回忆具体数值、名称、日期的场景下关键词检索的补充作用非常明显。3.4 MCP Server的工具定义MCP Server暴露给Agent的工具我定义了四个# 工具1存储记忆 { name: store_memory, description: 存储一条新的记忆记录, parameters: { content: string, 记忆正文, memory_type: string, 枚举值: working/episodic/semantic, metadata: object, 可选扩展字段, ttl: integer, 可选过期时间(秒) } } # 工具2检索记忆 { name: retrieve_memory, description: 根据查询检索相关记忆, parameters: { query: string, 查询文本, top_k: integer, 返回条数, 默认5, memory_type: string, 可选过滤类型 } } # 工具3提炼语义记忆 { name: summarize_episodes, description: 将多条情景记忆提炼为语义记忆, parameters: { episode_ids: array, 情景记忆ID列表, instruction: string, 提炼指令 } } # 工具4清理过期记忆 { name: cleanup_expired, description: 清理已过期的记忆记录, parameters: {} }Agent在对话过程中根据上下文自主决定何时调用这些工具。比如用户说“记住这个配置”Agent就调store_memory用户问“之前那个配置是什么”Agent就调retrieve_memory。4. 实操过程与核心环节实现从零搭建一个可用的记忆服务4.1 环境准备与Docker Compose编排先确保Docker Desktop已经装好并且能正常启动。Windows上如果遇到“Virtualization support not detected”的报错需要进BIOS开启虚拟化支持Intel VT-x或AMD-V然后在Windows功能里启用WSL2。Mac上相对简单装好Docker Desktop即可。项目目录结构如下hindsight-memory/ ├── docker-compose.yml ├── mcp-server/ │ ├── Dockerfile │ ├── requirements.txt │ └── src/ │ ├── main.py │ ├── storage.py │ ├── retrieval.py │ └── models.py ├── vector-db/ │ └── data/ └── config/ └── settings.yamldocker-compose.yml的内容version: 3.8 services: mcp-memory: build: ./mcp-server container_name: hindsight-mcp ports: - 8765:8765 volumes: - ./vector-db/data:/app/data - ./config:/app/config environment: - EMBEDDING_MODELbge-m3 - DB_PATH/app/data/memory.db - VECTOR_DIM768 networks: - agent-net restart: unless-stopped networks: agent-net: driver: bridgeDockerfileFROM python:3.11-slim WORKDIR /app RUN apt-get update apt-get install -y \ build-essential \ rm -rf /var/lib/apt/lists/* COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY src/ ./src/ COPY config/ ./config/ EXPOSE 8765 CMD [python, -m, src.main]requirements.txtmcp0.1.0 fastapi0.104.0 uvicorn0.24.0 sqlite-utils3.35 numpy1.24 sentence-transformers2.2.0 rank-bm250.2.2 pydantic2.04.2 存储层实现SQLite向量的混合方案存储层我用SQLite做结构化存储向量部分用numpy数组存在单独的文件里通过id关联。这样做的原因是SQLite对向量类型的支持还不够成熟而引入专门的向量数据库如Milvus、Qdrant又增加了部署复杂度。对于中小规模的记忆库几万条以内numpy数组加暴力检索完全够用。# storage.py 核心逻辑 import sqlite3 import numpy as np import json from datetime import datetime, timedelta from uuid import uuid4 class MemoryStore: def __init__(self, db_path, vector_dim768): self.conn sqlite3.connect(db_path, check_same_threadFalse) self.vector_dim vector_dim self._init_tables() self.vectors {} # id - np.array def _init_tables(self): self.conn.execute( CREATE TABLE IF NOT EXISTS memories ( id TEXT PRIMARY KEY, timestamp TEXT NOT NULL, session_id TEXT, memory_type TEXT NOT NULL, content TEXT NOT NULL, metadata TEXT, ttl INTEGER DEFAULT 0, access_count INTEGER DEFAULT 0 ) ) self.conn.execute( CREATE INDEX IF NOT EXISTS idx_type ON memories(memory_type) ) self.conn.execute( CREATE INDEX IF NOT EXISTS idx_timestamp ON memories(timestamp) ) self.conn.commit() def store(self, content, memory_type, session_idNone, metadataNone, ttl0, embeddingNone): mem_id str(uuid4()) ts datetime.utcnow().isoformat() self.conn.execute( INSERT INTO memories VALUES (?,?,?,?,?,?,?,?), (mem_id, ts, session_id, memory_type, content, json.dumps(metadata or {}), ttl, 0) ) self.conn.commit() if embedding is not None: self.vectors[mem_id] np.array(embedding, dtypenp.float32) return mem_id def cleanup_expired(self): now datetime.utcnow() rows self.conn.execute( SELECT id, timestamp, ttl FROM memories WHERE ttl 0 ).fetchall() expired [] for row in rows: ts datetime.fromisoformat(row[1]) if now ts timedelta(secondsrow[2]): expired.append(row[0]) for mem_id in expired: self.conn.execute(DELETE FROM memories WHERE id?, (mem_id,)) self.vectors.pop(mem_id, None) self.conn.commit() return len(expired)4.3 检索层实现混合排序算法检索层的核心是把向量相似度、关键词匹配、时间衰减、热度加权四个信号融合成一个综合得分。# retrieval.py 核心逻辑 import numpy as np from rank_bm25 import BM25Okapi from datetime import datetime import math class MemoryRetriever: def __init__(self, store, embedder): self.store store self.embedder embedder self.bm25 None self.bm25_ids [] self._build_bm25_index() def _build_bm25_index(self): rows self.store.conn.execute( SELECT id, content FROM memories ).fetchall() self.bm25_ids [r[0] for r in rows] corpus [r[1].lower().split() for r in rows] if corpus: self.bm25 BM25Okapi(corpus) def retrieve(self, query, top_k5, memory_typeNone): # 向量检索 query_vec self.embedder.encode(query) vec_scores {} for mem_id, vec in self.store.vectors.items(): sim np.dot(query_vec, vec) / ( np.linalg.norm(query_vec) * np.linalg.norm(vec) 1e-8 ) vec_scores[mem_id] float(sim) # 关键词检索 kw_scores {} if self.bm25: tokens query.lower().split() scores self.bm25.get_scores(tokens) for i, mem_id in enumerate(self.bm25_ids): kw_scores[mem_id] float(scores[i]) # 合并 all_ids set(vec_scores.keys()) | set(kw_scores.keys()) combined {} for mem_id in all_ids: v vec_scores.get(mem_id, 0) k kw_scores.get(mem_id, 0) # 归一化后加权 combined[mem_id] 0.7 * v 0.3 * min(k / 10.0, 1.0) # 时间衰减和热度加权 now datetime.utcnow() final {} for mem_id, score in combined.items(): row self.store.conn.execute( SELECT timestamp, access_count, memory_type FROM memories WHERE id?, (mem_id,) ).fetchone() if not row: continue if memory_type and row[2] ! memory_type: continue ts datetime.fromisoformat(row[0]) days_ago (now - ts).total_seconds() / 86400 decay math.exp(-0.01 * days_ago) heat math.log(1 row[1]) final[mem_id] score * decay * (1 0.1 * heat) # 排序取Top-K sorted_ids sorted(final.keys(), keylambda x: final[x], reverseTrue) results [] for mem_id in sorted_ids[:top_k]: row self.store.conn.execute( SELECT id, timestamp, content, memory_type, metadata FROM memories WHERE id?, (mem_id,) ).fetchone() results.append({ id: row[0], timestamp: row[1], content: row[2], memory_type: row[3], metadata: row[4], score: final[mem_id] }) # 更新访问计数 self.store.conn.execute( UPDATE memories SET access_count access_count 1 WHERE id?, (mem_id,) ) self.store.conn.commit() return results4.4 MCP Server的启动与接入main.py里用FastAPI起一个HTTP服务同时实现MCP协议的处理逻辑。MCP协议本身支持多种传输方式我用的是HTTPSSEServer-Sent Events因为Docker网络里HTTP最省事。# main.py 核心逻辑 from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse import json from src.storage import MemoryStore from src.retrieval import MemoryRetriever from src.models import get_embedder app FastAPI() store MemoryStore(/app/data/memory.db) embedder get_embedder(bge-m3) retriever MemoryRetriever(store, embedder) app.post(/mcp/tools/call) async def call_tool(request: Request): body await request.json() tool_name body.get(name) args body.get(arguments, {}) if tool_name store_memory: embedding embedder.encode(args[content]).tolist() mem_id store.store( contentargs[content], memory_typeargs.get(memory_type, episodic), metadataargs.get(metadata), ttlargs.get(ttl, 0), embeddingembedding ) return {result: {id: mem_id, status: stored}} elif tool_name retrieve_memory: results retriever.retrieve( queryargs[query], top_kargs.get(top_k, 5), memory_typeargs.get(memory_type) ) return {result: results} elif tool_name cleanup_expired: count store.cleanup_expired() return {result: {cleaned: count}} return {error: fUnknown tool: {tool_name}} app.get(/mcp/tools/list) async def list_tools(): return { tools: [ {name: store_memory, description: 存储记忆}, {name: retrieve_memory, description: 检索记忆}, {name: cleanup_expired, description: 清理过期记忆} ] }启动命令cd hindsight-memory docker compose up -d --build验证服务是否正常curl http://localhost:8765/mcp/tools/list返回工具列表就说明MCP Server跑起来了。然后在Agent端配置MCP连接指向http://localhost:8765即可。4.5 与Agent框架的对接示例以LangChain为例接入MCP记忆服务的代码大概长这样from langchain.agents import Tool, AgentExecutor from langchain_openai import ChatOpenAI import requests def store_memory(content: str, memory_type: str episodic) - str: resp requests.post(http://localhost:8765/mcp/tools/call, json{ name: store_memory, arguments: {content: content, memory_type: memory_type} }) return resp.json()[result][status] def retrieve_memory(query: str) - str: resp requests.post(http://localhost:8765/mcp/tools/call, json{ name: retrieve_memory, arguments: {query: query, top_k: 5} }) results resp.json()[result] return \n.join([r[content] for r in results]) tools [ Tool(namestore_memory, funcstore_memory, description存储重要信息到长期记忆), Tool(nameretrieve_memory, funcretrieve_memory, description从长期记忆中检索相关信息) ] llm ChatOpenAI(modelgpt-4) agent AgentExecutor.from_agent_and_tools( agentllm.bind_tools(tools), toolstools )这样Agent在对话过程中就能自主决定何时存储、何时检索了。5. 常见问题与排查技巧实录5.1 Docker相关的高频问题问题一Docker Desktop启动报“Virtualization support not detected”这是Windows上最常见的问题。原因通常是BIOS里没开虚拟化或者WSL2没装好。解决步骤重启进BIOS找到Intel VT-x或AMD-V选项设为Enabled回到Windows在“启用或关闭Windows功能”里勾选“虚拟机平台”和“适用于Linux的Windows子系统”重启后wsl --update更新内核。问题二容器间网络不通如果Agent跑在宿主机、MCP Server跑在Docker里用localhost:8765访问通常没问题。但如果Agent也在Docker里就需要用Docker网络的服务名访问比如http://mcp-memory:8765。检查方法docker network inspect agent-net看两个容器是否在同一个网络里。问题三SQLite数据库文件权限错误挂载volume的时候容器内用户和宿主机用户UID不一致会导致写不进去。解决办法是在Dockerfile里创建非root用户或者简单粗暴地chmod 777数据目录开发环境可以生产环境别这么干。5.2 记忆检索效果差的排查思路检索不准的时候按这个顺序排查排查项检查方法常见原因Embedding质量手动算两条相似文本的余弦相似度模型选错或维度不匹配向量是否正常存储查vectors字典的size存储时embedding为NoneBM25索引是否更新检查bm25_ids长度新记忆没重建索引时间衰减是否过强调大lambda值测试老记忆被过度惩罚权重分配是否合理调整0.7/0.3的比例向量和关键词权重失衡我遇到过一次检索完全失效的情况排查了半天发现是embedding模型加载失败返回了全零向量。全零向量之间的余弦相似度是NaN导致排序完全乱掉。后来在代码里加了断言embedding的norm小于1e-6就直接报错避免静默失败。5.3 记忆库膨胀的控制策略跑了一段时间之后记忆库会越来越大检索速度下降。几个控制手段第一严格执行TTL。Working Memory设1小时Episodic Memory设30天定期跑cleanup_expired。我写了个cron任务每天凌晨清理一次。第二去重合并。相似度超过0.95的记忆条目自动合并保留最新的时间戳和最高的access_count。这个逻辑我放在存储层每次store之前先查一下有没有高度相似的已有记忆。第三分层归档。超过90天的Episodic Memory转移到冷存储表检索时默认不查需要时手动指定。这样热数据保持精简检索速度快。第四语义提炼。把多条相关的情景记忆提炼成一条语义记忆原始情景记忆可以删除或归档。比如用户提了五次“喜欢用深色主题”提炼成一条“用户偏好深色主题”的语义记忆五条原始记录就可以清理了。5.4 MCP协议对接的坑MCP协议还在快速演进中不同版本的字段定义可能有差异。我遇到过的几个问题一是工具描述格式不兼容。有些客户端要求description字段必须有有些要求parameters必须是JSON Schema格式。建议严格按照MCP官方文档的示例来写别自己发挥。二是SSE连接超时。HTTPSSE的长连接如果超过一定时间没有数据传输中间的网络设备可能会断开。解决办法是加心跳每隔30秒发一个空事件保持连接。三是并发调用冲突。多个Agent同时调store_memory的时候SQLite的写锁可能导致超时。我的做法是在存储层加一个简单的队列写操作串行化。对于读多写少的场景这个方案够用了。提示MCP Server的日志一定要打详细每个工具调用的入参和出参都记下来。排查问题时日志是最可靠的线索。我用logging模块把日志同时输出到文件和stdoutDocker的docker logs命令就能直接看。5.5 性能优化的几个实测数据在我的测试环境MacBook Pro M1, 16GB RAM上几个关键指标单条记忆存储含embedding计算约50ms检索Top-5记忆库1000条约30ms检索Top-5记忆库10000条约120ms清理过期记忆10000条中清理100条约200ms当记忆库超过5万条时numpy暴力检索的延迟会超过500ms这时候就需要考虑上专门的向量数据库了。我的建议是先用SQLitenumpy快速验证等数据量真的上来了再迁移到Qdrant或Milvus不要过早优化。6. 几个我踩过的坑和对应的解法第一个坑是embedding模型的热加载。最开始每次检索都重新加载模型导致第一次检索要等好几秒。后来改成服务启动时加载一次全局复用检索延迟直接降到毫秒级。这个改动很简单但效果立竿见影。第二个坑是时间戳的时区问题。我用datetime.utcnow()存UTC时间但检索时用本地时间比较导致时间衰减计算错误。统一用UTC之后问题解决。建议所有时间戳都存UTC展示的时候再转本地时区。第三个坑是记忆内容的截断。有些记忆条目特别长比如一整段代码存进去之后检索出来占满了context window。后来加了长度限制超过500字符的内容自动摘要后再存储原始内容存到metadata里备查。第四个坑是MCP Server的重启丢数据。最开始向量存在内存里容器一重启就没了。改成持久化到磁盘后解决。SQLite的数据文件挂载到宿主机向量用numpy的save/load存成.npy文件重启后自动加载。第五个坑是并发写入的竞态条件。两个请求同时写同一条记忆的access_count导致计数丢失。加了个简单的线程锁之后解决。Python的GIL在IO密集型场景下保护有限涉及共享状态的地方该加锁就得加锁。这套记忆服务我跑了大概三个月累计存储了上万条记忆记录检索准确率稳定在90%以上。最大的体会是Agent Memory不是越复杂越好关键是找到适合当前场景的粒度。一开始就上全套的语义提炼、自动归档、多级缓存反而容易出bug。先用最简单的方案跑通遇到问题再逐步加机制这样每一步的收益和代价都清清楚楚。 
长河君 · 2026/10/3 11:42:52 · 阅读 1.2k 基于MCP协议与Docker的Agent Memory分层记忆架构实战
1. 从“hindsight”说起为什么我们需要给Agent装上“后视镜”“hindsight”这个词本身的意思就是“事后之明”——事情发生之后才明白当初应该怎么做。把这个词用在Agent Memory智能体记忆这个领域指向性非常明确我们要解决的是LLM驱动的Agent在长期运行中“记不住、想不起、用不对”的核心痛点。你如果最近在折腾LLM Agent相关的项目大概率已经踩过这样的坑——Agent在单轮对话里表现惊艳一旦对话轮次拉长到几十轮它就开始胡言乱语前面明确说过的约束条件被抛到九霄云外用户纠正过的错误反复再犯。这不是模型能力不够而是记忆架构没设计好。我最初接触这个方向是因为一个实际需求做一个能持续跟踪项目进展的辅助Agent它需要记住过去几周内用户提过的所有需求变更、技术决策和踩坑记录。用最朴素的方式——把全部历史对话塞进context window——很快就撞墙了。Token消耗爆炸不说模型对超长上下文的注意力衰减非常明显关键信息淹没在大量无关对话里。后来试过简单的摘要压缩又发现摘要会丢失细节等到真正需要某个具体参数的时候摘要里根本没保留。这就是“hindsight”要解决的核心问题如何让Agent像人一样在需要的时候能准确回忆起过去发生的关键事件并且理解这些事件之间的因果关系。这篇文章适合几类人看一是正在做LLM Agent应用开发、被记忆问题困扰的工程师二是对Agent架构设计感兴趣、想了解记忆模块怎么落地的人三是已经在用Docker部署各种服务、想把手头的LLM工具链整合起来的实践者。我会从整体设计思路讲起然后拆解核心细节接着给出完整的实操流程最后分享我在排查问题时积累的经验。文中涉及的技术栈包括LLM、MCP协议、Docker容器化部署以及Agent Memory的分层存储设计。2. 整体设计与思路拆解Agent Memory到底该怎么分层2.1 为什么不能只靠Context Window硬撑很多人刚开始做Agent的时候第一反应就是把所有历史对话都拼到prompt里。短对话没问题一旦超过模型的有效上下文长度要么截断丢信息要么就得换更大上下文的模型成本直线上升。更关键的是即使模型支持128K甚至更长的上下文中间部分的信息检索准确率也会显著下降——这是注意力机制的固有特性不是换个模型就能解决的。我做过一个简单的对比测试让Agent记住20轮对话前用户提到的一个特定数字然后在一段无关对话之后询问这个数字。把全部历史塞进context的方案准确率大概在60%左右而采用分层记忆架构、把关键信息单独存储并在需要时检索注入的方案准确率能到95%以上。这个差距在真实应用里就是“能用”和“不能用”的区别。所以核心思路很明确Context Window只放当前最相关的信息长期记忆放到外部存储通过检索机制按需注入。这就是hindsight这类Agent Memory系统的设计起点。2.2 三层记忆架构Working Memory、Episodic Memory、Semantic Memory参考认知科学里对人类记忆的分类我在实际项目中把Agent Memory分成三层Working Memory工作记忆当前对话轮次直接相关的信息放在context window里。这部分容量有限需要严格控制。通常只保留最近几轮对话的原文加上从长期记忆中检索出来的相关片段。Episodic Memory情景记忆具体发生过的事件记录。比如“用户在第三轮对话中要求把数据库从MySQL换成PostgreSQL”、“第七轮对话中用户反馈某个API返回格式不对”。这些记录带时间戳、带上下文按事件粒度存储。Semantic Memory语义记忆从多个事件中抽象出来的通用知识。比如“这个用户偏好使用ORM而不是手写SQL”、“这个项目的API设计遵循RESTful风格”。这部分是对情景记忆的进一步提炼存储的是结论而非原始事件。三层之间的流转关系是这样的对话进行时原始信息先进入Working Memory对话结束后关键事件被抽取出来存入Episodic Memory当同类事件积累到一定数量系统自动或手动触发提炼生成Semantic Memory。检索的时候优先查Semantic Memory因为信息密度高不够再查Episodic Memory最后才回退到Working Memory。注意不要一上来就追求全自动的语义提炼。我踩过的坑是自动提炼的语义记忆经常出现过度概括或错误归纳反而污染了记忆库。建议初期以手动或半自动方式管理Semantic Memory等积累足够多的样本后再考虑自动化。2.3 为什么选择MCP协议做记忆服务的接口层MCPModel Context Protocol是Anthropic推出的一个开放协议用来标准化LLM与外部工具、数据源之间的交互方式。我选择用MCP来封装记忆服务主要考虑几点第一解耦。记忆服务作为一个独立的MCP Server运行LLM Agent通过标准协议调用不依赖具体的Agent框架。今天用LangChain明天换AutoGen记忆服务不用改。第二可组合。MCP Server可以同时暴露多个工具比如store_memory、retrieve_memory、summarize_episodesAgent根据需要调用。而且多个MCP Server可以并行运行记忆服务可以和文件系统服务、数据库服务共存。第三生态兼容。现在越来越多的工具支持MCP协议包括各种IDE插件、浏览器扩展、命令行工具。把记忆服务做成MCP Server意味着任何支持MCP的客户端都能直接接入。具体实现上我用Python写了一个MCP Server底层存储用SQLite轻量、零配置加向量数据库用于语义检索。SQLite存结构化的事件记录向量库存embedding用于相似度搜索。两者通过事件ID关联。2.4 Docker化部署为什么不用裸机跑把记忆服务跑在Docker容器里好处是环境隔离和可移植性。我试过直接在宿主机上装依赖结果Python版本冲突、SQLite扩展加载失败、向量数据库的C库版本不匹配折腾了一整天。换成Docker之后所有依赖打包在镜像里换台机器docker compose up就能跑起来。另外Docker的网络管理让MCP Server和Agent之间的通信更可控。我可以把记忆服务放在一个内部网络里只暴露必要的端口Agent通过服务名访问不用关心IP地址。如果后续要加认证和限流在Docker网络层面做也比在应用层做更干净。3. 核心细节解析与实操要点从存储格式到检索策略3.1 记忆条目的数据结构设计每条记忆记录我设计了这些字段字段名类型说明idUUID全局唯一标识timestampISO8601事件发生时间session_idString所属会话标识memory_typeEnumworking/episodic/semanticcontentText记忆正文embeddingVector语义向量维度768metadataJSON扩展字段如来源、置信度、标签ttlInteger过期时间秒0表示永不过期access_countInteger被检索次数用于热度排序ttl字段的设计很关键。Working Memory的记录通常设置较短的TTL比如1小时过期自动清理。Episodic Memory的TTL可以设长一些比如30天Semantic Memory则永不过期。这样能自动控制存储增长避免记忆库无限膨胀。access_count用来做热度加权。检索的时候除了向量相似度还会考虑这条记忆被访问的频率。经常被用到的记忆说明价值高排序时应该靠前。3.2 向量化模型的选择与权衡Embedding模型我试过好几个最后选的是text-embedding-3-smallOpenAI和bge-m3本地部署两个方案并行。原因如下OpenAI的方案优点是稳定、无需维护适合快速验证。缺点是API调用有延迟和成本而且数据要出境如果在意的话。本地部署bge-m3的优点是数据不出本地、无API成本缺点是需要GPU资源推理速度取决于硬件。实际使用中我做了个路由策略对延迟敏感的检索走本地模型对精度要求高的走API模型。两者生成的向量存在不同的表里检索时分别查询再合并结果。提示embedding维度不一致会导致无法直接比较相似度。如果混用不同模型务必在metadata里标记embedding来源检索时按来源分组查询。3.3 检索策略向量相似度关键词时间衰减单纯的向量相似度检索有个问题它擅长语义匹配但对精确的关键词匹配不够敏感。比如用户问“上次说的那个端口号是多少”向量检索可能返回一堆关于“端口”的讨论但真正包含具体端口号的那条记录可能排在后面。我的做法是混合检索向量检索用query的embedding去向量库搜Top-KK取20。关键词检索用BM25算法在content字段上做全文搜索取Top-20。合并去重两路结果按id合并计算综合得分。时间衰减综合得分乘以时间衰减因子越久远的记忆得分越低。衰减公式我用的是score * exp(-lambda * days_ago)lambda取0.01意味着大约70天后权重降到一半。热度加权再乘以log(1 access_count)让高频访问的记忆获得加成。最终排序取Top-5返回给Agent。这套组合拳实测下来检索准确率比单用向量检索提升了大概20个百分点。尤其是在需要精确回忆具体数值、名称、日期的场景下关键词检索的补充作用非常明显。3.4 MCP Server的工具定义MCP Server暴露给Agent的工具我定义了四个# 工具1存储记忆 { name: store_memory, description: 存储一条新的记忆记录, parameters: { content: string, 记忆正文, memory_type: string, 枚举值: working/episodic/semantic, metadata: object, 可选扩展字段, ttl: integer, 可选过期时间(秒) } } # 工具2检索记忆 { name: retrieve_memory, description: 根据查询检索相关记忆, parameters: { query: string, 查询文本, top_k: integer, 返回条数, 默认5, memory_type: string, 可选过滤类型 } } # 工具3提炼语义记忆 { name: summarize_episodes, description: 将多条情景记忆提炼为语义记忆, parameters: { episode_ids: array, 情景记忆ID列表, instruction: string, 提炼指令 } } # 工具4清理过期记忆 { name: cleanup_expired, description: 清理已过期的记忆记录, parameters: {} }Agent在对话过程中根据上下文自主决定何时调用这些工具。比如用户说“记住这个配置”Agent就调store_memory用户问“之前那个配置是什么”Agent就调retrieve_memory。4. 实操过程与核心环节实现从零搭建一个可用的记忆服务4.1 环境准备与Docker Compose编排先确保Docker Desktop已经装好并且能正常启动。Windows上如果遇到“Virtualization support not detected”的报错需要进BIOS开启虚拟化支持Intel VT-x或AMD-V然后在Windows功能里启用WSL2。Mac上相对简单装好Docker Desktop即可。项目目录结构如下hindsight-memory/ ├── docker-compose.yml ├── mcp-server/ │ ├── Dockerfile │ ├── requirements.txt │ └── src/ │ ├── main.py │ ├── storage.py │ ├── retrieval.py │ └── models.py ├── vector-db/ │ └── data/ └── config/ └── settings.yamldocker-compose.yml的内容version: 3.8 services: mcp-memory: build: ./mcp-server container_name: hindsight-mcp ports: - 8765:8765 volumes: - ./vector-db/data:/app/data - ./config:/app/config environment: - EMBEDDING_MODELbge-m3 - DB_PATH/app/data/memory.db - VECTOR_DIM768 networks: - agent-net restart: unless-stopped networks: agent-net: driver: bridgeDockerfileFROM python:3.11-slim WORKDIR /app RUN apt-get update apt-get install -y \ build-essential \ rm -rf /var/lib/apt/lists/* COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY src/ ./src/ COPY config/ ./config/ EXPOSE 8765 CMD [python, -m, src.main]requirements.txtmcp0.1.0 fastapi0.104.0 uvicorn0.24.0 sqlite-utils3.35 numpy1.24 sentence-transformers2.2.0 rank-bm250.2.2 pydantic2.04.2 存储层实现SQLite向量的混合方案存储层我用SQLite做结构化存储向量部分用numpy数组存在单独的文件里通过id关联。这样做的原因是SQLite对向量类型的支持还不够成熟而引入专门的向量数据库如Milvus、Qdrant又增加了部署复杂度。对于中小规模的记忆库几万条以内numpy数组加暴力检索完全够用。# storage.py 核心逻辑 import sqlite3 import numpy as np import json from datetime import datetime, timedelta from uuid import uuid4 class MemoryStore: def __init__(self, db_path, vector_dim768): self.conn sqlite3.connect(db_path, check_same_threadFalse) self.vector_dim vector_dim self._init_tables() self.vectors {} # id - np.array def _init_tables(self): self.conn.execute( CREATE TABLE IF NOT EXISTS memories ( id TEXT PRIMARY KEY, timestamp TEXT NOT NULL, session_id TEXT, memory_type TEXT NOT NULL, content TEXT NOT NULL, metadata TEXT, ttl INTEGER DEFAULT 0, access_count INTEGER DEFAULT 0 ) ) self.conn.execute( CREATE INDEX IF NOT EXISTS idx_type ON memories(memory_type) ) self.conn.execute( CREATE INDEX IF NOT EXISTS idx_timestamp ON memories(timestamp) ) self.conn.commit() def store(self, content, memory_type, session_idNone, metadataNone, ttl0, embeddingNone): mem_id str(uuid4()) ts datetime.utcnow().isoformat() self.conn.execute( INSERT INTO memories VALUES (?,?,?,?,?,?,?,?), (mem_id, ts, session_id, memory_type, content, json.dumps(metadata or {}), ttl, 0) ) self.conn.commit() if embedding is not None: self.vectors[mem_id] np.array(embedding, dtypenp.float32) return mem_id def cleanup_expired(self): now datetime.utcnow() rows self.conn.execute( SELECT id, timestamp, ttl FROM memories WHERE ttl 0 ).fetchall() expired [] for row in rows: ts datetime.fromisoformat(row[1]) if now ts timedelta(secondsrow[2]): expired.append(row[0]) for mem_id in expired: self.conn.execute(DELETE FROM memories WHERE id?, (mem_id,)) self.vectors.pop(mem_id, None) self.conn.commit() return len(expired)4.3 检索层实现混合排序算法检索层的核心是把向量相似度、关键词匹配、时间衰减、热度加权四个信号融合成一个综合得分。# retrieval.py 核心逻辑 import numpy as np from rank_bm25 import BM25Okapi from datetime import datetime import math class MemoryRetriever: def __init__(self, store, embedder): self.store store self.embedder embedder self.bm25 None self.bm25_ids [] self._build_bm25_index() def _build_bm25_index(self): rows self.store.conn.execute( SELECT id, content FROM memories ).fetchall() self.bm25_ids [r[0] for r in rows] corpus [r[1].lower().split() for r in rows] if corpus: self.bm25 BM25Okapi(corpus) def retrieve(self, query, top_k5, memory_typeNone): # 向量检索 query_vec self.embedder.encode(query) vec_scores {} for mem_id, vec in self.store.vectors.items(): sim np.dot(query_vec, vec) / ( np.linalg.norm(query_vec) * np.linalg.norm(vec) 1e-8 ) vec_scores[mem_id] float(sim) # 关键词检索 kw_scores {} if self.bm25: tokens query.lower().split() scores self.bm25.get_scores(tokens) for i, mem_id in enumerate(self.bm25_ids): kw_scores[mem_id] float(scores[i]) # 合并 all_ids set(vec_scores.keys()) | set(kw_scores.keys()) combined {} for mem_id in all_ids: v vec_scores.get(mem_id, 0) k kw_scores.get(mem_id, 0) # 归一化后加权 combined[mem_id] 0.7 * v 0.3 * min(k / 10.0, 1.0) # 时间衰减和热度加权 now datetime.utcnow() final {} for mem_id, score in combined.items(): row self.store.conn.execute( SELECT timestamp, access_count, memory_type FROM memories WHERE id?, (mem_id,) ).fetchone() if not row: continue if memory_type and row[2] ! memory_type: continue ts datetime.fromisoformat(row[0]) days_ago (now - ts).total_seconds() / 86400 decay math.exp(-0.01 * days_ago) heat math.log(1 row[1]) final[mem_id] score * decay * (1 0.1 * heat) # 排序取Top-K sorted_ids sorted(final.keys(), keylambda x: final[x], reverseTrue) results [] for mem_id in sorted_ids[:top_k]: row self.store.conn.execute( SELECT id, timestamp, content, memory_type, metadata FROM memories WHERE id?, (mem_id,) ).fetchone() results.append({ id: row[0], timestamp: row[1], content: row[2], memory_type: row[3], metadata: row[4], score: final[mem_id] }) # 更新访问计数 self.store.conn.execute( UPDATE memories SET access_count access_count 1 WHERE id?, (mem_id,) ) self.store.conn.commit() return results4.4 MCP Server的启动与接入main.py里用FastAPI起一个HTTP服务同时实现MCP协议的处理逻辑。MCP协议本身支持多种传输方式我用的是HTTPSSEServer-Sent Events因为Docker网络里HTTP最省事。# main.py 核心逻辑 from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse import json from src.storage import MemoryStore from src.retrieval import MemoryRetriever from src.models import get_embedder app FastAPI() store MemoryStore(/app/data/memory.db) embedder get_embedder(bge-m3) retriever MemoryRetriever(store, embedder) app.post(/mcp/tools/call) async def call_tool(request: Request): body await request.json() tool_name body.get(name) args body.get(arguments, {}) if tool_name store_memory: embedding embedder.encode(args[content]).tolist() mem_id store.store( contentargs[content], memory_typeargs.get(memory_type, episodic), metadataargs.get(metadata), ttlargs.get(ttl, 0), embeddingembedding ) return {result: {id: mem_id, status: stored}} elif tool_name retrieve_memory: results retriever.retrieve( queryargs[query], top_kargs.get(top_k, 5), memory_typeargs.get(memory_type) ) return {result: results} elif tool_name cleanup_expired: count store.cleanup_expired() return {result: {cleaned: count}} return {error: fUnknown tool: {tool_name}} app.get(/mcp/tools/list) async def list_tools(): return { tools: [ {name: store_memory, description: 存储记忆}, {name: retrieve_memory, description: 检索记忆}, {name: cleanup_expired, description: 清理过期记忆} ] }启动命令cd hindsight-memory docker compose up -d --build验证服务是否正常curl http://localhost:8765/mcp/tools/list返回工具列表就说明MCP Server跑起来了。然后在Agent端配置MCP连接指向http://localhost:8765即可。4.5 与Agent框架的对接示例以LangChain为例接入MCP记忆服务的代码大概长这样from langchain.agents import Tool, AgentExecutor from langchain_openai import ChatOpenAI import requests def store_memory(content: str, memory_type: str episodic) - str: resp requests.post(http://localhost:8765/mcp/tools/call, json{ name: store_memory, arguments: {content: content, memory_type: memory_type} }) return resp.json()[result][status] def retrieve_memory(query: str) - str: resp requests.post(http://localhost:8765/mcp/tools/call, json{ name: retrieve_memory, arguments: {query: query, top_k: 5} }) results resp.json()[result] return \n.join([r[content] for r in results]) tools [ Tool(namestore_memory, funcstore_memory, description存储重要信息到长期记忆), Tool(nameretrieve_memory, funcretrieve_memory, description从长期记忆中检索相关信息) ] llm ChatOpenAI(modelgpt-4) agent AgentExecutor.from_agent_and_tools( agentllm.bind_tools(tools), toolstools )这样Agent在对话过程中就能自主决定何时存储、何时检索了。5. 常见问题与排查技巧实录5.1 Docker相关的高频问题问题一Docker Desktop启动报“Virtualization support not detected”这是Windows上最常见的问题。原因通常是BIOS里没开虚拟化或者WSL2没装好。解决步骤重启进BIOS找到Intel VT-x或AMD-V选项设为Enabled回到Windows在“启用或关闭Windows功能”里勾选“虚拟机平台”和“适用于Linux的Windows子系统”重启后wsl --update更新内核。问题二容器间网络不通如果Agent跑在宿主机、MCP Server跑在Docker里用localhost:8765访问通常没问题。但如果Agent也在Docker里就需要用Docker网络的服务名访问比如http://mcp-memory:8765。检查方法docker network inspect agent-net看两个容器是否在同一个网络里。问题三SQLite数据库文件权限错误挂载volume的时候容器内用户和宿主机用户UID不一致会导致写不进去。解决办法是在Dockerfile里创建非root用户或者简单粗暴地chmod 777数据目录开发环境可以生产环境别这么干。5.2 记忆检索效果差的排查思路检索不准的时候按这个顺序排查排查项检查方法常见原因Embedding质量手动算两条相似文本的余弦相似度模型选错或维度不匹配向量是否正常存储查vectors字典的size存储时embedding为NoneBM25索引是否更新检查bm25_ids长度新记忆没重建索引时间衰减是否过强调大lambda值测试老记忆被过度惩罚权重分配是否合理调整0.7/0.3的比例向量和关键词权重失衡我遇到过一次检索完全失效的情况排查了半天发现是embedding模型加载失败返回了全零向量。全零向量之间的余弦相似度是NaN导致排序完全乱掉。后来在代码里加了断言embedding的norm小于1e-6就直接报错避免静默失败。5.3 记忆库膨胀的控制策略跑了一段时间之后记忆库会越来越大检索速度下降。几个控制手段第一严格执行TTL。Working Memory设1小时Episodic Memory设30天定期跑cleanup_expired。我写了个cron任务每天凌晨清理一次。第二去重合并。相似度超过0.95的记忆条目自动合并保留最新的时间戳和最高的access_count。这个逻辑我放在存储层每次store之前先查一下有没有高度相似的已有记忆。第三分层归档。超过90天的Episodic Memory转移到冷存储表检索时默认不查需要时手动指定。这样热数据保持精简检索速度快。第四语义提炼。把多条相关的情景记忆提炼成一条语义记忆原始情景记忆可以删除或归档。比如用户提了五次“喜欢用深色主题”提炼成一条“用户偏好深色主题”的语义记忆五条原始记录就可以清理了。5.4 MCP协议对接的坑MCP协议还在快速演进中不同版本的字段定义可能有差异。我遇到过的几个问题一是工具描述格式不兼容。有些客户端要求description字段必须有有些要求parameters必须是JSON Schema格式。建议严格按照MCP官方文档的示例来写别自己发挥。二是SSE连接超时。HTTPSSE的长连接如果超过一定时间没有数据传输中间的网络设备可能会断开。解决办法是加心跳每隔30秒发一个空事件保持连接。三是并发调用冲突。多个Agent同时调store_memory的时候SQLite的写锁可能导致超时。我的做法是在存储层加一个简单的队列写操作串行化。对于读多写少的场景这个方案够用了。提示MCP Server的日志一定要打详细每个工具调用的入参和出参都记下来。排查问题时日志是最可靠的线索。我用logging模块把日志同时输出到文件和stdoutDocker的docker logs命令就能直接看。5.5 性能优化的几个实测数据在我的测试环境MacBook Pro M1, 16GB RAM上几个关键指标单条记忆存储含embedding计算约50ms检索Top-5记忆库1000条约30ms检索Top-5记忆库10000条约120ms清理过期记忆10000条中清理100条约200ms当记忆库超过5万条时numpy暴力检索的延迟会超过500ms这时候就需要考虑上专门的向量数据库了。我的建议是先用SQLitenumpy快速验证等数据量真的上来了再迁移到Qdrant或Milvus不要过早优化。6. 几个我踩过的坑和对应的解法第一个坑是embedding模型的热加载。最开始每次检索都重新加载模型导致第一次检索要等好几秒。后来改成服务启动时加载一次全局复用检索延迟直接降到毫秒级。这个改动很简单但效果立竿见影。第二个坑是时间戳的时区问题。我用datetime.utcnow()存UTC时间但检索时用本地时间比较导致时间衰减计算错误。统一用UTC之后问题解决。建议所有时间戳都存UTC展示的时候再转本地时区。第三个坑是记忆内容的截断。有些记忆条目特别长比如一整段代码存进去之后检索出来占满了context window。后来加了长度限制超过500字符的内容自动摘要后再存储原始内容存到metadata里备查。第四个坑是MCP Server的重启丢数据。最开始向量存在内存里容器一重启就没了。改成持久化到磁盘后解决。SQLite的数据文件挂载到宿主机向量用numpy的save/load存成.npy文件重启后自动加载。第五个坑是并发写入的竞态条件。两个请求同时写同一条记忆的access_count导致计数丢失。加了个简单的线程锁之后解决。Python的GIL在IO密集型场景下保护有限涉及共享状态的地方该加锁就得加锁。这套记忆服务我跑了大概三个月累计存储了上万条记忆记录检索准确率稳定在90%以上。最大的体会是Agent Memory不是越复杂越好关键是找到适合当前场景的粒度。一开始就上全套的语义提炼、自动归档、多级缓存反而容易出bug。先用最简单的方案跑通遇到问题再逐步加机制这样每一步的收益和代价都清清楚楚。