1. 从“hindsight”说起为什么我们需要给Agent装上“后视镜”“hindsight”这个词本身很有意思字面意思是“事后的洞察力”也就是我们常说的“后见之明”。放在LLM Agent的语境里它指向一个非常具体且要命的问题Agent的记忆。你肯定遇到过这种情况跟一个AI助手聊了半小时它突然忘了你五分钟前说过的关键约束或者一个自动化工作流跑了十几步中间某一步的中间结果丢了导致后面全盘崩溃。这不是模型不够聪明而是它的“记忆系统”设计得太粗糙。当前大多数Agent的记忆机制说白了就是往上下文窗口里硬塞对话历史塞满了就截断重要的和不重要的信息一起被冲进下水道。这种做法在短对话里勉强能用一旦任务链条拉长、信息密度升高立刻原形毕露。“hindsight”要解决的核心痛点就在这里。它不是简单地做一个“更大的上下文窗口”而是试图构建一套有选择、有结构、可检索、可遗忘的Agent记忆体系。你可以把它理解成给Agent装了一面后视镜——不是让它记住所有东西而是让它在该回头看的时候能精准地看到该看的东西。这套东西适合谁来折腾如果你正在做LLM应用开发尤其是涉及多轮对话、长任务编排、知识密集型工作流那Agent记忆就是你绕不过去的坎。如果你只是拿大模型聊聊天、写写文案那暂时用不上。但只要你开始认真考虑“让Agent记住用户偏好”“让Agent在跨会话场景下保持一致性”“让Agent从历史操作中学习”hindsight这类方案就值得你花时间研究。我接下来会从整体设计思路、核心机制拆解、实操落地、以及踩坑排查四个维度把hindsight这套Agent记忆方案讲透。中间会涉及MCP协议、Docker部署、存储分层这些具体技术点也会分享一些我在实际项目中积累的经验。文章偏长建议收藏后慢慢看。2. Agent记忆的整体设计与思路拆解2.1 为什么传统上下文窗口方案不够用先把这个事情说清楚不然后面很多设计决策你理解不了。LLM的上下文窗口本质上是一块固定大小的临时工作台。你往上面放东西放满了就得拿走一些才能放新的。现在主流模型动辄128K、200K token的窗口看起来很大但实际用起来消耗极快。一轮复杂的工具调用光是系统提示词、工具定义、few-shot示例就能吃掉几千token再加上多轮对话历史、工具返回结果、中间推理步骤几万token很快就没了。更关键的问题不是“装不下”而是“装不下该装的”。上下文窗口是一个无差别存储——它不区分哪些信息重要、哪些不重要、哪些该长期保留、哪些用完就该扔。这就导致两个典型故障信息淹没关键约束被大量无关对话稀释模型注意力分散输出质量下降。记忆断裂超出窗口的历史被直接丢弃Agent“忘记”了之前达成的共识或积累的中间结果。我见过太多项目在这上面翻车。一个客服Agent用户前面说了“我对花生过敏”聊了二十轮之后推荐零食时完全忘了这回事。这不是模型的问题是记忆架构的问题。2.2 hindsight的核心思路分层记忆与主动检索hindsight的设计哲学可以概括成一句话把记忆从上下文窗口里解放出来做成一个独立的外部系统。具体来说它把Agent的记忆分成几个层次记忆类型存储内容生命周期访问方式工作记忆当前任务上下文、临时变量单次会话直接注入上下文情景记忆历史对话、操作记录跨会话持久化语义检索语义记忆提炼后的事实、偏好、规则长期结构化查询程序记忆成功的工作流、工具调用模式长期模式匹配这个分层不是拍脑袋想出来的它对应的是认知科学里人类记忆的基本分类。工作记忆容量有限但访问极快情景记忆存储具体经历语义记忆存储抽象知识程序记忆存储技能。hindsight把这套模型搬到Agent上让不同类型的记忆各司其职。核心机制上hindsight做了三件关键的事第一写入时的信息提炼。不是把所有对话原封不动存进去而是在写入阶段就做一轮压缩和结构化。比如把“用户说他住在北京朝阳区家里有只猫叫咪咪对花生过敏”拆成三条独立的事实记录分别打上“位置”“宠物”“健康约束”的标签。这样做的好处是后续检索时能精准命中而不是把整段对话捞出来让模型自己找。第二检索时的相关性排序。当Agent需要回忆某些信息时hindsight不是简单按时间倒序取最近N条而是基于当前查询做语义相似度匹配再结合时间衰减、重要性权重、访问频率等因素做综合排序。这背后涉及向量检索和重排序模型的配合。第三遗忘机制。这个最容易被忽略但极其重要。记忆系统不能只进不出否则检索质量会随着数据量增长而持续下降。hindsight设计了基于时间、访问频率和重要性的衰减策略让低价值记忆逐渐淡出保持记忆库的“信噪比”。2.3 与MCP协议的关系为什么选择MCP作为接入层hindsight选择MCPModel Context Protocol作为对外接口这个决策值得展开说说。MCP本质上是一套标准化的工具调用协议它定义了LLM应用如何发现、调用外部服务。你可以把它类比成USB接口——不管你是键盘、鼠标还是U盘插上就能用不需要为每个设备单独写驱动。把Agent记忆做成MCP Server有几个明显好处解耦记忆系统独立运行不绑定特定的LLM框架。你今天用LangChain明天换AutoGen记忆层不用动。可复用同一个记忆服务可以同时给多个Agent使用共享用户偏好和知识库。可观测MCP协议天然支持工具调用的日志和追踪方便调试记忆的读写行为。生态兼容越来越多的开发工具和平台开始支持MCP接入成本低。当然MCP也不是没有代价。多一层协议就多一层网络开销和故障点对于延迟极度敏感的场景需要额外优化。但总体来看对于Agent记忆这种需要跨会话、跨应用共享的基础设施MCP带来的解耦价值远大于开销。2.4 Docker化部署的考量hindsight用Docker部署这个选择很务实。Agent记忆系统通常需要搭配向量数据库、关系型数据库、缓存服务等一堆组件裸机部署的依赖管理能把人逼疯。Docker Compose一把梭环境一致性问题直接消掉。而且记忆系统往往需要持久化存储Docker Volume的挂载机制让数据管理变得清晰。你可以把向量库的数据、关系库的数据、配置文件分别挂到不同卷上备份和迁移都方便。不过Docker部署也有坑后面实操部分我会详细讲。特别是Windows环境下Docker Desktop的虚拟化支持问题以及容器间网络通信的配置这两个是新手最容易卡住的地方。3. 核心细节解析与实操要点3.1 记忆的写入从原始对话到结构化知识写入是记忆系统的第一道关口这里的处理质量直接决定了后续检索的上限。hindsight在写入阶段做了几层处理我逐个拆解。原始输入捕获。每次Agent与用户的交互、每次工具调用的输入输出都会被完整记录下来。这部分是原始素材不做任何加工存到冷存储里备查。注意这里说的是“完整记录”包括时间戳、会话ID、角色标识、原始文本。这些元数据在后续检索和审计时非常关键。信息抽取与结构化。这是核心环节。hindsight会调用LLM对原始对话做一轮信息抽取把非结构化的文本转成结构化的事实条目。抽取的维度包括实体人名、地名、组织、产品等属性实体的特征描述关系实体之间的关联事件发生了什么、什么时候、涉及谁偏好用户的喜好、厌恶、约束条件抽取的prompt设计很讲究。我试过几种不同的抽取策略发现按类型分步抽取比一次性全抽效果更好。比如先抽实体和属性再基于已抽出的实体抽关系最后抽事件和偏好。这样做的好处是每一步的认知负荷更低抽取准确率明显提升。去重与合并。同一个事实可能在多轮对话中反复出现比如用户多次提到“我在北京”。如果每次都存一条记忆库会迅速膨胀且冗余。hindsight的做法是对新抽取的事实做相似度匹配如果和已有事实高度相似就更新已有条目的置信度和时间戳而不是新增一条。这里有个细节相似度阈值不能设太高。我一开始把阈值设到0.95结果“我在北京”和“我住在北京朝阳区”被判定为不同事实各存了一条。后来调到0.85左右合并效果就比较合理了。但也不能太低否则“我喜欢猫”和“我喜欢狗”可能被错误合并。这个阈值需要根据你的具体领域数据来调。重要性评分。不是所有记忆都同等重要。hindsight在写入时会给每条记忆打一个重要性分数后续检索和遗忘都会用到。评分维度包括信息类型健康约束、安全规则这类硬性约束权重最高情感强度用户表达强烈情绪的内容权重较高重复频率反复出现的信息权重递增时效性有时效性的信息如“我明天要出差”在过期后权重骤降重要性评分可以用规则引擎做也可以让LLM来打分。规则引擎的好处是稳定可控LLM的好处是能捕捉微妙语义。我目前的方案是规则打底、LLM微调兼顾稳定性和灵活性。3.2 记忆的检索如何让Agent“想起”该想起的检索是记忆系统价值兑现的环节。写得再好检索不出来等于零。hindsight的检索流程大致是查询理解 → 多路召回 → 重排序 → 上下文注入。查询理解。Agent发起检索时原始查询往往是一句模糊的自然语言比如“用户之前提到的那个偏好”。直接拿这句话去做向量检索效果不会好。hindsight会先用LLM对查询做一轮改写和扩展提取出关键实体和意图生成多个检索子查询。比如上面那句话可能被扩展成“用户 偏好 饮食 健康 约束”等多个查询向量。多路召回。单一检索策略很难覆盖所有场景。hindsight同时走几条路向量检索基于语义相似度召回关键词检索基于BM25等传统算法召回弥补向量检索对精确匹配的不足结构化查询基于标签、时间范围、实体类型做过滤图遍历沿着实体关系图扩散召回关联记忆这几路召回的结果合并后进入重排序阶段。重排序。召回阶段追求的是“不漏”重排序阶段追求的是“精准”。hindsight用一个交叉编码器cross-encoder对召回结果做精细打分综合考虑语义相关性、时间新鲜度、重要性权重、访问频率等因素。最终取Top-K条注入Agent的上下文。K值的选择是个权衡。太小了可能漏掉关键信息太大了会挤占上下文窗口且引入噪声。我的经验是动态K值简单查询取3-5条复杂查询取8-12条同时设置一个token预算上限确保注入的记忆不超过总上下文的20%。上下文注入格式。检索出来的记忆怎么放进prompt里也有讲究。hindsight的做法是把记忆按类型分组用结构化格式呈现而不是简单拼接。比如[用户偏好] - 饮食对花生过敏重要性高来源2024-01-15对话 - 居住北京朝阳区重要性中来源2024-01-10对话 [近期事件] - 2024-01-20用户提到下周要去上海出差这种格式让LLM能快速定位关键信息比一大段自然语言描述高效得多。3.3 存储层选型向量库、关系库与缓存的配合hindsight的存储层不是单一数据库而是多种存储的配合。每种存储负责它最擅长的部分。向量数据库负责语义检索。选型上Milvus、Qdrant、Weaviate、Chroma都是常见选项。我的建议是数据量小于100万条Chroma或Qdrant部署简单够用数据量100万到1亿条Milvus或Qdrant集群版需要混合检索向量标量过滤Qdrant或Weaviate关系型数据库负责结构化数据和元数据管理。PostgreSQL是稳妥选择配合pgvector扩展还能兼顾向量检索小规模场景可以少部署一个组件。MySQL也行但向量支持弱一些。缓存层负责热点记忆的快速访问。Redis是标配用来缓存高频访问的记忆条目和会话状态。TTL设置要根据记忆类型区分工作记忆TTL短分钟级语义记忆TTL长小时级或永久。图数据库是可选项用于存储实体关系。Neo4j或Nebula Graph都可以。如果你的Agent需要做复杂的关系推理比如“用户的同事的上级是谁”图数据库会很有价值。否则可以先不上用关系库的外键关联凑合。存储层之间的数据同步是个容易出问题的地方。我的做法是以关系库为主数据源向量库和图库作为索引。写入时先写关系库再异步同步到向量库和图库。检索时如果向量库挂了可以降级到关系库的关键词检索。这样保证了可用性。3.4 MCP接口设计工具定义与调用约定hindsight作为MCP Server对外暴露的工具集设计直接影响使用体验。核心工具大概有这几个工具名功能关键参数memory_write写入记忆content, type, importance, metadatamemory_search检索记忆query, top_k, filters, time_rangememory_forget删除或衰减记忆memory_id, decay_factormemory_update更新已有记忆memory_id, new_content, confidencememory_summarize对一段记忆做摘要session_id, time_range工具的参数设计要遵循最小必要原则。参数太多LLM调用时容易填错参数太少灵活性不够。我的经验是每个工具控制在3-5个参数必填参数不超过2个其余给合理默认值。MCP工具的description字段非常重要它是LLM决定是否调用、怎么调用的主要依据。description要写清楚这个工具做什么、什么时候用、参数含义、返回值格式。我见过很多MCP Server的description写得含糊其辞导致LLM要么不调用要么乱调用。还有一个容易忽略的点错误处理。MCP工具调用失败时返回的错误信息要足够清晰让LLM能理解发生了什么并决定下一步。比如“记忆写入失败向量库连接超时”比“Error 500”有用得多。4. 实操过程与核心环节实现4.1 环境准备Docker与依赖组件部署先把环境搭起来。以下操作基于Ubuntu 22.04Windows和macOS用户参考对应命令调整。安装Docker和Docker Compose。# 更新包索引 sudo apt update # 安装必要依赖 sudo apt install -y ca-certificates curl gnupg lsb-release # 添加Docker官方GPG密钥 sudo mkdir -p /etc/apt/keyrings curl -fsSL https://download.docker.com/linux/ubuntu/gpg | sudo gpg --dearmor -o /etc/apt/keyrings/docker.gpg # 添加Docker仓库 echo deb [arch$(dpkg --print-architecture) signed-by/etc/apt/keyrings/docker.gpg] https://download.docker.com/linux/ubuntu $(lsb_release -cs) stable | sudo tee /etc/apt/sources.list.d/docker.list /dev/null # 安装Docker Engine和Compose插件 sudo apt update sudo apt install -y docker-ce docker-ce-cli containerd.io docker-compose-plugin # 将当前用户加入docker组避免每次sudo sudo usermod -aG docker $USER newgrp docker # 验证安装 docker --version docker compose versionWindows用户如果用Docker Desktop注意两个常见问题一是需要开启WSL2后端二是在BIOS里开启虚拟化支持。如果启动时报“Virtualization support not detected”去BIOS里找Intel VT-x或AMD-V选项打开。如果报“Docker Desktop failed to start because virtualization support is not enabled”同样是虚拟化的问题。部署PostgreSQL含pgvector。# docker-compose.yml 片段 services: postgres: image: pgvector/pgvector:pg16 container_name: hindsight-postgres environment: POSTGRES_USER: hindsight POSTGRES_PASSWORD: your_strong_password POSTGRES_DB: hindsight ports: - 5432:5432 volumes: - pgdata:/var/lib/postgresql/data healthcheck: test: [CMD-SHELL, pg_isready -U hindsight] interval: 10s timeout: 5s retries: 5 volumes: pgdata:部署Redis。redis: image: redis:7-alpine container_name: hindsight-redis ports: - 6379:6379 volumes: - redisdata:/data command: redis-server --appendonly yes --maxmemory 512mb --maxmemory-policy allkeys-lru volumes: redisdata:部署Qdrant向量库。qdrant: image: qdrant/qdrant:latest container_name: hindsight-qdrant ports: - 6333:6333 - 6334:6334 volumes: - qdrantdata:/qdrant/storage volumes: qdrantdata:启动所有服务docker compose up -d docker compose ps确认所有容器状态是healthy或running。如果某个容器反复重启用docker compose logs service_name看日志排查。4.2 记忆写入流程的代码实现环境就绪后核心是写入逻辑。以下是一个简化版的Python实现展示从对话到结构化记忆的完整链路。import json import hashlib from datetime import datetime from typing import List, Dict, Any class MemoryWriter: def __init__(self, llm_client, vector_store, relational_store, cache): self.llm llm_client self.vector_store vector_store self.relational_store relational_store self.cache cache def extract_facts(self, conversation: str) - List[Dict[str, Any]]: 从对话中抽取结构化事实 prompt f从以下对话中抽取结构化事实。对每条事实输出JSON格式 {{ content: 事实内容, type: preference|constraint|event|entity|relation, entities: [涉及的实体], importance: 0.0-1.0, confidence: 0.0-1.0 }} 对话内容 {conversation} 只输出JSON数组不要其他内容。 response self.llm.generate(prompt) try: facts json.loads(response) except json.JSONDecodeError: # 容错尝试提取JSON部分 facts self._extract_json_fallback(response) return facts def deduplicate(self, new_fact: Dict, existing_facts: List[Dict]) - Dict: 去重合并如果新事实与已有事实高度相似合并而非新增 new_embedding self.vector_store.embed(new_fact[content]) for existing in existing_facts: similarity self._cosine_similarity( new_embedding, existing[embedding] ) if similarity 0.85: # 合并更新置信度和时间戳 existing[confidence] min( 1.0, existing[confidence] 0.1 ) existing[updated_at] datetime.now().isoformat() existing[content] self._merge_content( existing[content], new_fact[content] ) return existing return new_fact def write(self, conversation: str, session_id: str) - List[str]: 完整写入流程 # 1. 抽取事实 facts self.extract_facts(conversation) # 2. 获取相关已有事实用于去重 written_ids [] for fact in facts: # 检索相似已有事实 similar self.vector_store.search( fact[content], top_k5, threshold0.7 ) # 3. 去重合并 merged self.deduplicate(fact, similar) # 4. 生成唯一ID fact_id hashlib.md5( f{session_id}:{merged[content]}.encode() ).hexdigest()[:16] # 5. 写入关系库主数据源 self.relational_store.upsert( idfact_id, contentmerged[content], typemerged[type], entitiesjson.dumps(merged.get(entities, [])), importancemerged[importance], confidencemerged[confidence], session_idsession_id, created_atdatetime.now().isoformat(), updated_atdatetime.now().isoformat() ) # 6. 写入向量库索引 embedding self.vector_store.embed(merged[content]) self.vector_store.upsert( idfact_id, vectorembedding, payload{ content: merged[content], type: merged[type], importance: merged[importance], session_id: session_id } ) # 7. 更新缓存 self.cache.setex( fmemory:{fact_id}, 3600, json.dumps(merged) ) written_ids.append(fact_id) return written_ids def _cosine_similarity(self, a, b): dot sum(x*y for x, y in zip(a, b)) norm_a sum(x*x for x in a) ** 0.5 norm_b sum(x*x for x in b) ** 0.5 return dot / (norm_a * norm_b) if norm_a and norm_b else 0.0 def _merge_content(self, old: str, new: str) - str: 合并两条相似事实的内容 if new in old: return old if old in new: return new return f{old}{new} def _extract_json_fallback(self, text: str) - List[Dict]: 从非标准输出中提取JSON import re match re.search(r\[.*\], text, re.DOTALL) if match: try: return json.loads(match.group()) except json.JSONDecodeError: pass return []这段代码有几个关键设计点值得说明。抽取prompt的JSON约束。明确要求输出JSON数组并给出schema示例。实测下来这样能显著降低解析失败率。但即便如此仍然需要fallback逻辑因为LLM偶尔会加一些解释性文字。去重阈值0.85。这个值是调出来的。太低会误合并太高会漏合并。建议在你的领域数据上做一轮标注测试找到最优阈值。先写关系库再写向量库。关系库是主数据源向量库是索引。如果向量库写入失败关系库的数据还在可以后续重建索引。反过来就麻烦了。缓存TTL 3600秒。热点记忆缓存一小时平衡了内存占用和命中率。如果你的场景访问模式不同可以调整。4.3 检索流程的实现与参数调优检索比写入更考验工程能力因为它在请求路径上延迟直接影响用户体验。class MemoryRetriever: def __init__(self, llm_client, vector_store, relational_store, cache): self.llm llm_client self.vector_store vector_store self.relational_store relational_store self.cache cache def search(self, query: str, session_id: str None, top_k: int 8, time_range: tuple None) - List[Dict]: 多路召回 重排序 # 1. 查询理解与扩展 expanded_queries self._expand_query(query) # 2. 多路召回 candidates [] # 2a. 向量检索多查询 for q in expanded_queries: results self.vector_store.search( q, top_ktop_k * 2, filtersself._build_filters(session_id, time_range) ) candidates.extend(results) # 2b. 关键词检索 keyword_results self.relational_store.keyword_search( query, top_ktop_k * 2, session_idsession_id ) candidates.extend(keyword_results) # 3. 去重 seen_ids set() unique_candidates [] for c in candidates: if c[id] not in seen_ids: seen_ids.add(c[id]) unique_candidates.append(c) # 4. 重排序 reranked self._rerank(query, unique_candidates) # 5. 取Top-K final reranked[:top_k] # 6. 更新访问计数用于后续重要性调整 for item in final: self._record_access(item[id]) return final def _expand_query(self, query: str) - List[str]: 用LLM扩展查询生成多个检索子查询 prompt f将以下查询扩展为3-5个检索子查询覆盖不同角度。 每个子查询一行不要编号不要解释。 原始查询{query} response self.llm.generate(prompt) queries [q.strip() for q in response.strip().split(\n) if q.strip()] return [query] queries[:4] # 保留原始查询 最多4个扩展 def _rerank(self, query: str, candidates: List[Dict]) - List[Dict]: 综合打分重排序 scored [] now datetime.now() for c in candidates: # 语义相关性用交叉编码器或LLM打分 relevance self._semantic_score(query, c[content]) # 时间新鲜度指数衰减 age_hours (now - datetime.fromisoformat( c.get(updated_at, c.get(created_at)) )).total_seconds() / 3600 recency 0.5 ** (age_hours / 168) # 半衰期一周 # 重要性权重 importance c.get(importance, 0.5) # 访问频率归一化 access_count c.get(access_count, 0) frequency min(1.0, access_count / 10) # 综合得分 score ( 0.5 * relevance 0.2 * recency 0.2 * importance 0.1 * frequency ) scored.append({**c, score: score}) scored.sort(keylambda x: x[score], reverseTrue) return scored def _semantic_score(self, query: str, content: str) - float: 语义相关性打分实际可用交叉编码器替代 q_emb self.vector_store.embed(query) c_emb self.vector_store.embed(content) return self._cosine_similarity(q_emb, c_emb) def _build_filters(self, session_id, time_range): filters {} if session_id: filters[session_id] session_id if time_range: filters[created_at] { gte: time_range[0], lte: time_range[1] } return filters def _record_access(self, memory_id: str): 记录访问用于后续重要性调整 self.cache.incr(faccess:{memory_id}) self.relational_store.increment_access(memory_id) def _cosine_similarity(self, a, b): dot sum(x*y for x, y in zip(a, b)) norm_a sum(x*x for x in a) ** 0.5 norm_b sum(x*x for x in b) ** 0.5 return dot / (norm_a * norm_b) if norm_a and norm_b else 0.0权重分配的逻辑。语义相关性占0.5因为它是检索的核心目标。时间新鲜度占0.2半衰期设为一周意味着七天前的记忆权重减半。重要性占0.2让高价值记忆有更高优先级。访问频率占0.1作为辅助信号。这套权重不是固定的需要根据你的场景调。比如客服场景可能更看重时间新鲜度知识库场景可能更看重语义相关性。查询扩展的收益。我做过对比测试加查询扩展后召回率提升了约15%但延迟增加了约200ms因为多了一次LLM调用。如果你的场景对延迟敏感可以考虑用更轻量的方式做扩展比如基于同义词词典或预训练的查询改写模型。4.4 MCP Server的封装与接入把上面的写入和检索逻辑封装成MCP Server对外提供标准接口。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() - list[types.Tool]: return [ types.Tool( namememory_write, description将对话内容写入长期记忆。当用户提供了值得记住的信息偏好、约束、事实、事件时调用。, inputSchema{ type: object, properties: { content: { type: string, description: 要记住的原始内容 }, session_id: { type: string, description: 会话标识 }, importance: { type: number, description: 重要性0-1默认0.5, default: 0.5 } }, required: [content, session_id] } ), types.Tool( namememory_search, description检索长期记忆。当需要回忆用户偏好、历史约定或之前的事实信息时调用。, inputSchema{ type: object, properties: { query: { type: string, description: 检索查询用自然语言描述你想回忆什么 }, top_k: { type: integer, description: 返回条数默认8, default: 8 }, session_id: { type: string, description: 限定会话范围可选 } }, required: [query] } ), types.Tool( namememory_forget, description删除或衰减指定记忆。当用户要求忘记某些信息或记忆被确认过时时调用。, inputSchema{ type: object, properties: { memory_id: { type: string, description: 记忆ID }, decay_factor: { type: number, description: 衰减因子0-11为完全删除, default: 1.0 } }, required: [memory_id] } ) ] server.call_tool() async def handle_call_tool(name: str, arguments: dict): if name memory_write: writer MemoryWriter(...) ids writer.write( arguments[content], arguments[session_id] ) return [types.TextContent( typetext, textf已写入{len(ids)}条记忆{ids} )] elif name memory_search: retriever MemoryRetriever(...) results retriever.search( arguments[query], session_idarguments.get(session_id), top_karguments.get(top_k, 8) ) formatted format_memories_for_context(results) return [types.TextContent(typetext, textformatted)] elif name memory_forget: # 实现遗忘逻辑 ... raise ValueError(fUnknown tool: {name}) async def main(): async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, InitializationOptions( server_namehindsight-memory, server_version0.1.0 ) ) if __name__ __main__: import asyncio asyncio.run(main())工具description的写法。注意我在description里明确写了“什么时候调用”。这对LLM的工具选择至关重要。很多MCP Server只写“这个工具做什么”不写“什么时候用”导致LLM要么不调用要么在不该调用的时候调用。返回格式的设计。memory_search返回的是格式化后的文本而不是原始JSON。因为LLM对结构化文本的理解比JSON更好而且格式化文本可以直接注入上下文不需要额外解析。4.5 与Agent框架的集成MCP Server跑起来后需要在Agent框架里配置接入。以常见的配置方式为例{ mcpServers: { hindsight-memory: { command: python, args: [-m, hindsight.server], env: { POSTGRES_URL: postgresql://hindsight:passwordlocalhost:5432/hindsight, REDIS_URL: redis://localhost:6379/0, QDRANT_URL: http://localhost:6333, LLM_API_KEY: your_api_key } } } }配置完成后Agent在运行时会自动发现hindsight提供的工具并根据对话内容决定何时调用memory_write、何时调用memory_search。这里有个实践经验在系统提示词里明确告诉Agent记忆工具的存在和使用时机。比如加一段你可以使用memory_write工具记住用户的重要信息偏好、约束、事实。当用户提到个人信息、做出重要决定、或明确要求你记住某事时调用memory_write。在回答涉及用户历史信息的问题前先调用memory_search检索相关记忆。这段提示词能显著提升工具调用的准确率。不加的话Agent经常忘记使用记忆工具。5. 常见问题与排查技巧实录5.1 Docker环境问题速查问题现象可能原因排查步骤解决方案Docker Desktop启动失败提示虚拟化未检测到BIOS虚拟化未开启重启进BIOS检查VT-x/AMD-V开启虚拟化支持容器间网络不通不在同一networkdocker network inspect在compose中声明同一network端口冲突宿主机端口被占用netstat -tlnp | grep 端口修改映射端口或停止占用进程容器反复重启配置错误或依赖未就绪docker compose logs service根据日志修复配置加healthcheck数据丢失未挂载volumedocker inspect查看Mounts在compose中配置volume挂载镜像拉取慢网络问题docker pull手动测试配置镜像加速器Windows下Docker Desktop的虚拟化问题是最常见的入门障碍。除了BIOS设置还要确认WSL2已安装并设为默认。如果用的是Hyper-V后端确认Hyper-V功能已启用。这两个后端选一个就行不要同时开。容器网络不通的问题八成是因为服务不在同一个Docker network里。Docker Compose默认会创建一个network所有service都在里面。但如果你手动docker run启动的容器就需要显式指定--network。我的建议是统一用Compose管理省心。5.2 记忆检索质量差的排查思路检索质量差是反馈最多的问题。排查要按链路走第一步确认写入是否正常。查关系库看记忆条目是否真的写进去了。如果写入就有问题检索肯定好不了。常见写入问题包括LLM抽取失败返回了非JSON格式、去重逻辑误合并、向量化服务不可用。第二步确认向量维度一致。写入时用的embedding模型和检索时用的必须是同一个。如果中途换了模型向量维度对不上检索会直接报错或返回垃圾结果。我踩过这个坑换了embedding模型后忘了重建索引检索结果全是乱的。第三步检查查询扩展是否合理。查询扩展有时候会跑偏生成一些无关的子查询反而引入噪声。可以先把扩展关掉用原始查询检索对比效果。如果原始查询效果更好说明扩展逻辑需要调整。第四步看重排序权重。如果检索出来的东西语义相关但时间太久可能是recency权重太低。如果高重要性记忆没被排上来可能是importance权重不够。调权重是个细活建议做A/B测试。第五步检查数据量。记忆库太小比如只有几十条向量检索的优势体现不出来反而不如直接全量返回。数据量大了之后几万条以上检索质量才会明显提升。5.3 记忆膨胀与性能衰减的应对记忆库只增不减迟早会出问题。我见过一个项目跑了三个月记忆库膨胀到几百万条检索延迟从50ms涨到2秒而且召回质量明显下降。定期清理策略。设置一个定时任务每周跑一次清理删除重要性低于0.2且超过90天未访问的记忆合并高度相似的记忆条目相似度0.95对超过180天的记忆做摘要压缩多条合并为一条索引优化。向量库的索引类型和参数对性能影响很大。Qdrant的HNSW索引m参数控制图的连通度ef_construct控制构建时的搜索范围。数据量增长后适当调大这两个参数能提升召回率但会增加内存占用和构建时间。分片策略。如果单机扛不住可以按session_id或时间范围做分片。不同分片可以部署在不同节点上检索时并行查询再合并结果。5.4 MCP工具调用的典型故障工具不被调用。Agent该调用memory_write的时候不调用。排查检查工具description是否清晰、系统提示词是否提及记忆工具、LLM是否支持function calling。有些小模型对工具调用的支持不好换大一点的模型试试。工具被过度调用。每轮对话都调用memory_search浪费延迟和token。解决在description里明确“仅在需要回忆历史信息时调用”并在系统提示词里加约束。也可以在服务端做频率限制比如同一session 10秒内最多检索一次。参数填错。LLM把session_id填成了用户ID或者importance填了超出范围的值。解决在inputSchema里加约束minimum/maximum服务端做参数校验和默认值填充。超时。检索链路太长导致超时。解决设置合理的超时时间建议写入5秒、检索3秒超时后降级返回缓存结果或空结果不要让整个Agent卡死。5.5 我的实操心得与避坑清单最后分享几条踩坑换来的经验都是文档里不会写的。不要等到数据量大了才考虑索引。一开始就建好向量索引和关系库索引后期迁移成本很高。embedding模型的选择比向量库的选择更重要。换个好的embedding模型检索质量提升立竿见影。建议用MTEB榜单上排名靠前且维度适中的模型。记忆的写入时机很关键。不要每轮对话都写那样噪声太大。在对话告一段落、或用户明确表达重要信息时写入质量更高。测试环境一定要有记忆回放功能。能重现某次检索为什么返回了这些结果对调试至关重要。我一般会把每次检索的query、召回候选、重排序分数都记日志。MCP Server的日志要打到stderr不要打到stdout。stdout是协议通信通道打日志进去会破坏协议。这个坑我踩过排查了半天才发现是日志输出位置的问题。Docker Compose的depends_on不保证服务就绪。它只保证启动顺序不保证服务可用。要用healthcheck depends_on的condition形式确保依赖服务真正就绪后再启动。定期备份关系库。向量库可以重建关系库丢了就真丢了。设置每日自动备份保留最近30天。这套hindsight方案我在两个项目中实际跑过一个客服Agent、一个个人知识助手。客服场景下记忆检索的准确率大概在85%左右响应延迟控制在200ms以内。知识助手场景数据量更大检索延迟在500ms左右但通过缓存热点记忆P95延迟能压到300ms以下。整体来说这套架构在中等规模场景下是够用的再往上就需要考虑分布式和更精细的分片策略了。
hindsight实战：基于MCP与Docker构建LLM Agent分层记忆系统
1. 从“hindsight”说起为什么我们需要给Agent装上“后视镜”“hindsight”这个词本身很有意思字面意思是“事后的洞察力”也就是我们常说的“后见之明”。放在LLM Agent的语境里它指向一个非常具体且要命的问题Agent的记忆。你肯定遇到过这种情况跟一个AI助手聊了半小时它突然忘了你五分钟前说过的关键约束或者一个自动化工作流跑了十几步中间某一步的中间结果丢了导致后面全盘崩溃。这不是模型不够聪明而是它的“记忆系统”设计得太粗糙。当前大多数Agent的记忆机制说白了就是往上下文窗口里硬塞对话历史塞满了就截断重要的和不重要的信息一起被冲进下水道。这种做法在短对话里勉强能用一旦任务链条拉长、信息密度升高立刻原形毕露。“hindsight”要解决的核心痛点就在这里。它不是简单地做一个“更大的上下文窗口”而是试图构建一套有选择、有结构、可检索、可遗忘的Agent记忆体系。你可以把它理解成给Agent装了一面后视镜——不是让它记住所有东西而是让它在该回头看的时候能精准地看到该看的东西。这套东西适合谁来折腾如果你正在做LLM应用开发尤其是涉及多轮对话、长任务编排、知识密集型工作流那Agent记忆就是你绕不过去的坎。如果你只是拿大模型聊聊天、写写文案那暂时用不上。但只要你开始认真考虑“让Agent记住用户偏好”“让Agent在跨会话场景下保持一致性”“让Agent从历史操作中学习”hindsight这类方案就值得你花时间研究。我接下来会从整体设计思路、核心机制拆解、实操落地、以及踩坑排查四个维度把hindsight这套Agent记忆方案讲透。中间会涉及MCP协议、Docker部署、存储分层这些具体技术点也会分享一些我在实际项目中积累的经验。文章偏长建议收藏后慢慢看。2. Agent记忆的整体设计与思路拆解2.1 为什么传统上下文窗口方案不够用先把这个事情说清楚不然后面很多设计决策你理解不了。LLM的上下文窗口本质上是一块固定大小的临时工作台。你往上面放东西放满了就得拿走一些才能放新的。现在主流模型动辄128K、200K token的窗口看起来很大但实际用起来消耗极快。一轮复杂的工具调用光是系统提示词、工具定义、few-shot示例就能吃掉几千token再加上多轮对话历史、工具返回结果、中间推理步骤几万token很快就没了。更关键的问题不是“装不下”而是“装不下该装的”。上下文窗口是一个无差别存储——它不区分哪些信息重要、哪些不重要、哪些该长期保留、哪些用完就该扔。这就导致两个典型故障信息淹没关键约束被大量无关对话稀释模型注意力分散输出质量下降。记忆断裂超出窗口的历史被直接丢弃Agent“忘记”了之前达成的共识或积累的中间结果。我见过太多项目在这上面翻车。一个客服Agent用户前面说了“我对花生过敏”聊了二十轮之后推荐零食时完全忘了这回事。这不是模型的问题是记忆架构的问题。2.2 hindsight的核心思路分层记忆与主动检索hindsight的设计哲学可以概括成一句话把记忆从上下文窗口里解放出来做成一个独立的外部系统。具体来说它把Agent的记忆分成几个层次记忆类型存储内容生命周期访问方式工作记忆当前任务上下文、临时变量单次会话直接注入上下文情景记忆历史对话、操作记录跨会话持久化语义检索语义记忆提炼后的事实、偏好、规则长期结构化查询程序记忆成功的工作流、工具调用模式长期模式匹配这个分层不是拍脑袋想出来的它对应的是认知科学里人类记忆的基本分类。工作记忆容量有限但访问极快情景记忆存储具体经历语义记忆存储抽象知识程序记忆存储技能。hindsight把这套模型搬到Agent上让不同类型的记忆各司其职。核心机制上hindsight做了三件关键的事第一写入时的信息提炼。不是把所有对话原封不动存进去而是在写入阶段就做一轮压缩和结构化。比如把“用户说他住在北京朝阳区家里有只猫叫咪咪对花生过敏”拆成三条独立的事实记录分别打上“位置”“宠物”“健康约束”的标签。这样做的好处是后续检索时能精准命中而不是把整段对话捞出来让模型自己找。第二检索时的相关性排序。当Agent需要回忆某些信息时hindsight不是简单按时间倒序取最近N条而是基于当前查询做语义相似度匹配再结合时间衰减、重要性权重、访问频率等因素做综合排序。这背后涉及向量检索和重排序模型的配合。第三遗忘机制。这个最容易被忽略但极其重要。记忆系统不能只进不出否则检索质量会随着数据量增长而持续下降。hindsight设计了基于时间、访问频率和重要性的衰减策略让低价值记忆逐渐淡出保持记忆库的“信噪比”。2.3 与MCP协议的关系为什么选择MCP作为接入层hindsight选择MCPModel Context Protocol作为对外接口这个决策值得展开说说。MCP本质上是一套标准化的工具调用协议它定义了LLM应用如何发现、调用外部服务。你可以把它类比成USB接口——不管你是键盘、鼠标还是U盘插上就能用不需要为每个设备单独写驱动。把Agent记忆做成MCP Server有几个明显好处解耦记忆系统独立运行不绑定特定的LLM框架。你今天用LangChain明天换AutoGen记忆层不用动。可复用同一个记忆服务可以同时给多个Agent使用共享用户偏好和知识库。可观测MCP协议天然支持工具调用的日志和追踪方便调试记忆的读写行为。生态兼容越来越多的开发工具和平台开始支持MCP接入成本低。当然MCP也不是没有代价。多一层协议就多一层网络开销和故障点对于延迟极度敏感的场景需要额外优化。但总体来看对于Agent记忆这种需要跨会话、跨应用共享的基础设施MCP带来的解耦价值远大于开销。2.4 Docker化部署的考量hindsight用Docker部署这个选择很务实。Agent记忆系统通常需要搭配向量数据库、关系型数据库、缓存服务等一堆组件裸机部署的依赖管理能把人逼疯。Docker Compose一把梭环境一致性问题直接消掉。而且记忆系统往往需要持久化存储Docker Volume的挂载机制让数据管理变得清晰。你可以把向量库的数据、关系库的数据、配置文件分别挂到不同卷上备份和迁移都方便。不过Docker部署也有坑后面实操部分我会详细讲。特别是Windows环境下Docker Desktop的虚拟化支持问题以及容器间网络通信的配置这两个是新手最容易卡住的地方。3. 核心细节解析与实操要点3.1 记忆的写入从原始对话到结构化知识写入是记忆系统的第一道关口这里的处理质量直接决定了后续检索的上限。hindsight在写入阶段做了几层处理我逐个拆解。原始输入捕获。每次Agent与用户的交互、每次工具调用的输入输出都会被完整记录下来。这部分是原始素材不做任何加工存到冷存储里备查。注意这里说的是“完整记录”包括时间戳、会话ID、角色标识、原始文本。这些元数据在后续检索和审计时非常关键。信息抽取与结构化。这是核心环节。hindsight会调用LLM对原始对话做一轮信息抽取把非结构化的文本转成结构化的事实条目。抽取的维度包括实体人名、地名、组织、产品等属性实体的特征描述关系实体之间的关联事件发生了什么、什么时候、涉及谁偏好用户的喜好、厌恶、约束条件抽取的prompt设计很讲究。我试过几种不同的抽取策略发现按类型分步抽取比一次性全抽效果更好。比如先抽实体和属性再基于已抽出的实体抽关系最后抽事件和偏好。这样做的好处是每一步的认知负荷更低抽取准确率明显提升。去重与合并。同一个事实可能在多轮对话中反复出现比如用户多次提到“我在北京”。如果每次都存一条记忆库会迅速膨胀且冗余。hindsight的做法是对新抽取的事实做相似度匹配如果和已有事实高度相似就更新已有条目的置信度和时间戳而不是新增一条。这里有个细节相似度阈值不能设太高。我一开始把阈值设到0.95结果“我在北京”和“我住在北京朝阳区”被判定为不同事实各存了一条。后来调到0.85左右合并效果就比较合理了。但也不能太低否则“我喜欢猫”和“我喜欢狗”可能被错误合并。这个阈值需要根据你的具体领域数据来调。重要性评分。不是所有记忆都同等重要。hindsight在写入时会给每条记忆打一个重要性分数后续检索和遗忘都会用到。评分维度包括信息类型健康约束、安全规则这类硬性约束权重最高情感强度用户表达强烈情绪的内容权重较高重复频率反复出现的信息权重递增时效性有时效性的信息如“我明天要出差”在过期后权重骤降重要性评分可以用规则引擎做也可以让LLM来打分。规则引擎的好处是稳定可控LLM的好处是能捕捉微妙语义。我目前的方案是规则打底、LLM微调兼顾稳定性和灵活性。3.2 记忆的检索如何让Agent“想起”该想起的检索是记忆系统价值兑现的环节。写得再好检索不出来等于零。hindsight的检索流程大致是查询理解 → 多路召回 → 重排序 → 上下文注入。查询理解。Agent发起检索时原始查询往往是一句模糊的自然语言比如“用户之前提到的那个偏好”。直接拿这句话去做向量检索效果不会好。hindsight会先用LLM对查询做一轮改写和扩展提取出关键实体和意图生成多个检索子查询。比如上面那句话可能被扩展成“用户 偏好 饮食 健康 约束”等多个查询向量。多路召回。单一检索策略很难覆盖所有场景。hindsight同时走几条路向量检索基于语义相似度召回关键词检索基于BM25等传统算法召回弥补向量检索对精确匹配的不足结构化查询基于标签、时间范围、实体类型做过滤图遍历沿着实体关系图扩散召回关联记忆这几路召回的结果合并后进入重排序阶段。重排序。召回阶段追求的是“不漏”重排序阶段追求的是“精准”。hindsight用一个交叉编码器cross-encoder对召回结果做精细打分综合考虑语义相关性、时间新鲜度、重要性权重、访问频率等因素。最终取Top-K条注入Agent的上下文。K值的选择是个权衡。太小了可能漏掉关键信息太大了会挤占上下文窗口且引入噪声。我的经验是动态K值简单查询取3-5条复杂查询取8-12条同时设置一个token预算上限确保注入的记忆不超过总上下文的20%。上下文注入格式。检索出来的记忆怎么放进prompt里也有讲究。hindsight的做法是把记忆按类型分组用结构化格式呈现而不是简单拼接。比如[用户偏好] - 饮食对花生过敏重要性高来源2024-01-15对话 - 居住北京朝阳区重要性中来源2024-01-10对话 [近期事件] - 2024-01-20用户提到下周要去上海出差这种格式让LLM能快速定位关键信息比一大段自然语言描述高效得多。3.3 存储层选型向量库、关系库与缓存的配合hindsight的存储层不是单一数据库而是多种存储的配合。每种存储负责它最擅长的部分。向量数据库负责语义检索。选型上Milvus、Qdrant、Weaviate、Chroma都是常见选项。我的建议是数据量小于100万条Chroma或Qdrant部署简单够用数据量100万到1亿条Milvus或Qdrant集群版需要混合检索向量标量过滤Qdrant或Weaviate关系型数据库负责结构化数据和元数据管理。PostgreSQL是稳妥选择配合pgvector扩展还能兼顾向量检索小规模场景可以少部署一个组件。MySQL也行但向量支持弱一些。缓存层负责热点记忆的快速访问。Redis是标配用来缓存高频访问的记忆条目和会话状态。TTL设置要根据记忆类型区分工作记忆TTL短分钟级语义记忆TTL长小时级或永久。图数据库是可选项用于存储实体关系。Neo4j或Nebula Graph都可以。如果你的Agent需要做复杂的关系推理比如“用户的同事的上级是谁”图数据库会很有价值。否则可以先不上用关系库的外键关联凑合。存储层之间的数据同步是个容易出问题的地方。我的做法是以关系库为主数据源向量库和图库作为索引。写入时先写关系库再异步同步到向量库和图库。检索时如果向量库挂了可以降级到关系库的关键词检索。这样保证了可用性。3.4 MCP接口设计工具定义与调用约定hindsight作为MCP Server对外暴露的工具集设计直接影响使用体验。核心工具大概有这几个工具名功能关键参数memory_write写入记忆content, type, importance, metadatamemory_search检索记忆query, top_k, filters, time_rangememory_forget删除或衰减记忆memory_id, decay_factormemory_update更新已有记忆memory_id, new_content, confidencememory_summarize对一段记忆做摘要session_id, time_range工具的参数设计要遵循最小必要原则。参数太多LLM调用时容易填错参数太少灵活性不够。我的经验是每个工具控制在3-5个参数必填参数不超过2个其余给合理默认值。MCP工具的description字段非常重要它是LLM决定是否调用、怎么调用的主要依据。description要写清楚这个工具做什么、什么时候用、参数含义、返回值格式。我见过很多MCP Server的description写得含糊其辞导致LLM要么不调用要么乱调用。还有一个容易忽略的点错误处理。MCP工具调用失败时返回的错误信息要足够清晰让LLM能理解发生了什么并决定下一步。比如“记忆写入失败向量库连接超时”比“Error 500”有用得多。4. 实操过程与核心环节实现4.1 环境准备Docker与依赖组件部署先把环境搭起来。以下操作基于Ubuntu 22.04Windows和macOS用户参考对应命令调整。安装Docker和Docker Compose。# 更新包索引 sudo apt update # 安装必要依赖 sudo apt install -y ca-certificates curl gnupg lsb-release # 添加Docker官方GPG密钥 sudo mkdir -p /etc/apt/keyrings curl -fsSL https://download.docker.com/linux/ubuntu/gpg | sudo gpg --dearmor -o /etc/apt/keyrings/docker.gpg # 添加Docker仓库 echo deb [arch$(dpkg --print-architecture) signed-by/etc/apt/keyrings/docker.gpg] https://download.docker.com/linux/ubuntu $(lsb_release -cs) stable | sudo tee /etc/apt/sources.list.d/docker.list /dev/null # 安装Docker Engine和Compose插件 sudo apt update sudo apt install -y docker-ce docker-ce-cli containerd.io docker-compose-plugin # 将当前用户加入docker组避免每次sudo sudo usermod -aG docker $USER newgrp docker # 验证安装 docker --version docker compose versionWindows用户如果用Docker Desktop注意两个常见问题一是需要开启WSL2后端二是在BIOS里开启虚拟化支持。如果启动时报“Virtualization support not detected”去BIOS里找Intel VT-x或AMD-V选项打开。如果报“Docker Desktop failed to start because virtualization support is not enabled”同样是虚拟化的问题。部署PostgreSQL含pgvector。# docker-compose.yml 片段 services: postgres: image: pgvector/pgvector:pg16 container_name: hindsight-postgres environment: POSTGRES_USER: hindsight POSTGRES_PASSWORD: your_strong_password POSTGRES_DB: hindsight ports: - 5432:5432 volumes: - pgdata:/var/lib/postgresql/data healthcheck: test: [CMD-SHELL, pg_isready -U hindsight] interval: 10s timeout: 5s retries: 5 volumes: pgdata:部署Redis。redis: image: redis:7-alpine container_name: hindsight-redis ports: - 6379:6379 volumes: - redisdata:/data command: redis-server --appendonly yes --maxmemory 512mb --maxmemory-policy allkeys-lru volumes: redisdata:部署Qdrant向量库。qdrant: image: qdrant/qdrant:latest container_name: hindsight-qdrant ports: - 6333:6333 - 6334:6334 volumes: - qdrantdata:/qdrant/storage volumes: qdrantdata:启动所有服务docker compose up -d docker compose ps确认所有容器状态是healthy或running。如果某个容器反复重启用docker compose logs service_name看日志排查。4.2 记忆写入流程的代码实现环境就绪后核心是写入逻辑。以下是一个简化版的Python实现展示从对话到结构化记忆的完整链路。import json import hashlib from datetime import datetime from typing import List, Dict, Any class MemoryWriter: def __init__(self, llm_client, vector_store, relational_store, cache): self.llm llm_client self.vector_store vector_store self.relational_store relational_store self.cache cache def extract_facts(self, conversation: str) - List[Dict[str, Any]]: 从对话中抽取结构化事实 prompt f从以下对话中抽取结构化事实。对每条事实输出JSON格式 {{ content: 事实内容, type: preference|constraint|event|entity|relation, entities: [涉及的实体], importance: 0.0-1.0, confidence: 0.0-1.0 }} 对话内容 {conversation} 只输出JSON数组不要其他内容。 response self.llm.generate(prompt) try: facts json.loads(response) except json.JSONDecodeError: # 容错尝试提取JSON部分 facts self._extract_json_fallback(response) return facts def deduplicate(self, new_fact: Dict, existing_facts: List[Dict]) - Dict: 去重合并如果新事实与已有事实高度相似合并而非新增 new_embedding self.vector_store.embed(new_fact[content]) for existing in existing_facts: similarity self._cosine_similarity( new_embedding, existing[embedding] ) if similarity 0.85: # 合并更新置信度和时间戳 existing[confidence] min( 1.0, existing[confidence] 0.1 ) existing[updated_at] datetime.now().isoformat() existing[content] self._merge_content( existing[content], new_fact[content] ) return existing return new_fact def write(self, conversation: str, session_id: str) - List[str]: 完整写入流程 # 1. 抽取事实 facts self.extract_facts(conversation) # 2. 获取相关已有事实用于去重 written_ids [] for fact in facts: # 检索相似已有事实 similar self.vector_store.search( fact[content], top_k5, threshold0.7 ) # 3. 去重合并 merged self.deduplicate(fact, similar) # 4. 生成唯一ID fact_id hashlib.md5( f{session_id}:{merged[content]}.encode() ).hexdigest()[:16] # 5. 写入关系库主数据源 self.relational_store.upsert( idfact_id, contentmerged[content], typemerged[type], entitiesjson.dumps(merged.get(entities, [])), importancemerged[importance], confidencemerged[confidence], session_idsession_id, created_atdatetime.now().isoformat(), updated_atdatetime.now().isoformat() ) # 6. 写入向量库索引 embedding self.vector_store.embed(merged[content]) self.vector_store.upsert( idfact_id, vectorembedding, payload{ content: merged[content], type: merged[type], importance: merged[importance], session_id: session_id } ) # 7. 更新缓存 self.cache.setex( fmemory:{fact_id}, 3600, json.dumps(merged) ) written_ids.append(fact_id) return written_ids def _cosine_similarity(self, a, b): dot sum(x*y for x, y in zip(a, b)) norm_a sum(x*x for x in a) ** 0.5 norm_b sum(x*x for x in b) ** 0.5 return dot / (norm_a * norm_b) if norm_a and norm_b else 0.0 def _merge_content(self, old: str, new: str) - str: 合并两条相似事实的内容 if new in old: return old if old in new: return new return f{old}{new} def _extract_json_fallback(self, text: str) - List[Dict]: 从非标准输出中提取JSON import re match re.search(r\[.*\], text, re.DOTALL) if match: try: return json.loads(match.group()) except json.JSONDecodeError: pass return []这段代码有几个关键设计点值得说明。抽取prompt的JSON约束。明确要求输出JSON数组并给出schema示例。实测下来这样能显著降低解析失败率。但即便如此仍然需要fallback逻辑因为LLM偶尔会加一些解释性文字。去重阈值0.85。这个值是调出来的。太低会误合并太高会漏合并。建议在你的领域数据上做一轮标注测试找到最优阈值。先写关系库再写向量库。关系库是主数据源向量库是索引。如果向量库写入失败关系库的数据还在可以后续重建索引。反过来就麻烦了。缓存TTL 3600秒。热点记忆缓存一小时平衡了内存占用和命中率。如果你的场景访问模式不同可以调整。4.3 检索流程的实现与参数调优检索比写入更考验工程能力因为它在请求路径上延迟直接影响用户体验。class MemoryRetriever: def __init__(self, llm_client, vector_store, relational_store, cache): self.llm llm_client self.vector_store vector_store self.relational_store relational_store self.cache cache def search(self, query: str, session_id: str None, top_k: int 8, time_range: tuple None) - List[Dict]: 多路召回 重排序 # 1. 查询理解与扩展 expanded_queries self._expand_query(query) # 2. 多路召回 candidates [] # 2a. 向量检索多查询 for q in expanded_queries: results self.vector_store.search( q, top_ktop_k * 2, filtersself._build_filters(session_id, time_range) ) candidates.extend(results) # 2b. 关键词检索 keyword_results self.relational_store.keyword_search( query, top_ktop_k * 2, session_idsession_id ) candidates.extend(keyword_results) # 3. 去重 seen_ids set() unique_candidates [] for c in candidates: if c[id] not in seen_ids: seen_ids.add(c[id]) unique_candidates.append(c) # 4. 重排序 reranked self._rerank(query, unique_candidates) # 5. 取Top-K final reranked[:top_k] # 6. 更新访问计数用于后续重要性调整 for item in final: self._record_access(item[id]) return final def _expand_query(self, query: str) - List[str]: 用LLM扩展查询生成多个检索子查询 prompt f将以下查询扩展为3-5个检索子查询覆盖不同角度。 每个子查询一行不要编号不要解释。 原始查询{query} response self.llm.generate(prompt) queries [q.strip() for q in response.strip().split(\n) if q.strip()] return [query] queries[:4] # 保留原始查询 最多4个扩展 def _rerank(self, query: str, candidates: List[Dict]) - List[Dict]: 综合打分重排序 scored [] now datetime.now() for c in candidates: # 语义相关性用交叉编码器或LLM打分 relevance self._semantic_score(query, c[content]) # 时间新鲜度指数衰减 age_hours (now - datetime.fromisoformat( c.get(updated_at, c.get(created_at)) )).total_seconds() / 3600 recency 0.5 ** (age_hours / 168) # 半衰期一周 # 重要性权重 importance c.get(importance, 0.5) # 访问频率归一化 access_count c.get(access_count, 0) frequency min(1.0, access_count / 10) # 综合得分 score ( 0.5 * relevance 0.2 * recency 0.2 * importance 0.1 * frequency ) scored.append({**c, score: score}) scored.sort(keylambda x: x[score], reverseTrue) return scored def _semantic_score(self, query: str, content: str) - float: 语义相关性打分实际可用交叉编码器替代 q_emb self.vector_store.embed(query) c_emb self.vector_store.embed(content) return self._cosine_similarity(q_emb, c_emb) def _build_filters(self, session_id, time_range): filters {} if session_id: filters[session_id] session_id if time_range: filters[created_at] { gte: time_range[0], lte: time_range[1] } return filters def _record_access(self, memory_id: str): 记录访问用于后续重要性调整 self.cache.incr(faccess:{memory_id}) self.relational_store.increment_access(memory_id) def _cosine_similarity(self, a, b): dot sum(x*y for x, y in zip(a, b)) norm_a sum(x*x for x in a) ** 0.5 norm_b sum(x*x for x in b) ** 0.5 return dot / (norm_a * norm_b) if norm_a and norm_b else 0.0权重分配的逻辑。语义相关性占0.5因为它是检索的核心目标。时间新鲜度占0.2半衰期设为一周意味着七天前的记忆权重减半。重要性占0.2让高价值记忆有更高优先级。访问频率占0.1作为辅助信号。这套权重不是固定的需要根据你的场景调。比如客服场景可能更看重时间新鲜度知识库场景可能更看重语义相关性。查询扩展的收益。我做过对比测试加查询扩展后召回率提升了约15%但延迟增加了约200ms因为多了一次LLM调用。如果你的场景对延迟敏感可以考虑用更轻量的方式做扩展比如基于同义词词典或预训练的查询改写模型。4.4 MCP Server的封装与接入把上面的写入和检索逻辑封装成MCP Server对外提供标准接口。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() - list[types.Tool]: return [ types.Tool( namememory_write, description将对话内容写入长期记忆。当用户提供了值得记住的信息偏好、约束、事实、事件时调用。, inputSchema{ type: object, properties: { content: { type: string, description: 要记住的原始内容 }, session_id: { type: string, description: 会话标识 }, importance: { type: number, description: 重要性0-1默认0.5, default: 0.5 } }, required: [content, session_id] } ), types.Tool( namememory_search, description检索长期记忆。当需要回忆用户偏好、历史约定或之前的事实信息时调用。, inputSchema{ type: object, properties: { query: { type: string, description: 检索查询用自然语言描述你想回忆什么 }, top_k: { type: integer, description: 返回条数默认8, default: 8 }, session_id: { type: string, description: 限定会话范围可选 } }, required: [query] } ), types.Tool( namememory_forget, description删除或衰减指定记忆。当用户要求忘记某些信息或记忆被确认过时时调用。, inputSchema{ type: object, properties: { memory_id: { type: string, description: 记忆ID }, decay_factor: { type: number, description: 衰减因子0-11为完全删除, default: 1.0 } }, required: [memory_id] } ) ] server.call_tool() async def handle_call_tool(name: str, arguments: dict): if name memory_write: writer MemoryWriter(...) ids writer.write( arguments[content], arguments[session_id] ) return [types.TextContent( typetext, textf已写入{len(ids)}条记忆{ids} )] elif name memory_search: retriever MemoryRetriever(...) results retriever.search( arguments[query], session_idarguments.get(session_id), top_karguments.get(top_k, 8) ) formatted format_memories_for_context(results) return [types.TextContent(typetext, textformatted)] elif name memory_forget: # 实现遗忘逻辑 ... raise ValueError(fUnknown tool: {name}) async def main(): async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, InitializationOptions( server_namehindsight-memory, server_version0.1.0 ) ) if __name__ __main__: import asyncio asyncio.run(main())工具description的写法。注意我在description里明确写了“什么时候调用”。这对LLM的工具选择至关重要。很多MCP Server只写“这个工具做什么”不写“什么时候用”导致LLM要么不调用要么在不该调用的时候调用。返回格式的设计。memory_search返回的是格式化后的文本而不是原始JSON。因为LLM对结构化文本的理解比JSON更好而且格式化文本可以直接注入上下文不需要额外解析。4.5 与Agent框架的集成MCP Server跑起来后需要在Agent框架里配置接入。以常见的配置方式为例{ mcpServers: { hindsight-memory: { command: python, args: [-m, hindsight.server], env: { POSTGRES_URL: postgresql://hindsight:passwordlocalhost:5432/hindsight, REDIS_URL: redis://localhost:6379/0, QDRANT_URL: http://localhost:6333, LLM_API_KEY: your_api_key } } } }配置完成后Agent在运行时会自动发现hindsight提供的工具并根据对话内容决定何时调用memory_write、何时调用memory_search。这里有个实践经验在系统提示词里明确告诉Agent记忆工具的存在和使用时机。比如加一段你可以使用memory_write工具记住用户的重要信息偏好、约束、事实。当用户提到个人信息、做出重要决定、或明确要求你记住某事时调用memory_write。在回答涉及用户历史信息的问题前先调用memory_search检索相关记忆。这段提示词能显著提升工具调用的准确率。不加的话Agent经常忘记使用记忆工具。5. 常见问题与排查技巧实录5.1 Docker环境问题速查问题现象可能原因排查步骤解决方案Docker Desktop启动失败提示虚拟化未检测到BIOS虚拟化未开启重启进BIOS检查VT-x/AMD-V开启虚拟化支持容器间网络不通不在同一networkdocker network inspect在compose中声明同一network端口冲突宿主机端口被占用netstat -tlnp | grep 端口修改映射端口或停止占用进程容器反复重启配置错误或依赖未就绪docker compose logs service根据日志修复配置加healthcheck数据丢失未挂载volumedocker inspect查看Mounts在compose中配置volume挂载镜像拉取慢网络问题docker pull手动测试配置镜像加速器Windows下Docker Desktop的虚拟化问题是最常见的入门障碍。除了BIOS设置还要确认WSL2已安装并设为默认。如果用的是Hyper-V后端确认Hyper-V功能已启用。这两个后端选一个就行不要同时开。容器网络不通的问题八成是因为服务不在同一个Docker network里。Docker Compose默认会创建一个network所有service都在里面。但如果你手动docker run启动的容器就需要显式指定--network。我的建议是统一用Compose管理省心。5.2 记忆检索质量差的排查思路检索质量差是反馈最多的问题。排查要按链路走第一步确认写入是否正常。查关系库看记忆条目是否真的写进去了。如果写入就有问题检索肯定好不了。常见写入问题包括LLM抽取失败返回了非JSON格式、去重逻辑误合并、向量化服务不可用。第二步确认向量维度一致。写入时用的embedding模型和检索时用的必须是同一个。如果中途换了模型向量维度对不上检索会直接报错或返回垃圾结果。我踩过这个坑换了embedding模型后忘了重建索引检索结果全是乱的。第三步检查查询扩展是否合理。查询扩展有时候会跑偏生成一些无关的子查询反而引入噪声。可以先把扩展关掉用原始查询检索对比效果。如果原始查询效果更好说明扩展逻辑需要调整。第四步看重排序权重。如果检索出来的东西语义相关但时间太久可能是recency权重太低。如果高重要性记忆没被排上来可能是importance权重不够。调权重是个细活建议做A/B测试。第五步检查数据量。记忆库太小比如只有几十条向量检索的优势体现不出来反而不如直接全量返回。数据量大了之后几万条以上检索质量才会明显提升。5.3 记忆膨胀与性能衰减的应对记忆库只增不减迟早会出问题。我见过一个项目跑了三个月记忆库膨胀到几百万条检索延迟从50ms涨到2秒而且召回质量明显下降。定期清理策略。设置一个定时任务每周跑一次清理删除重要性低于0.2且超过90天未访问的记忆合并高度相似的记忆条目相似度0.95对超过180天的记忆做摘要压缩多条合并为一条索引优化。向量库的索引类型和参数对性能影响很大。Qdrant的HNSW索引m参数控制图的连通度ef_construct控制构建时的搜索范围。数据量增长后适当调大这两个参数能提升召回率但会增加内存占用和构建时间。分片策略。如果单机扛不住可以按session_id或时间范围做分片。不同分片可以部署在不同节点上检索时并行查询再合并结果。5.4 MCP工具调用的典型故障工具不被调用。Agent该调用memory_write的时候不调用。排查检查工具description是否清晰、系统提示词是否提及记忆工具、LLM是否支持function calling。有些小模型对工具调用的支持不好换大一点的模型试试。工具被过度调用。每轮对话都调用memory_search浪费延迟和token。解决在description里明确“仅在需要回忆历史信息时调用”并在系统提示词里加约束。也可以在服务端做频率限制比如同一session 10秒内最多检索一次。参数填错。LLM把session_id填成了用户ID或者importance填了超出范围的值。解决在inputSchema里加约束minimum/maximum服务端做参数校验和默认值填充。超时。检索链路太长导致超时。解决设置合理的超时时间建议写入5秒、检索3秒超时后降级返回缓存结果或空结果不要让整个Agent卡死。5.5 我的实操心得与避坑清单最后分享几条踩坑换来的经验都是文档里不会写的。不要等到数据量大了才考虑索引。一开始就建好向量索引和关系库索引后期迁移成本很高。embedding模型的选择比向量库的选择更重要。换个好的embedding模型检索质量提升立竿见影。建议用MTEB榜单上排名靠前且维度适中的模型。记忆的写入时机很关键。不要每轮对话都写那样噪声太大。在对话告一段落、或用户明确表达重要信息时写入质量更高。测试环境一定要有记忆回放功能。能重现某次检索为什么返回了这些结果对调试至关重要。我一般会把每次检索的query、召回候选、重排序分数都记日志。MCP Server的日志要打到stderr不要打到stdout。stdout是协议通信通道打日志进去会破坏协议。这个坑我踩过排查了半天才发现是日志输出位置的问题。Docker Compose的depends_on不保证服务就绪。它只保证启动顺序不保证服务可用。要用healthcheck depends_on的condition形式确保依赖服务真正就绪后再启动。定期备份关系库。向量库可以重建关系库丢了就真丢了。设置每日自动备份保留最近30天。这套hindsight方案我在两个项目中实际跑过一个客服Agent、一个个人知识助手。客服场景下记忆检索的准确率大概在85%左右响应延迟控制在200ms以内。知识助手场景数据量更大检索延迟在500ms左右但通过缓存热点记忆P95延迟能压到300ms以下。整体来说这套架构在中等规模场景下是够用的再往上就需要考虑分布式和更精细的分片策略了。
