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

1. 从“hindsight”说起：为什么记忆是 Agent 落地的最后一公里

“hindsight”这个词本身很有意思，字面意思是“事后的洞察力”，也就是我们常说的“后见之明”。把这个词放到 Agent Memory 这个语境里，它指向的东西非常具体：一个 Agent 在完成一轮任务之后，能不能把刚才发生的事、踩过的坑、验证过的结论沉淀下来，在下一轮任务里直接调用，而不是每次都从零开始。

我接触过不少做 LLM 应用的朋友，大家一开始都把精力砸在 Prompt 调优和模型选型上，等到真正要跑一个持续性的任务流时才发现，Agent 的“失忆”才是最大的拦路虎。你让它帮你分析一份财报，它分析得头头是道；第二天你再让它基于昨天的结论做延伸，它一脸茫然地反问你“什么财报”。这不是模型能力的问题，是记忆架构的问题。

“hindsight”这个项目标题，我理解它要解决的核心痛点就是：让 Agent 具备跨会话、跨任务的长期记忆能力，并且这种记忆不是简单的聊天记录堆砌，而是经过结构化、可检索、可推理的知识沉淀。它适合谁看？如果你正在用 LLM 框架搭 Agent，或者你在用 MCP 协议做工具编排，又或者你单纯对“Agent 怎么记住东西”这件事好奇，那这篇内容就是写给你的。

我下面会从整体设计思路、核心细节拆解、实操落地、问题排查几个维度，把“hindsight”这类 Agent Memory 方案的里里外外讲透。中间会涉及 MCP 协议、Docker 部署、存储分层这些具体技术点，也会分享一些我自己踩过的坑。

2. Agent Memory 的整体设计与思路拆解

2.1 为什么传统 RAG 撑不起 Agent 的长期记忆

很多人一提到“让 LLM 记住东西”，第一反应就是上 RAG：把历史对话切块、向量化、存进向量库，下次检索 top-k 塞进上下文。这个方案在问答场景里够用，但放到 Agent 场景里就捉襟见肘了。

原因有三。第一，RAG 是无状态的，它不知道哪些记忆是“已经过时”的，哪些是“被修正过”的。你昨天告诉 Agent “项目 A 的预算是 100 万”，今天改成 “120 万”，RAG 会把两条都检索出来，模型自己都懵。第二，RAG 缺乏时间维度，它检索的是语义相似度，不是时间新鲜度，导致旧信息经常压过新信息。第三，RAG 不做推理，它只是把原文片段捞出来，不会把“用户上周提到喜欢简洁风格”和“用户今天要求写一份报告”这两条信息合成为“报告要写得简洁”。

“hindsight”这类方案的设计出发点，就是要把记忆从“被动检索”升级为“主动管理”。它需要回答三个核心问题：存什么、怎么存、怎么取。

2.2 记忆分层：Working Memory 与 Long-term Memory 的职责划分

我在实际项目里最常用的分层方式是参考认知科学的模型，把 Agent 记忆分成三层：

Working Memory 就是热数据，读写要快，通常放内存或者 Redis，会话结束就可以清理。Episodic Memory 是“发生了什么”，比如“2024-06-01 用户让我分析了一份新能源行业报告，结论是产能过剩”。Semantic Memory 是“我知道了什么”，比如“用户偏好数据驱动的分析风格”。

“hindsight”这个名字暗示的其实是 Episodic 到 Semantic 的转化过程——事后回顾，从具体事件中提炼出可复用的知识。这个转化过程才是 Agent Memory 真正的技术壁垒，而不是简单的向量检索。

2.3 为什么选 MCP 作为记忆服务的接入协议

MCP（Model Context Protocol）这两年在 Agent 工具编排领域火得很快，它的核心价值是把工具调用标准化。你可以把它理解成“AI 世界的 USB-C 接口”——不管后端是什么服务，只要实现了 MCP Server，任何支持 MCP 的客户端都能直接调用。

把 Agent Memory 做成一个 MCP Server，好处非常明显。第一，解耦，记忆服务独立部署，Agent 框架换了大模型或者换了编排逻辑，记忆层不用动。第二，复用，同一个记忆服务可以同时给多个 Agent 用，比如你的写作 Agent 和数据分析 Agent 共享一套用户偏好记忆。第三，可观测，MCP 协议本身有标准的请求响应格式，调试和监控都方便。

我试过把记忆逻辑直接写死在 Agent 代码里，也试过抽成独立服务，实测下来后者在维护成本上低太多了。尤其是当你有三四个 Agent 在跑的时候，统一记忆服务几乎是唯一可行的方案。

2.4 Docker 化部署的取舍：为什么不用 Serverless

记忆服务有个特点：它需要持久化存储，而且对延迟敏感。Serverless 方案冷启动动辄几百毫秒，对于每次对话都要查记忆的场景来说，这个延迟是不可接受的。Docker 部署可以保证服务常驻，配合 Docker Compose 或者 Kubernetes 做编排，既方便又稳定。

另外，记忆服务通常要连向量库、关系库、缓存，这些依赖在 Docker 网络里配置起来比 Serverless 的环境变量注入要直观得多。我下面会给出一个完整的 Docker Compose 配置，你可以直接抄。

3. 核心细节解析与实操要点

3.1 记忆的写入策略：什么时候该记，什么时候不该记

这是最容易被忽视的环节。很多方案上来就把所有对话都存进去，结果记忆库膨胀得飞快，检索质量还越来越差。我的经验是，写入要过三道过滤：

第一道，重要性过滤。不是每句话都值得记。用户说“你好”不需要记，用户说“我以后所有报告都要加数据来源”必须记。可以用一个轻量的 LLM 调用做重要性打分，或者用规则匹配关键词。

第二道，去重过滤。新记忆写入前，先跟已有记忆做相似度比对，如果相似度超过阈值（我一般设 0.92），就做合并而不是新增。合并策略可以是“新覆盖旧”或者“取并集”，取决于记忆类型。

第三道，时效性过滤。有些记忆是有有效期的，比如“用户这周在出差”，过了这周就该失效。写入时打上 TTL 标签，检索时自动过滤过期项。

def should_write_memory(content, existing_memories, threshold=0.92): # 重要性打分，可以用小模型或者规则 importance = score_importance(content) if importance < 0.3: return False, "low_importance" # 去重检查 for mem in existing_memories: sim = cosine_similarity(embed(content), mem.embedding) if sim > threshold: return False, f"duplicate_of_{mem.id}" return True, "ok"

注意：去重阈值不要设太低，否则会把相关但不相同的记忆误合并。我一开始设 0.85，结果“用户喜欢 Python”和“用户喜欢 Python 的简洁语法”被合并了，丢失了细节。

3.2 记忆的检索策略：不只是向量相似度

检索环节决定了 Agent 能不能“想起”该想起的东西。单纯用向量相似度检索有三个坑：时间盲区、关系盲区、意图盲区。

时间盲区是指，用户问“我上次说的那个方案”，向量检索可能召回半年前的相似内容，而不是最近的那次。解决办法是混合排序，把时间衰减因子加进打分公式：

final_score = semantic_similarity * 0.7 + time_decay * 0.2 + importance * 0.1

其中time_decay = exp(-lambda * days_since_creation)，lambda 一般取 0.05 到 0.1。

关系盲区是指，记忆之间有关联但向量检索看不出来。比如“项目 A 的负责人是张三”和“张三偏好敏捷开发”，这两条记忆单独检索都可能召回，但它们的关联关系需要图数据库来维护。我通常会用 Neo4j 或者轻量的 NetworkX 存实体关系，检索时做一跳或两跳扩展。

意图盲区是指，用户的问题可能对应多种记忆类型。比如“帮我写个报告”，可能需要召回“用户偏好”“历史报告模板”“当前项目背景”三类记忆。这时候需要查询改写，把用户 query 拆成多个子查询，分别检索再合并。

3.3 记忆的更新与遗忘：比写入更难的是维护

记忆库不是只增不减的。我见过一个项目，跑了三个月，记忆库里有 20 万条记录，检索一次要 3 秒，Agent 响应慢得没法用。问题就出在没有遗忘机制。

遗忘策略我一般分三种：

• TTL 过期：给每条记忆打上有效期标签，到期自动归档或删除。适合临时性信息。
• 冲突消解：新记忆与旧记忆矛盾时，标记旧记忆为“已废弃”，检索时降权而不是直接删，保留审计线索。
• 容量淘汰：记忆库超过阈值时，按“重要性 × 最近访问时间”排序，淘汰尾部。这个类似操作系统的 LRU 算法。

-- 冲突消解示例：标记旧记忆为废弃 UPDATE memories SET status = 'deprecated', deprecated_at = NOW(), deprecated_by = :new_memory_id WHERE id IN ( SELECT id FROM memories WHERE entity = :entity AND attribute = :attribute AND status = 'active' );

提示：废弃不要物理删除，因为 Agent 有时候需要知道“曾经有过什么错误认知”，这对调试和审计很重要。

3.4 MCP Server 的接口设计：三个核心 Tool

把记忆服务暴露成 MCP Server，我建议至少实现三个 Tool：

1. memory_write：写入记忆，参数包括 content、memory_type、importance、ttl、metadata。
2. memory_search：检索记忆，参数包括 query、top_k、time_range、memory_types。
3. memory_forget：主动遗忘，参数包括 memory_id 或过滤条件。

接口设计的关键是参数要少而精。我见过有人设计了二十几个参数，结果调用方根本不知道该传什么。MCP 的 Tool 定义要像好的 API 一样，让调用者一眼看懂。

{ "name": "memory_search", "description": "检索 Agent 长期记忆", "inputSchema": { "type": "object", "properties": { "query": {"type": "string", "description": "检索查询"}, "top_k": {"type": "integer", "default": 5}, "memory_types": { "type": "array", "items": {"enum": ["episodic", "semantic", "working"]} } }, "required": ["query"] } }

4. 实操过程与核心环节实现

4.1 环境准备：Docker 与依赖服务

先把基础环境搭起来。我假设你用的是 Ubuntu 或者 macOS，Windows 用户建议用 WSL2，因为 Docker Desktop 在 Windows 上的网络配置有时候会抽风。

# 安装 Docker（Ubuntu） curl -fsSL https://get.docker.com | sh sudo usermod -aG docker $USER newgrp docker # 验证 docker --version docker compose version

如果你用 Windows，装完 Docker Desktop 后记得在设置里开启 WSL2 集成，否则容器访问宿主机文件系统会很慢。我踩过的坑是：Docker Desktop 默认的资源限制太保守，跑向量库的时候经常 OOM，建议在 Settings 里把内存调到 8GB 以上。

接下来是依赖服务。记忆服务通常需要三个后端：PostgreSQL（存结构化记忆）、Redis（存 working memory）、Qdrant（存向量）。用 Docker Compose 一键拉起：

version: '3.8' services: postgres: image: postgres:16 environment: POSTGRES_USER: agent POSTGRES_PASSWORD: agent_pass POSTGRES_DB: memory ports: - "5432:5432" volumes: - pg_data:/var/lib/postgresql/data redis: image: redis:7-alpine ports: - "6379:6379" command: redis-server --maxmemory 512mb --maxmemory-policy allkeys-lru qdrant: image: qdrant/qdrant:latest ports: - "6333:6333" volumes: - qdrant_data:/qdrant/storage volumes: pg_data: qdrant_data:

启动命令：

docker compose up -d docker compose ps # 确认三个服务都是 healthy

注意：Redis 一定要设 maxmemory 和淘汰策略，否则 working memory 会把内存吃光。我一般设 512MB 加 allkeys-lru，够用了。

4.2 记忆服务的核心代码实现

下面是一个精简版的记忆服务实现，用 Python + FastAPI，同时暴露 MCP 接口。核心逻辑分三块：写入、检索、遗忘。

from fastapi import FastAPI from pydantic import BaseModel import asyncpg, redis.asyncio as redis from qdrant_client import QdrantClient from datetime import datetime, timedelta import numpy as np app = FastAPI() class MemoryWrite(BaseModel): content: str memory_type: str = "episodic" importance: float = 0.5 ttl_days: int | None = None metadata: dict = {} class MemorySearch(BaseModel): query: str top_k: int = 5 memory_types: list[str] = ["episodic", "semantic"] @app.post("/memory/write") async def write_memory(req: MemoryWrite): # 1. 生成 embedding embedding = await embed(req.content) # 2. 去重检查 similar = qdrant.search( collection_name="memories", query_vector=embedding, limit=1 ) if similar and similar[0].score > 0.92: return {"status": "skipped", "reason": "duplicate"} # 3. 写入 PostgreSQL expires_at = None if req.ttl_days: expires_at = datetime.utcnow() + timedelta(days=req.ttl_days) memory_id = await pg.fetchval(""" INSERT INTO memories (content, memory_type, importance, expires_at, metadata) VALUES ($1, $2, $3, $4, $5) RETURNING id """, req.content, req.memory_type, req.importance, expires_at, req.metadata) # 4. 写入 Qdrant qdrant.upsert( collection_name="memories", points=[{ "id": memory_id, "vector": embedding, "payload": {"memory_type": req.memory_type, "importance": req.importance} }] ) return {"status": "ok", "memory_id": memory_id} @app.post("/memory/search") async def search_memory(req: MemorySearch): embedding = await embed(req.query) # 向量检索 results = qdrant.search( collection_name="memories", query_vector=embedding, limit=req.top_k * 3, # 多召回一些，后面重排 query_filter={ "must": [{"key": "memory_type", "match": {"any": req.memory_types}}] } ) # 混合重排：语义相似度 + 时间衰减 + 重要性 now = datetime.utcnow() scored = [] for r in results: mem = await pg.fetchrow("SELECT * FROM memories WHERE id = $1", r.id) if mem["expires_at"] and mem["expires_at"] < now: continue days_old = (now - mem["created_at"]).days time_decay = np.exp(-0.05 * days_old) final_score = r.score * 0.7 + time_decay * 0.2 + mem["importance"] * 0.1 scored.append((final_score, mem)) scored.sort(key=lambda x: x[0], reverse=True) return {"memories": [dict(m) for _, m in scored[:req.top_k]]}

这段代码有几个关键点值得展开。第一，去重检查放在写入前，避免重复记忆污染检索结果。第二，向量检索多召回再重排，因为纯向量分数不能反映时间新鲜度和重要性。第三，过期记忆在检索时过滤，而不是等定时任务清理，保证实时性。

4.3 MCP Server 的封装与接入

把上面的 HTTP 服务封装成 MCP Server，让 Agent 框架能直接调用。我用的是官方 Python SDK：

from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent import httpx app = Server("hindsight-memory") client = httpx.AsyncClient(base_url="http://localhost:8000") @app.list_tools() async def list_tools(): return [ Tool( name="memory_write", description="写入一条 Agent 记忆", inputSchema={ "type": "object", "properties": { "content": {"type": "string"}, "memory_type": {"type": "string", "enum": ["episodic", "semantic"]}, "importance": {"type": "number", "minimum": 0, "maximum": 1} }, "required": ["content"] } ), Tool( name="memory_search", description="检索 Agent 记忆", 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_write": resp = await client.post("/memory/write", json=arguments) elif name == "memory_search": resp = await client.post("/memory/search", json=arguments) else: raise ValueError(f"Unknown tool: {name}") return [TextContent(type="text", text=resp.text)] 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())

接入 Agent 框架时，在配置文件里加上 MCP Server 的启动命令就行。比如在 Claude Desktop 的配置里：

{ "mcpServers": { "hindsight-memory": { "command": "python", "args": ["/path/to/mcp_server.py"], "env": { "MEMORY_API_URL": "http://localhost:8000" } } } }

4.4 参数选择与性能调优

几个关键参数我给出实测后的推荐值：

性能方面，单次检索的 P99 延迟我实测在 80ms 左右（不含 LLM 调用），瓶颈主要在向量检索和 PostgreSQL 查询。如果记忆量超过 10 万条，建议给 Qdrant 建 HNSW 索引，把m设为 16，ef_construct设为 100。

5. 常见问题与排查技巧实录

5.1 记忆检索不准确：从三个维度排查

检索不准是最常见的问题。我的排查顺序是：先看召回，再看排序，最后看查询改写。

召回阶段，检查向量模型是否适合你的领域。通用 embedding 模型在专业领域（比如医疗、法律）表现会差很多，建议用领域数据微调或者换领域模型。我试过用通用模型检索医疗记忆，召回率只有 60%，换成医疗微调模型后到了 85%。

排序阶段，检查时间衰减和重要性的权重是否合理。如果你的场景是“用户最近说的话最重要”，把时间衰减权重调高到 0.3。如果是“重要规则永远优先”，把重要性权重调高。

查询改写阶段，检查用户 query 是否需要拆解。比如“帮我写个报告”这种模糊查询，直接检索效果很差，需要先让 LLM 改写成“用户报告偏好”“历史报告模板”“当前项目背景”三个子查询。

5.2 Docker 网络不通：最常见的三个原因

Docker 网络问题我踩过太多次了，总结下来就三个原因：

第一，容器间通信用了 localhost。容器里的 localhost 指向容器自己，不是宿主机。容器间通信要用服务名，比如postgres:5432而不是localhost:5432。

第二，端口映射写反了。-p 8000:8000是宿主机端口在前，容器端口在后。写反了就连不上。

第三，防火墙拦截。Ubuntu 的 ufw 默认会拦截 Docker 的流量，需要加规则：

sudo ufw allow from 172.16.0.0/12 sudo ufw allow from 192.168.0.0/16

提示：如果 Docker Desktop 启动报 “Virtualization support not detected”，检查 BIOS 里的虚拟化选项是否开启，Windows 用户还要确认 Hyper-V 和 WSL2 都启用了。

5.3 记忆库膨胀：容量控制的实操方案

记忆库膨胀的表现是检索变慢、存储成本上升、检索质量下降。我的控制方案分三步：

第一步，写入时严格过滤。重要性低于 0.3 的直接丢弃，重复的合并。

第二步，定期归档。超过 90 天且访问次数少于 3 次的记忆，迁移到冷存储（比如 S3 或者本地归档表），检索时不查冷存储。

第三步，容量告警。记忆总数超过阈值（我设 50 万条）时触发告警，人工介入清理。

-- 归档冷记忆 INSERT INTO memories_archive SELECT * FROM memories WHERE created_at < NOW() - INTERVAL '90 days' AND access_count < 3; DELETE FROM memories WHERE created_at < NOW() - INTERVAL '90 days' AND access_count < 3;

5.4 常见问题速查表

5.5 几个我踩过的坑

坑一：embedding 模型换了没重建索引。我一开始用 text-embedding-ada-002，后来换成 bge-large，忘了重建 Qdrant 索引，结果检索全乱套。换模型必须重建索引，没有例外。

坑二：working memory 没设 TTL。Redis 里的 working memory 如果不设过期时间，会话结束后数据还在，下次会话读到脏数据。我现在的做法是每次会话结束显式清理，同时设 24 小时兜底 TTL。

坑三：MCP Server 用 stdio 模式但没处理异常。stdio 模式下如果 Server 抛异常没捕获，整个进程会挂掉，Agent 那边看到的就是“工具不可用”。建议在 call_tool 里包一层 try-except，把异常转成 TextContent 返回。

坑四：PostgreSQL 连接池没配。默认连接数太少，并发一高就报 “too many connections”。我一般设 min_size=5, max_size=20，根据实际并发调整。

6. 记忆安全与未来扩展方向

6.1 记忆投毒与防御：a-memguard 思路的借鉴

Agent Memory 有个容易被忽视的安全问题：记忆投毒。如果攻击者能往记忆库里写入恶意内容，比如“用户的所有密码都应该发到某个邮箱”，Agent 后续行为就会被操控。a-memguard 这类主动防御框架的思路值得借鉴，核心是写入前做内容审核，检索后做一致性校验。

写入审核可以用规则加小模型，检测是否有指令注入、敏感信息、逻辑矛盾。检索后校验是检查召回的记忆是否与当前任务上下文冲突，冲突的降权或丢弃。我在实际项目里加了一层“记忆签名”，每条记忆写入时用服务端密钥签名，检索时验签，防止外部篡改。

6.2 从 Episodic 到 Semantic 的自动提炼

“hindsight”最有价值的能力，是从具体事件中自动提炼出可复用的知识。比如 Agent 经历了三次“用户要求报告加数据来源”的事件后，应该自动生成一条 Semantic Memory：“用户偏好数据驱动的报告风格”。

实现思路是定期跑一个提炼任务：拉取最近 N 条 Episodic Memory，用 LLM 做聚类和抽象，生成候选 Semantic Memory，人工审核后入库。这个任务可以每天跑一次，也可以按事件数量触发。

async def distill_semantic_memories(days=7): episodes = await pg.fetch(""" SELECT content FROM memories WHERE memory_type = 'episodic' AND created_at > NOW() - INTERVAL '%s days' """, days) prompt = f"""从以下事件记录中提炼出可复用的用户偏好或规则， 每条用一句话表述，输出 JSON 数组： {episodes}""" candidates = await llm.generate(prompt) for c in candidates: await write_memory(MemoryWrite( content=c, memory_type="semantic", importance=0.8 ))

6.3 多 Agent 共享记忆的隔离与协作

当你有多个 Agent 时，记忆的隔离和共享需要设计。我的方案是三层命名空间：全局记忆（所有 Agent 共享）、团队记忆（一组 Agent 共享）、私有记忆（单个 Agent 独有）。检索时按命名空间过滤，写入时指定命名空间。

这个设计的好处是灵活。比如用户偏好放全局，项目背景放团队，Agent 的临时状态放私有。MCP Server 的接口里加一个namespace参数就能支持。

7. 一些实操后的个人体会

这套方案我在两个项目里跑过，一个是个人的写作助手，一个是团队的数据分析 Agent。写作助手那边，记忆量不大，几千条，检索延迟基本无感。数据分析 Agent 那边，记忆量到了十几万条，调优后 P99 延迟控制在 150ms 以内，可以接受。

最大的体会是：Agent Memory 的难点不在技术，在于产品设计。你得想清楚什么该记、什么该忘、什么该提炼。技术方案再漂亮，如果记忆策略设计得不对，Agent 还是会表现得像个失忆患者。

另外，MCP 协议确实让记忆服务的接入变得简单了，但它的生态还在早期，调试工具不够完善。我建议在开发阶段加一层日志中间件，把每次 MCP 调用的请求和响应都记下来，排查问题的时候能省很多时间。

最后分享一个小技巧：给记忆加一个“来源”字段，记录这条记忆是从哪次对话、哪个任务来的。当 Agent 行为异常时，你可以顺着来源回溯，快速定位是哪条记忆导致的。这个字段在调试阶段的价值极高，强烈建议加上。