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。这个例子选得好是因为它同时覆盖了长期记忆的三种典型形态记忆类型示例内容存储方式用户事实小杨是科技组的编辑偏好3000字深度稿不喜欢标题党结构化存储SQLite业务状态2025年3月已确定选题DeerFlow二次开发实战、AI可观测性工具对比结构化存储向量索引历史经验上次写Agent相关选题时读者反馈偏重原理的阅读量高于纯工具教程语义记忆向量库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:2208:00 } }写入流程分四步对话结束事件触发提取模型生成候选记忆条目。候选条目经过dedup检查查询已有记忆中是否存在语义相似度超过阈值的记录。如果存在相似记录走合并更新路径保留最新信息并累加置信度。如果不存在写入新记录并同步生成向量索引。你可能注意到confidence这个字段。我一开始没在意它后来发现它很有用当某条记忆被多次对话反复印证置信度会上升检索权重也会提高如果一条记忆很久没被访问置信度会随时间衰减最终被清理。这相当于给记忆系统加了一个新陈代谢机制避免了废旧记忆无限堆积。3.3 记忆更新与冲突处理记忆更新是很多初学者容易忽略的坑。举个例子小杨第一次说我是科技组编辑第二次说我转岗到产品组了。如果你只做新增不做更新记忆库里就会同时存在两条矛盾事实检索时模型不知道该信哪条。DeerFlow的做法是在写入前执行冲突检测对同一user_id下涉及同一主体的记忆做一致性判断。实现上并不复杂核心逻辑是对结构化记忆按subject字段分组同主体下内容变更时旧记录标记为superseded不物理删除保留审计轨迹。对语义记忆靠向量相似度找相关旧记录由模型判断是补充还是替代。值得说明的是DeerFlow并没有把冲突判断做得特别重它更倾向于让模型判断用元数据兜底。我在二次开发时遇到过这么个问题小杨的岗位信息更新后旧记忆虽然被标记过期但语义检索时偶尔还会被召回。后来我在检索函数里加了一个过滤器默认排除statussuperseded的记录这个问题就消失了。这种小改动很常见相当于给框架做定制而DeerFlow的数据结构设计给这类定制留了充分空间。4. 记忆读取链路检索、排序、注入写入做得再好读取端不给力记忆也发挥不了作用。读取链路解决的问题是面对一堆记忆怎么在每次对话前选出最相关的几条塞进模型上下文。4.1 检索策略混合召回DeerFlow默认做的是混合召回结构化记忆按key直接查语义记忆走向量相似度。举个例子用户问我们组这周还有什么好选题这个query会同时触发两条检索路径结构化路径user_iduser_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 核心依赖 # deerflow2.0.0 # httpx0.27 # openai1.30 # 用于embedding与对话模型初始化Agent配置这里关键是把记忆系统打开并指定存储路径from deerflow import DeerFlowAgent, MemoryConfig agent DeerFlowAgent( api_keyos.getenv(DEEPSEEK_API_KEY), modeldeepseek-chat, memoryMemoryConfig( enabledTrue, storage_typesqlite, # 本地SQLite vector_storelancedb, # 向量库 db_path./data/topic_bot_memory.db, embedding_modelbge-large-zh, # 中文场景推荐 max_memory_entries8, # 每次注入的条目上限 ), )这里我用的模型是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_iduser_id, querymessage) system_prompt self.agent.build_prompt( memory_contextmemory ) # 2. 发起流式请求 payload { model: self.agent.model, messages: [ {role: system, content: system_prompt}, {role: user, content: message}, ], stream: True, } headers {Authorization: fBearer {self.agent.api_key}} async with httpx.AsyncClient(timeout60) as client: async with client.stream( POST, f{self.agent.base_url}/chat/completions, jsonpayload, headersheaders, ) 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 fdata: {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_typetext/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. 实战中遇到的坑与排查方法这一节把我实际踩过、或者看社区同学踩过的问题汇总一下做成一个速查表。每个坑都附带排查思路比盲目调参高效。问题现象可能原因排查与解决Agent在隔天对话中完全不记得用户记忆未写入或检索未命中先看写入审计视图确认是否有记录再看检索命中视图确认相似度阈值是否过高记忆库里全是废话记忆提取模型的判断标准太宽松调整提取提示词增加是否对未来对话有持久价值的判定要求新旧记忆互相矛盾冲突检测未生效检查是否有superseded记录被召回给检索函数加状态过滤注入记忆后回复质量反而下降注入条数过多或相关性排序不对调低max_memory_entries对比5条、8条、12条的回复效果SSE流式返回中断未处理agent.memory_action事件导致解析异常确认解析器覆盖所有事件类型控制事件在后端消费记忆写入延迟影响首字返回记忆写入在回复前同步执行把写入改为异步或先在流式结束后再执行写入向量检索召回不准embedding模型与业务领域不匹配中文业务换bge-large-zh英文换专用模型并测试分块长度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的数据结构设计上是有这个扩展空间的等我把这一版跑稳了再回来分享具体做法。