☰
AstrBot Mnemosyne长期记忆链路:抽取、召回与注入实践
2026/11/2 2:55:54 网站建设 项目流程

1. 长期记忆为什么是聊天机器人的分水岭

给聊天机器人加长期记忆这件事,我在 AstrBot 上折腾了不短时间,Mnemosyne 这个插件是我目前用得最顺手的一套方案。它要做的事情说白了很朴素:让机器人记住上一次、上个月甚至半年前聊过的内容,并且在合适的时候把该想起来的东西想起来。听起来像一句话需求,真做起来却横跨了对话管理、向量检索、存储设计、提示词工程和成本控制五个方向,任何一个环节没处理好,最终表现出来的都是"这机器人怎么突然开始胡言乱语"。

这篇内容适合三类人看:一是已经在 AstrBot 里装了 Mnemosyne,但发现它有时灵有时不灵的;二是想自己动手改记忆链路、加自定义存储的;三是压根没打算用这个插件,但想给自己那套对话系统补一块长期记忆模块的。我会把 Mnemosyne 的架构思路、关键参数、写入与召回的具体实现、以及我踩过的那些坑,从头到尾拆一遍。代码部分用 Python 写,存储用 SQLite 加本地向量计算,哪怕你不用 AstrBot,照着逻辑也能移植到自己的项目里。

1.1 上下文窗口再大也算不上记忆

很多人对长期记忆有个误解,觉得模型上下文窗口涨到几十万 token 之后,记忆问题就自动解决了。我实测下来的结论是:窗口大小解决的是"一次对话里能塞多少东西",而记忆解决的是"跨越多次对话之后还记得什么"。这两件事压根不是一个维度。

单次会话的上下文是线性增长的,聊到第 50 轮的时候,前面 49 轮的内容要么被截断,要么被摘要压缩,压缩的过程本身就是信息有损的。更要命的是会话结束之后,上下文一清空,机器人对你的了解就归零了。下一次你连上来,它依然会问"请问有什么可以帮您",哪怕你们昨天刚聊了三个小时。

长期记忆干的活,是把对话流里那些值得留存的信息,抽出来、压缩好、存到外部存储里,等下一次对话开始前,再挑相关的部分塞回上下文。它是一个跨越会话边界的持久化问题,跟窗口大小没有替代关系。窗口大一点,能塞回去的记忆多一些,仅此而已。

1.2 Mnemosyne 在 AstrBot 里的定位和边界

AstrBot 本身是个多平台适配的对话机器人框架,它负责消息接入、会话管理、模型调用这些主干流程。Mnemosyne 作为插件挂在它上面,只负责一件事:在消息进来之前和模型回复之后,插一脚做记忆的读写。

这个定位很重要,它决定了插件的边界。Mnemosyne 不会改变模型本身的能力,它只是在提示词里加了料;它也不会替你判断什么信息该记、什么不该记——这个判断是它内部用一层抽取逻辑做的,但抽取质量取决于你给的提示词和模型本身的水平。所以当记忆效果不好的时候,问题往往出在抽取环节和召回环节,而不是插件"坏了"。

我见过不少人装上插件就期待它秒变私人助理,结果发现它记住了"今天天气不错"这种废话,却漏掉了"我对花生过敏"这种关键信息。这不是插件不行,是抽取策略没调,后面我会专门讲怎么改这块。

2. 整体架构拆解:一条记忆的完整生命周期

在动手改任何东西之前,得先搞清楚一条记忆从产生到消亡走了哪几站。Mnemosyne 的链路我把它归纳成五个阶段:抽取、归一、向量化、存储、召回。每个阶段的输入输出都很清晰,理解了这几站,后面调参就是按图索骥。

2.1 五个阶段各自在干什么

抽取是最容易出问题的一站。对话流进来的是用户原话和机器人回复,里面有寒暄、有语气词、有重复表达,真正有价值的信息可能只有一句"我下周三要去深圳出差"。抽取这一步就是要从这一堆噪音里把"用户下周三去深圳出差"这句话拉出来,并且标注成一条带时间属性的记忆。

归一化处理的是表述不一致的问题。"我在深圳"和"我人在深圳"和"我目前在深圳工作",语义上高度重合,如果不做归一,库里会堆一堆同义记忆,召回时互相打架。常见的做法是把抽取出来的内容再交给模型润色成统一格式,或者干脆用结构化字段存,比如{subject: 用户, predicate: 所在地, object: 深圳, valid_from: ...}。

向量化和存储是工程活,选好模型和库就行,但要注意分块策略和元数据设计。召回是最后一步,也是用户感知最直接的一步,召回不准的表现是"答非所问"或者"该记的没记住"。

2.2 存储选型:为什么是向量库加关系型元数据

只存向量是不够的。向量能解决"语义相似",但解决不了"这条记忆是什么时候的""属于哪个用户""置信度多少""有没有被覆盖"这些问题。这些是结构化查询的活儿,交给关系型数据库干效率高得多。

所以 Mnemosyne 实际用的是双存储:向量用来做语义召回,元数据存在关系表里做过滤和排序。我自己的实践里,元数据表至少要有这几个字段:

字段类型作用备注
id自增整数主键便于增量更新
user_id文本用户隔离多用户场景必须
content文本记忆原文召回后直接拼进提示词
embedding二进制或文本向量也可以存外部向量库
memory_type枚举事实/偏好/事件决定注入优先级
importance浮点 0-1重要性权重参与最终打分
created_at时间戳创建时间支持时间衰减
last_hit_at时间戳上次被召回时间用于热度统计
confidence浮点 0-1置信度抽取模型给的分
status枚举有效/失效/合并软删除用

这张表看着简单,但每个字段都是被坑出来的。比如importance这个字段,一开始我没加,结果所有记忆平权,召回时"用户喜欢猫"和"用户下周要去医院复查"权重一样,明显不合理。status字段也是后来补的,因为记忆会过时,直接物理删除又怕误删,软删除是最稳的。

2.3 写读分离:别在用户等回复的时候做重活

这是个很关键的工程决策。写入链路要做的事包括调用模型抽取、调用嵌入模型算向量、写两张表,整个过程少说几百毫秒,多则几秒。如果放在用户消息处理的同步链路上,用户会明显感觉到"这机器人反应好慢"。

我的做法是把写入做成异步任务队列:消息处理完、回复发出去之后,再把这一轮对话丢进队列,后台慢慢抽。这样用户侧的延迟只受召回影响,而召回一般是毫秒到百毫秒级别,感知不明显。

代价是写入有延迟,可能出现"刚说完的话,下一句还记不住"的情况。这个用一个小技巧补:在内存里维护一个短期缓冲区,把最近 N 轮对话的原文放在里面,召回时先查缓冲区再查库。缓冲区设置成十几轮足够覆盖写入延迟窗口。

3. 核心机制:抽取、向量化、召回、注入

这一节是全篇的重点,我会把每个环节的原理和可调参数都摊开讲。理解了这几步,你就能判断自己的记忆效果差是差在哪一环。

3.1 记忆抽取:从对话流里捞出真正有用的东西

抽取质量直接决定记忆质量,这是整条链路上最需要花心思的地方。Mnemosyne 默认的抽取方式是给模型一段系统提示,让它从对话里挑出"值得长期记住的信息",输出成结构化 JSON。听起来简单,实际效果取决于三件事:提示词写得细不细、触发时机对不对、过滤规则严不严。

提示词这块,我改过好几版,最后稳定下来的是一个分类清单式的写法:把记忆分成事实类(用户的身份、职业、所在地、家庭成员)、偏好类(喜欢什么、讨厌什么、习惯什么)、事件类(有明确时间点的事情)、关系类(用户提到的人和用户的关系)。每一类给两三个正例,模型抽出来的准确率会明显提升。

触发时机也很讲究。每轮对话都抽一次,成本和噪音都高;隔很多轮才抽一次,又会漏。我现在的策略是"三条件触发":对话轮数达到阈值、单轮消息长度超过阈值、或者对话里出现了明显的记忆信号词("记住""我一直""我以后""我从不"这类)。三个条件满足任意一个就跑一次抽取。

过滤规则是我踩坑最多的地方。早期我把抽取结果直接入库,结果库里堆了一堆"用户说你好""用户问了几点"这种垃圾记忆。后来加了三层过滤:长度过滤(少于 8 个字的丢掉)、语义过滤(纯寒暄、纯确认类丢弃)、重复过滤(跟已有记忆的向量相似度超过 0.95 的,走合并流程而不是新增)。

提示:抽取用的模型可以比主对话模型小一号。抽取任务本身不难,用小模型能省下大量成本,我实测下来质量差异远小于成本差异。

3.2 向量化与索引:维度、距离度量和分块

嵌入模型的选择决定了语义召回的上限。这里有个常见误区是"维度越高越好",实际上维度高意味着存储更大、计算更慢,而在短文本记忆场景下,768 维和 1536 维的召回效果差异往往在几个百分点以内。我的建议是先用一个中等维度的模型跑通全链路,等真的遇到召回瓶颈再换。

距离度量方面,绝大多数嵌入模型是配合余弦相似度训练的,所以直接用余弦就行。如果你用欧氏距离,记得先把向量做归一化,否则长度不同的向量之间比较会失真。这个细节很多人忽略,结果召回结果莫名其妙。

分块策略在记忆场景下和文档问答不太一样。文档问答需要把长文档切成小块,记忆场景恰恰相反——一条记忆应该是一个完整语义单元,切成两半反而破坏语义。所以我的分块原则是"一条记忆一个块,超过 200 字的记忆在抽取阶段就该被拆分或压缩"。如果一条记忆有 300 字,说明抽取环节做得不好,把整段对话抄进去了。

索引方面,记忆量在一万条以内的时候,直接全量加载进内存算相似度完全够用,速度是毫秒级。超过十万条再考虑引入专门的向量索引结构。我见过有人几百条记忆就上分布式向量库,纯属给自己找麻烦。

3.3 多路召回与重排:别只依赖向量相似度

纯向量召回的典型问题是:语义相近但时间不对。比如用户三个月前说"我在准备考研",昨天说"我上岸了",两条记忆语义都跟"考研"相关,向量召回会把两条都拉出来,然后机器人可能回一句"祝你考研顺利",场面相当尴尬。

解决办法是多路召回再融合。我现在用的是三路:

第一路是向量召回,按余弦相似度取 TopK,这是主干。

第二路是关键词召回,用简单的中文分词加倒排匹配,专门捞那些专有名词、人名、地名这类向量容易糊掉的精确信息。

第三路是时间召回,取最近 N 条记忆,保证机器人对用户最近的状态有感知。

三路结果合并去重后,再走一次重排。重排的打分公式我用的是加权求和:

score = 0.55 * 语义相似度 + 0.20 * 重要性 + 0.15 * 时间新鲜度 + 0.10 * 置信度

时间新鲜度用指数衰减算,exp(-天数 / 半衰期),半衰期我设的是 30 天,也就是说一个月前的记忆权重衰减到 0.37。这个参数对不同类型的记忆要区别对待:偏好类记忆衰减慢(半衰期设 180 天),事件类衰减快(半衰期设 14 天)。

3.4 上下文注入:预算怎么分才不挤爆提示词

召回出来的记忆最终要拼进提示词,但提示词是有预算的。我的分配方案是:系统提示占 15%,记忆块占 25%,历史对话占 45%,当前输入和预留输出占 15%。记忆块超了就按重排分数从低到高砍。

拼装格式也很影响效果。直接堆一段文字,模型容易忽略;用带标记的结构化格式,遵从率会高不少。我用的格式大概是这样:

[记忆片段 - 事实类] - 用户常驻城市:深圳(置信度 0.92,更新于 2025-03-11) - 用户职业:后端开发(置信度 0.88,更新于 2025-02-02) [记忆片段 - 偏好类] - 用户不吃香菜(置信度 0.95)

每一条后面带上置信度和更新时间,模型在做判断时会自然倾向于用新的、高置信度的信息。这个细节看着不起眼,但对减少"记忆冲突"引发的胡言乱语非常有效。

注意:注入的记忆一定要带"这是历史记忆,可能过时"这样的元提示,否则模型会把记忆当成当前事实,用户搬家了它还在说旧地址。

4. 实操:从零跑通一条记忆链路

理论讲完,直接上手。这一节我给的是可以复制粘贴跑起来的最小实现,用 SQLite 存元数据,向量直接在内存里算,不需要任何外部服务。规模小的时候这套方案完全够用,跑通之后再按需替换组件。

4.1 环境准备与配置项清单

依赖很轻:sqlite3是标准库,numpy用来算向量,jieba用来做关键词召回的分词。嵌入模型你自己接,本地跑的或者调 API 的都行,只要最终拿到一个list[float]。

配置项我整理成了一张表,这几个参数是真正会影响效果的,其余的保持默认就行:

配置项建议值说明
chunk_max_len200单条记忆最大字数,超了触发压缩
extract_min_len8短于这个长度的抽取结果直接丢弃
dedup_threshold0.95相似度超过走合并而非新增
recall_topk20每路召回的候选数
inject_topn6最终注入提示词的条数
time_half_life30时间衰减半衰期,单位天
buffer_rounds12内存短期缓冲的轮数

4.2 写入链路:抽取、去重、落库

建表语句先放这里,字段跟前面那张元数据表对得上:

import sqlite3 def init_db(path="memory.db"): conn = sqlite3.connect(path) conn.execute(""" CREATE TABLE IF NOT EXISTS memories ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id TEXT NOT NULL, content TEXT NOT NULL, embedding BLOB, memory_type TEXT DEFAULT 'fact', importance REAL DEFAULT 0.5, confidence REAL DEFAULT 0.8, created_at INTEGER, last_hit_at INTEGER, status TEXT DEFAULT 'active' ) """) conn.execute("CREATE INDEX IF NOT EXISTS idx_user ON memories(user_id, status)") conn.commit() return conn

向量用numpy.tobytes()存成 BLOB,读出来再frombuffer还原,比存 JSON 数组省一半以上空间。

写入主流程分四步,关键是去重那一步不能省:

import time import numpy as np def write_memory(conn, user_id, raw_text, embed_fn, dedup_threshold=0.95): # 第一步:抽取,这里用你自己的模型调用替换 extracted = extract_with_llm(raw_text) if not extracted or len(extracted["content"]) < 8: return None content = extracted["content"][:200] vec = np.array(embed_fn(content), dtype=np.float32) # 第二步:去重检查,只跟同用户的 active 记忆比 rows = conn.execute( "SELECT id, content, embedding FROM memories WHERE user_id=? AND status='active'", (user_id,) ).fetchall() for rid, old_content, old_blob in rows: old_vec = np.frombuffer(old_blob, dtype=np.float32) sim = float(np.dot(vec, old_vec) / (np.linalg.norm(vec) * np.linalg.norm(old_vec) + 1e-8)) if sim >= dedup_threshold: # 走合并:保留新的,旧标记为 merged conn.execute("UPDATE memories SET status='merged' WHERE id=?", (rid,)) break # 第三步:落库 conn.execute( """INSERT INTO memories (user_id, content, embedding, memory_type, importance, confidence, created_at, last_hit_at, status) VALUES (?,?,?,?,?,?,?,?,?)""", (user_id, content, vec.tobytes(), extracted.get("type", "fact"), extracted.get("importance", 0.5), extracted.get("confidence", 0.8), int(time.time()), int(time.time()), "active") ) conn.commit() return content

注意去重那里我只标记了旧记忆为merged而没有物理删除,这是为了保留追溯能力。真出问题的时候能查到"我当时记的到底是什么",比直接删掉强太多。

4.3 召回链路:三路合并加重排

召回的实现比写入稍微复杂一点,但逻辑清晰:

def recall(conn, user_id, query, embed_fn, topk=20, topn=6): q_vec = np.array(embed_fn(query), dtype=np.float32) rows = conn.execute( "SELECT id, content, embedding, memory_type, importance, confidence, created_at " "FROM memories WHERE user_id=? AND status='active'", (user_id,) ).fetchall() now = int(time.time()) scored = [] for rid, content, blob, mtype, imp, conf, ctime in rows: vec = np.frombuffer(blob, dtype=np.float32) sim = float(np.dot(q_vec, vec) / (np.linalg.norm(q_vec) * np.linalg.norm(vec) + 1e-8)) days = max((now - ctime) / 86400.0, 0) half_life = 14.0 if mtype == "event" else 30.0 fresh = float(np.exp(-days / half_life)) score = 0.55 * sim + 0.20 * imp + 0.15 * fresh + 0.10 * conf scored.append((score, rid, content)) scored.sort(key=lambda x: -x[0]) picked = scored[:topn] # 更新被命中的时间戳,给热度统计留数据 ids = [p[1] for p in picked] if ids: conn.executemany("UPDATE memories SET last_hit_at=? WHERE id=?", [(now, i) for i in ids]) conn.commit() return [p[2] for p in picked]

这一段里half_life的分类是我自己加的,Mnemosyne 默认给的是一个统一值。分开之后事件类记忆过期得更快,实际体验里减少了不少"用三个月前的行程回答今天的问题"这种事。

4.4 参数怎么定:算一遍心里才有数

参数不能拍脑袋,我拿分块和 TopK 举两个例子。

分块大小:假设嵌入模型的最大输入是 512 token,中文大致 1 token 对 1.5 个字,那么理论上限是 768 字。但记忆场景要留压缩空间,而且太长会稀释语义,所以取 200 字左右比较合适。这个数不是我拍的,是把"模型上限 × 语义纯度系数(约 0.3)"算出来的。

TopK 与注入条数:假设每条记忆平均 40 字,注入 6 条就是 240 字,加上标记大概 300 字,在 4000 token 的记忆预算里占不到 15%,很安全。如果注入 20 条就是 1000 字,超过 25% 的预算,会挤占历史对话的空间,得不偿失。所以inject_topn维持在 5 到 8 之间是合理区间,别贪多。

召回候选数:recall_topk设 20 是个经验值。设太小容易漏,设太大重排成本上去,而且低分候选会污染重排结果。20 这个数在几千条记忆的规模下,重排耗时通常在 5 毫秒以内。

5. 常见问题与排查实录

这一节是我真正想写的部分,因为坑都是自己踩出来的,别人写的文档里基本不会提。

5.1 记忆串味:多用户场景下最危险的问题

最严重的一次事故,是测试环境里两个用户的记忆混到了一起,机器人把 A 用户的职业说成了 B 用户的。排查下来是user_id在某条异步写入路径里被覆盖成了默认值。

这个问题本质上不是插件 bug,而是异步链路上的上下文丢失。修法很简单:异步任务提交的时候显式把user_id作为参数传进去,绝对不要依赖任何全局变量或请求上下文。我现在所有写入函数的第一个参数都是user_id,且加了断言,为空直接抛异常。

另外建议在数据库层加一道保险:记忆查询永远带WHERE user_id = ?,不要写"先全查再内存过滤"的代码。内存过滤只要有一处忘了判断,就是数据越界。

5.2 召回不准的三种典型表现和对策

表现一:答非所问。用户问"我上次说的那个项目怎么样了",机器人回了一堆不相关的内容。原因是召回时查询语句太短、语义太泛,"那个项目"这种指代表达向量化之后跟具体项目名差很远。对策是把最近几轮对话一起作为查询向量,或者在抽取阶段就把指代消解掉,存成"用户在做的项目:XX"而不是"那个项目"。

表现二:该记的没记住。大概率是抽取阶段丢了。排查方式是打开插件的调试日志,看抽取模型对那段对话输出了什么。我遇到过一次是提示词里要求"只抽取明确陈述的信息",结果用户说"我可能下周要去趟杭州","可能"两个字被判定为不确定,整条被丢弃。后来把提示词改成允许记录带不确定标记的推测类记忆,问题解决。

表现三:记住了旧的,忽略了新的。这是时间衰减参数的问题。如果half_life设得太大(比如 365 天),新记忆和旧记忆权重几乎一样,重排时容易被旧的高相似度记忆压住。调小到 30 天以内,情况会明显改善。

5.3 延迟和成本失控怎么止住

延迟问题一般出在两个地方:嵌入模型调用太慢、召回时全表扫描。

嵌入模型调用是同步阻塞的话,一轮对话里抽一次、查一次,就是两次模型调用。我的做法是嵌入模型本地部署,用一个小模型就行,单次推理十几毫秒。如果只能用 API,那就做批量:写入时把一轮对话里的多条记忆攒一批算,召回时如果查询很短可以加个 LRU 缓存,相同相似查询直接命中缓存。

成本问题主要来自抽取用的模型。早期我用主对话模型做抽取,一次对话几毛钱。换成小模型之后成本降了一个数量级,质量只掉了几个点。判断标准很简单:如果你发现抽取出来的东西经常需要人工纠正,那才升级模型;如果大部分都挺准,就说明小模型够用了。

5.4 记忆库膨胀:该删的时候要敢删

跑了一个月之后我的库涨到了十几万条,召回开始变慢,重排结果也越来越杂。清理策略有三条:

第一条是按状态清,status='merged'的记录保留 30 天后物理删除,这些是被覆盖掉的旧版本,留着只占空间。

第二条是按命中率清,连续 90 天last_hit_at没有更新的低重要性记忆,直接软删。这条规则有点激进,但实测下来被删的基本都是"用户说了句今天好累"这种,删了没有损失。

第三条是按类型清,事件类记忆过了有效时间窗口就自动失效。这个需要在写入时给事件类记忆带上valid_until字段,到期自动标失效。

注意:清理脚本一定要先跑一遍 dry-run,把要删的条目打印出来人工扫一眼。我有一次清理规则写错,把一个用户的所有偏好类记忆全删了,找回来花了半天。

5.5 常见问题速查表

现象最可能的原因排查动作修复方向
记忆串用户user_id 在异步链路丢失查写入日志里的 user_id显式传参加断言
记住了废话抽取提示词太宽松打开抽取结果日志加长度和类型过滤
语义相近但答错纯向量召回无重排检查重排是否生效加时间衰减和重要性权重
刚说的话没记住写入异步有延迟看队列积压情况加内存短期缓冲
回复变慢嵌入调用阻塞打点统计各阶段耗时本地化模型或加缓存
库越来越大无清理策略统计 status 分布按状态和命中率清理
旧信息覆盖新信息时间衰减没配检查 half_life按类型分设半衰期
提示词超长注入条数过多统计提示词 token 数调低 inject_topn

6. 我自己踩出来的几条经验

写到这里,把几个零散但很关键的点补充一下,这些属于文档里不会写、但实际用起来影响很大的细节。

第一,记忆的置信度字段比想象中重要。抽取模型对每条记忆给一个 0 到 1 的置信度,这个值在重排里只占 0.1 的权重,看起来不起眼,但它真正的作用是给你一个排查入口。当用户投诉"它记错了"的时候,你去看那条记忆的置信度,如果低于 0.6,说明抽取环节本身就不确定,问题在抽取;如果高于 0.9,说明抽取很确信但事实变了,问题在时效。这个区分能让排查效率提升一大截。

第二,别迷信自动抽取,给用户一个手动入口。Mnemosyne 支持通过指令手动写入记忆,这个功能看起来原始,实际很有用。用户主动说"记住我对海鲜过敏",比任何自动抽取都准。我在实际部署里会引导用户用这个入口补充关键信息,自动抽取只作为兜底。

第三,测试记忆功能要用长周期数据。刚装上的时候测几轮对话,感觉效果挺好。真跑一个月才发现各种时效问题。建议在测试阶段就人工构造带时间戳的历史数据,把"三个月前的记忆"和"昨天的记忆"混在一起灌进去,专门测冲突场景。

第四,日志一定要打全。记忆系统是个黑盒,出问题时你只能靠日志还原。我的日志埋点是:抽取输入、抽取输出、去重判断结果、召回候选列表和分数、最终注入的记忆。这五个点打全了,任何异常都能在五分钟内定位到环节。

最后分享一个我最近在试的扩展方向:给记忆加一层"关系图谱"。现在所有记忆是平铺的,用户提到"我女朋友"和"小雅"是两条独立记忆,模型不知道这俩是同一个人。如果在抽取阶段就把实体和关系抽出来存成图结构,召回时沿着关系扩展一跳,能捞到很多平铺存储捞不到的信息。这个方案我还在验证,主要难点是实体消歧的准确率,等跑出稳定结果再来分享。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询