拓十年匠心定制 · 商业建站与技术教学双线并行 咨询热线:400-886-1026 service@lmnt.cn
ARTICLE DETAIL

资讯详情

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

LLM Agent长期记忆系统实战:从记忆抽取到MCP协议集成与Docker部署

LLM Agent长期记忆系统实战:从记忆抽取到MCP协议集成与Docker部署

1. 从“hindsight”说起:为什么我们需要给Agent装上记忆

“hindsight”这个词本身很有意思,字面意思是“事后的洞察力”,也就是我们常说的“后见之明”。放在LLM Agent的语境里,它指向一个非常具体且棘手的问题:Agent如何记住过去发生过的事,并在后续决策中真正用上这些经验。这不是一个简单的“存个日志”就能解决的问题,它涉及到记忆的编码、存储、检索、更新和遗忘一整条链路。

我最初接触这个方向,是因为在实际项目里反复遇到同一个尴尬场景:用户跟Agent聊了半小时,中间明确说过“我对花生过敏”,结果推荐餐厅时Agent还是推了一家花生酱招牌菜的店。用户骂它“没脑子”,但本质上不是模型不够聪明,而是它根本没有一个可靠的记忆机制。LLM的上下文窗口再大,也扛不住长对话的累积,更别说跨会话的场景了。所以“hindsight”这个项目标题,我理解它的核心使命就是:让Agent拥有跨会话、可检索、可推理的长期记忆能力。

这篇文章适合谁看?如果你正在做LLM Agent相关的产品、正在选型记忆方案、或者单纯想搞清楚“Agent Memory”到底该怎么落地,那接下来的内容应该能帮你省下不少试错时间。我会从整体设计思路讲到具体实现细节,包括存储结构、检索策略、MCP协议集成、Docker部署,以及我在实际调试中踩过的坑。不堆概念,只讲能跑起来的东西。

2. 整体设计思路:Agent Memory到底该怎么架构

2.1 为什么不能只靠上下文窗口和向量数据库

很多人第一反应是:记忆嘛,不就是把对话历史塞进向量数据库,需要的时候检索一下?我一开始也这么想,但实际跑下来发现三个致命问题。

第一,向量检索的粒度太粗。你把一整段对话embedding成一个向量,检索出来的是“某次对话的模糊印象”,而不是“用户在第3轮明确说过对花生过敏”这种精确事实。Agent拿到这种模糊记忆,很容易产生幻觉式的补全。

第二,缺乏结构化推理能力。记忆不应该是孤立的碎片,而应该是有关系的网络。比如“用户对花生过敏”和“用户喜欢川菜”这两条记忆,在推荐餐厅时需要联合推理,纯向量检索做不到这种关系推导。

第三,没有遗忘和更新机制。用户上周说喜欢某家店,这周说那家店倒闭了,记忆系统得知道旧信息该失效了。向量数据库的“相似度检索”天然不支持这种时序更新。

所以“hindsight”的设计思路,我倾向于把它拆成三层:工作记忆(Working Memory)、情景记忆(Episodic Memory)、语义记忆(Semantic Memory)。工作记忆就是当前会话的上下文,情景记忆是带时间戳的具体事件,语义记忆是从事件中抽象出来的稳定事实。三层各司其职,检索时按需调用。

2.2 核心存储结构:Key-Query-Value三元组的设计哲学

热词里有一条我印象很深:“LLM的token三个点:key我是谁、query我在找什么、value我能提供什么”。这其实是在用注意力机制的隐喻来解释记忆检索。我把它落地成具体的存储结构:

字段含义示例
key记忆的主体标识user:12345:allergy
query检索时的意图向量“推荐餐厅时需要考虑的约束”
value具体记忆内容“对花生过敏,严重程度:中度”
timestamp记忆写入时间2025-01-15T10:30:00Z
confidence置信度0.95
source来源会话IDsession:abc-123
ttl过期时间无(永久有效)

这个结构的好处是,检索时可以用key做精确匹配,用query做语义匹配,用value做内容返回,三者解耦。比如用户问“今晚吃什么”,Agent先用query=“饮食偏好”去检索,命中key=user:12345:allergy,返回value=“对花生过敏”,然后推理时自动排除含花生的餐厅。

注意:key的设计一定要有命名空间隔离,否则多用户场景下会串数据。我见过有人直接用“allergy”做key,结果A用户的过敏信息被B用户检索到了,这是生产事故级别的bug。

2.3 记忆的生命周期管理

记忆不是写进去就完事了,它有一个完整的生命周期:写入 → 索引 → 检索 → 更新 → 衰减 → 归档/删除。

写入阶段,我建议用LLM做一次“记忆抽取”,把原始对话转成结构化的三元组。比如用户说“我上周去了趟成都,吃了火锅,但第二天肚子不舒服”,抽取出来应该是:

  • 事件:2025-01-08 去成都
  • 事件:吃了火锅
  • 事实:吃火锅后肠胃不适(置信度0.7,因为可能是其他原因)

索引阶段,key做倒排索引,query做向量索引,timestamp做时序索引。检索阶段,根据当前对话意图,决定走哪条索引路径。更新阶段,如果新记忆和旧记忆冲突,按置信度和时间戳做合并。衰减阶段,给每条记忆一个“热度分”,长期不被检索到的记忆逐渐降低权重。归档阶段,超过一定时间的低热度记忆移到冷存储。

这套机制听起来复杂,但用Docker把各个组件容器化之后,运维成本其实可控。后面我会讲具体怎么部署。

3. 核心细节解析:从记忆抽取到检索排序的完整链路

3.1 记忆抽取:怎么让LLM输出结构化的三元组

记忆抽取的质量直接决定整个系统的上限。我的做法是给LLM一个严格的输出模板,用few-shot引导它按格式输出。提示词大概长这样:

EXTRACTION_PROMPT = """ 你是一个记忆抽取器。从以下对话中提取结构化记忆。 对话内容: {conversation} 请按以下JSON格式输出,不要输出任何其他内容: { "memories": [ { "key": "命名空间:主体:属性", "query": "这条记忆可能被什么意图检索到", "value": "具体内容", "confidence": 0.0-1.0, "memory_type": "episodic|semantic" } ] } 规则: 1. 只提取确定的事实,不确定的降低confidence 2. key必须包含用户ID命名空间 3. 一条对话可能提取出多条记忆 4. 如果对话中没有值得记忆的内容,返回空数组 """

实测下来,这个提示词在GPT-4级别的模型上抽取准确率能到85%左右。剩下的15%主要是两类问题:一是模型过度抽取,把寒暄也当成记忆;二是key的命名不规范。我的解决办法是加一层后处理校验,用正则检查key的格式,不合规的直接丢弃。

实操心得:抽取温度建议设成0,不要让它发挥创造力。记忆抽取要的是稳定和准确,不是多样性。

3.2 检索排序:为什么不能只看向量相似度

检索阶段最容易犯的错就是“唯向量相似度论”。我一开始也是这么做的,结果发现检索出来的记忆经常是“语义相似但实际无关”的。比如用户问“推荐个电影”,向量检索可能返回“用户上次说喜欢诺兰的电影”,这没问题;但也可能返回“用户上次说电影院停车很难”,这就跑偏了。

后来我改成多路召回+重排序的架构:

  1. 精确匹配路:用当前对话的实体做key前缀匹配,召回强相关记忆
  2. 语义匹配路:用query向量做ANN检索,召回语义相关记忆
  3. 时序匹配路:召回最近N条记忆,保证时效性
  4. 重排序:用一个小的cross-encoder模型对三路召回结果做统一打分

重排序的打分公式我设计成:

final_score = 0.4 * semantic_similarity + 0.3 * recency_score + 0.2 * confidence + 0.1 * access_frequency

recency_score用指数衰减:exp(-λ * hours_since_access),λ取0.01的话,大约3天后权重降到一半。access_frequency是这条记忆被检索到的次数,高频记忆说明它重要。

这套组合拳下来,检索准确率比纯向量方案提升了大概30%。代价是多了一次重排序推理,延迟增加约50ms,但在可接受范围内。

3.3 记忆冲突处理:新信息来了,旧信息怎么办

这是最容易被忽略的环节。用户先说“我喜欢咖啡”,后来说“我戒咖啡了”,记忆系统得知道后者覆盖前者。我的处理策略是:

  • 同key冲突:新记忆直接覆盖旧记忆,但旧记忆保留在历史版本中,标记为superseded
  • 矛盾但不同key:比如“喜欢咖啡”和“戒咖啡了”,key不同但语义矛盾,用LLM做一次冲突检测,生成一条新的“状态变更”记忆
  • 置信度冲突:新记忆置信度低于旧记忆时,不覆盖,而是追加一条“存疑”标记
def resolve_conflict(old_memory, new_memory): if old_memory.key == new_memory.key: # 同key直接更新,保留历史 archive(old_memory) return new_memory elif is_contradictory(old_memory, new_memory): # 语义矛盾,生成状态变更记忆 return create_state_change_memory(old_memory, new_memory) elif new_memory.confidence < old_memory.confidence: # 新记忆置信度低,标记存疑 new_memory.status = "pending_verification" return new_memory else: return new_memory

注意:冲突检测不要用规则硬匹配,一定要用LLM做语义判断。“戒咖啡了”和“喜欢咖啡”字面上不矛盾,但语义上矛盾,规则匹配搞不定。

4. 实操部署:用Docker把整套记忆系统跑起来

4.1 环境准备与Docker安装要点

这套系统我建议用Docker Compose编排,因为涉及多个组件:记忆存储(PostgreSQL + pgvector)、缓存(Redis)、记忆服务(Python FastAPI)、MCP网关。先确保Docker环境正常。

Windows用户装Docker Desktop时,最常见的坑是“Virtualization support not detected”。这个报错的意思是BIOS里没开虚拟化。重启进BIOS,找到Intel VT-x或AMD-V,设为Enabled。如果开了还报错,检查是不是Hyper-V和WSL2冲突了,在“启用或关闭Windows功能”里把Hyper-V关掉,只留WSL2。

Ubuntu用户相对简单:

# 安装Docker curl -fsSL https://get.docker.com | sh # 把当前用户加入docker组,避免每次sudo sudo usermod -aG docker $USER # 重新登录后验证 docker run hello-world

实操心得:国内环境拉镜像慢的话,配置一下镜像加速器。但注意不要用那些来路不明的加速地址,用云厂商官方提供的。

4.2 docker-compose编排文件详解

version: '3.8' services: postgres: image: pgvector/pgvector:pg16 environment: POSTGRES_DB: agent_memory POSTGRES_USER: memory_user POSTGRES_PASSWORD: ${DB_PASSWORD} volumes: - pgdata:/var/lib/postgresql/data ports: - "5432:5432" healthcheck: test: ["CMD-SHELL", "pg_isready -U memory_user"] interval: 10s retries: 5 redis: image: redis:7-alpine command: redis-server --appendonly yes volumes: - redisdata:/data ports: - "6379:6379" memory-service: build: ./memory-service environment: DATABASE_URL: postgresql://memory_user:${DB_PASSWORD}@postgres:5432/agent_memory REDIS_URL: redis://redis:6379/0 LLM_API_KEY: ${LLM_API_KEY} depends_on: postgres: condition: service_healthy redis: condition: service_started ports: - "8000:8000" mcp-gateway: build: ./mcp-gateway environment: MEMORY_SERVICE_URL: http://memory-service:8000 depends_on: - memory-service ports: - "8080:8080" volumes: pgdata: redisdata:

这个编排文件里,postgres用pgvector镜像,直接支持向量检索,省得单独装向量数据库。healthcheck确保postgres完全启动后再启动memory-service,避免连接失败。

4.3 数据库表结构设计

CREATE EXTENSION IF NOT EXISTS vector; CREATE TABLE memories ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), key TEXT NOT NULL, query_vector vector(1536), value TEXT NOT NULL, memory_type TEXT CHECK (memory_type IN ('episodic', 'semantic')), confidence FLOAT DEFAULT 1.0, source_session TEXT, access_count INT DEFAULT 0, last_accessed_at TIMESTAMPTZ DEFAULT NOW(), created_at TIMESTAMPTZ DEFAULT NOW(), status TEXT DEFAULT 'active', superseded_by UUID REFERENCES memories(id) ); CREATE INDEX idx_memories_key ON memories(key); CREATE INDEX idx_memories_vector ON memories USING ivfflat (query_vector vector_cosine_ops) WITH (lists = 100); CREATE INDEX idx_memories_created ON memories(created_at DESC);

ivfflat索引的lists参数,经验值是数据量的平方根。10万条记忆的话,lists设316左右。数据量小的时候(<1万),不建向量索引反而更快,因为索引本身有开销。

4.4 MCP协议集成:让Agent通过标准协议访问记忆

MCP(Model Context Protocol)是当前Agent工具调用的事实标准。把记忆系统封装成MCP Server,好处是任何支持MCP的Agent框架都能直接接入,不用改代码。

MCP Server的核心是暴露几个工具:

from mcp.server import Server, NotificationOptions from mcp.server.models import InitializationOptions server = Server("hindsight-memory") @server.tool() async def store_memory(key: str, value: str, query: str, memory_type: str = "semantic", confidence: float = 1.0) -> str: """存储一条记忆""" memory_id = await memory_service.store( key=key, value=value, query=query, memory_type=memory_type, confidence=confidence ) return f"Memory stored: {memory_id}" @server.tool() async def retrieve_memories(query: str, top_k: int = 5) -> str: """根据查询检索相关记忆""" memories = await memory_service.retrieve(query, top_k) return format_memories(memories) @server.tool() async def forget_memory(key: str) -> str: """删除指定记忆""" await memory_service.delete(key) return f"Memory deleted: {key}"

Agent侧只需要配置MCP Server地址,就能自动发现这三个工具。我实测过Claude Desktop和几个开源Agent框架,接入都很顺。

注意:MCP Server的token认证一定要做。热词里那个wss://api.xiaozhi.me/mcp/?token=...的格式就是典型做法,token放在URL参数或Header里。生产环境建议用Header,URL参数容易在日志里泄露。

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

5.1 记忆检索不准的排查思路

这是最高频的问题。排查顺序我总结成一张表:

现象可能原因排查方法解决
检索不到任何记忆向量维度不匹配检查embedding模型输出维度与表定义是否一致统一维度,重建索引
检索到无关记忆query向量质量差打印query向量,看是否和预期语义一致换embedding模型或加query改写
检索结果排序不合理重排序权重失衡打印各路召回分数调整权重系数
新记忆检索不到索引未更新检查ivfflat索引是否需要重建定期REINDEX
记忆重复抽取阶段重复写入查key是否有唯一约束加唯一索引或去重逻辑

我遇到最坑的一次是embedding模型换了,从text-embedding-ada-002换到text-embedding-3-small,维度从1536变成1536(3-small默认也是1536,但可以调),结果旧记忆的向量和新query的向量不在同一空间,检索全乱。后来加了模型版本字段,换模型时自动触发全量重embedding。

5.2 Docker网络不通的经典场景

Docker Compose里服务之间通信用服务名,不是localhost。我见过有人把DATABASE_URL写成postgresql://user:pass@localhost:5432/db,在容器里localhost指向容器自己,当然连不上。正确写法是@postgres:5432,postgres是compose里的服务名。

另一个坑是端口映射。ports: - "5432:5432"是把容器端口映射到宿主机,容器之间通信不需要这个映射,直接用服务名+容器端口就行。映射多了反而容易和宿主机已有服务冲突。

如果确实需要从宿主机访问容器内的服务做调试,用docker exec -it container_name bash进去查,比映射端口更安全。

5.3 记忆膨胀导致性能下降

跑了一段时间后,记忆表可能涨到几十万条,检索延迟从50ms涨到500ms。我的处理策略:

  1. 冷热分离:超过30天未被访问的记忆移到memories_archive表,主表只保留热数据
  2. 向量索引调优:ivfflat的probes参数,查询时设置SET ivfflat.probes = 10,在召回率和速度之间平衡
  3. 定期归档:写个定时任务,每天凌晨跑一次归档
  4. 压缩低频记忆:多条相似的低频记忆合并成一条摘要记忆
-- 归档30天未访问的记忆 INSERT INTO memories_archive SELECT * FROM memories WHERE last_accessed_at < NOW() - INTERVAL '30 days' AND access_count < 3; DELETE FROM memories WHERE last_accessed_at < NOW() - INTERVAL '30 days' AND access_count < 3;

实操心得:归档前一定要备份。我有次手抖把WHERE条件写错了,差点把活跃记忆全删了。现在归档脚本里强制加LIMIT 1000,分批处理,每批确认后再继续。

5.4 MCP工具调用超时的处理

MCP协议本身有超时机制,但记忆检索如果走了LLM做query改写,延迟可能超过默认超时。我的做法是:

  • 记忆检索的MCP工具设置独立超时,比默认值长
  • query改写做成可选的,简单查询直接走向量检索
  • 加缓存,相同query在5分钟内直接返回缓存结果
from functools import lru_cache import hashlib @lru_cache(maxsize=1000) def cached_retrieve(query_hash: str): # 实际检索逻辑 pass async def retrieve(query: str, top_k: int): query_hash = hashlib.md5(f"{query}:{top_k}".encode()).hexdigest() return cached_retrieve(query_hash)

缓存命中率在实际场景里能到40%左右,因为用户的追问往往语义相近。

6. 记忆系统的安全防护与未来扩展

6.1 记忆投毒与防御思路

Agent记忆系统有一个容易被忽视的攻击面:记忆投毒。如果攻击者能往记忆里写入虚假信息,比如“用户说他的密码是xxx”,后续Agent就可能泄露敏感信息。热词里提到的“a-memguard”就是针对这个问题的主动防御框架。

我的防御策略分三层:

第一层,写入校验。所有记忆写入前过一遍敏感信息检测,密码、身份证号、银行卡号这类直接拒绝写入。用正则+LLM双重检测。

第二层,来源可信度。每条记忆记录来源会话的可信等级。用户直接输入的可信度高,Agent从网页抓取的可信度低。检索时低可信记忆降权。

第三层,异常检测。监控记忆写入频率,如果某个会话突然写入大量记忆,触发人工审核。正常对话每分钟写入1-3条记忆,超过10条就可疑。

SENSITIVE_PATTERNS = [ r'\b\d{16,19}\b', # 银行卡号 r'\b\d{17}[\dXx]\b', # 身份证号 r'password\s*[:=]\s*\S+', # 密码 ] def is_sensitive(content: str) -> bool: for pattern in SENSITIVE_PATTERNS: if re.search(pattern, content, re.IGNORECASE): return True return False

6.2 多Agent共享记忆的隔离设计

如果一个系统里有多个Agent,记忆要不要共享?我的建议是默认隔离,按需共享。每个Agent有自己的命名空间,共享记忆放在公共命名空间,通过显式授权访问。

命名空间设计:

agent:{agent_id}:private:{key} # 私有记忆 shared:{domain}:{key} # 共享记忆 user:{user_id}:{key} # 用户级记忆,跨Agent共享

检索时,Agent默认只能查自己的私有记忆和用户级记忆,共享记忆需要显式声明。这样既保证了隔离性,又支持协作场景。

6.3 后续可以扩展的方向

这套系统跑通之后,有几个方向可以继续深挖。一是记忆的可视化,做一个Web界面,让用户能看到Agent记住了什么,还能手动编辑和删除,这对建立用户信任很重要。二是记忆的跨模态扩展,现在只存文本,未来可以存图片、音频的embedding,让Agent记住用户发过的图片。三是记忆的联邦学习,多个部署实例之间在不共享原始数据的前提下,共享记忆的统计规律,提升整体检索质量。

我个人最看好的方向是记忆可视化。现在Agent的记忆是个黑盒,用户不知道它记住了什么、忘了什么,出了问题也没法排查。把记忆变成用户可感知、可控制的东西,才是Agent真正走向实用的关键一步。我下一个项目就打算做这个,到时候再跟大家分享。

返回列表