1. 从“hindsight”说起：为什么我们需要给Agent装上“后视镜”
“hindsight”这个词本身很有意思，字面意思是“事后的洞察力”，也就是我们常说的“后见之明”。放在LLM Agent的语境里，它指向一个非常具体且要命的问题：Agent的记忆机制。你肯定遇到过这种情况——跟一个AI助手聊了半小时，它突然忘了你五分钟前说过的关键约束；或者一个自动化工作流跑到第三步，把第一步的中间结果丢得一干二净。这不是模型不够聪明，而是记忆架构没搭对。
我最初接触这个方向，是因为在做一个多步骤的代码审查Agent。当时用的是一个主流LLM框架，对话轮次一多，上下文窗口就爆了，要么截断历史，要么成本飙升。更麻烦的是，Agent在“回忆”之前做过什么的时候，经常张冠李戴，把不同会话的内容混在一起。后来我意识到，Agent的记忆不能只靠一个不断增长的对话数组，它需要一套分层的、可检索的、带时效性的存储体系。而“hindsight”这个概念，恰好概括了这套体系的核心目标：让Agent能够像人一样，在需要的时候“回头看”，精准提取过去的相关经验，而不是把所有东西都塞在脑子里。
这篇文章适合谁看？如果你正在用LLM搭建Agent、工作流或者任何需要多轮交互的应用，并且被“记忆”问题折磨过，那接下来的内容就是为你准备的。我会从架构设计、存储分层、检索策略、MCP协议集成、Docker化部署这几个维度，把“hindsight”这个项目拆开揉碎，讲清楚每一步为什么这么做，以及我踩过的那些坑。全文基于我实际落地的经验，代码和配置都可以直接抄作业。
2. 核心架构拆解：Agent记忆到底该怎么分层
2.1 为什么单一上下文窗口是死路一条
先算一笔账。假设你用的是一个128K上下文窗口的模型，每轮对话平均消耗500个token，那么理论上能撑256轮。但实际情况是，Agent的每一轮输出往往比输入更长，加上工具调用的返回结果、系统提示词、few-shot示例，实际可用轮次可能不到50轮。而且，上下文越长，模型的注意力越容易涣散，中间部分的信息被忽略的概率显著上升。这就是所谓的“lost in the middle”现象。
更致命的是成本。以某主流API为例，输入token的价格是输出的三分之一左右，但如果你每轮都把全部历史塞进去，输入token会呈平方级增长。聊到第100轮的时候，单次调用的成本可能是第一轮的几十倍。所以，把记忆全部放在上下文窗口里，既贵又慢还不准。
“hindsight”的思路是：上下文窗口只放“当前最相关”的信息，其余的全部外置到存储层，按需检索。这就像你工作时不会把过去十年的所有邮件都摊在桌上，而是需要哪封就翻哪封。
2.2 三层记忆模型：Working、Episodic、Semantic
基于常见实践，我把Agent记忆分为三层，这也是“hindsight”项目里最核心的设计：
第一层：Working Memory（工作记忆）。这是Agent当前正在处理的任务上下文，直接放在LLM的上下文窗口里。它的特点是容量小、时效短、访问快。比如用户当前的问题、最近三轮的对话、当前步骤的工具返回结果。这一层不需要持久化，任务结束就可以丢弃。
第二层：Episodic Memory（情景记忆）。这是Agent过去执行过的具体任务记录，包括时间戳、任务描述、执行步骤、结果状态。它的特点是按时间线组织，支持“最近发生了什么”这类查询。比如“上周三我处理的那个退款请求最后是什么结果”。这一层需要持久化存储，通常用关系型数据库或者带时间索引的文档数据库。
第三层：Semantic Memory（语义记忆）。这是从大量情景记忆中抽象出来的知识和规律，比如“用户A偏好用邮件沟通”“处理这类报错需要先检查配置文件”。它的特点是与具体时间无关，支持语义检索。这一层通常用向量数据库存储，通过embedding做相似度匹配。
这三层的访问频率和存储介质完全不同。Working Memory在内存里，Episodic Memory在PostgreSQL或SQLite里，Semantic Memory在向量数据库里。分层的意义在于，让每一层用最适合它的方式存储和检索，而不是一刀切。
2.3 记忆的写入与读取路径设计
写入路径：Agent每完成一个步骤，就把关键信息写入Episodic Memory。这里有个关键决策——写什么。我的经验是，不要写原始对话，而是写“结构化摘要”。比如，不要存“用户说：我想查一下订单123的状态”，而是存{action: "query_order", order_id: "123", timestamp: "...", result: "shipped"}。这样后续检索时，可以直接按字段过滤，而不是做全文扫描。
读取路径：当Agent需要回忆时，先用当前任务的关键词去Episodic Memory做时间范围过滤，再去Semantic Memory做向量相似度检索，最后把两边的结果合并、去重、按相关性排序，取Top-K注入上下文。这里的关键是“混合检索”——纯向量检索容易漏掉时间敏感的信息，纯时间过滤又无法处理语义相关但时间久远的记忆。
注意：写入频率要控制。不是每一步都写，而是只在“状态发生实质变化”时写。比如工具调用成功、用户给出新约束、任务阶段切换。否则Episodic Memory会迅速膨胀，检索效率急剧下降。
3. 关键技术点实操：从存储到检索的完整实现
3.1 用Docker快速拉起PostgreSQL和Redis
“hindsight”的存储层依赖两个核心组件：PostgreSQL用于Episodic Memory，Redis用于Working Memory的缓存和会话状态。用Docker Compose可以一键拉起，省去手动配置的麻烦。
version: '3.8' services: postgres: image: postgres:16-alpine environment: POSTGRES_USER: agent POSTGRES_PASSWORD: agent_pass POSTGRES_DB: hindsight ports: - "5432:5432" volumes: - pg_data:/var/lib/postgresql/data healthcheck: test: ["CMD-SHELL", "pg_isready -U agent"] interval: 5s timeout: 5s retries: 5 redis: image: redis:7-alpine ports: - "6379:6379" command: redis-server --appendonly yes volumes: - redis_data:/data volumes: pg_data: redis_data:启动命令就一行：docker compose up -d。等健康检查通过后，PostgreSQL里建一张episodic_memory表：
CREATE TABLE episodic_memory ( id BIGSERIAL PRIMARY KEY, session_id VARCHAR(64) NOT NULL, step_index INT NOT NULL, action_type VARCHAR(32) NOT NULL, content JSONB NOT NULL, embedding VECTOR(1536), created_at TIMESTAMPTZ DEFAULT NOW() ); CREATE INDEX idx_session_time ON episodic_memory(session_id, created_at DESC); CREATE INDEX idx_embedding ON episodic_memory USING ivfflat (embedding vector_cosine_ops);这里用了pgvector扩展来存embedding。如果你用的PostgreSQL镜像不带这个扩展，需要换成pgvector/pgvector:pg16镜像。embedding维度1536对应的是某主流embedding模型的输出维度，如果你用别的模型，记得改。
3.2 记忆写入的批处理与去重策略
Agent运行过程中，如果每产生一条记忆就写一次数据库，IO开销会很大。我的做法是攒批写入：在内存里维护一个缓冲区，每积累10条或者每隔5秒，批量flush到PostgreSQL。这样既降低了写入频率，又保证了记忆不会丢失太多。
去重策略也很关键。同一个任务步骤可能因为重试而被多次记录，如果不做去重，检索时会返回大量重复内容。我的方案是：用session_id + step_index + action_type作为唯一键，写入时用ON CONFLICT DO UPDATE，只保留最新版本。
import asyncpg import json from datetime import datetime class EpisodicWriter: def __init__(self, pool, batch_size=10, flush_interval=5): self.pool = pool self.buffer = [] self.batch_size = batch_size self.flush_interval = flush_interval self.last_flush = datetime.now() async def add(self, session_id, step_index, action_type, content, embedding): self.buffer.append((session_id, step_index, action_type, json.dumps(content), embedding)) if len(self.buffer) >= self.batch_size or (datetime.now() - self.last_flush).seconds >= self.flush_interval: await self.flush() async def flush(self): if not self.buffer: return async with self.pool.acquire() as conn: await conn.executemany(""" INSERT INTO episodic_memory (session_id, step_index, action_type, content, embedding) VALUES ($1, $2, $3, $4::jsonb, $5) ON CONFLICT (session_id, step_index, action_type) DO UPDATE SET content = EXCLUDED.content, embedding = EXCLUDED.embedding """, self.buffer) self.buffer.clear() self.last_flush = datetime.now()实操心得：批处理的大小不要设太大。我试过设成100，结果Agent崩溃时丢了大量记忆。10到20是比较稳妥的范围。另外，flush操作要放在
try/finally里，确保异常时也能把缓冲区写出去。
3.3 混合检索：时间过滤加向量相似度
检索是“hindsight”最核心的能力。我的实现是两阶段检索：
第一阶段，用session_id和时间范围做粗筛，从Episodic Memory里取出最近N条记录。这一步用SQL就能完成，速度极快。
第二阶段，对粗筛结果做向量相似度排序。把当前查询的embedding和每条记忆的embedding做余弦相似度计算，取Top-K。如果粗筛结果太多，可以先在数据库层面用pgvector的<=>操作符做近似最近邻搜索。
async def retrieve_memory(pool, session_id, query_embedding, top_k=5, time_window_hours=24): async with pool.acquire() as conn: rows = await conn.fetch(""" SELECT id, action_type, content, embedding, 1 - (embedding <=> $1) AS similarity FROM episodic_memory WHERE session_id = $2 AND created_at > NOW() - INTERVAL '%s hours' ORDER BY embedding <=> $1 LIMIT $3 """ % time_window_hours, query_embedding, session_id, top_k) return [dict(row) for row in rows]这里有个细节：<=>是pgvector的余弦距离操作符，1 - 距离就是相似度。时间窗口我一般设24小时，但如果是长期运行的Agent，可以放宽到7天。关键是不要全表扫描，一定要有session_id和时间索引。
3.4 与MCP协议集成：让记忆成为可调用的工具
MCP（Model Context Protocol）是当前Agent生态里很火的一个协议，它让LLM可以像调用函数一样调用外部工具。“hindsight”的记忆检索能力，完全可以封装成一个MCP Server，这样任何支持MCP的Agent框架都能直接使用。
MCP Server的核心是定义工具描述和输入schema。我用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="retrieve_memory", description="根据查询内容检索Agent的历史记忆", inputSchema={ "type": "object", "properties": { "session_id": {"type": "string"}, "query": {"type": "string"}, "top_k": {"type": "integer", "default": 5} }, "required": ["session_id", "query"] } ), types.Tool( name="write_memory", description="将当前步骤的关键信息写入记忆", inputSchema={ "type": "object", "properties": { "session_id": {"type": "string"}, "step_index": {"type": "integer"}, "action_type": {"type": "string"}, "content": {"type": "object"} }, "required": ["session_id", "step_index", "action_type", "content"] } ) ] @server.call_tool() async def handle_call_tool(name, arguments): if name == "retrieve_memory": embedding = await get_embedding(arguments["query"]) results = await retrieve_memory(pool, arguments["session_id"], embedding, arguments.get("top_k", 5)) return [types.TextContent(type="text", text=json.dumps(results, ensure_ascii=False))] elif name == "write_memory": embedding = await get_embedding(json.dumps(arguments["content"])) await writer.add(arguments["session_id"], arguments["step_index"], arguments["action_type"], arguments["content"], embedding) return [types.TextContent(type="text", text="ok")]这样，Agent在需要回忆时，只需要调用retrieve_memory工具，传入当前查询，就能拿到相关的历史记忆。MCP的好处是解耦——记忆服务独立部署，Agent框架不需要关心底层用的是PostgreSQL还是别的什么。
注意：MCP Server的启动方式要和你的Agent框架匹配。如果是stdio模式，Server作为子进程启动；如果是SSE模式，Server需要监听一个HTTP端口。我一般用stdio模式，简单直接。
4. Docker化部署与生产环境调优
4.1 完整Docker Compose编排
把PostgreSQL、Redis、MCP Server打包成一个Compose文件，一键部署：
version: '3.8' services: postgres: image: pgvector/pgvector:pg16 environment: POSTGRES_USER: agent POSTGRES_PASSWORD: agent_pass POSTGRES_DB: hindsight volumes: - pg_data:/var/lib/postgresql/data - ./init.sql:/docker-entrypoint-initdb.d/init.sql healthcheck: test: ["CMD-SHELL", "pg_isready -U agent"] interval: 5s retries: 5 redis: image: redis:7-alpine command: redis-server --appendonly yes --maxmemory 256mb --maxmemory-policy allkeys-lru volumes: - redis_data:/data mcp-server: build: ./mcp-server environment: DATABASE_URL: postgresql://agent:agent_pass@postgres:5432/hindsight REDIS_URL: redis://redis:6379/0 EMBEDDING_API_KEY: ${EMBEDDING_API_KEY} depends_on: postgres: condition: service_healthy redis: condition: service_started stdin_open: true tty: true volumes: pg_data: redis_data:init.sql里放建表和扩展的语句。MCP Server的Dockerfile很简单：
FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD ["python", "-m", "hindsight_mcp"]4.2 资源限制与性能调优
生产环境一定要设资源限制，否则一个Agent跑飞了能把整台机器拖垮。我的配置参考：
| 服务 | CPU限制 | 内存限制 | 说明 |
|---|---|---|---|
| PostgreSQL | 2核 | 2GB | 主要吃内存做缓存，2GB够用 |
| Redis | 1核 | 512MB | 设了LRU淘汰，不会爆 |
| MCP Server | 2核 | 1GB | embedding计算是CPU密集型的 |
PostgreSQL的调优参数：shared_buffers设为内存的25%，work_mem设64MB，maintenance_work_mem设256MB。如果记忆量很大，ivfflat索引的lists参数要调大，一般设为rows / 1000。
Redis这边，maxmemory-policy设成allkeys-lru，保证Working Memory的缓存不会无限增长。Working Memory的TTL我一般设30分钟，超过这个时间的会话状态自动过期。
4.3 监控与日志：别等崩了才查
Docker Compose默认的日志驱动是json-file，时间长了会占满磁盘。我一般改成local驱动，并限制大小：
logging: driver: local options: max-size: "10m" max-file: "3"监控方面，PostgreSQL的pg_stat_statements扩展一定要开，能看到哪些查询最慢。MCP Server里加一个简单的健康检查端点，返回数据库连接状态和缓冲区大小。我踩过最大的坑是embedding API超时导致整个写入流程阻塞，后来加了超时和重试机制才解决。
5. 常见问题与排查实录
5.1 记忆检索不准：为什么Agent总是“记错”
这是最常见的问题。Agent检索到的记忆和当前任务不相关，或者相关但排序不对。排查思路：
第一，检查embedding模型是否一致。写入和检索必须用同一个模型，否则向量空间不对齐，相似度计算完全没意义。我见过有人写入用OpenAI的embedding，检索用本地的，结果检索出来的全是噪声。
第二，检查时间窗口是否合理。如果时间窗口设得太短，可能把相关记忆过滤掉了；设得太长，又引入了太多噪声。我的经验是，先用24小时，如果检索结果太少再放宽。
第三，检查Top-K的值。K太小会漏，K太大会引入不相关的内容。一般5到10比较合适。如果发现检索结果里有很多重复，说明去重没做好。
5.2 Docker网络不通：容器间通信的坑
用Docker Compose时，服务之间通过服务名通信。比如MCP Server连PostgreSQL，host要写postgres而不是localhost。这个坑我踩过好几次，明明本地能连，一进容器就报连接拒绝。
排查步骤：
- 进入MCP Server容器：
docker exec -it mcp-server bash - 测试DNS解析：
ping postgres - 测试端口连通性：
nc -zv postgres 5432 - 检查PostgreSQL是否在监听：
docker exec postgres pg_isready -U agent
如果DNS解析失败，检查Compose文件里服务是否在同一个网络。默认情况下，Compose会创建一个共享网络，所有服务都在里面。如果手动指定了网络，要确保服务都加入了。
5.3 内存泄漏：缓冲区没清导致的OOM
MCP Server跑久了内存持续增长，最后被OOM Killer干掉。原因通常是缓冲区没清或者连接池没释放。我的排查方法：
用docker stats看内存曲线，如果是阶梯式上升，说明有对象没释放。重点检查EpisodicWriter的buffer是否在flush后清空，以及数据库连接是否归还到连接池。Python的asyncpg连接池一定要用async with上下文管理器，否则连接不会自动归还。
还有一个隐蔽的坑：embedding向量占内存很大。1536维的float32向量，一条就是6KB。如果缓冲区里攒了1000条，就是6MB。所以批处理大小不能设太大，10到20是安全范围。
5.4 常见问题速查表
| 现象 | 可能原因 | 解决方法 |
|---|---|---|
| 检索结果为空 | 时间窗口太短或session_id不匹配 | 放宽时间窗口，检查session_id传递 |
| 检索结果重复 | 去重键设置不当 | 用session_id+step_index+action_type做唯一键 |
| 写入速度慢 | 单条写入或embedding API延迟 | 改批处理，加embedding缓存 |
| 容器间连接失败 | 用了localhost或网络不共享 | 用服务名，检查Compose网络配置 |
| 内存持续增长 | 缓冲区未清或连接泄漏 | 检查flush逻辑，用async with管理连接 |
| embedding维度不匹配 | 写入和检索用了不同模型 | 统一embedding模型，检查维度配置 |
避坑技巧：在开发阶段，把embedding结果缓存到Redis里，key用内容的MD5。这样重复内容不需要重复调用embedding API，既省钱又快。生产环境可以设一个较短的TTL，比如1小时。
6. 记忆安全与未来扩展方向
6.1 记忆污染与防御思路
Agent的记忆一旦被污染，后续所有基于记忆的决策都会出错。常见的污染来源包括：用户恶意输入诱导Agent写入错误记忆、工具返回的异常数据被当作正常结果存储、多个会话之间的记忆串扰。
防御思路有几个层面。写入前做校验：对工具返回的结果做schema验证，不符合预期的直接丢弃，不写入记忆。会话隔离：每个session_id的记忆严格隔离，检索时强制带session_id过滤，防止跨会话污染。记忆过期：Episodic Memory设TTL，比如30天，过期的自动归档或删除。异常检测：如果某个session在短时间内写入大量记忆，触发告警，可能是被攻击了。
6.2 从Episodic到Semantic的抽象
目前“hindsight”主要实现了Episodic Memory，Semantic Memory的抽象还在探索中。我的思路是定期跑一个离线任务，把Episodic Memory里的高频模式提取出来，生成Semantic Memory。比如，如果Agent发现某个用户连续多次要求“用表格输出”，就可以抽象出一条语义记忆：“用户偏好表格格式”。
这个抽象过程可以用LLM来做：把最近N条Episodic Memory喂给LLM，让它总结出规律。但要注意抽象的频率不能太高，否则Semantic Memory会变得不稳定。我一般一周跑一次。
6.3 多Agent共享记忆的架构
如果多个Agent需要共享记忆，架构就要调整。我的方案是：每个Agent有自己的Working Memory和Episodic Memory，但Semantic Memory是共享的。共享的Semantic Memory放在独立的向量数据库里，所有Agent通过MCP Server访问。
这样设计的好处是，Agent之间的经验可以沉淀到共享层，但各自的会话上下文不会互相干扰。关键是权限控制——不是所有Agent都能写入共享Semantic Memory，只有经过验证的抽象结果才能写入。
最后分享一个我在实际部署中的小技巧：给记忆加一个“置信度”字段。写入时根据来源打标，比如工具返回的结果置信度0.9，LLM自己总结的置信度0.6。检索时按置信度加权排序，这样低置信度的记忆不会轻易影响决策。这个字段加下去之后，Agent的“记错”问题明显减少了。
