★ FEATURED ARTICLE 1. 从“hindsight”说起为什么记忆是Agent落地的最后一公里“hindsight”这个词本身很有意思字面意思是“事后的洞察力”也就是我们常说的“后见之明”。把这个词放在Agent Memory的语境下它指向一个非常具体且长期被低估的问题一个LLM Agent在完成一轮任务之后能不能把这一轮里发生的关键信息沉淀下来在下一轮、下一个会话、甚至下一个完全不同的任务里重新调用换句话说Agent有没有“记住自己做过什么”的能力。我接触过不少做Agent落地的团队大家一开始的注意力几乎都放在模型选型、Prompt工程、工具调用链编排上这些当然重要但真正让一个Demo变成可用产品的分水岭往往就是记忆系统。一个没有记忆的Agent每次对话都像失忆患者重新问诊用户要反复交代背景工具调用结果无法跨轮次复用多步任务做到一半断了就得从头再来。这种体验在演示阶段还能靠精心设计的脚本糊弄过去一旦进入真实场景立刻原形毕露。hindsight这个项目标题结合agent memory、LLM、MCP、Docker这几个关键词我判断它要解决的核心问题是为LLM Agent构建一套可持久化、可检索、可跨会话复用的记忆层并且通过MCP协议把这个能力标准化地暴露给上层Agent框架同时用Docker保证部署的一致性和可移植性。这套组合拳打下来目标很明确——让记忆不再是每个Agent项目各自造轮子的私活而是一个可以插拔的基础设施组件。适合读这篇内容的人我大致分三类。第一类是正在做Agent应用开发、被多轮对话状态管理折磨的工程师第二类是对MCP协议感兴趣、想搞清楚它到底怎么落地的人第三类是想用Docker快速搭一套可复现环境、不想在依赖上浪费时间的实践派。不管你是哪一类接下来的内容都会从设计思路一路讲到实操细节尽量把每个选择背后的“为什么”说清楚。2. 记忆系统的整体设计与方案选型2.1 为什么不是简单的向量数据库加个表很多人一提到Agent Memory第一反应就是“上个向量库不就行了”。这个思路不能说错但太粗糙。向量检索解决的是语义相似度召回它擅长回答“哪些历史片段和当前query语义接近”但它不擅长回答“这个任务当前进行到哪一步了”“上一轮工具调用返回了什么结构化结果”“用户明确说过的偏好是什么”。这些信息有的是时序性的有的是结构化的有的是需要精确匹配的单靠一个向量索引覆盖不了。hindsight这类项目通常会把记忆拆成几个层次来设计。最底层是原始事件流记录每一轮交互的完整上下文包括用户输入、模型输出、工具调用参数和返回值这一层追求的是完整性和可追溯性通常用关系型数据库或者文档数据库存。中间层是工作记忆也就是当前任务会话内的短期状态它需要快速读写往往放在内存或者Redis这类缓存里生命周期和会话绑定。最上层是长期记忆从历史事件里抽取出来的事实、偏好、结论经过向量化之后存入向量索引供跨会话检索。这样分层的好处是各司其职。工作记忆保证当前任务不丢状态长期记忆保证Agent越用越懂用户原始事件流保证出了问题能回溯。如果只用一个向量库硬扛你会发现工作记忆的精确读写需求和高频更新会把向量库搞得很难受而长期记忆的语义召回又用不上关系型数据库的强一致性两头不讨好。2.2 MCP在这里扮演什么角色MCP全称是Model Context Protocol它是一个软件协议不是硬件协议。你可以把它理解成Agent世界里的USB-C接口标准——以前每个工具、每个数据源都要为不同的Agent框架写一套适配层现在大家约定一个协议Agent通过MCP Client去调用MCP Server暴露的能力工具方只需要实现一次MCP Server就能被所有支持MCP的Agent框架使用。把记忆系统做成MCP Server好处非常直接。你的Agent不管是基于哪个框架写的只要它支持MCP就能通过标准接口读写记忆。记忆的存储后端、检索策略、抽取逻辑全部封装在Server内部对Agent透明。这意味着你可以独立升级记忆系统换向量库、换抽取模型、调整检索策略都不需要动Agent本身的代码。这种解耦在快速迭代阶段价值巨大。具体到接口设计一个记忆MCP Server通常会暴露这么几类工具写入记忆接收一段文本或结构化事件决定存到哪一层、检索记忆根据query召回相关片段、更新工作记忆设置当前任务状态、清除或归档记忆会话结束时的清理逻辑。每个工具的输入输出schema要设计得足够通用不能绑死某个具体业务。2.3 Docker带来的部署确定性Agent Memory系统涉及的东西不少关系型数据库、向量数据库、缓存、可能还有嵌入模型服务。如果每个开发者都在自己机器上手动装一遍版本差异、配置差异、端口冲突能消耗掉大量时间。Docker和Docker Compose在这里的价值就是把整个技术栈打包成一组可复现的容器编排。我自己的习惯是任何涉及超过两个有状态服务的项目第一天就把docker-compose.yml写好。记忆系统正好符合这个特征。用Compose定义好各个服务的镜像、端口映射、数据卷、环境变量、依赖关系新同学拉下代码执行一条命令就能跑起来这比写十页安装文档都管用。而且Compose文件本身就是最好的架构文档谁依赖谁、数据存在哪、暴露什么端口一目了然。选型上关系型存储我倾向PostgreSQL它对JSON字段的支持足够好存原始事件流很顺手生态也成熟。向量检索可以用pgvector插件直接在PostgreSQL里做省得再维护一个独立的向量数据库对于中小规模记忆量完全够用。缓存用Redis工作记忆的读写延迟要求高Redis是稳妥选择。嵌入模型可以本地跑一个小模型也可以调外部API这个根据你的隐私要求和成本预算决定。3. 核心细节解析与实操要点3.1 记忆的写入策略什么时候记、记什么写入策略是记忆系统里最容易被做烂的部分。我见过一些实现把每一轮对话原封不动全量塞进向量库结果检索出来的全是噪音。记忆的价值不在于多而在于准。hindsight这类项目通常会在写入环节做几件事。第一是事件切分。一轮完整的交互可能包含用户消息、模型思考、工具调用、工具返回、模型最终回复这些不应该揉成一大坨文本。合理的做法是按语义单元切分工具调用和它的返回值绑定成一个事件模型回复单独成一个事件用户消息单独成一个事件。每个事件带上时间戳、会话ID、任务ID这些元数据方便后续过滤。第二是重要性判定。不是所有事件都值得进入长期记忆。用户随口说的一句“今天天气不错”没有长期价值但“我下周三之前要完成这份报告”就是关键约束。判定重要性可以用规则比如包含时间、数字、明确指令的事件加权也可以用一个小模型做分类。我实测下来规则加轻量模型兜底的组合性价比最高纯规则太死板纯模型成本高且不稳定。第三是抽取与压缩。原始事件流保留完整信息但进入长期记忆之前应该做一次抽取把事实性内容提炼成简洁的陈述句。比如原始事件是“用户说帮我查一下北京到上海的航班要明天上午的最好国航”抽取后可能是“用户计划明天上午从北京飞上海偏好国航”。这样检索时匹配效率更高也节省存储。注意抽取环节一定要保留原始事件的引用ID否则后续发现抽取有误时无法回溯修正。这个坑我踩过早期版本没存引用后来想重新抽取历史记忆发现原始数据已经和抽取结果对不上了。3.2 检索策略怎么让Agent找到该找的记忆检索不是简单的向量相似度Top-K。实际用下来纯向量召回在记忆场景下有几个明显问题一是时间衰减没考虑三个月前的相似记忆和昨天的相似记忆权重一样不合理二是精确约束容易被忽略用户说“用我上次说的那个模板”向量检索可能召回一堆模板相关但并非“上次那个”的内容三是多跳推理缺失有些记忆需要先找到A再通过A的关联找到B。hindsight这类项目一般会做混合检索。向量召回负责语义相关性关键词检索负责精确匹配时间衰减因子负责给近期记忆加权元数据过滤负责限定范围。最终得分是这几项的加权组合。权重怎么定没有标准答案要根据你的场景调。任务型Agent可能时间衰减权重要高一些知识型Agent可能语义相关性权重要高一些。另一个关键是检索结果的组装。召回一堆记忆片段之后不能直接全塞进Prompt那样会挤占上下文窗口还引入噪音。通常的做法是做一个重排序用交叉编码器或者小模型对召回结果精排取Top-N然后按时间或重要性组织成一段连贯的上下文。组织方式也有讲究按时间正序排列适合让模型理解事件发展脉络按重要性排列适合让模型快速抓住关键约束。3.3 工作记忆的生命周期管理工作记忆和长期记忆的管理逻辑完全不同。长期记忆追求持久和可检索工作记忆追求快速和准确。一个任务会话开始时工作记忆初始化可能从长期记忆里加载一些相关背景任务进行中每一步的状态更新都写入工作记忆任务结束时工作记忆里的关键结论沉淀到长期记忆然后工作记忆清空或归档。这里有个容易忽略的点工作记忆的并发控制。如果一个用户同时开了多个会话或者一个会话里有多个子任务并行工作记忆的读写需要隔离。简单的做法是按会话ID分key复杂一点的可能需要事务支持。Redis的原子操作能覆盖大部分场景但如果你的工作记忆更新逻辑涉及多步读写就要考虑用Lua脚本或者分布式锁保证一致性。还有一个实践中的坑工作记忆的过期策略。如果会话异常中断没有走到清理逻辑工作记忆会一直占着内存。所以一定要设置TTL作为兜底同时定期扫描孤儿会话做清理。TTL设多长取决于你的任务典型时长我一般设24小时足够覆盖绝大多数场景又不会让垃圾数据堆积太久。4. 实操过程与核心环节实现4.1 用Docker Compose搭建基础环境先把环境跑起来再谈代码。下面这份Compose文件是我在多个项目里迭代出来的基础版本包含PostgreSQL带pgvector、Redis和记忆服务本身。你可以直接拿去改。version: 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: 3s retries: 5 redis: image: redis:7-alpine container_name: hindsight-redis ports: - 6379:6379 volumes: - redisdata:/data command: redis-server --appendonly yes memory-server: build: . container_name: hindsight-memory depends_on: postgres: condition: service_healthy redis: condition: service_started environment: DATABASE_URL: postgresql://hindsight:hindsight_devpostgres:5432/hindsight REDIS_URL: redis://redis:6379/0 EMBEDDING_MODEL: BAAI/bge-small-zh-v1.5 ports: - 8080:8080 volumes: pgdata: redisdata:几个关键点解释一下。pgvector的官方镜像已经预装了扩展省得自己编译。healthcheck很重要memory-server依赖postgres如果不加condition容器启动顺序不保证memory-server可能在postgres还没ready的时候就尝试连接然后崩掉。Redis开了appendonly持久化工作记忆虽然可以丢但能持久化总归更稳。嵌入模型先用一个小型中文模型本地跑不需要GPU适合开发阶段。启动命令就一条docker compose up -d第一次跑会拉镜像、构建memory-server视网络情况可能需要几分钟。起来之后用docker compose ps确认所有服务都是healthy状态。如果memory-server起不来先看日志docker compose logs memory-server大概率是数据库连接问题检查postgres的healthcheck是否通过。提示Windows上装Docker Desktop如果报“virtualization support not detected”先去BIOS里确认CPU虚拟化开了然后在Windows功能里确认Hyper-V或WSL2启用了。这个报错和Docker本身没关系是系统层面的虚拟化没开。4.2 数据库表结构设计PostgreSQL里至少需要三张表事件表、记忆表、工作记忆表。事件表存原始交互记忆表存抽取后的长期记忆工作记忆表存会话状态。CREATE EXTENSION IF NOT EXISTS vector; CREATE TABLE events ( id BIGSERIAL PRIMARY KEY, session_id TEXT NOT NULL, task_id TEXT, event_type TEXT NOT NULL, content TEXT NOT NULL, metadata JSONB DEFAULT {}, created_at TIMESTAMPTZ DEFAULT NOW() ); CREATE INDEX idx_events_session ON events(session_id); CREATE INDEX idx_events_task ON events(task_id); CREATE TABLE memories ( id BIGSERIAL PRIMARY KEY, source_event_id BIGINT REFERENCES events(id), content TEXT NOT NULL, embedding vector(512), importance REAL DEFAULT 0.5, access_count INT DEFAULT 0, last_accessed TIMESTAMPTZ, created_at TIMESTAMPTZ DEFAULT NOW() ); CREATE INDEX idx_memories_embedding ON memories USING ivfflat (embedding vector_cosine_ops) WITH (lists 100); CREATE TABLE working_memory ( session_id TEXT PRIMARY KEY, state JSONB NOT NULL DEFAULT {}, updated_at TIMESTAMPTZ DEFAULT NOW(), expires_at TIMESTAMPTZ DEFAULT NOW() INTERVAL 24 hours );embedding维度512对应bge-small-zh-v1.5的输出维度如果你换模型要同步改。ivfflat索引的lists参数一般取行数的平方根左右初期数据少可以设小一点数据量上来了再重建。working_memory的expires_at给了默认24小时配合定期清理任务使用。4.3 记忆写入的代码实现写入逻辑的核心是把一轮交互拆成事件判定重要性抽取长期记忆。下面是一个简化但可运行的Python实现。import json import time from datetime import datetime from typing import Optional import psycopg2 import redis from sentence_transformers import SentenceTransformer class MemoryWriter: def __init__(self, db_url: str, redis_url: str, model_name: str): self.conn psycopg2.connect(db_url) self.redis redis.from_url(redis_url) self.model SentenceTransformer(model_name) def write_event(self, session_id: str, event_type: str, content: str, task_id: Optional[str] None, metadata: Optional[dict] None) - int: with self.conn.cursor() as cur: cur.execute( INSERT INTO events (session_id, task_id, event_type, content, metadata) VALUES (%s, %s, %s, %s, %s) RETURNING id, (session_id, task_id, event_type, content, json.dumps(metadata or {})) ) event_id cur.fetchone()[0] self.conn.commit() return event_id def score_importance(self, content: str, event_type: str) - float: score 0.5 if event_type user_message: score 0.1 if any(kw in content for kw in [记住, 重要, 必须, 截止]): score 0.3 if any(ch.isdigit() for ch in content): score 0.1 return min(score, 1.0) def extract_memory(self, content: str) - Optional[str]: # 实际项目里这里调LLM做抽取这里用规则示意 if len(content) 10: return None return content.strip() def write_memory(self, event_id: int, content: str, importance: float) - Optional[int]: extracted self.extract_memory(content) if not extracted: return None embedding self.model.encode(extracted).tolist() with self.conn.cursor() as cur: cur.execute( INSERT INTO memories (source_event_id, content, embedding, importance) VALUES (%s, %s, %s, %s) RETURNING id, (event_id, extracted, embedding, importance) ) memory_id cur.fetchone()[0] self.conn.commit() return memory_id def update_working_memory(self, session_id: str, state: dict): key fwm:{session_id} self.redis.setex(key, 86400, json.dumps(state))这段代码里score_importance用的是规则打分实际项目里可以替换成小模型分类。extract_memory同理规则版只是占位。update_working_memory用Redis的setexTTL 86400秒即24小时和数据库里的expires_at保持一致。4.4 检索接口的实现检索要做混合召回下面是一个基础版本。import numpy as np class MemoryRetriever: def __init__(self, db_url: str, model_name: str): self.conn psycopg2.connect(db_url) self.model SentenceTransformer(model_name) def vector_search(self, query: str, top_k: int 20): q_emb self.model.encode(query).tolist() with self.conn.cursor() as cur: cur.execute( SELECT id, content, importance, created_at, 1 - (embedding %s::vector) AS similarity FROM memories ORDER BY embedding %s::vector LIMIT %s, (q_emb, q_emb, top_k) ) return cur.fetchall() def keyword_search(self, query: str, top_k: int 20): with self.conn.cursor() as cur: cur.execute( SELECT id, content, importance, created_at, ts_rank(to_tsvector(simple, content), plainto_tsquery(simple, %s)) AS rank FROM memories WHERE to_tsvector(simple, content) plainto_tsquery(simple, %s) ORDER BY rank DESC LIMIT %s, (query, query, top_k) ) return cur.fetchall() def hybrid_search(self, query: str, top_k: int 10, time_decay_days: float 30.0): vec_results {r[0]: r for r in self.vector_search(query)} kw_results {r[0]: r for r in self.keyword_search(query)} all_ids set(vec_results) | set(kw_results) now datetime.now().timestamp() scored [] for mid in all_ids: vec_score vec_results.get(mid, (0,)*5)[4] if mid in vec_results else 0 kw_score kw_results.get(mid, (0,)*5)[4] if mid in kw_results else 0 importance (vec_results.get(mid) or kw_results.get(mid))[2] created (vec_results.get(mid) or kw_results.get(mid))[3] age_days (now - created.timestamp()) / 86400 decay np.exp(-age_days / time_decay_days) final 0.5 * vec_score 0.2 * kw_score 0.2 * importance 0.1 * decay scored.append((mid, final)) scored.sort(keylambda x: x[1], reverseTrue) top_ids [s[0] for s in scored[:top_k]] return self.fetch_by_ids(top_ids) def fetch_by_ids(self, ids): if not ids: return [] with self.conn.cursor() as cur: cur.execute( SELECT id, content, importance FROM memories WHERE id ANY(%s), (ids,) ) return cur.fetchall()混合检索的权重是拍脑袋定的实际用的时候要拿真实query做A/B测试调。时间衰减用指数函数time_decay_days控制衰减速度30天意味着30天前的记忆权重降到约0.37。这个参数对任务型Agent可以调小比如7天让近期记忆更突出。4.5 把记忆能力包装成MCP ServerMCP Server的实现取决于你用的语言和SDK。核心是把上面这些能力暴露成标准工具。下面用伪代码示意工具定义。# 工具1写入事件 { name: write_event, description: 记录一轮交互中的事件, inputSchema: { type: object, properties: { session_id: {type: string}, event_type: {type: string, enum: [user_message, assistant_message, tool_call, tool_result]}, content: {type: string}, task_id: {type: string}, metadata: {type: object} }, required: [session_id, event_type, content] } } # 工具2检索记忆 { name: search_memory, description: 根据query检索相关长期记忆, inputSchema: { type: object, properties: { query: {type: string}, top_k: {type: integer, default: 10}, session_id: {type: string} }, required: [query] } } # 工具3更新工作记忆 { name: update_working_memory, description: 更新当前会话的工作记忆状态, inputSchema: { type: object, properties: { session_id: {type: string}, state: {type: object} }, required: [session_id, state] } }工具描述要写得让LLM能理解什么时候该调用。write_event的description里最好说明“在每轮交互的关键节点调用”search_memory说明“在需要回忆历史信息时调用”。这些描述直接影响Agent的工具调用准确率值得反复打磨。5. 常见问题与排查技巧实录5.1 记忆检索召回不准的排查思路召回不准是最常见的问题表现是Agent该想起的没想起或者想起了一堆无关的。排查按这个顺序走。先看写入环节。如果原始事件就没写进去检索自然无从谈起。查events表里有没有对应session的记录确认写入逻辑没有被异常吞掉。我遇到过因为metadata里有不可序列化的对象导致写入静默失败的情况后来加了写入异常日志才定位到。再看抽取环节。如果事件写了但memories表里没有对应记录说明抽取逻辑把它过滤掉了。检查抽取的阈值是不是太严或者抽取模型对某类内容有偏见。可以临时把抽取阈值调低看召回是否改善。然后看检索环节。如果memories表里有数据但检索不出来先单独跑向量检索和关键词检索看是哪一路出了问题。向量检索不出来通常是嵌入模型和查询语义不匹配试试换个模型或者对query做改写。关键词检索不出来通常是分词问题中文场景下simple配置按空格分词效果很差需要换成支持中文的分词器或者用pg_jieba扩展。最后看排序环节。如果召回的内容相关但排序靠后调权重。把时间衰减调弱、把语义相似度权重调高试试。排序问题没有银弹只能拿真实数据反复调。5.2 Docker环境下的典型故障Docker Desktop在Windows上启动失败报虚拟化相关的错前面提过去BIOS开虚拟化、启用WSL2。如果WSL2启用后还是不行试试wsl --update更新内核。容器间网络不通表现是memory-server连不上postgres。先确认它们在同一个Compose网络里Compose默认会创建一个网络所有服务都在里面。然后用服务名而不是localhost做主机名Compose内置了DNS解析。如果还不行进容器里ping postgres看解析是否正常。数据卷权限问题PostgreSQL容器启动时报权限错误。这通常是因为宿主机上的数据卷目录属主不对。删掉数据卷重新初始化或者手动chown。开发环境直接docker compose down -v清掉重来最快。镜像拉取慢或者失败配置镜像加速器。这个在Docker Desktop的设置里有入口填一个可用的加速地址就行。如果公司网络有代理也要在Docker Desktop里配好代理否则容器内部访问外部服务会失败。5.3 记忆膨胀与性能衰减系统跑一段时间之后memories表越来越大检索变慢召回质量下降。这是记忆系统必然会遇到的问题要提前设计清理策略。一个做法是定期归档。超过一定时间且access_count很低的记忆移到归档表主表只保留活跃记忆。归档表不建向量索引需要时再查。另一个做法是记忆合并把多条相似记忆合并成一条更概括的记忆减少冗余。合并可以用聚类加摘要的方式做但要注意保留原始引用。access_count这个字段很有用每次检索命中就加一。它反映了记忆的实际价值清理时优先保留高access_count的。last_accessed配合使用很久没被访问且重要性低的可以优先清理。提示清理策略一定要做成可配置、可回滚的。我见过直接硬删的后来发现删错了想恢复没有备份只能认栽。归档比删除安全至少数据还在。5.4 MCP接入时的常见报错Agent框架报找不到MCP Server先确认Server进程在跑端口在监听。然后确认Agent框架的MCP配置里地址和端口写对了。如果是本地stdio方式的MCP确认启动命令的路径和参数正确。工具调用报schema不匹配检查inputSchema的required字段和实际传入的参数是否一致。有些框架对additionalProperties敏感如果schema里没定义但传了额外字段会报错可以在schema里加additionalProperties: false明确禁止或者干脆不限制。工具调用超时记忆检索如果涉及大量数据或者嵌入模型推理慢可能超过框架默认超时。优化检索性能或者调大超时配置。嵌入模型首次加载会慢可以考虑启动时预热。6. 一些实操心得与后续扩展方向记忆系统的调优是个持续过程没有一劳永逸的配置。我自己的习惯是建一个小的评测集收集几十条真实query和期望召回的记忆每次调整检索策略就跑一遍评测看召回率和准确率的变化。没有评测集的话调参就是盲人摸象今天觉得好了明天又觉得差了完全凭感觉。嵌入模型的选择上中文场景我试过几个bge-small-zh在速度和效果之间平衡得不错适合开发和小规模生产。如果对效果要求更高且预算允许可以上更大的模型或者调外部嵌入API。但要注意换模型意味着所有历史记忆的向量都要重新生成这个迁移成本要提前考虑。所以初期选型时宁可多花点时间对比也别频繁换。工作记忆和长期记忆的边界有时候会模糊。我的判断标准是这个信息在当前任务结束后还有没有价值有就沉淀到长期记忆没有就随工作记忆一起清理。比如工具调用的中间结果任务结束就没用了不用沉淀。但工具调用揭示的用户偏好比如“用户喜欢用表格展示数据”这个有长期价值要沉淀。后续扩展的话有几个方向值得探索。一是记忆的主动遗忘不是所有旧记忆都该保留有些过时的信息留着反而干扰需要一套机制识别并淡化它们。二是跨Agent的记忆共享多个Agent协作时记忆能不能在它们之间安全地流转。三是记忆的可解释性当Agent基于某条记忆做出决策时能不能向用户展示这条记忆的来源和推理链这在需要审计的场景里很重要。最后分享一个小技巧在开发阶段给记忆系统加一个调试接口输入session_id就能看到这个会话的所有事件、抽取出的记忆、以及每次检索的召回结果和得分。这个接口在排查问题时能省下大量时间比翻日志高效得多。上线前记得把这个接口关掉或者加权限控制。
阅读完成 · 觉得有帮助？
