你有没有遇到过这种情况:智能体每次对话都要重新自我介绍,用户一小时前提的需求转头就忘,明明投喂了大量行业文档,关键时刻一个细节都想不起来。我最早用OpenClaw做智能体项目的时候,记忆这块完全是空白,每次会话都是一次“失忆重启”,用户问第二次同样的问题,它仍然一脸茫然地重新开始。后来我花了不少时间把记忆系统彻底重做了一遍,沉淀出一套可以复制、可以改、可以直接拿去用的搭建模板,今天就把这套模板和踩过的坑一次性讲清楚。
这套模板解决的核心问题很直接:让OpenClaw具备三种记忆能力——会话内的短期记忆、跨会话的长期事实记忆、以及用于压缩上下文的摘要记忆。适合正在用OpenClaw做智能客服、知识助手、个人AI助理,或者任何需要“记住用户”的场景的开发者参考。不管你是刚接触OpenClaw的小白,还是已经跑过几个Demo想完善生产方案的开发者,这套模板都能直接套用。
1. 记忆系统的整体设计与模块划分
1.1 记忆到底解决什么问题
先明确一个概念:记忆不是把聊天记录存下来那么简单。很多人一开始就把所有历史消息倒进向量库,以为这样就“有记忆”了,结果跑起来又慢又费钱,回答质量反而更差。记忆系统真正的目标是在合适的时间、把合适的记忆、以合适的形态交给模型。
打个比方,人的大脑不会把所有经历都原封不动存着。你记得昨天中午吃了什么,但不会记得三年前某天中午吃的什么;你记得好朋友的生日,但不会记得上周三他和你说过的每句话的每个字。智能体的记忆也应该是分层的:重要的长期留存,次要的短期保留,过期的自动遗忘。
OpenClaw作为智能体调度框架,本身不自带成熟的记忆方案,但它的插件机制和上下文管理接口给了我们自己做记忆系统的空间。我搭建的这套模板,本质上是在OpenClaw的外部封装了一层记忆管理层,负责拦截、存储、检索和注入记忆。
1.2 三层记忆架构:短期、长期、工作记忆
我推荐的架构分三层,每一层的职责单一清晰:
- 短期记忆(Short-term Memory):存放当前会话内的对话记录,采用滑动窗口机制,控制Token开销。
- 长期记忆(Long-term Memory):存放跨会话的关键事实、用户偏好、历史知识,采用向量库+结构化存储双通道。
- 工作记忆(Working Memory):每次请求时从短期和长期记忆中筛选出来的、注入到当前提示词的那部分内容。
工作记忆是“最终被模型看到的内容”,另外两层是“候选池”。所有记忆管理的核心逻辑,都是在决定候选池里的哪些内容能进入工作记忆。
1.3 模板目录结构与配置骨架
这套模板的目录结构如下,我建议你直接照抄:
openclaw-memory-template/ ├── config/ │ └── memory_config.yaml ├── memory/ │ ├── __init__.py │ ├── short_term.py │ ├── long_term.py │ ├── summary.py │ ├── working_memory.py │ └── storage/ │ ├── vector_store.py │ └── sqlite_store.py ├── utils/ │ ├── token_counter.py │ └── time_decay.py ├── tests/ │ └── test_memory.py └── main.py配置文件的骨架决定了整个记忆系统的行为边界,我习惯把所有参数外置,避免硬编码。下面这份配置是我目前在生产环境使用的版本,每项参数的含义在后面章节逐一展开:
# memory_config.yaml short_term: max_tokens: 4000 # 短期记忆最大Token预算 window_size: 20 # 最大保留的消息轮数 summary_threshold: 3000 # 达到此Token数触发摘要压缩 long_term: vector_dim: 768 # 向量维度,需与Embedding模型对齐 top_k: 5 # 默认检索条数 min_score: 0.45 # 相关性阈值,低于此分不召回 time_decay_factor: 0.02 # 时间衰减因子 max_facts_per_user: 500 # 每个用户最多存储的结构化事实数 storage: vector_db: "sqlite-vec" # 可替换为其他向量库 sqlite_path: "./data/memory.db" embedding_model: "bge-small-zh" # 中文场景推荐注意:所有路径和模型名都要和你实际环境对齐,向量维度不一致会导致存储故障,这是最容易被忽视的问题之一。
2. 核心模块:短期会话记忆的实现
2.1 滑动窗口与Token预算
短期记忆的第一要务是控本。如果你的模型上下文窗口是8K,对话历史全量塞进去,再好的Prompt也救不回来,模型要么截断,要么报错。我的做法是设定一个Token预算,比如4000,预留一半空间给系统指令、检索到的长期记忆和用户当前输入。
滑动窗口的逻辑很简单:每次新消息进来,把消息加入队列,然后从最旧的消息开始淘汰,直到总Token数降到预算以内。但这里有一个关键细节:不能无脑按条数淘汰,因为有的消息很长、有的很短。必须按Token数逐一减去,而不是按条数硬切。
具体实现时我封装了一个带Token计数的环形队列:
# memory/short_term.py from collections import deque class ShortTermMemory: def __init__(self, max_tokens: int, window_size: int): self.max_tokens = max_tokens self.window_size = window_size self.messages = deque(maxlen=window_size) self.token_count = 0 def add_message(self, message: dict): msg_tokens = estimate_tokens(message["content"]) # 先按条数窗口淘汰 if len(self.messages) == self.window_size: old = self.messages.popleft() self.token_count -= estimate_tokens(old["content"]) self.messages.append(message) self.token_count += msg_tokens # 再按Token预算淘汰 while self.token_count > self.max_tokens and len(self.messages) > 1: old = self.messages.popleft() self.token_count -= estimate_tokens(old["content"]) def get_recent(self, max_tokens: int): """返回不超过max_tokens的最近消息""" result = [] tokens = 0 for msg in reversed(self.messages): t = estimate_tokens(msg["content"]) if tokens + t > max_tokens: break result.append(msg) tokens += t return list(reversed(result))这里有个值得注意的细节:淘汰顺序要从最旧开始,但返回时要返回最近的。deque的双端特性刚好满足这两个方向的需求,比用列表手动管理索引干净得多。
2.2 会话摘要如何触发
当Token预算被占满,且当前对话涉及重要信息不能简单丢掉时,就需要摘要机制上场。摘要的本质是牺牲细节换容量:把十轮对话压缩成一段概述,释放出空间给后续对话。
触发时机我建议设在前面配置的summary_threshold。每次写短期记忆时检查:当前Token数超过阈值,且消息条数达到窗口上限,就触发摘要。摘要本身调用一次模型,用专门的Prompt把既有消息压缩成“结构化摘要+关键事实列表”,然后把原始消息清空,把摘要作为一条特殊消息存入短期记忆。
摘要Prompt我踩过好几次坑,最初用的是“请总结这段对话”,结果模型经常丢关键信息。后来改成带约束的版本,效果好很多:
SUMMARY_PROMPT = """ 请将以下对话压缩为结构化的会话摘要,要求: 1. 保留用户的核心诉求、明确给出的偏好、未完成事项 2. 保留已确认的关键决策和具体细节(时间、数量、名称等) 3. 删除寒暄、重复、与主题无关的内容 4. 输出格式:背景概述(100字内)+ 关键事实列表(每行一条) 对话内容: {conversation} """摘要不仅要给模型看,也要给检索用。所以摘要生成后,我同时写入长期记忆的向量库,这样跨会话也可以检索到“之前聊过什么”,弥补短期记忆只有本会话可见的局限。
2.3 短期记忆的持久化设计
短期记忆默认在内存里,进程一退就没了。如果只是Demo无所谓,但生产环境必须落盘。我的做法是把短期记忆的每条消息同时写进SQLite,表结构带session_id、role、content、created_at,重启后通过session_id恢复。
这里有三个字段强烈建议加索引:session_id、created_at,以及两者组成的联合索引。不要觉得数据量小无所谓,我遇到过会话数到几万条之后,恢复会话时查询直接超时的情况,加索引之后从几百毫秒降到个位数毫秒。
3. 长期记忆:向量存储与混合检索
3.1 向量库选型与对比
长期记忆的核心是向量检索。向量库的选型决定了你能支撑的用户量级和查询延迟。我按场景分成三档:
| 场景 | 推荐方案 | 优点 | 缺点 |
|---|---|---|---|
| 个人项目/原型 | SQLite + sqlite-vec | 零部署、文件即库、随应用打包 | 不支持分布式,容量有限 |
| 中小团队/几十万条记忆 | 独立向量库服务 | 延迟稳定、支持过滤、生态成熟 | 需要运维,多一个依赖 |
| 大规模生产/千万级记忆 | 分布式向量数据库 | 高可用、水平扩展、混合索引 | 成本高,部署复杂 |
我个人最常用的是“本地SQLite+vec”起步,用脚本压测发现瓶颈后再迁移。过早引入分布式组件只会拖慢迭代速度,这跟“先用手写记账本,账多了再换财务系统”是一个道理。
选型时还有一个容易忽略的点:向量库和Embedding模型必须匹配。有些向量库对向量归一化有要求,有些要求指定距离算法。如果你把OpenAI的Embedding结果塞进为Cosine距离优化的库,问题不大;但如果库内部默认L2距离,而你用的是不归一化的向量,召回质量会明显下降。解决方案是入库前统一做L2归一化,并显式声明距离算法。
3.2 记忆写入:切分、向量化、元数据
长期记忆不是把整段对话原样丢进向量库,那会稀释检索精度。写入前必须做三件事:切分、向量化、附加元数据。
切分方面,我按语义段落切,而不是按固定字符数硬切。固定切分会把一个完整事实劈成两半,检索时哪一半都匹配不上。我的做法是用标点符号和换行做锚点,段落到250字左右收束到最近的句号。这样每个chunk都是一个语义完整的单元。
向量化方面,中文场景我目前用bge-small-zh,兼顾效果和速度。维度768,单条文本在CPU上向量化大约几毫秒,算力需求不高。如果你面向英文为主,可以考虑同类英文模型。值得警惕的是:不要在写入和检索时用不同的Embedding模型,哪怕只是版本升级,也要做一次全量重新向量化,否则检索结果会“飘”。我吃过这个亏,升级模型后忘了重刷存量数据,线上召回率掉了一半。
写入时元数据比向量本身更重要,因为元数据决定你能做到多精确的过滤。我每条记忆记录包含以下字段:
memory_entry = { "id": "mem_xxxxx", "user_id": "user_001", "session_id": "session_002", "type": "fact|preference|behavior|summary", # 记忆类型 "content": "用户提到他对价格敏感,优先推荐性价比路线", "embedding": [0.01, 0.02, ...], # 768维向量 "importance": 3, # 重要度 1-5 "access_count": 0, # 被访问次数 "created_at": 1708230000, "last_accessed_at": 1708230000 }importance和access_count是后面做时间衰减和淘汰策略的关键输入。如果没有这两个字段,所有记忆在系统看来价值相等,那就根本无法做“遗忘”。
3.3 混合检索:相似度 + 关键词 + 时间衰减
向量检索解决“语义相似”的问题,但单靠它远远不够。举一个我实际遇到的例子:用户问“上次你说那个便宜又耐用的方案”,向量检索可以召回包含“便宜”“耐用”的对话片段,但如果库里相似内容很多,排序就会失准。这时候需要关键词匹配和时间衰减来辅助重排。
我的做法是三路召回加一套加权融合:
- 向量召回:按余弦相似度取Top 20。
- 关键词召回:对用户问题做分词,提取关键名词,从记忆表中做倒排匹配,取Top 20。
- 最近活跃召回:取该用户最近3天内产生的高重要度记忆,不管相关性如何,先捞回来。
融合公式我经过多轮实验调整参数,目前稳定在下面的形态:
final_score = 0.55 * vector_score + 0.30 * keyword_score + 0.15 * recency_score其中recency_score由最后访问时间和时间衰减因子计算。衰减函数我用的是指数衰减,越久没被访问,分越低。这模拟了人的遗忘曲线,长时间不用的记忆即便相关也不应该占据靠前位置。
这里有一个心得:min_score阈值一定要设。如果没有阈值,拿一个完全不相关的问题去检索,系统也会硬把最相似的一条拿出来注入上下文,导致模型被无关记忆干扰。我把默认阈值设为0.45,实际使用中这个值在0.4到0.5之间比较平衡,太低噪声太多,太高容易漏召回。
4. 结构化记忆:用户画像与事实库
4.1 动态抽取用户偏好
光有向量记忆还不够,因为向量适合“模糊回忆”,不适合“精确查证”。比如用户说“我住南山”,你需要在调用某个服务时精确知道地址,靠向量检索可能召回一段关于南山的闲聊,但拿不到准确的“居住地=南山”这个事实。所以要单独建一张结构化事实表,专门存这种“一句话能说清、且不容混淆”的事实。
抽取环节我采用“事后抽取”策略:每一轮对话结束后,调一次轻量模型,用固定Prompt从对话中抽取结构化事实。注意,不是每句话都要抽,也不是整个会话结束后再抽,而是每个对话回合结束后增量抽取,避免上下文太长导致抽取质量下降。
抽取的Prompt我经过几轮调整,核心约束是“只抽稳定的、跨会话有用的事实,不抽临时状态”:
FACT_EXTRACT_PROMPT = """ 从以下对话中抽取用户的关键事实,包括: - 用户提供的身份信息(姓名、职业、所在地等) - 用户的明确偏好(价格敏感度、风格取向、沟通习惯) - 用户当前正在进行的长期事务(项目、计划、周期性任务) 规则: 1. 只抽取明确陈述或强烈暗示的内容,不做过度推断 2. 删除一次性临时信息(如今天天气、当前心情) 3. 冲突时保留最新陈述 4. 输出JSON数组,格式:[{"type": "preference", "fact": "用户对价格敏感", "certainty": 0.9}] 对话内容: {conversation} """4.2 事实库的更新与冲突消解
事实库最大的坑是冲突。用户第一轮说“我在上海”,第三轮说“我下个月搬到北京”。如果两句话都存进去,模型就不知道用户到底在哪。解决办法是“同主键覆盖”:将事实按(user_id, fact_type, attribute)做唯一键,新事实写入时检查同键旧事实,若存在且certainty不低于旧值,则覆盖;若新事实certainty很低(比如用户只是随口一说),则保留旧值并记录疑点。
这个规则听起来简单,但实际工程中要加上冲突检测提示。我在抽取模型返回时加了字段replaces,让模型显式指出它认为哪条旧事实应该被替换。这样冲突消解从“写时被动处理”变成“模型主动声明”,准确率提升非常明显。
4.3 记忆回写与过期清理策略
记忆系统不能只进不出,否则会无限膨胀。我在配置里设了max_facts_per_user: 500,超过上限后,按重要度和最近访问时间排序,把低重要度且长期未访问的记录归档到冷存储表,不再参与在线检索。
同时还有一套定期清理任务,每周跑一次:
- 超过180天未访问且重要度低于3的事实,直接删除。
- 向量库中对应的向量一并删除,不然会产生“有向量无事实”的孤儿记录。
- 摘要记忆保留最近30天的,更早的只保留按周聚合的周摘要。
这套策略是跟一个做推荐系统的朋友聊天时受到的启发:没有遗忘机制的记忆系统,就像一个不打扫的房间,时间越久越是寸步难行。别怕丢数据,有价值的记忆会被重复访问,自然留存;长期不用的记忆丢掉成本其实很低。
5. 实际操作:从零搭建一套记忆系统模板
5.1 环境准备与依赖安装
开始之前先把基础环境准备好。我是在Linux服务器上跑,Python 3.10+,虚拟环境隔离。核心依赖就四个:OpenClaw框架本体、向量存储、SQLite驱动、Embedding模型加载库。
# 创建虚拟环境 python3 -m venv venv source venv/bin/activate # 安装核心依赖 pip install openclaw pip install sqlite-vec pip install sentence-transformers # 验证安装 python -c "import openclaw, sqlite_vec, sentence_transformers; print('ok')"如果你的网络环境拉取模型困难,提前把Embedding模型下载到本地目录,然后通过环境变量指定模型路径,避免每次启动都检查下载。这一点在受限网络环境下尤为重要。
5.2 OpenClaw接入配置
OpenClaw的接入配置是记忆系统能否真正生效的关键。需要在OpenClaw的配置文件中,把记忆模块注册为中间件,并指定上下文的注入方式:
# openclaw_config.yaml middlewares: - name: memory_middleware module: memory.middleware config: short_term_max_tokens: 4000 long_term_top_k: 5 working_memory_budget: 1500 context: inject_memory: true memory_position: "before_user_input" system_prefix: | 【历史记忆】 {memory_block} 【当前输入】memory_position我建议放在用户输入之前、系统指令之后。原因很实际:模型对越靠前的内容注意力越强,核心系统指令必须占据最前位置,记忆块紧跟其后,这样记忆的指令遵循度最高。放在系统指令前面,会削弱系统指令的约束力,这个顺序别搞反。
5.3 核心代码实现:记忆管理中间件
记忆系统的核心是一个中间件,它拦截每一次对话请求,走“读取工作记忆 → 注入上下文 → 响应后写入新记忆”的流程。这里我给出一版可以运行的简版实现:
# memory/middleware.py import time from .short_term import ShortTermMemory from .long_term import LongTermMemory from .summary import generate_summary class MemoryMiddleware: def __init__(self, config): self.config = config self.short_term = ShortTermMemory( max_tokens=config["short_term_max_tokens"], window_size=config.get("window_size", 20) ) self.long_term = LongTermMemory( top_k=config["long_term_top_k"] ) def before_request(self, user_id, session_id, user_input): """请求前:组装工作记忆并注入上下文""" # 1. 从短期记忆中获取最近对话 recent = self.short_term.get_recent(1500) # 2. 从长期记忆中检索相关事实 related = self.long_term.search( query=user_input, user_id=user_id, top_k=self.config["long_term_top_k"] ) # 3. 组装工作记忆块 memory_block = "" if related: memory_block += "【相关历史记忆】\n" for mem in related: memory_block += f"- {mem['content']}\n" if recent: memory_block += "【最近对话】\n" for msg in recent: memory_block += f"{msg['role']}: {msg['content']}\n" return memory_block def after_response(self, user_id, session_id, user_input, response): """响应后:写入短期记忆,并异步抽取长期事实""" # 1. 写入短期记忆 self.short_term.add_message({"role": "user", "content": user_input}) self.short_term.add_message({"role": "assistant", "content": response}) # 2. 短期记忆超阈值则生成摘要 if self.short_term.token_count > self.config.get("summary_threshold", 3000): summary = generate_summary(self.short_term.get_recent(20000)) self.short_term.reset_with_summary(summary) # 3. 异步抽取长期事实 self.long_term.extract_and_store(user_id, session_id, user_input, response)实际生产版本还需要把SQLite落盘、向量重排、并发锁这些细节补上,但整体的流程骨架就是这样。把这个中间件注册到OpenClaw之后,记忆系统就真正跑起来了。
5.4 压测与效果验证
搭好之后不要急着上线,我建议做三轮验证。
第一轮是功能验证:连续对话20轮,确认模型能记住中段提到的信息,而不是只记得最近两条。第二轮是跨会话验证:新开会话,问“我之前提过一个需求……”,确认长期记忆能召回相关内容。第三轮是压力验证:模拟50个用户并发对话,观察写入和检索的延迟。
我的经验数据供参考:SQLite+本地向量方案在个人电脑上,单条记忆写入约15ms,检索Top 5约30ms,50并发下延迟涨幅在2倍以内,没有出现死锁或超时。如果你发现延迟暴涨,优先排查是不是每次请求都重新加载了Embedding模型,务必把模型常驻内存。
6. 常见问题与排查技巧实录
6.1 记忆失效:模型还是“不认识你”
症状:明明存了用户的名字和偏好,新会话一问三不知。
排查步骤:
- 先查长期记忆表,确认用户ID下是否有记录。有时问题出在用户ID传递不一致,同一用户在Web端和API端可能拿到不同的匿名ID,记忆当然对不上。
- 再查检索逻辑。看
min_score阈值是不是设得过高,导致所有记忆都被过滤。我把0.45作为默认值,但如果你用的是不同Embedding模型,分数分布可能差异很大,可以临时把阈值降到0.1观察召回情况。 - 确认记忆块真的注入到了提示词中。OpenClaw有个调试模式可以打印最终发送给模型的完整提示词,看
memory_block是否为空。
这三点检查完,大部分“失忆”问题都能定位到原因。其中用户ID不一致是最隐蔽的,也是我踩过最深的一个坑。
6.2 向量检索召回不准
症状:召回的记忆和当前问题语义相关度不高,甚至完全跑偏。
我遇到过三次这种情况,原因各不相同:第一次是切分粒度太粗,一段记忆包含多个主题,检索时噪声大于信号;第二次是Embedding模型和领域不匹配,通用模型在专业领域表现平平;第三次是缺乏关键词召回,单纯靠向量扛不住精确术语匹配。
解决路径也很清晰:细化切分粒度(250字左右)、换领域匹配的Embedding模型、增加关键词召回通道。三者都做完,召回质量会有质的提升。
6.3 记忆膨胀与Token开销过大
症状:系统运行一周后,响应越来越慢,Token消耗越来越高。
这个问题的根源通常是记忆写入太“贪婪”。我的排查经验是:打开记忆明细,按用户统计记忆条数和Token占用,找出膨胀最严重的用户,去看他的记忆类型分布。大概率会发现大量低价值记录,比如“用户说好的谢谢”也被当成事实存了进去。
根治办法是提高事实抽取的certainty阈值,并在写入前加一道“价值判断”:这条记忆如果不存,三个月后用户问起,模型能不能承受丢失的代价?不能承受才值得存。这道题问下来,能拦掉至少一半的垃圾记忆。
6.4 多并发场景下的记忆串线
症状:A用户的问题,模型回答中出现了B用户的信息。
这是最严重的安全事故级故障。原因通常出在缓存和连接池没有按用户隔离。我排查过一起此类问题,最终定位在向量库的连接复用时,没有在每次查询前清空上一用户的过滤条件,导致检索结果混入了其他用户的数据。
修复办法很简单:检索前强制重新设置user_id过滤参数,并且在代码审查清单里加一条“所有记忆读写必须显式携带user_id,禁止使用无过滤条件的默认查询”。这个清单项之后帮我避免了好几轮潜在事故。
6.5 常见问题速查表
| 症状 | 可能原因 | 排查优先级 | 解决方案 |
|---|---|---|---|
| 记忆完全失效 | 用户ID不一致 | 高 | 统一会话身份标识 |
| 记忆完全失效 | min_score过高 | 高 | 降低阈值或调模型 |
| 召回不准 | 切分粒度过粗 | 中 | 按语义段落切分 |
| 召回不准 | Embedding不匹配 | 中 | 统一模型并重刷向量 |
| Token膨胀 | 低价值记忆过多 | 中 | 提高certainty阈值 |
| 并发串线 | 缺少用户过滤器 | 紧急 | 强制显式user_id过滤 |
这套速查表是我在维护记忆系统过程中一点一点沉淀下来的,每次线上出问题,先按表排查,大多数时候能在五分钟内定位根因。
最后再分享一个我的个人体会。当初设计这套模板的时候,我追求的是“大而全”,想把所有记忆能力一次性做进去。后来被现实教育了几次才明白,记忆系统的核心不是功能多少,而是边界是否清晰:什么时候该记、什么时候该忘、什么记忆该给模型看,比“记住一切”重要得多。这套模板我迭代了好几版,从最初只能存会话记录,到后来能主动维护用户画像,每一版都在做减法。如果你也在搭类似的系统,建议先跑通短期记忆和跨会话召回,再加摘要、再上事实库,一步步来,别一口吃成胖子。