DeepAgent 实时交互与长期记忆改造：SSE 流式输出与 LangChain 记忆实战

1. 从SSE 已上线说起DeepAgent 的实时交互链路到底怎么搭DeepAgent 这个项目最近在圈子里讨论度不低尤其是它把 SSE 流式输出跑通之后前端能实时看到大模型一个字一个字往外蹦体验上确实比等一个完整 JSON 返回要舒服太多。但真正上手拆过它的人会发现一个尴尬的事实SSE 这条链路是通的长期记忆却还是个半成品。这不是 DeepAgent 一家的问题而是当前大量基于 LangChain FastAPI Vue 3 搭建的 Agent 项目普遍存在的状态——交互层做得漂亮记忆层还在凑合。我先把这篇要聊的范围划清楚。DeepAgent 本质上是一个 Agent 应用框架它的技术栈组合是LangChainAgent 编排 FastAPI后端服务 SSE流式传输 Vue 3前端渲染。这套组合在 2024 年下半年到 2025 年几乎成了国内做 AI 应用的标准配方原因很直接LangChain 提供了现成的 Agent、Tool、Memory 抽象FastAPI 天然支持异步和流式响应SSE 比 WebSocket 更轻量、更容易穿透各种中间层Vue 3 的响应式系统配合流式数据渲染几乎不需要额外状态管理。但标准配方不等于标准答案。SSE 上线只是解决了看得见的问题长期记忆解决的是记得住的问题而后者才是 Agent 从玩具变成工具的分水岭。我见过太多项目SSE 调通当天团队很兴奋两周后开始有人问为什么它记不住我昨天说过的话然后发现记忆模块是硬编码的、是内存态的、是每次重启就清空的。这篇文章我会按四条线来拆先讲 SSE 这条链路在 DeepAgent 里到底怎么落地、有哪些坑再讲长期记忆为什么是半成品、半成品具体半在哪里然后讲 LangChain 和 LangGraph 在记忆这件事上的分工与差异最后给一套我自己验证过的、能把记忆从半成品推到可用的改造路径。适合正在做 Agent 应用、已经跑通流式输出、但被记忆问题卡住的开发者。2. SSE 流式输出在 DeepAgent 里的完整落地链路2.1 为什么选 SSE 而不是 WebSocket这个问题几乎每个项目立项时都会被问一遍。我的答案一直很明确Agent 场景下 SSE 是更优解除非你有双向实时通信的硬需求。先看两者的本质差异。WebSocket 是全双工协议建立连接后双方可以随时互发消息SSE 是单向的服务端推、客户端收基于普通 HTTP 长连接。Agent 交互的典型模式是什么用户发一次请求服务端流式返回一段回答结束。这个模式里客户端在回答生成期间几乎不需要再发数据双向通信的能力是浪费的。再看工程成本。SSE 走的是标准 HTTP意味着你现有的鉴权中间件、日志、网关、负载均衡基本不用改Nginx 加一行proxy_buffering off就能用。WebSocket 需要协议升级握手很多企业网关默认不放行排查起来很烦。我在一个项目里遇到过 WebSocket 连接在某个中间层被静默断开、前端只能靠心跳检测发现的情况换成 SSE 之后这类问题直接消失。还有一个容易被忽略的点SSE 天然支持断线重连和事件 ID。浏览器端的EventSource会自动重连配合Last-Event-ID头可以实现断点续传。虽然 DeepAgent 当前未必用到了这个能力但协议层面留了口子后续做回答生成到一半网络断了、重连后接着显示就有基础。当然 SSE 也有硬限制。最典型的是浏览器对同域名 HTTP/1.1 连接数的限制通常 6 个如果用户同时开多个标签页跑 Agent可能把连接占满。解法是上 HTTP/2多路复用之后这个限制就没了。另外 SSE 只能传文本二进制数据要 base64 编码不过 Agent 场景基本都是文本影响不大。2.2 FastAPI 侧的实现细节与常见坑FastAPI 实现 SSE 的核心是StreamingResponse配合异步生成器。看起来简单但有几个坑我踩过不止一次。第一个坑是响应头。必须显式设置media_typetext/event-stream同时加上Cache-Control: no-cache和Connection: keep-alive。少了no-cache某些中间层会缓存整个响应前端就变成等全部生成完才一次性显示流式效果全没了。这个现象很隐蔽因为本地开发直连没问题一上生产经过网关就出问题。第二个坑是生成器的异常处理。异步生成器里如果抛异常FastAPI 不会自动把它转成正常的错误响应而是直接断开连接前端只能看到一个莫名其妙的结束。正确做法是在生成器内部 try/except把异常转成一条 SSE 事件推给前端比如event: error加错误信息然后再正常结束。第三个坑是客户端断开后的资源清理。用户点了停止生成、或者直接关了页面服务端的生成器还在跑LangChain 的 Agent 还在调 LLM、还在烧 token。必须监听request.is_disconnected()在每次 yield 之前检查一旦断开就break并清理资源。这个检查不能只在循环开头做一次因为 LLM 单次调用可能耗时好几秒要在每个 chunk 之间都检查。async def event_generator(request: Request, agent, query: str): try: async for chunk in agent.astream(query): if await request.is_disconnected(): break yield fdata: {json.dumps({content: chunk})}\n\n yield data: [DONE]\n\n except Exception as e: yield fevent: error\ndata: {json.dumps({msg: str(e)})}\n\n这段代码里[DONE]这个约定很重要。SSE 协议本身没有结束事件连接关闭就是结束但前端EventSource在连接正常关闭时会自动重连导致请求被重复触发。发一个显式的[DONE]标记前端收到后主动close()才能干净地结束。2.3 Vue 3 前端的流式渲染与 abort 控制前端这块很多人第一反应是用EventSource但我要泼盆冷水Agent 场景下EventSource往往不够用建议直接用fetchReadableStream。原因有三。一是EventSource只支持 GET 请求Agent 的 query 通常比较长塞 URL 里不合适而且很多场景需要 POST 传结构化参数。二是EventSource不能自定义请求头鉴权 token 只能塞 URL不安全。三是EventSource的自动重连在 Agent 场景下是负担前面说的重复触发问题就是它带来的。用fetch手动解析 SSE 流配合AbortController做中断是目前最灵活的方案const controller new AbortController() const response await fetch(/api/agent/chat, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ query }), signal: controller.signal }) const reader response.body.getReader() const decoder new TextDecoder() let buffer while (true) { const { done, value } await reader.read() if (done) break buffer decoder.decode(value, { stream: true }) const lines buffer.split(\n\n) buffer lines.pop() for (const line of lines) { if (line.startsWith(data: )) { const data line.slice(6) if (data [DONE]) return appendToMessage(JSON.parse(data).content) } } }AbortController是这里的关键。用户点停止生成时调controller.abort()fetch 会抛AbortError同时服务端因为连接断开触发is_disconnected两边同步停止。这个链路必须打通否则前端停了后端还在烧钱。Vue 3 的响应式在这里有个小技巧不要每收到一个 chunk 就整体替换消息内容那样会触发大量无谓的 DOM diff。更好的做法是把消息内容做成ref字符串用追加Vue 的响应式系统会做最小化更新。如果消息很长还可以考虑用shallowRef配合手动触发更新进一步降低开销。2.4 SSE 链路的性能与稳定性经验跑通之后要关注的是稳定性。我总结了几条实战经验。心跳不能少。有些中间层会在连接空闲 60 秒后主动断开而 LLM 生成第一个 token 之前可能有几秒到几十秒的思考时间尤其是带工具调用的 Agent。解法是服务端在等待期间定期发注释行: keepalive\n\n这种以冒号开头的行会被 SSE 客户端忽略但能保持连接活跃。chunk 粒度要控制。LangChain 的astream默认按 token 返回粒度很细每个 token 一个 SSE 事件网络开销不小。可以在服务端做一层缓冲攒够一定字符数或者间隔一定时间再发减少事件数量。但缓冲不能太大否则流式的实时感就没了我的经验是 20 到 50 毫秒或者 10 到 20 个字符触发一次比较合适。错误要分级。LLM 调用失败、工具调用失败、网络失败这三类错误前端应该有不同的展示。建议在 SSE 事件里带上error_type字段前端据此决定是提示重试、还是提示换个问法、还是提示稍后再试。3. 长期记忆为什么是半成品拆开看半在哪里3.1 内存态记忆重启即失忆的默认实现大部分 Agent 项目起步时用的都是 LangChain 的ConversationBufferMemory或者它的异步版本本质就是一个内存里的列表每轮对话往里 append。这个方案在 demo 阶段完全够用但一上生产就暴露问题进程重启、多实例部署、水平扩容任何一个都会让记忆丢失或错乱。我见过一个项目开发环境单实例跑得好好的上线后开了 4 个 worker用户发现 Agent 时而记得上一句、时而完全不记得排查半天才反应过来是请求被负载均衡打到了不同实例每个实例的内存记忆是独立的。这个坑非常典型本质是把会话状态这个有状态的东西放在了无状态的服务层。内存态记忆还有个隐性问题是没有上限。对话轮数一多整个历史都塞进 prompttoken 消耗线性增长很快就撞上上下文窗口。有些实现会做简单的截断只保留最近 N 轮但截断策略很粗暴可能把关键信息比如用户前面说的偏好也截掉了。3.2 向量检索记忆的召回率陷阱意识到内存态不行之后下一步通常是上向量数据库把历史对话做 embedding 存起来每轮根据当前 query 检索相关历史。这个方向是对的但召回率是个大坑。第一个问题是检索时机。很多实现是每轮对话都检索一次把 top-k 相关历史塞进 prompt。但 Agent 场景下当前 query 可能和任何历史都不相关比如用户突然问一个全新话题这时候检索出来的相关历史其实是噪声反而干扰模型。更合理的做法是让模型自己决定要不要检索记忆或者用更复杂的路由逻辑。第二个问题是embedding 的语义漂移。用户说帮我改一下那个配置那个指代的是上一轮提到的某个文件但单独把这句话做 embedding检索出来的历史可能完全不相关。对话中的指代、省略、上下文依赖是向量检索的天然弱项。第三个问题是写入策略。是每轮对话都写、还是只写关键信息、还是定期总结后写每轮都写会导致记忆库膨胀且充满冗余只写关键信息需要额外的判断逻辑定期总结则可能丢失细节。DeepAgent 当前大概率是每轮都写或者简单截断这就是半成品的一个具体表现。3.3 记忆的读写时机什么时候该记、什么时候该忘长期记忆真正难的不是存储是读写时机的判断。人脑的记忆机制其实很有参考价值不是所有信息都值得长期记住重要的会被强化不重要的会自然遗忘。Agent 的记忆也应该有类似的机制。我倾向于把记忆分成三层层级内容存储生命周期工作记忆当前对话的最近几轮内存/Redis会话期间情景记忆关键事件、用户偏好、重要结论向量库结构化库长期语义记忆从多次交互中提炼的规律结构化库长期定期更新工作记忆保证对话连贯情景记忆保证跨会话的个性化语义记忆保证 Agent 越用越懂用户。DeepAgent 当前基本只有工作记忆而且工作记忆还是内存态的这就是半成品的核心含义。写入时机上我的经验是不要每轮都写。可以设几个触发条件用户显式表达了偏好我喜欢简洁的回答、对话中出现了需要长期记住的事实我的项目用的是 FastAPI、或者一轮对话结束时用一个小模型做一次总结判断。读取时机上也不要每轮都读可以让 Agent 在需要的时候主动调用一个recall_memory工具。3.4 多用户隔离与记忆污染还有一个容易被忽视的问题多用户场景下的记忆隔离。如果记忆库没有严格的 user_id 隔离A 用户的偏好可能被检索到 B 用户的对话里这是严重的数据泄露。隔离要在三个层面做存储层每个向量必须带 user_id 元数据检索层必须强制过滤 user_id应用层要确保 user_id 从鉴权信息里来、不能被前端伪造。我见过有项目把 user_id 放在请求体里传前端随便改一个就能读到别人的记忆这个必须避免。另外还有记忆污染的问题如果 Agent 自己生成的内容也被写进记忆库下一轮检索时可能把自己的幻觉当成事实。所以写入记忆前最好做一次判断区分用户说的和Agent 说的后者要谨慎写入。4. LangChain 与 LangGraph 在记忆这件事上的分工4.1 两者的定位差异不是替代关系社区里经常有人问LangChain 和 LangGraph 到底啥区别、该用哪个这个问题本身就问偏了。它们不是替代关系是不同层次的抽象。LangChain 提供的是组件级的抽象LLM、Prompt、Tool、Retriever、Memory每个都是可以单独用的积木。LangGraph 提供的是编排级的抽象把多个步骤组织成有状态图节点是处理逻辑边是流转条件整个图有共享的 state。放到记忆这件事上LangChain 的 Memory 组件负责存和取这个动作LangGraph 的 state 和 checkpointer 负责在图的执行过程中保持状态。DeepAgent 如果用的是 LangChain 的 AgentExecutor那记忆就是外挂的如果用的是 LangGraph记忆可以内化成图的一部分。4.2 LangGraph 的 checkpointer 才是长期记忆的正解LangGraph 的checkpointer机制是我认为目前做 Agent 长期记忆最优雅的方案。它的核心思想是把整个图的 state 在每个节点执行后持久化下次用同一个 thread_id 进来时自动恢复。这意味着什么意味着你不需要手动管理哪些历史要存、怎么存、怎么取只要把对话历史放进 statecheckpointer 会自动帮你持久化和恢复。thread_id就是会话 ID同一个 thread_id 的多次调用共享同一份 state。from langgraph.checkpoint.postgres import PostgresSaver checkpointer PostgresSaver.from_conn_string(DB_URL) graph builder.compile(checkpointercheckpointer) config {configurable: {thread_id: user_session_id}} result graph.invoke({messages: [user_input]}, config)这段代码背后LangGraph 会把 state 存进 Postgres下次同 thread_id 调用时自动加载。相比自己写一套记忆管理这个方案的好处是状态和图执行是原子绑定的不会出现图执行到一半、记忆没存上的不一致。但 checkpointer 也不是银弹。它解决的是会话状态持久化不是跨会话的语义记忆。同一个 thread_id 内的历史能恢复但用户开新会话时之前会话里的偏好怎么带过来这还是要靠向量库那套。所以我的建议是checkpointer 管会话内向量库管跨会话两者配合。4.3 从 AgentExecutor 迁移到 LangGraph 的代价如果 DeepAgent 当前用的是AgentExecutor想迁移到 LangGraph 做记忆代价要提前评估。迁移的主要工作量在于把 AgentExecutor 的隐式循环显式化成图。AgentExecutor 内部是一个 while 循环调 LLM、判断要不要调工具、调工具、把结果塞回去、再调 LLM直到没有工具调用为止。在 LangGraph 里这个循环要拆成节点和条件边。好处是显式化之后你可以在任意节点插入记忆读写、人工审核、分支逻辑。比如在调 LLM 之前插入一个检索相关记忆节点在工具调用之后插入一个判断是否值得记忆节点。这些在 AgentExecutor 里很难做因为它的循环是黑盒。代价是代码量增加、调试复杂度上升。我的经验是如果项目还在早期、记忆需求不复杂先用 AgentExecutor 加外挂记忆也能跑如果已经明确要做复杂的记忆逻辑、多轮工具调用、人工介入那早点迁到 LangGraph 更划算。4.4 工具调用与记忆的协同设计记忆和工具调用其实可以协同。一个很实用的模式是把记忆检索做成一个工具让 Agent 自己决定什么时候调用。tool def recall_memory(query: str) - str: 检索与当前问题相关的历史记忆。当用户提到之前讨论过的内容、 或者需要个性化回答时调用。 results vector_store.similarity_search(query, k5, filter{user_id: current_user}) return \n.join([r.page_content for r in results])这样做的好处是检索时机由模型判断避免了每轮都检索、检索出噪声的问题。模型在需要的时候主动调不需要的时候不调更符合实际对话的节奏。写入也可以做成工具或者做成一个后置节点。我倾向于写入用后置节点自动做因为让模型判断什么值得记容易漏而后置节点可以用规则小模型结合的方式更可控。5. 把记忆从半成品推到可用的改造路径5.1 第一步把内存态换成持久化存储改造的第一步永远是把内存态干掉。不管后面做多复杂的记忆逻辑先把重启不丢这个底线守住。如果已经在用 LangGraph直接上 checkpointerPostgres 或 Redis 都行。Postgres 的好处是可以用 SQL 查询和分析记忆数据Redis 的好处是快、适合高频读写。我的选择是 Postgres因为记忆数据量不会太大而且需要做复杂的过滤查询。如果还在用 AgentExecutor那就自己实现一个持久化的 Memory 类把历史存进数据库每次调用时按 session_id 加载。这里要注意加载时的截断策略不能把所有历史都塞进 prompt要按 token 数或者轮数做限制同时保留最近几轮和关键信息。class PersistentMemory: def load(self, session_id: str, max_tokens: int 2000): history db.query_messages(session_id, limit50) # 从最近往前累加直到接近 token 上限 selected [] total 0 for msg in reversed(history): tokens count_tokens(msg.content) if total tokens max_tokens: break selected.append(msg) total tokens return list(reversed(selected))5.2 第二步设计记忆的写入触发条件持久化之后要解决记什么的问题。我的方案是规则触发 模型判断双管齐下。规则触发负责明确的场景用户消息里出现记住以后都我的偏好是这类关键词直接标记为需要长期记忆。模型判断负责模糊的场景一轮对话结束后用一个小模型比如本地跑的小参数模型判断这轮对话里有没有值得长期记住的信息输出结构化结果。判断的 prompt 可以这样设计让模型输出 JSON包含should_remember布尔、memory_type偏好/事实/事件、content要记的内容、importance1-5 分。这样写入的记忆带元数据后续检索时可以按类型和重要性过滤。要注意的是判断本身也有成本。每轮都调一次小模型token 和时间开销不小。可以优化成每 N 轮判断一次或者只在对话结束时判断具体看场景。5.3 第三步混合检索替代纯向量检索纯向量检索的召回率问题前面说过了解法是混合检索向量检索 关键词检索 元数据过滤三路结果融合。向量检索负责语义相似关键词检索BM25 之类负责精确匹配元数据过滤负责按 user_id、时间、类型做筛选。三路结果用 RRFReciprocal Rank Fusion之类的算法融合排序取 top-k。这里有个细节不同来源的分数不能直接比向量相似度是 0 到 1BM25 是无界的必须用 RRF 这种基于排名的融合方法而不是简单加权求和。RRF 的公式很简单每个文档的得分是sum(1 / (k rank))k 通常取 60。融合之后还要做去重。同一段记忆可能被多路检索到要去掉重复。去重不能只按 ID因为不同来源的 ID 可能不同要按内容相似度去重。我见过有实现只按 ID 去重结果同一段话因为来源不同被塞进 prompt 两次浪费 token。5.4 第四步记忆的更新与遗忘机制记忆不是只增不减的。过时的记忆要更新无用的记忆要遗忘否则记忆库会越来越臃肿检索质量越来越差。更新机制当新信息和旧记忆冲突时比如用户之前说喜欢 Python现在说改用 Go 了要能识别冲突并更新。简单做法是写入时检索相似记忆如果相似度超过阈值就让模型判断是新增还是更新。复杂做法是维护记忆的版本链保留历史但标记最新。遗忘机制可以按时间衰减老记忆的检索权重降低也可以按访问频率长期没被检索到的记忆降低权重甚至归档。我倾向于组合策略重要性高的记忆不衰减重要性低的按时间衰减超过一定时间没访问的归档到冷存储。5.5 实测中的性能与成本权衡最后说说实测数据。我在一个中等规模的项目里做过对比记忆模块从内存态改造成Postgres checkpointer 向量库混合检索之后单轮对话的额外延迟从 0 增加到约 80 到 150 毫秒主要是检索和写入跨会话的个性化准确率人工评估从基本不可用提升到约 75%存储成本每月增加约几十元Postgres 向量库token 消耗因为检索结果塞进 prompt 增加了约 20%这个代价我认为是值得的但要注意延迟。80 到 150 毫秒在流式场景下用户基本感知不到因为首 token 本来就要等但如果做的是实时性要求极高的场景就要考虑把检索做成异步预取或者用更快的向量库。还有一个成本点是embedding 调用。每轮对话都要对 query 做 embedding如果用的是外部 API调用量和费用都不小。可以考虑本地部署 embedding 模型虽然效果可能略差但成本和延迟都可控。6. 几个我踩过的坑和对应的解法6.1 SSE 连接被中间层缓冲这个坑我踩过两次两次都是上线后才暴露。现象是本地流式正常生产环境变成等全部生成完一次性显示。根因是中间层Nginx、某些云网关默认会缓冲响应。Nginx 的解法是加配置location /api/agent/ { proxy_pass http://backend; proxy_buffering off; proxy_cache off; proxy_set_header Connection ; proxy_http_version 1.1; chunked_transfer_encoding off; }云网关的话要看具体厂商的文档通常有关闭响应缓冲或者流式模式的开关。排查方法是用 curl 直接打后端接口如果 curl 能看到流式、经过网关就没了那基本就是缓冲问题。6.2 记忆检索把 Agent 自己的幻觉也检索出来了这个坑很隐蔽。Agent 某轮生成了一段错误信息被写进记忆库下一轮检索时把这段错误信息当成事实塞进 prompt导致错误被强化。用户会觉得它怎么越用越离谱。解法是写入时区分来源。用户明确说的内容标记为source: userAgent 生成的内容标记为source: assistant检索时对 assistant 来源的记忆降权或者只在高置信度时才使用。更严格的做法是 assistant 来源的内容不直接写入长期记忆只写入工作记忆。6.3 多实例部署下 thread_id 冲突用 LangGraph checkpointer 时如果 thread_id 生成逻辑有问题不同用户的会话可能撞到同一个 thread_id导致记忆串台。thread_id 必须是全局唯一的建议用user_id session_id组合或者直接用 UUID。不要用时间戳高并发下会撞不要用自增 ID多实例下会撞。我现在的做法是前端生成 UUID 作为 session_id后端拼上 user_id 作为 thread_id双重保证唯一。6.4 记忆写入阻塞了流式响应早期我把记忆写入放在流式响应的生成器里同步做结果用户看到回答生成完了但连接迟迟不关闭因为还在写记忆。体验上就是回答完了但转圈还在转。解法是写入异步化。流式响应结束后把写入任务丢进后台队列Celery、FastAPI 的 BackgroundTasks 都行立即关闭连接。写入失败也不影响用户记个日志重试就行。这个改动很小但体验提升明显。7. 关于 DeepAgent 这类项目的一点个人判断拆完 DeepAgent 的 SSE 链路和记忆现状我最大的感受是当前 Agent 项目的竞争点正在从能不能流式输出转移到能不能记住用户。SSE 已经是标配LangChain FastAPI Vue 3 这套组合也已经被验证过无数遍技术选型上很难做出差异化。真正拉开差距的是记忆的质量——能不能准确记住该记的、忘掉该忘的、在合适的时候想起来。长期记忆是半成品这件事其实不是 DeepAgent 的锅是整个行业都还在摸索。向量检索、checkpointer、混合检索、记忆更新这些技术单独看都不新鲜但怎么组合成一套稳定可用的记忆系统还没有公认的最佳实践。我的建议是不要追求一步到位先把持久化做了、再把写入触发条件设计好、然后逐步上混合检索和遗忘机制每一步都能带来可感知的提升。最后分享一个我自己的判断标准如果一个 Agent 用了一周之后你感觉它比第一天更懂你了那记忆就做对了如果一周后感觉和第一天没区别那记忆就是摆设。这个标准很主观但比任何技术指标都更接近用户的真实感受。