ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

hindsight:基于MCP与Docker的LLM Agent三层记忆系统实战

hindsight:基于MCP与Docker的LLM Agent三层记忆系统实战 1. 为什么“事后复盘”这件事值得单独做一个项目做过几年开发的人都有一个共同的体会真正让人成长的往往不是写代码那一刻而是事后回头看的那一刻。英文里有个词叫 hindsight直译就是“后见之明”说白了就是事后诸葛亮。但这个词在工程语境里其实是个褒义词——它代表一种能力在事情发生之后能够回看整个过程找出当时没看到的关键信息然后把它变成下一次的直觉。我把这个项目命名为 hindsight核心目标只有一个给 LLM Agent 做一套真正能用的记忆系统。不是那种把对话历史一股脑塞进上下文窗口的假记忆而是能存、能查、能忘、能关联的工程化记忆层。热搜词里出现了 agent memory、working memory、MCP、Docker 这些关键词说明大家关心的方向是一致的——Agent 要能记住东西而且要记得有结构、有层次、有取舍。这个项目适合谁看如果你正在做 Agent 相关的开发被上下文窗口限制折磨过或者你用过 MCP 协议接过工具但还没想清楚记忆该怎么管那这篇内容会对你有直接帮助。如果你只是听说过 LLM 但没动过手也没关系我会把每个概念用生活化的方式讲清楚保证你能看懂整个思路。先说结论hindsight 这套记忆系统的核心设计思路是把 Agent 的记忆分成三层——工作记忆、情景记忆、语义记忆分别对应“正在做的事”“做过的事”“知道的事”。三层之间通过 MCP 协议暴露接口底层用 Docker 做环境隔离和部署。下面我会把这套东西从设计到落地完整拆一遍。2. 记忆系统的整体设计与分层思路2.1 为什么不能只靠上下文窗口很多人做 Agent 的第一反应是上下文窗口不是有 128K 甚至 1M token 了吗直接把历史对话全塞进去不就行了我一开始也是这么想的实测下来问题很多。第一个问题是成本。每次请求都把几万 token 的历史带上费用是线性增长的对话轮次一多账单会非常难看。第二个问题是注意力稀释。上下文里塞的东西越多模型对关键信息的注意力就越分散实测下来回答质量反而下降。第三个问题是无法持久化。上下文窗口是会话级的会话一结束就没了下次开新会话Agent 又是白纸一张。所以记忆系统要解决的核心矛盾是如何在有限的上下文预算内让 Agent 获得尽可能准确、尽可能相关的历史信息。这就引出了分层设计。2.2 三层记忆的职责划分我把记忆分成三层这个划分参考了认知科学里对人类记忆的分类但在工程上做了简化。工作记忆Working Memory对应“我现在正在干什么”。它保存的是当前任务的即时状态比如用户刚说的话、当前正在调用的工具、中间产生的临时结果。工作记忆的特点是容量小、更新快、生命周期短任务结束就可以丢弃。在实现上工作记忆就是当前会话的上下文但我会做主动的压缩和摘要而不是无脑堆砌。情景记忆Episodic Memory对应“我过去做过什么”。它保存的是一次次具体的交互事件比如“上周三用户让我查了某个数据”“昨天这个任务失败了因为参数不对”。情景记忆的特点是带时间戳、带上下文、可检索。当 Agent 遇到类似任务时可以检索出过去的相关经历作为参考。语义记忆Semantic Memory对应“我知道什么”。它保存的是从多次交互中提炼出来的稳定知识比如“这个用户的偏好是用简洁的回答”“这个 API 的调用需要先鉴权”。语义记忆是经过抽象和归纳的不依赖具体某次交互是 Agent 的长期知识库。这三层的划分不是拍脑袋定的它直接决定了存储结构、检索策略和淘汰机制。工作记忆用内存队列情景记忆用带时间索引的向量库语义记忆用结构化的键值存储加向量检索。下面会详细讲。2.3 为什么选 MCP 作为对外接口MCP 这个词在热搜里反复出现很多人问“MCP 是软件协议还是硬件协议那个概念”。简单说MCP 是一个软件层的协议全称是 Model Context Protocol它的作用是标准化 LLM 和外部工具、数据源之间的交互方式。你可以把它理解成“AI 世界的 USB 接口”——不管后面接的是数据库、文件系统还是某个 API只要符合 MCP 规范模型就能用统一的方式调用。hindsight 选择 MCP 作为对外接口理由有三个。第一是解耦。记忆系统的实现可以随便换只要 MCP 接口不变上层的 Agent 就不用改。第二是复用。现在很多工具和框架都开始支持 MCP接进来就能用不用自己造轮子。第三是标准化。MCP 定义了工具描述、参数 schema、返回格式这些规范让记忆的读写操作变得可预测、可测试。具体来说hindsight 通过 MCP 暴露了这么几个工具memory_store用于写入记忆memory_recall用于检索记忆memory_forget用于删除或降权记忆memory_summarize用于触发摘要压缩。每个工具都有明确的参数定义Agent 可以根据当前需要自主决定调用哪个。2.4 Docker 在整套方案里的角色热搜里 docker 相关的词特别多docker 安装、docker desktop、docker compose、docker 网络不通说明这是很多人的痛点。在 hindsight 项目里Docker 承担的是环境一致性和部署简化的角色。记忆系统依赖几个组件向量数据库、关系型数据库、缓存。如果每个都手动装光是版本兼容就能折腾一天。用 Docker Compose 把这些组件编排起来一条命令就能拉起整套环境换台机器也能复现。而且记忆系统通常要和 Agent 主程序隔离部署Docker 的网络和卷机制正好适合这种场景。我用的组合是PostgreSQL 加 pgvector 做向量存储Redis 做工作记忆的缓存记忆服务本身打包成一个独立的容器。下面会给出完整的 compose 配置。3. 核心细节解析与实操要点3.1 工作记忆的压缩策略工作记忆最容易踩的坑就是“什么都往里塞”。我的做法是给工作记忆设一个 token 预算比如 4000 token超过就触发压缩。压缩不是简单截断而是分两步走。第一步是识别关键信息。当前任务的目标、用户的最新指令、正在调用的工具及其参数这些是必须保留的。中间的推理过程、已经完成的子任务结果这些可以压缩成一句话摘要。第二步是生成摘要。我会用一个便宜的小模型来做摘要而不是用主模型这样成本可控。摘要的 prompt 大概是这样的SUMMARY_PROMPT 把下面的对话历史压缩成不超过 200 字的摘要保留 1. 用户的原始目标 2. 已经完成的关键步骤 3. 当前卡在哪里 4. 下一步要做什么 丢弃寒暄、重复确认、中间推理细节。 对话历史 {history} 这里有个经验摘要一定要保留“当前卡在哪里”和“下一步要做什么”这两个信息在后续推理里最容易被用到。我试过只保留目标不保留进度结果 Agent 会重复做已经做过的事。3.2 情景记忆的存储结构情景记忆的每一条记录我设计了这么几个字段字段名类型说明iduuid唯一标识timestampdatetime事件发生时间task_typestring任务类型标签summarytext事件摘要embeddingvector摘要的向量表示outcomestring成功/失败/部分成功metadatajsonb其他结构化信息这里的关键设计是 task_type 和 outcome。task_type 用于粗筛比如“数据查询”“文件操作”“代码生成”检索时可以先按类型过滤再算向量相似度速度会快很多。outcome 用于给记忆加权成功的经验权重高失败的教训权重也不低——因为失败案例往往更有参考价值能帮 Agent 避开坑。embedding 我用的是 768 维的模型实测下来在中文场景下效果够用而且存储成本比 1536 维低一半。如果你主要处理英文可以用更大的维度。3.3 语义记忆的提炼机制语义记忆不是手动写的而是从情景记忆里自动提炼的。我设了一个触发条件当某个 task_type 下的情景记忆超过 10 条就触发一次提炼。提炼的逻辑是让模型读这 10 条记录找出其中的共性规律。比如 10 次数据查询里有 8 次都因为没加时间范围而超时那就可以提炼出一条语义记忆“数据查询任务必须指定时间范围”。提炼出来的语义记忆存成键值对加向量。键是规律的一句话描述值是具体的操作建议向量用于检索。这样当 Agent 遇到新任务时可以先检索语义记忆看看有没有现成的经验可用。注意语义记忆一定要设过期时间或者置信度衰减。因为业务在变三个月前的规律现在可能不适用了。我的做法是给每条语义记忆一个 confidence 分数每次被成功使用就加分长时间不用就减分低于阈值就归档。3.4 MCP 工具的参数设计MCP 工具的参数 schema 设计很关键设计得不好模型根本不知道怎么调。我踩过的坑是参数名太抽象比如用q表示查询模型经常传错。后来改成query_text、task_type_filter、time_range这种自解释的名字调用成功率明显提升。memory_recall这个工具的参数是这样的{ name: memory_recall, description: 检索历史记忆用于查找过去类似任务的经验, parameters: { query_text: 描述你要找什么用自然语言, memory_layer: episodic 或 semantic默认 episodic, task_type_filter: 可选按任务类型过滤, top_k: 返回条数默认 5最大 20, time_range: 可选格式 2024-01-01~2024-01-31 } }description 里我特意写了“用于查找过去类似任务的经验”这是给模型看的提示能引导它在合适的时机调用。实测下来description 写得好模型的调用准确率能提升不少。4. 实操过程与核心环节实现4.1 环境准备Docker Compose 编排先把环境搭起来。我假设你已经装好了 Docker 和 Docker Compose。如果还没装Windows 用户装 Docker Desktop 就行注意要开启 WSL2 后端不然性能很差。装完之后用docker --version和docker compose version确认一下。下面是 hindsight 的 compose 文件我放在项目根目录的docker-compose.ymlversion: 3.9 services: postgres: image: pgvector/pgvector:pg16 container_name: hindsight-postgres environment: POSTGRES_USER: hindsight POSTGRES_PASSWORD: hindsight_dev POSTGRES_DB: hindsight ports: - 5432:5432 volumes: - pgdata:/var/lib/postgresql/data healthcheck: test: [CMD-SHELL, pg_isready -U hindsight] interval: 5s timeout: 5s retries: 5 redis: image: redis:7-alpine container_name: hindsight-redis ports: - 6379:6379 volumes: - redisdata:/data command: redis-server --appendonly yes memory-service: build: ./service container_name: hindsight-service ports: - 8080:8080 environment: DATABASE_URL: postgresql://hindsight:hindsight_devpostgres:5432/hindsight REDIS_URL: redis://redis:6379/0 EMBEDDING_MODEL: bge-base-zh depends_on: postgres: condition: service_healthy redis: condition: service_started volumes: pgdata: redisdata:这里有几个细节值得说。pgvector 我直接用了官方带 pgvector 的镜像省得自己编译扩展。healthcheck 是必须的不然 memory-service 可能在 postgres 还没就绪时就启动导致连接失败。depends_on 配合 condition 能保证启动顺序。启动命令就一句docker compose up -d第一次启动会拉镜像可能要几分钟。启动后用docker compose ps看状态三个服务都是 running 就对了。提示如果遇到 docker 网络不通的问题先检查是不是代理配置干扰了。Docker Desktop 的设置里有个 Resources 选项里面的 proxy 配置如果和系统代理冲突会导致容器拉不到镜像。把 Docker 的代理关掉用系统代理就行。4.2 数据库初始化Postgres 起来之后需要建表和索引。我写了一个初始化脚本放在service/init.sqlCREATE EXTENSION IF NOT EXISTS vector; CREATE TABLE episodic_memory ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), timestamp TIMESTAMPTZ NOT NULL DEFAULT now(), task_type VARCHAR(64) NOT NULL, summary TEXT NOT NULL, embedding vector(768), outcome VARCHAR(16) NOT NULL, metadata JSONB DEFAULT {}::jsonb ); CREATE INDEX idx_episodic_task_type ON episodic_memory(task_type); CREATE INDEX idx_episodic_timestamp ON episodic_memory(timestamp DESC); CREATE INDEX idx_episodic_embedding ON episodic_memory USING ivfflat (embedding vector_cosine_ops) WITH (lists 100); CREATE TABLE semantic_memory ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), key_text TEXT NOT NULL, value_text TEXT NOT NULL, embedding vector(768), confidence FLOAT DEFAULT 1.0, last_used_at TIMESTAMPTZ, created_at TIMESTAMPTZ NOT NULL DEFAULT now() ); CREATE INDEX idx_semantic_embedding ON semantic_memory USING ivfflat (embedding vector_cosine_ops) WITH (lists 50);ivfflat 索引的 lists 参数经验值是数据量的平方根。我预估情景记忆会到万级别所以设 100。如果数据量小设太大反而影响召回率。4.3 记忆写入的实现记忆写入的核心是把一次交互转成结构化记录。我写了一个store_episodic函数import uuid from datetime import datetime from typing import Optional def store_episodic( conn, task_type: str, summary: str, outcome: str, embedding: list[float], metadata: Optional[dict] None ) - str: memory_id str(uuid.uuid4()) with conn.cursor() as cur: cur.execute( INSERT INTO episodic_memory (id, timestamp, task_type, summary, embedding, outcome, metadata) VALUES (%s, %s, %s, %s, %s, %s, %s) , ( memory_id, datetime.utcnow(), task_type, summary, embedding, outcome, metadata or {} ) ) conn.commit() return memory_idsummary 的生成我用的是主模型因为这一步的质量直接影响后续检索效果。prompt 是这样的EPISODIC_SUMMARY_PROMPT 把下面这次任务交互总结成一句话不超过 80 字必须包含 - 任务目标 - 关键操作 - 最终结果 交互记录 {interaction} 80 字是个经验值太短信息不够太长向量检索的精度会下降。4.4 记忆检索的实现检索分两步先粗筛再精排。粗筛用 task_type 和时间范围过滤精排用向量相似度。def recall_episodic( conn, query_embedding: list[float], task_type: Optional[str] None, top_k: int 5, time_range: Optional[tuple] None ) - list[dict]: conditions [] params [] if task_type: conditions.append(task_type %s) params.append(task_type) if time_range: conditions.append(timestamp BETWEEN %s AND %s) params.extend(time_range) where_clause AND .join(conditions) if conditions else TRUE sql f SELECT id, timestamp, task_type, summary, outcome, metadata, 1 - (embedding %s::vector) AS similarity FROM episodic_memory WHERE {where_clause} ORDER BY embedding %s::vector LIMIT %s params [query_embedding] params [query_embedding, top_k] with conn.cursor() as cur: cur.execute(sql, params) rows cur.fetchall() return [ { id: row[0], timestamp: row[1], task_type: row[2], summary: row[3], outcome: row[4], metadata: row[5], similarity: row[6] } for row in rows ]这里是 pgvector 的余弦距离操作符1 - 距离就是相似度。ORDER BY 用距离升序等价于相似度降序。注意向量检索的 top_k 不要设太大。我试过设 20结果返回一堆低相似度的噪声反而干扰模型判断。5 到 8 是比较合适的范围具体看你的场景。4.5 MCP 服务的暴露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_recall, description检索历史记忆用于查找过去类似任务的经验, inputSchema{ type: object, properties: { query_text: {type: string}, memory_layer: { type: string, enum: [episodic, semantic], default: episodic }, task_type_filter: {type: string}, top_k: {type: integer, default: 5} }, required: [query_text] } ), Tool( namememory_store, description存储一次任务交互到记忆系统, inputSchema{ type: object, properties: { task_type: {type: string}, summary: {type: string}, outcome: { type: string, enum: [success, failure, partial] } }, required: [task_type, summary, outcome] } ) ] app.call_tool() async def call_tool(name: str, arguments: dict): if name memory_recall: query_emb embed(arguments[query_text]) results recall_episodic( conn, query_emb, task_typearguments.get(task_type_filter), top_karguments.get(top_k, 5) ) return [TextContent(typetext, textformat_results(results))] elif name memory_store: emb embed(arguments[summary]) memory_id store_episodic( conn, arguments[task_type], arguments[summary], arguments[outcome], emb ) return [TextContent(typetext, textfstored: {memory_id})]这个服务跑在 8080 端口Agent 那边通过 MCP 客户端连过来就能用。4.6 语义记忆的自动提炼提炼任务我做成一个定时任务每小时跑一次def extract_semantic_memory(conn, task_type: str, min_samples: int 10): with conn.cursor() as cur: cur.execute( SELECT summary, outcome FROM episodic_memory WHERE task_type %s ORDER BY timestamp DESC LIMIT %s , (task_type, min_samples) ) samples cur.fetchall() if len(samples) min_samples: return None samples_text \n.join( f- [{outcome}] {summary} for summary, outcome in samples ) prompt f 下面是 {task_type} 类型任务的 {len(samples)} 次执行记录。 请提炼出 1-3 条共性规律每条用一句话描述格式 规律xxx 建议xxx 记录 {samples_text} result call_llm(prompt) patterns parse_patterns(result) for pattern in patterns: emb embed(pattern[key]) with conn.cursor() as cur: cur.execute( INSERT INTO semantic_memory (key_text, value_text, embedding) VALUES (%s, %s, %s) , (pattern[key], pattern[value], emb) ) conn.commit()这个提炼机制实测下来效果不错尤其是对于重复性高的任务类型能提炼出很实用的规律。5. 常见问题与排查技巧实录5.1 Docker 环境相关的坑问题一Docker Desktop 启动报 virtualization support not detected。这个在 Windows 上很常见原因是 BIOS 里的虚拟化没开。进 BIOS 找 Intel VT-x 或 AMD-V开启就行。另外 Windows 的 Hyper-V 和 WSL2 功能也要在“启用或关闭 Windows 功能”里勾上。问题二docker compose up 之后 postgres 一直 unhealthy。先看日志docker compose logs postgres。常见原因是端口 5432 被本机的 postgres 占了。解决办法是把 compose 里的端口映射改成5433:5432然后连接串里的端口也改成 5433。问题三容器之间网络不通。Docker Compose 默认会创建一个 bridge 网络同 compose 文件里的服务用服务名就能互相访问。如果你发现 memory-service 连不上 postgres先确认连接串里用的是服务名postgres而不是localhost。在容器里localhost 指的是容器自己不是宿主机。5.2 记忆检索效果差的排查症状Agent 检索出来的记忆和当前任务不相关。排查顺序是这样的。第一检查 embedding 模型是否一致。写入和检索必须用同一个模型用不同的模型算出来的向量不在一个空间里相似度没有意义。第二检查 summary 的质量。如果 summary 写得太泛比如“完成了一个任务”那检索出来肯定不准。第三检查 top_k 和相似度阈值。我一般会加一个相似度下限低于 0.6 的直接丢弃宁可返回空也不要返回噪声。症状检索速度慢。先看数据量。万级数据用 ivfflat 索引查询应该在几十毫秒。如果慢检查索引有没有建上用EXPLAIN ANALYZE看执行计划。另外 task_type 过滤能大幅减少候选集尽量用上。5.3 MCP 调用失败的排查问题模型不调用 MCP 工具或者调用时参数传错。这个问题的根源通常在工具描述上。description 要写得让模型一看就知道什么时候该用。参数名要自解释别用缩写。另外 required 字段要明确不然模型可能漏传。问题MCP 服务连不上。先确认服务在跑curl http://localhost:8080/health看有没有响应。然后检查 MCP 客户端的配置传输方式stdio 还是 SSE和服务端是否匹配。如果是 SSE确认端口和路径对得上。5.4 常见问题速查表问题现象可能原因排查方法解决方案postgres 启动失败端口占用docker compose logs postgres改端口映射容器间不通用了 localhostdocker exec进容器 ping改用服务名检索结果不相关embedding 模型不一致检查写入和检索的模型统一模型检索慢索引缺失EXPLAIN ANALYZE建 ivfflat 索引MCP 不调用description 不清看模型输出改工具描述语义记忆过时没有衰减机制看 confidence 分布加衰减和归档5.5 几个实操心得第一个心得是关于摘要的。我一开始让模型自由发挥写摘要结果格式五花八门有的长有的短检索效果很不稳定。后来我强制要求摘要必须包含“目标、操作、结果”三个要素格式统一之后检索准确率明显提升。第二个心得是关于记忆的遗忘。很多人做记忆系统只想着怎么存不想着怎么删。实际上遗忘和存储一样重要。我的做法是给每条记忆一个访问计数超过 30 天没被访问且计数低于阈值的自动归档到冷存储。这样热数据保持精简检索速度不会随数据量增长而下降。第三个心得是关于测试。记忆系统的效果很难用单元测试验证因为它是概率性的。我的做法是建一个评测集准备 50 个查询和对应的期望记忆每次改动后跑一遍看召回率和准确率的变化。这个评测集是慢慢积累的但一旦有了改代码就有底气了。6. 记忆系统的扩展方向与个人体会这套 hindsight 系统目前跑在我自己的几个 Agent 项目上稳定运行了几个月。如果要说还能往哪些方向扩展我脑子里有几个想法。一个是记忆的关联图谱。现在情景记忆和语义记忆是分开存的但它们之间其实有关联。比如某条语义记忆是从哪几条情景记忆提炼出来的这个关系如果能存下来检索时就能顺着关系找到更多相关记忆。实现上可以用图数据库或者在 Postgres 里加一张关系表。另一个是跨 Agent 的记忆共享。现在每个 Agent 的记忆是独立的但如果多个 Agent 协作它们其实可以共享一部分语义记忆。这需要设计一套权限和隔离机制避免记忆污染。热搜里提到的 agentpoison 就是这类安全问题记忆被投毒会直接影响 Agent 的行为所以共享记忆一定要做来源验证。还有一个是多模态记忆。现在只存文本但实际任务里经常有图片、表格、文件。把这些也纳入记忆系统检索时就能跨模态匹配。技术上就是把不同模态的内容都 embed 到同一个向量空间用多模态模型来做。最后分享一个我在实际使用中的体会记忆系统的价值不在于存了多少而在于检索时能不能在对的时机给出对的信息。我见过太多项目把记忆做成了“什么都存、什么都查”结果模型被无关信息干扰表现反而不如没有记忆。真正好的记忆系统是知道什么时候该想起来也知道什么时候该忘掉。这个平衡点需要根据你的具体场景慢慢调没有通用答案。
返回列表