DeerFlow长期记忆全链路解析：从提取到注入的工程实践

1. 先理清楚：DeerFlow的长期记忆，到底在解决什么问题

如果你做过AI Agent的应用，一定见过这类场景：用户上午跟智能体说"我喜欢简洁的回复风格，不要铺垫"，下午再问"帮我写个晨会纪要"，模型完全忘了上午的偏好，照样给你输出一大段客套话。

这不是某一个模型的问题，而是AI Agent的普遍困境：默认情况下，模型本身是无状态的。每一次对话对模型来说都是一次全新的开始，除非你在上下文中显式带上历史信息。短期会话里还能靠"把所有聊天记录拼进prompt"解决，可一旦会话跨天、跨周，或者涉及多个用户，历史记录越攒越长，要么爆token上限，要么把Agent拖到响应超时。

所以长期记忆，本质上是给Agent装了一个"外置大脑"：把值得记住的信息从对话流里抽出来，结构化存档，下次对话时把相关部分检索出来，再注入到模型上下文里。DeerFlow是字节跳动开源的一个AI Agent框架，它把这件事做到了框架层面，而不是让每个开发者自己去拼RAG。我最近在一个内部工具项目里用了DeerFlow做智能体二次开发，从封装SSE流式接口到处理长期记忆写入，整个过程踩了不少坑，也把它的实现思路摸了个七七八八。

这篇文章不打算复述官方文档，我会用一个贯穿全文的业务示例，把DeerFlow长期记忆从提取、写入、更新到检索、注入的全链路拆给你看。代码基于我实际部署的DeerFlow 2.0版本整理，不同小版本的API可能有微调，但核心设计是一致的。

1.1 无状态对话的痛点

先看一个最直观的问题。假设你给团队做了一个"周报助手"Agent，用户在周一说过"我是前端组的老王，每周五下午提交周报，周报里需要包含风险项和下周计划"。到了周四，他直接说"帮我起草这周的周报"。

没有长期记忆的Agent，此时问号脸：他是谁？周报模板是什么？风险项要写在哪？你不得不让用户重新把所有背景描述一遍。更麻烦的是，如果这个Agent服务多个团队，你甚至不知道该调哪套周报模板。

短期方案是"把所有聊天记录拼进上下文"。这在单轮内有效，但跨会话之后就失效了。你总不能把所有用户的所有历史对话都塞给模型，那既不经济也不可行。这也是为什么DeerFlow选择"抽取式记忆"而不是"全量上下文"：它只保存被判定为有长期价值的信息片段，而不是把流水账原文都存下来。

1.2 DeerFlow长期记忆的两类核心载体

从存储层面看，DeerFlow的长期记忆不是单一实现，而是分了两类配合使用。一类是结构化记忆，存到SQLite里，记录用户偏好、任务状态、明确说过的事实，字段清晰，按key查询很快。另一类是语义记忆，内容先过embedding模型转成向量，存进向量库，检索时按相似度召回。这两种各有分工：结构化记忆回答"这个用户叫什么、他偏好什么风格"这种明确问题；语义记忆回答"之前有没有聊过和这个主题类似的内容"这种模糊问题。

这个设计和人的记忆机制很像。你记得同事的手机号（结构化），也记得"上次开会讨论过预算问题的那个氛围"（语义化）。DeerFlow把两者做成了统一接口，对上层Agent来说，只需要调用agent.memory.write和agent.memory.search，至于底层走SQLite还是向量库，由框架根据内容自动判断。

2. 一个贯穿全文的业务示例：让它记住"你是谁"

为了讲清楚实现，我先固定一个场景，后面所有代码都围绕它展开。

场景：做一个面向内容团队的"选题助手"Agent。它需要长期记住三件事：用户身份和偏好、历史选题记录、团队的选题规范。我给它起了个名字叫topic-bot。

这个例子选得好，是因为它同时覆盖了长期记忆的三种典型形态：

2.1 示例场景设定

具体来说，这个Agent要支持以下对话流：

• 用户首次使用："我叫小杨，科技组编辑，平时写AI方向选题，喜欢3000字左右的长文。"
• 几小时后再次登录："帮我看看我们组这周还有什么好选题。"
• Agent要能回忆起小杨的身份，并结合他之前提过的偏好给出选题建议。
• 如果小杨说"这个方向写过了"，Agent要能更新记忆：这个选题已经被处理，避免下次再推。

你会发现，这里面的关键不是"模型聪明不聪明"，而是"记忆系统能不能把这些信息可靠地存下来、按需取出来"。模型能力再强，记不住就是记不住。

2.2 没有长期记忆时的表现

我先跑了一下不加记忆的基线版本。第一次对话还正常，小杨把背景信息说完，Agent给出了几个选题方向。第二次对话，我故意隔了半小时再问"我们组这周还有什么好选题"，Agent完全不知道"我们组"是哪个组，也不知道小杨是科技组的，甚至反问"请问你的团队是做什么的"。

这就是很多Agent项目上线的真实状态：单次对话演示效果很好，一进入真实业务就露馅。用户没有耐心每天把背景信息重新讲一遍，产品留存自然上不去。可以说，长期记忆不是"加分项"，而是Agent从demo走向生产的必要能力。

2.3 开启长期记忆后的表现

开启DeerFlow长期记忆后，同一段对话的表现明显不同。第二次提问时，Agent的回复里直接带出了小杨的身份信息，还结合记忆里"AI方向""3000字深度稿"这些历史记录给出了候选选题。我特意在测试里验证了一次记忆更新：让小杨说"区块链那个选题已经写过了，别再推荐了"，之后再问相关方向时，该选题确实不再出现。

这个结果看起来不稀奇，但背后的实现链路值得拆开看。下面两节分别讲写入链路和读取链路，这是DeerFlow长期记忆的核心。

3. 记忆写入链路：怎么从对话里"提炼"出值得记住的信息

写入链路是长期记忆最关键的一环，也是最容易做砸的一环。很多自研记忆系统死在两个字上：一个是"杂"，什么对话都往记忆里塞，导致记忆库里全是垃圾；另一个是"漏"，该记住的没记住。DeerFlow在这块的设计思路是：让大模型自己决定哪些信息值得存，而不是靠规则硬匹配。

3.1 记忆提取的判断标准

每次对话结束后，DeerFlow会调用一次提取模型，把当前这轮对话的关键信息抽出来。它判定"值得存"的依据大致有三条：

• 主体确定性：信息是否涉及明确的用户事实，比如姓名、职业、偏好、约束条件。
• 时间持久性：这个信息在未来几周、几个月内是否还有效。比如"我今天心情不好"就不适合进长期记忆，而"我偏好深度稿"就适合。
• 决策关联性：是否影响后续任务执行，比如"确定选题不再推荐某方向"这类业务状态。

判断逻辑本身是靠提示词实现的。DeerFlow会构造一段提取指令，要求模型输出JSON格式的记忆条目，再经过校验后写入存储。我在二次开发时调整过这段提取逻辑，加了一条规则：涉及"已写过""已排除""不要推荐"这类否定性描述时，强制走记忆更新路径而不是新增路径，避免新旧记忆打架。

3.2 写入流程与数据结构

写入时，记忆条目会带上几个关键元数据字段，我用实际调试过的结构举例：

{ "memory_id": "mem_8f3a2c91", "user_id": "user_xiaoyang", "content": "小杨是科技组编辑，偏好3000字左右深度稿，研究方向为AI Agent应用", "memory_type": "user_fact", "metadata": { "source_turn": "session_20250312_004", "confidence": 0.92, "last_accessed": "2025-03-12T15:04:22+08:00" } }

写入流程分四步：

1. 对话结束事件触发提取模型，生成候选记忆条目。
2. 候选条目经过dedup检查：查询已有记忆中是否存在语义相似度超过阈值的记录。
3. 如果存在相似记录，走合并更新路径，保留最新信息并累加置信度。
4. 如果不存在，写入新记录，并同步生成向量索引。

你可能注意到confidence这个字段。我一开始没在意它，后来发现它很有用：当某条记忆被多次对话反复印证，置信度会上升，检索权重也会提高；如果一条记忆很久没被访问，置信度会随时间衰减，最终被清理。这相当于给记忆系统加了一个"新陈代谢"机制，避免了废旧记忆无限堆积。

3.3 记忆更新与冲突处理

记忆更新是很多初学者容易忽略的坑。举个例子：小杨第一次说"我是科技组编辑"，第二次说"我转岗到产品组了"。如果你只做新增不做更新，记忆库里就会同时存在两条矛盾事实，检索时模型不知道该信哪条。

DeerFlow的做法是，在写入前执行冲突检测，对同一user_id下涉及同一主体的记忆做一致性判断。实现上并不复杂，核心逻辑是：

• 对结构化记忆，按subject字段分组，同主体下内容变更时，旧记录标记为superseded，不物理删除，保留审计轨迹。
• 对语义记忆，靠向量相似度找相关旧记录，由模型判断是补充还是替代。

值得说明的是，DeerFlow并没有把冲突判断做得特别重，它更倾向于"让模型判断，用元数据兜底"。我在二次开发时遇到过这么个问题：小杨的岗位信息更新后，旧记忆虽然被标记过期，但语义检索时偶尔还会被召回。后来我在检索函数里加了一个过滤器，默认排除status=superseded的记录，这个问题就消失了。这种小改动很常见，相当于给框架做定制，而DeerFlow的数据结构设计给这类定制留了充分空间。

4. 记忆读取链路：检索、排序、注入

写入做得再好，读取端不给力，记忆也发挥不了作用。读取链路解决的问题是：面对一堆记忆，怎么在每次对话前选出最相关的几条，塞进模型上下文。

4.1 检索策略：混合召回

DeerFlow默认做的是混合召回：结构化记忆按key直接查，语义记忆走向量相似度。举个例子，用户问"我们组这周还有什么好选题"，这个query会同时触发两条检索路径：

• 结构化路径：user_id=user_xiaoyang查他的偏好记录，命中"科技组""深度稿"这些字段。
• 语义路径：把query转成向量，召回历史选题记录中语义接近的条目，比如"3月已确定选题"里和"AI方向"相关的记录。

两条路径的结果合并后再统一排序。这个设计好理解：结构化检索精确但覆盖面窄，语义检索宽泛但容易召回噪音，两者互补。

4.2 排序与筛选：控制注入质量

合并后的候选记忆不是全塞进上下文，而是要经过排序和筛选。我调试时看到，DeerFlow的排序因子主要有三个：

• 相关度：语义相似度或结构匹配度，这是最重要的因子。
• 时效性：last_accessed越近的记录权重越高，这与人类记忆规律一致。
• 置信度：被反复印证过的记忆优先。

实际项目中，我用过一个比较粗暴的调参方式：把相关度的权重调高，同时把召回条数上限设为6条。因为从我自己的测试经验看，超过6条记忆注入后，模型的回复质量不升反降，反而开始混淆一些不相关的细节。这个数字没有普适性，跟你的业务复杂度相关，建议自己跑几组对比测试找到合适的值。

4.3 上下文注入与token控制

注入环节有一个容易忽略的点：记忆不是简单拼在用户消息前面。DeerFlow会在系统提示词里专门划分一个memory_context区域，用明确的标记提示模型"以下是该用户的历史记忆，可供参考，但不要直接复述"。这样做的原因是，如果记忆和用户消息混在一起，模型容易把记忆内容误当成用户当前指令，产生幻觉。

另一个工程细节是token控制。DeerFlow在注入前会估算记忆片段的token长度，超出预算时优先丢弃低相关度条目。我实测过，长对话场景下，合理的记忆注入预算大约占整体上下文窗口的15%-20%。你觉得模型"突然忘了"，很多时候不是真忘了，而是注入的记忆被截断或排序掉了，排查时先看这里。

5. 结合SSE流式输出的完整落地代码

理论链路清楚了，接下来是代码实操。这一节我用的是DeerFlow 2.0的调用方式，重点讲怎么把长期记忆和SSE流式接口封装到一起。这也是社区里问得最多的问题：记忆系统跑通了，但一上流式就各种问题。

5.1 环境准备与初始化

先准备环境。我本地的Python是3.10，用uv管理依赖，安装DeerFlow的客户端包后初始化：

# requirements.txt 核心依赖 # deerflow>=2.0.0 # httpx>=0.27 # openai>=1.30 # 用于embedding与对话模型

初始化Agent配置，这里关键是把记忆系统打开，并指定存储路径：

from deerflow import DeerFlowAgent, MemoryConfig agent = DeerFlowAgent( api_key=os.getenv("DEEPSEEK_API_KEY"), model="deepseek-chat", memory=MemoryConfig( enabled=True, storage_type="sqlite", # 本地SQLite vector_store="lancedb", # 向量库 db_path="./data/topic_bot_memory.db", embedding_model="bge-large-zh", # 中文场景推荐 max_memory_entries=8, # 每次注入的条目上限 ), )

这里我用的模型是DeepSeek，因为内网部署方便，DeerFlow也兼容OpenAI接口格式。如果你用的是别的模型，只要走标准接口就行。bge-large-zh是中文embedding里性价比不错的方案，英文场景可以换text-embedding-3-small。

5.2 核心调用逻辑封装

DeerFlow原生支持流式返回，但直接裸调在业务代码里会很乱。我的做法是封装一个统一的stream_chat函数，内部处理好：拼接历史、触发记忆检索、发起SSE请求、解析事件流。这一步对应热搜词里说的"封装SSE流式接口调用逻辑"。

import json import httpx class TopicBotClient: def __init__(self, agent): self.agent = agent async def stream_chat(self, user_id: str, message: str): """ 封装SSE流式调用： 1. 注入长期记忆 2. 发起流式请求 3. 按事件类型解析返回 """ # 1. 检索并注入长期记忆 memory = self.agent.memory.search(user_id=user_id, query=message) system_prompt = self.agent.build_prompt( memory_context=memory ) # 2. 发起流式请求 payload = { "model": self.agent.model, "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": message}, ], "stream": True, } headers = {"Authorization": f"Bearer {self.agent.api_key}"} async with httpx.AsyncClient(timeout=60) as client: async with client.stream( "POST", f"{self.agent.base_url}/chat/completions", json=payload, headers=headers, ) as resp: async for line in resp.aiter_lines(): event = self._parse_sse_line(line) if event: yield event def _parse_sse_line(self, line: str): """SSE流式消息解析""" if not line.startswith("data:"): return None data = line[5:].strip() if data == "[DONE]": return {"type": "done"} try: raw = json.loads(data) except json.JSONDecodeError: return None # DeerFlow在流式返回中会穿插记忆写入事件 event_type = raw.get("type", "message") if event_type == "agent.memory_action": return {"type": "memory", "content": raw.get("memory")} if event_type == "agent.message": return {"type": "message", "content": raw.get("content", "")} if event_type == "agent.tool_call": return {"type": "tool_call", "content": raw.get("tool_name")} return {"type": "unknown", "content": raw}

这段代码有三个关键点。第一，memory.search放在构造payload之前，确保用户的历史记忆在请求发出前就注入。第二，SSE解析时不只处理message类型，还处理agent.memory_action事件，因为DeerFlow会在流式返回的中途触发异步记忆写入，不解析的话会把JSON字符串直接吐给前端。第三，事件类型判断要放在[DONE]判断之后，顺序错了容易漏掉结束标记。

5.3 流式消息解析与前端对接

前端对接时，服务端用的是SSE协议，核心是text/event-stream。我在FastAPI里把上面的client暴露为接口：

from fastapi import FastAPI from fastapi.responses import StreamingResponse app = FastAPI() client = TopicBotClient(agent) @app.post("/api/chat") async def chat(user_id: str, message: str): async def event_generator(): async for event in client.stream_chat(user_id, message): if event["type"] == "message": # 只把正常消息文本透传给前端 yield f"data: {json.dumps({'content': event['content']})}\n\n" elif event["type"] == "memory": # 记忆写入事件在这里做本地标记，不直接透传 print(f"[memory update] {event['content']}") elif event["type"] == "done": yield "data: [DONE]\n\n" return StreamingResponse(event_generator(), media_type="text/event-stream")

有一个我在实际中踩过的坑：刚开始我把agent.memory_action事件也透传给了前端，结果前端把类似{"storage": "sqlite"}的JSON渲染到了聊天界面里，用户看到一堆乱码。后来改成在后端消费记忆事件，前端只收message和DONE，问题就干净了。这个经验值得记下来：SSE流里的控制事件和业务事件要分层，不是所有事件都适合推给前端。

如果你不是用DeerFlow自带的流式协议，而是要对接外部大模型API，那_parse_sse_line里的解析逻辑就需要适配对方的协议格式。但总体思路一致：一个方法负责建连、一个方法负责逐行解析、一个方法负责事件分发。

6. 可观测性与人机协同：记忆系统不能是黑盒

话题再拉高一层。光把读写链路跑通不算完，生产环境里你还得能回答两个问题：Agent到底记住了什么？记错了怎么办？这对应热搜词里"deerflow可观测"和"deerflow人机协同"两个方向。

6.1 记忆可视化与调试

DeerFlow提供了一套可观测的调试面板，我实际用过之后觉得最有价值的是这几个视图：

• 记忆列表视图：按用户维度展示所有记忆条目，能看到内容、类型、置信度、最后访问时间。
• 检索命中视图：每次对话结束后，展示本次检索召回了哪些记忆、排序结果如何。
• 写入审计视图：记录每次写入的来源对话，方便回溯"这条记忆是哪句话产生的"。

这些视图对应的底层数据都来自前面提到的元数据字段。source_turn记录来源于哪一轮对话，last_accessed记录最后使用时间，confidence记录置信度。我自己调试时发现，检索命中视图是最有用的排查工具：当用户反馈"Agent怎么不记得我说过什么"，第一件事不是去改prompt，而是去查检索环节有没有把对应记忆召回出来。如果记忆在库里但没被召回，那基本是embedding相似度阈值或者排序权重的问题。

6.2 人工修正与记忆管理

人机协同，在记忆系统里的体现是"人可以对记忆进行干预"。DeerFlow允许运营者或用户在面板里直接做三件事：

• 锁定（pin）：某些关键记忆，比如合规要求、用户明确的身份信息，设为不可被衰减清理。
• 删除（remove）：用户明确要求"忘掉这条"，系统立即删除相关记忆和向量索引。
• 修正（edit）：直接改记忆内容，框架会同步更新元数据并重新生成向量。

第三个功能对内容运营场景特别重要。如果Agent记错了一个事实，比如把小杨的岗位记成"科技组编辑"，而实际上他转岗了，运营人员直接在面板修正这条记忆，远比让用户反复纠正来得快。我在项目里还加了一个"记忆确认"流程：当Agent对某个新记忆的置信度低于阈值时，会在回复末尾追加一句"我记下了你的偏好，如果不对可以纠正我"。这算是人机协同的一种轻量形态，实测用户接受度不错，也降低了错误记忆的负面影响。

6.3 DeerFlow 2.0在记忆方面的改进

最后聊一下DeerFlow 2.0的变化。我在社区看到不少讨论，结合自己升级后的体感，最明显的几点是：记忆合并策略更智能了，不再只是简单的覆盖写，而是能识别"同一主题下的渐进式补充"；检索时引入了一步重排，会根据当前对话目标对候选记忆重新打分；另外2.0的流式协议更规范了，事件类型从字符串改成了带namespace的枚举，这对接入方的类型安全更友好。

不过升级也有成本。我升级到2.0后，原先写的一些针对旧版事件类型的解析代码全部要改。如果你在二次开发中深度依赖了内部事件结构，建议先在小流量上灰度，不要一把梭。这是所有框架升级的通病，DeerFlow在这个阶段版本迭代快，锁定版本号做升级测试是个好习惯。

7. 实战中遇到的坑与排查方法

这一节把我实际踩过、或者看社区同学踩过的问题汇总一下，做成一个速查表。每个坑都附带排查思路，比盲目调参高效。

7.1 记忆检索不准的深层排查

检索不准确是最难排查的一类问题，因为表面上看代码都跑通了，但结果就是不对。我遇到过一个典型场景：用户说"帮我找一个之前聊过的关于内容生产的记忆"，但Agent没有召回。查检索视图发现，那条记忆的向量相似度只有0.71，而阈值是0.75，刚好被卡掉了。

这种边缘情况靠调阈值解决不了根本问题。更有效的做法是：第一，检查embedding模型的输入是否有截断，超长文本截断后会丢失语义；第二，对记忆条目做规范化，把"科技组""技术编辑"这类同义表达在写入时就归一化；第三，在检索时做关键词兜底，如果向量召回结果为空，退化到SQLite的LIKE查询，保证至少有个结果。

7.2 记忆写入过多过杂的治理

另一个让我头疼的问题是记忆污染。项目上线一周后，记忆库超过了两千条，其中一半是"用户今天问了XXX"这种价值极低的记录。原因是我把提取模型的prompt写得太宽松，它把每轮对话的摘要都当成了记忆。

后面我加了两道闸门：一道是规则过滤，长度过短、不包含任何实体的内容直接丢弃；另一道是二次确认，让提取模型对每条候选记忆打一个retention_score，低于0.6就不入库。这两道闸门加完，记忆库一周只增长了三百条有效记录，检索质量明显提升。

7.3 流式输出与记忆写入的时序问题

最后提一个和本文主题强相关的问题：流式场景下，记忆写入的时机该怎么安排。刚开始我把记忆写入放在"用户消息进来之后、模型回复之前"执行，结果每次对话首字返回都慢了两秒。后来改成在模型回复结束、SSE流关闭之后异步执行写入，首字延迟问题解决，但带来了新的问题：如果用户在流式回复过程中就发来了新消息，上一轮的记忆还没写入，这轮就检索不到。

DeerFlow的agent.memory_action事件正是为了解决这个时序问题设计的：它在回复流中可以动态触发记忆更新，而不是等整轮结束。但这个能力需要接入方正确解析事件并处理异步写入。我的建议是：对记忆实时性要求高的业务，采用"流中写入"模式，即解析到memory_action事件时立即执行写入；对实时性要求不高的业务，保持结束后写入即可，实现更简单不容易出错。

8. 最后分享一点我个人实操中的体会

如果你正准备基于DeerFlow做智能体二次开发，我的建议是你先别急着写业务代码，花两天时间把记忆系统的读写链路和数据模型摸透。DeerFlow的设计把长期记忆做得足够通用，但再通用的框架也要适配你的业务：什么样的信息值得记、多少条记忆注入合适、哪些记忆需要人工干预，这些参数只有在你自己的数据上跑过才能定下来。

我在这个选题助手项目里最大的感受是：长期记忆不是一个独立模块，它和流式输出、可观测性、人工审核都耦合在一起。单独把记忆读写跑通不算难，难的是让它在真实业务里稳定运转、不被垃圾数据污染、不影响响应延迟。这也是为什么我在这篇文章里花了大量篇幅讲排查方法而不是只贴代码——代码只是表象，对记忆系统的理解才是你在生产环境里真正用得上的东西。

后续我还在考虑给这个Agent加上记忆导出和跨用户迁移的能力：比如把一组记忆从一个Agent实例复制到另一个，配合团队模板快速搭建新选题助手。DeerFlow的数据结构设计上是有这个扩展空间的，等我把这一版跑稳了，再回来分享具体做法。