企业官网建设流程全解析

1. 从一次多轮对话的“失忆”说起

前阵子我在做一个基于智能体框架的客服助手二次开发，需求很朴素：用户先说自己买的是哪款设备，再问“它的保修期多久”，助手得能接得住。结果第一版跑下来，助手每次回复都像第一次见面，用户上一句说的设备型号，下一句就忘得干干净净。这个“失忆”问题，本质上就是长期记忆没做对。

DeerFlow 是字节跳动开源的一个深度研究类智能体框架，它把“人机协同”和“多轮研究”这两件事串了起来。很多人第一次看它的代码，会被那一堆节点、状态、检查点绕晕，但如果只盯着一个目标——它到底怎么让智能体记住跨轮次、跨会话的信息——整条链路反而会清晰很多。这篇我就用一个具体示例，把 DeerFlow 的长期记忆实现方案拆开讲透，顺带把可观测、流式接口、二次开发这些周边问题一起聊了。适合正在做智能体二次开发、被多轮上下文折磨过的同学，也适合想理解“记忆”这件事在工程上到底怎么落地的人。

先说结论方向：DeerFlow 的记忆不是靠一个“大数组”硬塞，而是分层的——短期靠状态图里的消息通道，长期靠可持久化的存储与检索，中间用检查点做衔接。下面我按这个思路一层层剥。

2. DeerFlow 记忆方案的整体设计与选型考量

2.1 为什么不是“把所有历史都塞进 Prompt”

新手最容易想到的方案，就是把历史对话拼成一个超长字符串，每次请求都带上。我早期也这么干过，实测下来有三个致命问题。第一是成本，上下文越长，token 消耗越夸张，多轮下来账单很难看。第二是注意力稀释，模型对超长上下文中间部分的召回率会下降，前面说的设备型号可能被淹没。第三是无法跨会话，用户关掉页面再回来，历史就没了。

DeerFlow 走的是另一条路：它把“当前这轮需要什么”和“历史上存过什么”分开处理。当前轮次的消息走状态通道，历史信息走存储层，需要的时候再检索回来。这个拆分是整个方案的地基，理解了它，后面所有代码都好读。

2.2 分层记忆的核心结构

我把 DeerFlow 的记忆相关部分归纳成三层，方便记忆：

• 会话内短期记忆：基于状态图的消息列表，保存当前会话的对话轮次，随图执行流动。
• 跨会话长期记忆：落到外部存储（如向量库或键值存储），按用户或会话维度归档。
• 检查点与恢复：把图执行到某一步的状态快照存下来，支持中断续跑和回溯。

这三层不是孤立的。短期记忆在会话结束时，会被“提炼”成长期记忆写入存储；下次会话开始时，再按需检索回短期通道。检查点则保证这个过程中间断了也能接上。

2.3 选型背后的取舍逻辑

为什么用状态图而不是简单的列表？因为智能体的执行是有分支的——它可能先检索、再推理、再调用工具、再反思。状态图能把每一步的输入输出都挂到统一的 state 上，记忆自然就跟着 state 走。为什么长期记忆要用检索而不是全量加载？因为长期记忆会无限增长，全量加载既慢又贵，检索能只取相关片段。

提示：如果你只是做单轮问答，这套分层是过度设计；但只要涉及多轮、跨会话、人机协同，分层几乎是必选项。

3. 核心细节解析：记忆到底存在哪、怎么取

3.1 状态通道里的消息是怎么累积的

DeerFlow 的状态定义里，消息字段通常用带“追加”语义的 reducer。普通字段是覆盖式更新，新值直接替换旧值；而消息字段是累加式，每一步产出的新消息会被 append 到列表尾部。这个设计很关键——它保证了多轮对话的历史不会被后一步覆盖掉。

我踩过的一个坑：早期我自己写状态时用了普通字段存消息，结果每次节点执行完，历史只剩最后一条。排查了半天才反应过来是 reducer 的问题。所以你在二次开发时，凡是需要“累积”的字段，一定要确认它的 reducer 是追加而不是覆盖。

3.2 长期记忆的写入时机

长期记忆不是每轮都写，那样太频繁。常见做法是在会话结束或达到一定轮次时触发写入。写入前一般会做一次“提炼”：把冗长的对话压缩成结构化的要点，比如“用户设备型号：X200”“用户偏好：简洁回复”。提炼的好处是存储体积小、检索命中率高。

这里有个经验：提炼用的提示词要明确要求输出键值对或短句，不要让模型自由发挥写一大段。我试过让模型自由总结，结果存进去的全是废话，检索时根本匹配不上。

3.3 检索回来的记忆怎么用

检索到长期记忆后，不能直接一股脑塞进 Prompt，得做相关性过滤和排序。DeerFlow 这类框架通常会给检索结果打分，只取 top-k。然后把这些片段作为“背景信息”注入到当前轮的上下文里，和当前消息区分开。

我一般会把注入格式设计成类似“已知背景：……”的独立区块，让模型清楚哪些是历史、哪些是当前问题。实测这样比混在一起效果好很多。

3.4 检查点与记忆的关系

检查点存的是图执行的完整状态快照，包括消息列表、中间变量等。它和长期记忆的区别在于：检查点是“过程存档”，用于恢复执行；长期记忆是“知识存档”，用于跨会话召回。两者可以共用一套存储后端，但用途不同，别混为一谈。

4. 实操过程：从零搭一个带长期记忆的最小示例

4.1 环境与依赖准备

先明确一点，下面是我基于常见实践补全的一套可复现方案，不是逐行照抄官方仓库，但思路和 DeerFlow 一致。你需要准备 Python 环境、一个状态图框架（如 LangGraph 风格）、一个向量存储（本地可用轻量方案）。

pip install langgraph langchain-openai chromadb

向量库我选 Chroma 是因为它本地跑、零配置，适合做示例。生产环境你可以换成更重的方案，但接口逻辑是一样的。

4.2 定义带记忆的状态结构

from typing import Annotated from typing_extensions import TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] # 追加式短期记忆 user_id: str # 用于检索长期记忆 long_term_context: str # 检索回来的背景

注意add_messages这个 reducer，它就是保证消息累积的关键。user_id是长期记忆的索引键，没有它就没法按用户隔离记忆。

4.3 写入长期记忆的节点

def save_memory(state: AgentState): history = state["messages"] summary = llm.invoke( f"把以下对话提炼成不超过3条要点，每条为'键: 值'格式：\n{history}" ) vector_store.add( documents=[summary.content], metadatas=[{"user_id": state["user_id"]}] ) return {}

这里的提炼提示词我特意限定了格式和条数。为什么要限条数？因为不限的话模型会写一大堆，存储和检索都变重。

4.4 检索长期记忆的节点

def load_memory(state: AgentState): results = vector_store.query( query_texts=[state["messages"][-1].content], where={"user_id": state["user_id"]}, n_results=3 ) context = "\n".join(results["documents"][0]) return {"long_term_context": context}

n_results=3是我反复调出来的经验值。取太少容易漏，取太多会引入噪声，3 到 5 之间比较稳。

4.5 把记忆注入到生成环节

def generate(state: AgentState): prompt = f"""已知背景： {state.get('long_term_context', '无')} 请基于以上背景回答用户问题。""" response = llm.invoke(prompt + state["messages"][-1].content) return {"messages": [response]}

把背景和当前问题用明确的分隔符隔开，模型对“哪些是历史”的辨识度会明显提升。

4.6 参数选择与计算过程

检索的相似度阈值我一般设在 0.7 左右。这个值怎么来的？我做过一组对比：阈值 0.5 时召回多但噪声大，阈值 0.85 时经常召回为空。0.7 是在我的测试集上准确率和召回率比较平衡的点。当然这跟你的嵌入模型有关，换模型要重新标定。

5. 可观测与流式接口：让记忆过程看得见

5.1 为什么记忆需要可观测

记忆这东西最怕“黑盒”。你不知道它到底存了什么、取回了什么，出了问题只能瞎猜。我在项目里加了一层日志，把每次写入的提炼结果和每次检索的命中片段都打出来。有一次线上反馈“助手记错设备型号”，一看日志发现是提炼时把型号写错了，问题立刻定位。

5.2 流式接口下的记忆处理

DeerFlow 支持流式输出，这对记忆有个特殊要求：记忆的检索要在流式开始之前完成。因为检索是同步的、需要完整结果，而流式是逐字吐出的。所以正确的顺序是：先 load_memory，再进入流式生成。如果你把检索塞进流式过程里，会出现“前半句没背景、后半句才有”的割裂感。

5.3 SSE 流式消息的解析要点

封装 SSE 接口时，要注意消息是按data:行分块传输的，解析时要处理跨块拼接。我遇到过一个典型问题：一个 JSON 被拆到两个数据块里，直接逐块解析会报错。解决办法是维护一个缓冲区，遇到完整的分隔符再解析。

注意：流式解析里“无效字符”和“多字节字符集”相关的报错，多半是分块时把一个多字节字符从中间截断了。处理方式是按字节缓冲，凑齐完整字符再解码。

6. 常见问题与排查技巧实录

6.1 记忆相关的高频问题速查

6.2 几个我踩过的坑

第一个坑是记忆污染。有次测试时用了同一个 user_id，结果不同测试用例的记忆互相干扰，排查了很久。后来我养成习惯，测试时给每个用例分配独立 user_id。

第二个坑是提炼丢失关键信息。模型提炼时可能把“型号 X200”简化成“某设备”，导致后续检索匹配不上。解决办法是在提示词里明确要求“保留所有专有名词和数字”。

第三个坑是检索结果重复。同一段记忆被多次写入，检索时返回一堆重复内容。我加了一层去重，按内容哈希过滤。

6.3 独家避坑技巧

• 写入长期记忆前，先做一次去重检查，避免同一信息反复入库。
• 检索结果注入前，做一次长度截断，防止背景信息挤占当前问题的空间。
• 给记忆加时间戳，检索时可以按时间加权，近期记忆优先。
• 定期做记忆清理，把过期或低价值的记忆归档，控制存储规模。

7. 二次开发中的扩展思路

7.1 记忆的优先级与衰减

不是所有记忆都同等重要。我在二次开发时加了一个简单的权重机制：用户明确说过的偏好权重高，闲聊内容权重低。检索时按权重排序，效果比纯相似度好。衰减则是给老记忆降权，避免几个月前的信息干扰当前判断。

7.2 人机协同场景下的记忆共享

DeerFlow 强调人机协同，这意味着记忆可能需要在人和智能体之间共享。我的做法是把人工修正的结果也写入长期记忆，并标记来源。这样智能体下次就能“学到”人的修正，而不是每次都靠人重复纠正。

7.3 多智能体之间的记忆隔离与传递

如果你在一个系统里跑多个智能体，记忆要按智能体隔离，否则会串味。但有些场景又需要传递，比如研究智能体的结论要给写作智能体用。这时候可以用一个共享的记忆空间，配合命名空间做隔离。

8. 我对这套方案的实际体会

这套分层记忆方案我用了大半年，最大的感受是：记忆的难点不在存，而在取和用。存进去容易，但要在正确的时机取回正确的内容，并且以模型能理解的方式注入，才是真正花功夫的地方。DeerFlow 给了一个不错的骨架，但具体到你的业务，提炼提示词、相似度阈值、注入格式这些都得自己调。

最后分享一个小技巧：如果你不确定记忆方案有没有生效，先别急着看代码，直接打印每次请求的完整 Prompt。我很多次都是靠看 Prompt 里到底有没有背景信息，一眼就定位了问题。这个笨办法，比任何调试工具都直接。