Hindsight 项目实战：为 AI Agent 构建分层记忆与事后复盘系统

1. 项目缘起：为什么“事后复盘”值得被单独做成一个项目

“hindsight”这个词本身很有意思，字面意思是“后见之明”，也就是事情发生之后才明白过来的那种洞察。放在 AI Agent 和 LLM 的语境里，它指向一个非常具体、也非常痛的问题：Agent 的记忆到底该怎么管，才能让它在事后真正“学到东西”，而不是每次对话都从零开始。

我接触过不少做 Agent 的团队，大家一开始都很兴奋，觉得只要把 LLM 接上工具、接上知识库，Agent 就能干活了。但跑一段时间就会发现，Agent 的“记忆”是个巨大的坑。它要么什么都不记，每次对话都是白纸一张；要么什么都记，把一堆无关紧要的废话全塞进上下文，token 烧得飞快，效果还越来越差。更麻烦的是，当 Agent 做错了一件事，你很难让它从这次错误里吸取教训——下次遇到类似场景，它大概率还会犯同样的错。

“hindsight”这个项目，本质上就是在解决这个问题。它不是简单地做一个“记忆存储”，而是围绕Agent Memory做一套完整的“事后复盘与经验固化”机制。你可以把它理解成给 Agent 装了一个“错题本”加“经验库”：每次任务执行完之后，系统会自动回顾整个过程，把值得记住的东西提炼出来，存进一个结构化的记忆层，下次遇到类似任务时再精准地调出来用。

这个项目适合谁？如果你正在做 LLM 应用、Agent 开发、RAG 系统，或者你只是对“怎么让 AI 记住东西”这件事感兴趣，那这套思路都值得你花时间研究。它不依赖某个特定框架，核心逻辑是通用的，你可以用 Python 自己实现，也可以结合现有的 MCP 协议、Docker 部署方案来落地。接下来我会从设计思路、核心细节、实操过程、问题排查几个层面，把这件事拆开讲透。

2. 整体设计思路：Agent Memory 到底该怎么分层

2.1 为什么“一个向量库”解决不了记忆问题

很多人做 Agent 记忆的第一反应是：搞个向量数据库，把对话历史 embedding 一下存进去，需要的时候检索出来塞进 prompt。这个方案能跑，但跑不长。原因很简单：记忆不是同质的。你对话里既有“用户今天心情不好”这种短期情绪信息，也有“用户对花生过敏”这种长期事实，还有“上次用某个 API 调用失败了，原因是参数格式不对”这种操作性经验。这三类东西的存储方式、检索方式、生命周期完全不一样，全塞进一个向量库，检索出来的结果必然是混乱的。

“hindsight”的设计思路是把记忆分成三层，我把它叫做工作记忆、情景记忆、语义记忆。这个分层借鉴了认知科学里对人类记忆的研究，但在工程上做了简化，保证可落地。

• 工作记忆（Working Memory）：当前任务执行过程中的临时状态，比如“我现在正在查订单”“我已经拿到了用户 ID”。它的生命周期就是一次任务，任务结束就清空或者归档。
• 情景记忆（Episodic Memory）：具体发生过的事件，比如“2024 年 3 月 5 日，用户让我查订单 A123，我调用了订单查询接口，返回了物流信息”。它带时间戳、带上下文，是“发生了什么”的记录。
• 语义记忆（Semantic Memory）：从多次情景中提炼出来的通用知识，比如“这个用户偏好用邮件接收通知”“订单查询接口在参数缺失时会返回 400”。它是“我知道了什么”的沉淀。

这个分层的核心价值在于：检索的时候可以按需选择层级。当前任务需要临时状态，就只查工作记忆；需要参考历史案例，就查情景记忆；需要通用规则，就查语义记忆。这样既省 token，又提高准确率。

2.2 为什么选择 MCP 作为记忆的接入层

MCP（Model Context Protocol）是最近很热的一个协议，它的核心作用是让 LLM 能够以标准化的方式访问外部工具和数据源。在“hindsight”项目里，我把记忆系统做成了一个 MCP Server，这样任何支持 MCP 的客户端（比如 Claude Desktop、各种 IDE 插件、自研 Agent 框架）都能直接调用记忆能力，而不需要每个项目都重新实现一遍记忆逻辑。

这个选择背后的考量是解耦。记忆系统本身是一个独立服务，它不应该和某个具体的 Agent 框架绑死。今天你用 LangChain，明天换 AutoGen，记忆层不应该跟着重写。MCP 提供了一个相对标准的接口，把“记忆的读写”抽象成几个工具调用，比如memory_store、memory_retrieve、memory_reflect。Agent 只需要知道怎么调这些工具，不需要关心底层是用向量库还是图数据库。

另外，MCP 的生态正在快速扩张，很多工具都已经支持 MCP 协议。把记忆做成 MCP Server，意味着它可以和现有的工具链无缝组合。比如你可以让 Agent 先用 Playwright MCP 去抓网页，然后把抓到的关键信息通过 hindsight MCP 存进记忆，下次直接检索，不用重新抓。

2.3 Docker 化部署：为什么这是必选项

记忆系统是一个有状态的服务，它需要持久化存储、需要稳定的运行环境、可能需要和多个 Agent 实例共享。如果用裸机部署，环境依赖、版本冲突、数据迁移都是麻烦事。Docker 化之后，整个记忆系统可以打包成一个镜像，一条命令启动，数据卷挂载到宿主机，升级和迁移都变得可控。

而且，现在很多 Agent 开发环境本身就是 Docker 化的。比如你在 Docker Desktop 里跑一个 Agent 容器，再跑一个 hindsight 容器，两者通过 Docker 网络通信，整个环境是隔离的、可复现的。这对于团队协作和持续集成来说，价值很大。

3. 核心细节解析：记忆的写入、检索与反思

3.1 记忆写入：什么该记，什么不该记

这是整个系统里最关键的决策点。记多了，噪声大、检索慢、token 贵；记少了，Agent 学不到东西。我的经验是遵循一个“三问原则”：

1. 这个信息在未来类似场景下还会用到吗？如果只是一次性的临时状态，比如“当前时间戳”，那就不需要长期存储。
2. 这个信息是否具有可复用性？比如“用户说‘好的’”这种确认性回复，没有复用价值；但“用户明确表示不喜欢电话沟通”就有。
3. 这个信息是否会影响未来的决策？如果会，那就值得记；如果只是记录事实但不影响行为，优先级就低。

在具体实现上，我设计了一个记忆评分函数，每次任务结束后，系统会对候选记忆条目打分，分数超过阈值的才写入长期存储。评分维度包括：

维度	说明	权重示例
复用频率	类似场景出现的频率	0.3
决策影响	是否改变后续行为	0.3
信息稳定性	是否长期有效	0.2
独特性	是否与已有记忆重复	0.2

这个评分函数不需要很精确，它的作用是提供一个可解释的过滤机制，而不是靠拍脑袋决定记什么。

3.2 记忆检索：三个点 key 的设计

热搜词里有一个很有意思的说法：“LLM 的 token 三个点 key：我是谁、query 我在找什么、value 我能提供什么”。这其实是在说，记忆检索的时候，需要三个维度的信息来精准匹配：

• 我是谁（Agent Identity）：当前 Agent 的角色是什么？是客服 Agent、是代码助手、还是数据分析师？不同角色的记忆应该隔离，客服 Agent 不需要知道代码助手上次调试的 bug。
• 我在找什么（Query Intent）：当前任务的意图是什么？是查询事实、是寻求建议、还是执行操作？意图决定了检索的策略。
• 我能提供什么（Context）：当前上下文里已经有哪些信息？这些信息可以作为检索的过滤条件，缩小范围。

在 hindsight 里，我把这三个维度编码成一个检索请求结构：

retrieval_request = { "agent_id": "customer_service_agent", "intent": "order_inquiry", "context": { "user_id": "U12345", "order_id": "A123", "time_range": "last_30_days" }, "memory_types": ["episodic", "semantic"], "top_k": 5 }

检索的时候，系统会先用agent_id和intent做粗筛，然后用context做精排，最后返回 top_k 条记忆。这样既保证了相关性，又控制了返回数量。

3.3 记忆反思：让 Agent 自己总结“我学到了什么”

这是 hindsight 最有价值的部分。每次任务结束后，系统会触发一个反思流程：把本次任务的工作记忆和情景记忆拿出来，让 LLM 做一次总结，提炼出可以沉淀到语义记忆的规则。

比如，一次订单查询任务结束后，反思流程可能会输出：

本次任务中，订单查询接口在order_id格式不正确时返回了 400 错误，错误信息为“Invalid order ID format”。建议：在调用接口前，先校验order_id是否符合^[A-Z]\d{3}$的格式。

这条规则会被写入语义记忆，下次再遇到订单查询任务时，Agent 会先检查格式，避免重复犯错。

反思流程的 prompt 设计很关键。我的经验是，不要问“这次任务有什么值得记住的”，太开放了，LLM 会输出一堆废话。要问具体的问题，比如：

• 这次任务中出现了哪些错误或异常？
• 这些错误的原因是什么？
• 下次遇到类似情况，应该采取什么不同的做法？
• 这次任务中，用户的偏好或约束有哪些是之前不知道的？

这样引导之后，LLM 的输出会聚焦得多。

4. 实操过程：从零搭建一个 hindsight 记忆服务

4.1 环境准备与 Docker 部署

先说一下基础环境。我用的是一台 Ubuntu 22.04 的机器，Docker 版本 24.0 以上。如果你在 Windows 上，建议用 Docker Desktop，但要注意开启 WSL2 后端，否则性能会差很多。安装 Docker 的步骤这里不展开，网上教程很多，核心就是：

# Ubuntu 上安装 Docker sudo apt update sudo apt install docker.io docker-compose-plugin sudo systemctl enable docker sudo systemctl start docker

装完之后，验证一下：

docker --version docker compose version

接下来是 hindsight 服务的部署。我把它拆成两个容器：一个是记忆存储服务（用 PostgreSQL + pgvector 做向量存储），一个是MCP Server（提供记忆读写接口）。

docker-compose.yml大概长这样：

version: '3.8' services: memory-db: image: pgvector/pgvector:pg16 environment: POSTGRES_DB: hindsight POSTGRES_USER: hindsight POSTGRES_PASSWORD: hindsight_dev volumes: - ./data/pgdata:/var/lib/postgresql/data ports: - "5432:5432" networks: - hindsight-net mcp-server: build: ./mcp-server environment: DB_HOST: memory-db DB_PORT: 5432 DB_NAME: hindsight DB_USER: hindsight DB_PASSWORD: hindsight_dev LLM_API_KEY: ${LLM_API_KEY} ports: - "8080:8080" depends_on: - memory-db networks: - hindsight-net networks: hindsight-net: driver: bridge

这里有几个细节值得注意：

• pgvector 镜像：直接用pgvector/pgvector:pg16，省得自己编译扩展。版本选 pg16 是因为它对向量索引的支持更成熟。
• 数据卷挂载：./data/pgdata挂到容器里，这样容器删了数据还在。我踩过的坑是忘了挂载，结果docker compose down之后数据全没了，白跑一天。
• 网络配置：两个容器在同一个 bridge 网络里，MCP Server 可以直接用memory-db这个主机名访问数据库，不需要暴露数据库端口到宿主机。如果你本地调试需要连数据库，再把 5432 端口映射出来。

启动命令：

docker compose up -d

启动之后，检查一下日志：

docker compose logs -f mcp-server

如果看到 “MCP Server listening on 8080” 和 “Database connected”，就说明服务起来了。

4.2 数据库表结构设计

记忆存储的核心表结构我设计了四张表：

-- 工作记忆表 CREATE TABLE working_memory ( id SERIAL PRIMARY KEY, agent_id VARCHAR(64) NOT NULL, session_id VARCHAR(64) NOT NULL, content TEXT NOT NULL, created_at TIMESTAMP DEFAULT NOW(), expires_at TIMESTAMP ); -- 情景记忆表 CREATE TABLE episodic_memory ( id SERIAL PRIMARY KEY, agent_id VARCHAR(64) NOT NULL, session_id VARCHAR(64), event_type VARCHAR(32), content TEXT NOT NULL, embedding vector(1536), metadata JSONB, created_at TIMESTAMP DEFAULT NOW() ); -- 语义记忆表 CREATE TABLE semantic_memory ( id SERIAL PRIMARY KEY, agent_id VARCHAR(64) NOT NULL, rule_type VARCHAR(32), content TEXT NOT NULL, embedding vector(1536), confidence FLOAT DEFAULT 0.5, source_episodes INTEGER[], created_at TIMESTAMP DEFAULT NOW(), updated_at TIMESTAMP DEFAULT NOW() ); -- 记忆检索日志 CREATE TABLE retrieval_log ( id SERIAL PRIMARY KEY, agent_id VARCHAR(64), query TEXT, retrieved_ids INTEGER[], feedback FLOAT, created_at TIMESTAMP DEFAULT NOW() );

几个设计要点：

• embedding 维度 1536：这是 OpenAI text-embedding-3-small 的维度。如果你用别的 embedding 模型，改这个数字就行。
• confidence 字段：语义记忆的置信度，初始 0.5，每次被成功使用后增加，被证伪后降低。这样系统可以自动淘汰低置信度的规则。
• source_episodes：记录这条语义记忆是从哪些情景记忆里提炼出来的，方便追溯。
• retrieval_log：记录每次检索的 query 和返回结果，用于后续优化检索策略。这个表一开始我觉得可有可无，后来发现它对于调试和评估太重要了。

4.3 MCP Server 的核心接口实现

MCP Server 我用 Python 写，基于mcp这个库。核心暴露三个工具：

from mcp.server import Server, NotificationOptions from mcp.server.models import InitializationOptions import mcp.server.stdio import mcp.types as types server = Server("hindsight-memory") @server.list_tools() async def handle_list_tools(): return [ types.Tool( name="memory_store", description="存储一条记忆", inputSchema={ "type": "object", "properties": { "agent_id": {"type": "string"}, "memory_type": {"type": "string", "enum": ["working", "episodic", "semantic"]}, "content": {"type": "string"}, "metadata": {"type": "object"} }, "required": ["agent_id", "memory_type", "content"] } ), types.Tool( name="memory_retrieve", description="检索记忆", inputSchema={ "type": "object", "properties": { "agent_id": {"type": "string"}, "query": {"type": "string"}, "memory_types": {"type": "array", "items": {"type": "string"}}, "top_k": {"type": "integer", "default": 5} }, "required": ["agent_id", "query"] } ), types.Tool( name="memory_reflect", description="触发反思流程，从情景记忆中提炼语义记忆", inputSchema={ "type": "object", "properties": { "agent_id": {"type": "string"}, "session_id": {"type": "string"} }, "required": ["agent_id", "session_id"] } ) ]

memory_store的实现逻辑：

async def handle_store(agent_id, memory_type, content, metadata=None): embedding = await get_embedding(content) if memory_type == "working": # 工作记忆，设置过期时间 await db.execute( "INSERT INTO working_memory (agent_id, session_id, content, expires_at) VALUES ($1, $2, $3, NOW() + INTERVAL '1 hour')", agent_id, metadata.get("session_id"), content ) elif memory_type == "episodic": await db.execute( "INSERT INTO episodic_memory (agent_id, session_id, event_type, content, embedding, metadata) VALUES ($1, $2, $3, $4, $5, $6)", agent_id, metadata.get("session_id"), metadata.get("event_type"), content, embedding, metadata ) elif memory_type == "semantic": # 语义记忆需要先检查是否重复 existing = await db.fetch( "SELECT id, confidence FROM semantic_memory WHERE agent_id = $1 AND content = $2", agent_id, content ) if existing: # 更新置信度 await db.execute( "UPDATE semantic_memory SET confidence = LEAST(confidence + 0.1, 1.0), updated_at = NOW() WHERE id = $1", existing[0]["id"] ) else: await db.execute( "INSERT INTO semantic_memory (agent_id, rule_type, content, embedding, confidence) VALUES ($1, $2, $3, $4, 0.5)", agent_id, metadata.get("rule_type"), content, embedding )

memory_retrieve的实现逻辑：

async def handle_retrieve(agent_id, query, memory_types=None, top_k=5): query_embedding = await get_embedding(query) results = [] if not memory_types or "episodic" in memory_types: episodic = await db.fetch( """ SELECT id, content, metadata, 1 - (embedding <=> $1) AS similarity FROM episodic_memory WHERE agent_id = $2 ORDER BY embedding <=> $1 LIMIT $3 """, query_embedding, agent_id, top_k ) results.extend([dict(r) for r in episodic]) if not memory_types or "semantic" in memory_types: semantic = await db.fetch( """ SELECT id, content, confidence, 1 - (embedding <=> $1) AS similarity FROM semantic_memory WHERE agent_id = $2 AND confidence > 0.3 ORDER BY embedding <=> $1 LIMIT $3 """, query_embedding, agent_id, top_k ) results.extend([dict(r) for r in semantic]) # 按相似度排序，返回 top_k results.sort(key=lambda x: x["similarity"], reverse=True) return results[:top_k]

这里用到了 pgvector 的<=>操作符，它计算的是余弦距离。1 - 余弦距离就是余弦相似度。注意在检索语义记忆时加了一个confidence > 0.3的过滤，避免低质量的规则干扰结果。

4.4 反思流程的 Prompt 设计

反思流程是整个系统里最“玄学”的部分，prompt 写得好不好，直接决定提炼出来的规则质量。我试过很多版本，最后稳定下来的 prompt 结构是这样的：

REFLECTION_PROMPT = """你是一个 Agent 经验总结助手。请根据以下任务执行记录，提炼出可复用的经验规则。 任务执行记录： {episodes} 请回答以下问题： 1. 这次任务中出现了哪些错误、异常或低效的环节？ 2. 这些问题的根本原因是什么？ 3. 下次遇到类似任务时，应该采取什么不同的做法？ 4. 这次任务中，有哪些用户偏好或环境约束是之前不知道的？ 输出格式要求： - 每条规则一行，以 "RULE: " 开头 - 规则要具体、可操作，避免模糊表述 - 如果某条规则与已有规则重复，请标注 "DUPLICATE: " - 最多输出 5 条规则 """

这个 prompt 的关键在于限制输出数量和格式。如果不限制，LLM 会输出一大堆泛泛而谈的东西。限制 5 条之后，它会优先输出最重要的。RULE:前缀方便后续解析。

解析逻辑：

def parse_reflection_output(output): rules = [] for line in output.split("\n"): line = line.strip() if line.startswith("RULE: "): rules.append({ "content": line[6:], "rule_type": "learned_rule" }) elif line.startswith("DUPLICATE: "): # 跳过重复规则 continue return rules

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

5.1 Docker 网络不通怎么办

这是部署阶段最常见的问题。症状是 MCP Server 容器启动时报错 “could not connect to server: Connection refused”，或者连接超时。

排查思路按顺序来：

1. 检查容器是否在同一网络：docker network inspect hindsight-net，看看两个容器是不是都在里面。如果不在，检查docker-compose.yml里的networks配置。
2. 检查数据库是否就绪：docker compose logs memory-db，看 PostgreSQL 是否完成了初始化。有时候 MCP Server 启动太快，数据库还没准备好，就会连接失败。解决办法是在depends_on里加condition: service_healthy，并给数据库配置 healthcheck。
3. 检查主机名解析：在 MCP Server 容器里执行ping memory-db，看能不能解析。如果不行，说明网络配置有问题。
4. 检查端口：PostgreSQL 默认监听 5432，确认没有改过。

我踩过的一个坑是：在 Windows 上用 Docker Desktop，有时候 WSL2 的网络会出问题，容器之间 ping 不通。解决办法是重启 Docker Desktop，或者执行wsl --shutdown后重新启动。

5.2 记忆检索结果不相关怎么调

检索不相关通常有三个原因：

• embedding 模型不适合：如果你用的是通用的 embedding 模型，它可能对领域术语不敏感。解决办法是换一个在相关领域微调过的模型，或者自己微调一个。
• 检索策略太单一：纯向量检索有时候会漏掉关键词匹配的结果。我的做法是混合检索：向量相似度 + 关键词匹配（BM25），然后加权融合。pgvector 本身不支持 BM25，但可以用 PostgreSQL 的全文检索功能配合。
• 记忆内容质量差：如果写入的记忆本身就是一堆废话，检索出来自然没用。回到 3.1 节的评分函数，把阈值调高，只让高质量记忆进入长期存储。

一个实用的调试技巧：把retrieval_log表里的记录拿出来，人工标注哪些是相关的、哪些是不相关的，然后计算准确率。如果准确率低于 70%，就需要调整检索策略。

5.3 反思流程输出质量不稳定怎么办

LLM 的输出本身就有随机性，反思流程的输出质量波动是正常的。我的应对策略是：

• 温度调低：反思流程用temperature=0.3，不要用默认的 0.7。低温度让输出更稳定、更聚焦。
• 多次采样取交集：对同一个任务记录，跑 3 次反思，取三次都出现的规则作为高置信度规则。这个做法成本高一些，但对于关键规则值得。
• 人工审核：在系统初期，所有语义记忆都经过人工审核再入库。等积累了一定量的标注数据后，可以训练一个小的分类器来自动过滤。

5.4 常见问题速查表

问题现象	可能原因	排查方法	解决方案
容器启动失败	端口冲突	docker compose logs	改端口映射
数据库连接超时	网络不通	docker network inspect	检查网络配置
检索结果为空	embedding 未生成	查数据库 embedding 字段	检查 embedding 服务
记忆膨胀过快	写入阈值太低	查 working_memory 数量	调高评分阈值
反思无输出	prompt 格式错误	查 LLM 返回内容	修正 prompt
语义记忆冲突	规则矛盾	查 semantic_memory 表	加冲突检测逻辑

6. 记忆系统的扩展方向与个人体会

6.1 从单 Agent 到多 Agent 共享记忆

现在这套系统是单 Agent 的，每个 Agent 有自己的记忆空间。但在多 Agent 协作的场景下，记忆需要共享。比如一个客服 Agent 和一个订单 Agent 协作处理用户请求，客服 Agent 了解到的用户偏好应该能被订单 Agent 看到。

扩展的思路是引入记忆命名空间和访问控制。每个记忆条目除了agent_id，还有一个namespace字段。命名空间可以是global（所有 Agent 可见）、team:xxx（团队内可见）、private（仅自己可见）。检索的时候根据当前 Agent 的权限过滤。

这个扩展不复杂，但需要仔细设计权限模型，避免记忆泄露。

6.2 记忆的遗忘机制

人脑会遗忘，Agent 的记忆系统也需要遗忘。不是所有记忆都值得永久保留。我的做法是给每条记忆加一个衰减因子，随着时间推移，未被检索到的记忆权重逐渐降低，低于阈值后自动归档或删除。

具体实现可以用一个定时任务，每天跑一次：

UPDATE episodic_memory SET weight = weight * 0.95 WHERE last_accessed_at < NOW() - INTERVAL '30 days'; DELETE FROM episodic_memory WHERE weight < 0.1;

这个机制可以防止记忆库无限膨胀，同时保留真正重要的记忆。

6.3 我个人的一些体会

做这个项目最大的感受是：Agent 的记忆问题，本质上不是技术问题，而是产品问题。你得先想清楚，这个 Agent 需要记住什么、不需要记住什么、记住之后怎么用。技术只是实现手段。

另外，不要追求一步到位。我一开始想做一个完美的记忆系统，结果拖了很久没落地。后来改成先跑通最小闭环：能存、能取、能反思，然后再逐步优化。这个策略让我在两周内就有了一个可用的版本，后续的迭代也有了基础。

最后分享一个小技巧：在开发阶段，把每次记忆读写都打上详细日志，包括原始内容、embedding 向量、检索分数。这些日志在调试的时候非常有用，能帮你快速定位是写入的问题还是检索的问题。等系统稳定了，再把日志级别调低。

这个项目后续还可以往几个方向扩展：接入更多类型的记忆（比如视觉记忆、音频记忆）、支持记忆的版本控制、做记忆的可视化面板。但这些都是后话，先把核心闭环跑通，比什么都重要。