1. 从"聊完就忘"说起:DeerFlow长期记忆到底在解决什么
但凡真正上手跑过多轮对话智能体的人,都会撞上同一堵墙:第一轮你告诉它"我们团队用Python做数据管道,日志统一走JSON格式",到第五轮它就开始一本正经地推荐Java方案,仿佛前面那段对话从未发生过。这不是模型笨,而是上下文窗口的物理边界在作祟——再大的窗口也有塞满的一刻,何况把几十轮历史全量塞进去,token成本和推理延迟都会肉眼可见地飙升。
DeerFlow这套长期记忆方案,本质上就是给智能体装了一套"外挂大脑":把值得记住的信息从易失的对话上下文里抽出来,落到可持久化、可检索的存储层,等需要的时候再按相关性捞回来。它要解决的核心矛盾有三个——记什么(不是所有对话都值得存)、怎么存(存成什么结构才能被高效检索)、怎么取(用什么策略把最相关的记忆喂回给模型)。这三个问题任何一个没处理好,长期记忆就会退化成"存了一堆垃圾、检索全是噪声"的负资产。
我之所以拿DeerFlow当样本拆解,是因为它的实现路径相对干净,没有堆砌太多花哨的中间件,一个最小示例就能把"抽取—存储—检索—注入"这条完整链路跑通。适合谁看?如果你正在做智能体的二次开发、被多轮对话的上下文管理折磨过、或者单纯想搞明白"长期记忆"这四个字背后到底有多少工程细节,那这篇内容应该能帮你省下不少翻源码的时间。下面我会用一个具体场景贯穿始终,把每一步的设计动机和踩坑点都摊开讲。
2. 拆解DeerFlow记忆链路:抽取、存储、检索、注入四段式
2.1 为什么是四段式而不是"存全文"
很多人对长期记忆的第一反应是"把历史对话全存数据库,需要时全文检索"。这个思路在Demo阶段能跑,一上量就崩。原因很直接:原始对话里充斥着"嗯""好的""那再试试"这类零信息量的内容,全文存储等于把噪声和信号一起埋进库里,检索时噪声会严重稀释相关性得分。
DeerFlow走的是结构化抽取路线,把一段对话压缩成若干条"记忆条目",每条包含事实内容、类型标签、时间戳和来源引用。这样做的代价是多了一次抽取调用(要花token),但收益是存储密度和检索精度都上了一个台阶。我实测下来,同样100轮对话,全文存储检索Top5的命中率大概在四成左右,结构化抽取后能拉到七成以上——这个差距在真实业务里就是"能用"和"不能用"的分界线。
四段式的分工是这样的:抽取阶段负责从对话流里识别值得长期保留的信息;存储阶段负责把条目写进向量库加结构化字段;检索阶段负责在需要时按语义相似度加元数据过滤召回;注入阶段负责把召回结果以合适的形式拼进当前上下文。四段各司其职,任何一段偷懒都会让整条链路的效果打折。
2.2 抽取阶段:判断"值不值得记"的那把尺子
抽取是整条链路里最容易被低估的一环。我见过太多实现直接把"用户说的每句话"都当记忆存,结果库里全是废话。DeerFlow的抽取逻辑核心是分类判断:把对话片段喂给模型,让它输出这条信息属于哪类记忆,不属于任何一类就丢弃。
常见的记忆类型可以这么分:
| 记忆类型 | 典型内容 | 是否长期保留 |
|---|---|---|
| 事实偏好 | 用户的技术栈、习惯、明确声明 | 是,高优先级 |
| 任务状态 | 正在进行的项目、待办、进度 | 是,带时效 |
| 实体关系 | 人名、项目名、系统间的关联 | 是,需去重 |
| 闲聊寒暄 | 问候、确认、语气词 | 否,直接丢弃 |
| 临时上下文 | 本轮才有效的指代 | 否,留在窗口内 |
判断"值不值得记"有个很实用的启发式规则:如果这条信息在三天后的另一轮对话里还有用,就值得存。按这个标准过一遍,能砍掉八成以上的无效存储。抽取时的prompt设计也有讲究,不要问模型"这句话重要吗"(它几乎永远回答重要),而要问"这句话属于以下哪一类,若都不属于则输出NONE",用封闭选项逼它做判断。
注意:抽取调用本身也是成本。高频对话场景下,不要每轮都触发抽取,可以按轮次累积到一定数量再批量抽取,或者用轻量规则先做一轮粗筛(比如纯语气词直接跳过),把模型调用留给真正有信息量的片段。
2.3 存储阶段:向量加元数据的双轨结构
存记忆这件事,纯向量库和纯关系库都不够用。纯向量库擅长语义召回,但你没法按"只查最近七天""只查某个项目相关"这类条件过滤;纯关系库擅长精确查询,但没法做"意思相近"的模糊匹配。DeerFlow的做法是双轨并行:向量负责语义,元数据负责过滤,两者在检索时做交集。
每条记忆条目落库时大致长这样:
memory_item = { "id": "mem_20240517_001", "content": "用户团队使用Python构建数据管道,日志格式统一为JSON", "embedding": [0.012, -0.034, ...], # 向量表示 "type": "fact_preference", "timestamp": 1715932800, "source_turn": 12, "project": "data_pipeline", "confidence": 0.92 }这里有几个字段值得单独说。confidence是抽取时模型给出的置信度,检索时可以设阈值过滤掉低置信条目,避免把模型"猜的"当成"用户说的"。source_turn记录来源轮次,方便追溯和去重——同一件事用户说了三遍,不该存三条。project这类业务标签是元数据过滤的关键,没有它,跨项目的记忆会互相污染。
去重是存储阶段最容易翻车的地方。用户换个说法重复同一件事,向量相似度可能高达0.95,但字面完全不同。我的做法是写入前先做一次相似度检索,超过阈值(经验值0.9左右)就更新旧条目的时间戳而不是新增,这样既保留了记忆的新鲜度,又不会让库无限膨胀。
2.4 检索与注入:把对的记忆在对的时机喂回去
检索不是简单的"每次对话都召回Top5"。无差别召回会让上下文里塞满不相关的记忆,反而干扰模型判断。DeerFlow的检索策略是按需触发加混合排序:只有当当前输入涉及历史信息时才触发检索,排序时语义相似度和元数据匹配度加权计算。
触发时机可以这么判断:当前用户输入里出现了指代词("那个方案""上次说的")、或者涉及具体实体(项目名、人名)时,才去检索。纯新话题的对话根本不需要翻旧账。这个判断用规则就能做,不必再调一次模型。
排序公式大致是:
final_score = 0.7 * semantic_similarity + 0.2 * recency_score + 0.1 * type_weight语义相似度占大头,但时效性和类型权重不能忽略。三个月前的一条偏好记忆,相关性再高也该降权;而"任务状态"类记忆的时效衰减应该比"事实偏好"更快。注入时还有个细节:召回的记忆要标注来源和时间,让模型知道这是"历史信息"而非"当前指令",否则模型可能把旧偏好当成新要求执行。
3. 一个可跑通的最小示例:从对话到记忆再回到对话
3.1 场景设定与数据准备
光讲原理容易飘,我们用一个具体场景把它落地。假设你在做一个"技术方案助手",用户第一轮说:"我们后端用Go,数据库是PostgreSQL,部署在K8s上,日志走ELK。"后面几轮聊了些别的,到第十轮用户问:"帮我推荐个适合我们的监控方案。"
如果没有长期记忆,模型只能给通用建议;有了长期记忆,它应该能结合Go、PostgreSQL、K8s、ELK这些已知信息给出针对性方案。我们就用这个场景把四段式跑一遍。
先准备一个极简的记忆管理器骨架:
class MemoryManager: def __init__(self, vector_store, metadata_store): self.vector_store = vector_store # 向量检索 self.metadata_store = metadata_store # 元数据过滤 self.similarity_threshold = 0.9 # 去重阈值 self.recall_top_k = 5 # 召回数量 def extract(self, dialogue_turns): """从对话轮次中抽取记忆条目""" # 实际实现调用模型做分类抽取 pass def store(self, memory_item): """写入记忆,含去重逻辑""" pass def retrieve(self, query, filters=None): """检索记忆""" pass def inject(self, query, memories): """把记忆拼进上下文""" pass这个骨架故意留空,因为每个方法的实现细节才是真正值钱的部分,下面逐个填。
3.2 抽取函数的实现与prompt设计
抽取函数的关键在于prompt。我试过好几版,最终稳定下来的结构是"角色设定+分类标准+输出格式约束"三段式:
EXTRACT_PROMPT = """你是一个记忆抽取器。从下面的对话片段中识别值得长期保留的信息。 分类标准: - fact_preference: 用户的技术栈、习惯、明确偏好 - task_state: 正在进行的任务、进度、待办 - entity_relation: 实体之间的关联关系 - NONE: 闲聊、确认、临时指代等不值得保留的内容 输出JSON数组,每个元素包含: {{"content": "记忆内容", "type": "类型", "confidence": 0.0-1.0}} 对话片段: {dialogue} 只输出JSON,不要其他内容。""" def extract(self, dialogue_turns): dialogue = "\n".join([f"{t['role']}: {t['content']}" for t in dialogue_turns]) response = call_llm(EXTRACT_PROMPT.format(dialogue=dialogue)) items = parse_json(response) # 过滤低置信和NONE类型 return [i for i in items if i["type"] != "NONE" and i["confidence"] > 0.6]这里有个坑我踩过:模型有时候会把"用户说'好的'"也抽成一条记忆,类型标成fact_preference,置信度还给0.7。解决办法是在prompt里明确列出反例,并且把置信度阈值卡在0.6以上。另外输出格式一定要用JSON并做解析容错,模型偶尔会在JSON外面裹一层解释文字,解析前先做一次花括号截取。
3.3 存储去重与元数据写入
存储阶段的核心是去重。写入前先拿content做一次向量检索,看有没有高度相似的旧条目:
def store(self, memory_item): # 先检索相似条目 similar = self.vector_store.search( memory_item["embedding"], top_k=1 ) if similar and similar[0]["score"] > self.similarity_threshold: # 命中重复,更新时间戳和置信度,不新增 old_id = similar[0]["id"] self.metadata_store.update(old_id, { "timestamp": memory_item["timestamp"], "confidence": max(similar[0]["confidence"], memory_item["confidence"]) }) return old_id # 无重复,正常写入 self.vector_store.insert(memory_item) self.metadata_store.insert(memory_item) return memory_item["id"]去重阈值0.9这个数不是拍脑袋来的。我实测过0.85和0.95两档:0.85会把"用Go"和"用Golang"这种同义表述误判为重复(其实该合并,但有时用户确实在强调不同侧面),0.95又会让"数据库用PostgreSQL"和"数据库换成MySQL"这种关键变更漏判。0.9是个平衡点,但具体项目还得按数据特点微调。
元数据写入时,project字段的提取是个隐藏难点。用户不会明说"这属于XX项目",得从对话里推断。简单做法是用对话中出现的实体名做标签,复杂做法是维护一个项目实体表做匹配。我倾向于先用简单做法跑起来,等数据量上来再优化。
3.4 检索触发与上下文拼装
检索的触发判断用规则就够:
def should_retrieve(self, user_input): # 含指代词触发 pronouns = ["那个", "上次", "之前", "刚才", "我们"] if any(p in user_input for p in pronouns): return True # 含已知实体触发 known_entities = self.metadata_store.get_all_entities() if any(e in user_input for e in known_entities): return True return False拼装上下文时,格式很重要。我习惯用带标注的块结构:
def inject(self, query, memories): if not memories: return query memory_block = "\n".join([ f"[历史记忆|{m['type']}|{format_time(m['timestamp'])}] {m['content']}" for m in memories ]) return f"""以下是与当前问题相关的历史信息,供参考: {memory_block} 当前问题:{query}"""标注类型和时间是为了让模型区分"这是历史"和"这是现在"。实测下来,不加标注时模型有约15%的概率把历史偏好当成当前指令执行,加了标注后这个比例降到3%以下。
4. 实测中暴露的三个真问题与应对
4.1 记忆污染:错误信息一旦入库就反复被召回
这是长期记忆最阴险的坑。假设某轮对话里用户开玩笑说"我们准备把数据库换成MongoDB",抽取器忠实地把它存成了fact_preference。之后每次检索都会把这条捞出来,模型就会持续认为用户在用MongoDB。更糟的是,如果这条错误记忆被召回后模型又基于它生成了新内容,新内容可能又被抽取成记忆,形成错误放大循环。
应对办法有三层。第一层是来源标记:抽取时记录这条记忆是"用户明确声明"还是"模型推断",前者置信度高,后者低。第二层是冲突检测:新记忆入库时,如果和已有记忆在同一实体上矛盾(比如数据库类型从PostgreSQL变成MongoDB),不要直接覆盖,而是标记旧条目为"可能过期",检索时降权。第三层是人工纠错入口:给用户一个"这条记错了"的反馈通道,反馈后直接删除或修正对应条目。三层里第一层最容易做,第三层最有效但需要产品配合。
4.2 时效衰减:三个月前的偏好还该不该信
记忆不是越全越好,过期的记忆比没有记忆更危险。用户三个月前说"我们主要用Python",现在团队可能已经转Go了,但检索时这条旧记忆的语义相似度依然很高,会被召回并误导模型。
DeerFlow的思路是给记忆加时效权重,不同类型的记忆衰减速度不同:
| 记忆类型 | 半衰期 | 说明 |
|---|---|---|
| fact_preference | 90天 | 技术栈类偏好变化慢 |
| task_state | 7天 | 任务状态变化快 |
| entity_relation | 180天 | 实体关系相对稳定 |
检索排序时把时效权重乘进去,超过半衰期两倍的记忆直接不召回。这个策略的代价是可能漏掉一些"虽然旧但仍然有效"的记忆,但相比被过期信息误导,漏召回的代价更小。我个人的经验是宁可少召回,也不要召回错的。
4.3 检索噪声:Top5里三条不相关怎么办
召回数量设成5是个经验值,但实际场景里经常出现"5条里只有2条相关"的情况。这时候如果把5条全塞进上下文,噪声会稀释信号。解决办法是动态截断:按final_score排序后,从高分往低分累加,当某条得分低于最高分的60%时就截断,不再往下取。
def dynamic_truncate(self, memories, ratio=0.6): if not memories: return [] memories.sort(key=lambda m: m["final_score"], reverse=True) top_score = memories[0]["final_score"] result = [] for m in memories: if m["final_score"] < top_score * ratio: break result.append(m) return result这个0.6的比值也是调出来的。设0.8太严,经常只剩一条;设0.4太松,噪声又进来了。0.6在多数场景下能保留2到3条高质量记忆,是个比较稳的默认值。
5. 二次开发时值得提前想清楚的几件事
5.1 记忆的粒度:一条记忆装多少信息
粒度太粗,一条记忆塞进整段对话,检索时相关性被稀释;粒度太细,一句话拆成五条,检索时又得拼半天。我的经验是一条记忆对应一个独立事实,判断标准是"这条信息能否独立成立、独立被引用"。"用户用Go"是一条,"用户用PostgreSQL"是另一条,不该合并成"用户用Go和PostgreSQL"。合并后如果只问数据库相关,整条记忆的相关性得分会被Go那部分拉低。
但也不能无限细。像"部署在K8s上,日志走ELK"这种紧密关联的部署信息,拆开反而丢失了上下文。实践中我会把同一实体同一维度的信息合并,不同维度拆开。这个规则听起来抽象,落到代码里就是抽取prompt里加一句"同一实体的同类属性合并为一条,不同属性分开输出"。
5.2 冷启动:没有历史记忆时怎么过渡
新用户进来时记忆库是空的,检索永远返回空,长期记忆形同虚设。这时候需要冷启动策略:前几轮对话不做检索,但抽取照常进行,快速把库填起来。等积累到一定条目数(比如10条)再开启检索。这样用户感知上是从第三四轮开始"助手变聪明了",而不是一上来就期待它有记忆。
另一个冷启动技巧是预置领域记忆。如果产品面向特定行业,可以预置一批通用事实(比如"K8s是容器编排系统"),让检索在早期就有内容可召回。但预置记忆要标记来源为"系统预置",置信度设低一些,避免和用户真实信息冲突时抢占排序。
5.3 可观测性:怎么知道记忆系统在正常工作
记忆系统最大的问题是它坏了你不容易发现。检索返回空、返回错、返回一堆噪声,从用户视角看都只是"回答质量下降",很难定位到是记忆环节出了问题。所以可观测性必须提前埋点。
我通常会记录这几个指标:抽取命中率(多少轮对话抽出了记忆)、检索触发率(多少轮触发了检索)、召回相关性(召回条目和当前问题的实际相关比例,可以抽样人工标注)、注入后效果(注入记忆的回答和未注入的回答质量对比)。这几个指标里,召回相关性最关键,建议每周抽样50条人工看一眼,比任何自动指标都靠谱。
提示:调试阶段可以把每次检索的query、召回条目、final_score都打到日志里,出问题时直接翻日志比猜快得多。上线后再把日志级别调低,避免存储成本失控。
6. 我在实际项目里踩过的两个具体坑
第一个坑是向量维度和模型不匹配。早期我用A模型生成embedding,后来换了B模型做对话,忘了embedding也得跟着换,结果新旧向量在同一个空间里根本不可比,检索结果乱七八糟。排查了两天才发现是维度对不上。教训是embedding模型和对话模型要作为一个整体版本管理,换一个就得考虑另一个。
第二个坑是并发写入导致的重复记忆。多轮对话并发抽取时,两条相同内容的记忆可能同时通过去重检查(因为彼此还没写入),结果库里出现重复条目。解决办法是在存储层加唯一约束,或者用写入锁串行化去重检查。这个坑在低并发时根本遇不到,一上量就暴露,属于典型的"测试环境永远正常"问题。
这两个坑的共同点是:它们都不在算法层面,而在工程层面。长期记忆方案的技术难点,一半在抽取和检索的策略设计,另一半在存储、并发、版本管理这些"不性感"的工程细节上。很多人把精力全花在调prompt上,结果被工程问题拖垮,这个优先级得摆正。
最后分享一个我常用的验证方法:拿一段真实的多轮对话,手动标注出"哪些信息应该被记住",然后跑一遍你的记忆系统,对比抽取结果和标注的差异。这个对比能同时暴露抽取漏召、误召、粒度不当三类问题,比单看指标直观得多。我每次调整抽取prompt后都会跑一遍这个对比,基本能覆盖八成以上的回归问题。