企业官网建设流程全解析

DeepAgent 的 SSE 流式输出上线跑了一阵子，整体链路算是通了，但长期记忆这块我评估下来仍然是个半成品。这篇文章把这次实战的完整过程拆开讲清楚：SSE 怎么接、Abort 怎么处理、记忆体系怎么设计、以及为什么说长期记忆还差得远。内容偏工程落地向，适合正在做 AI Agent 应用、需要把大模型交互封装成标准化服务的开发者参考。

先说结论：如果你只是想快速把 LLM 的流式输出接到前端，DeepAgent 这套 SSE 方案可以直接抄作业；但如果你指望它有成熟的长期记忆能力，那还要自己动手补不少东西。

1. 项目背景与技术选型思路

1.1 DeepAgent 到底是什么场景的产物

DeepAgent 本质上是一个面向多智能体协作场景的应用编排层，底层对接大模型能力，向上提供统一的业务接口。和直接调 OpenAI API 不同，它更像一个中间件：帮你把 Prompt 管理、工具调用、多轮会话、流式输出这些繁琐的逻辑收敛到一层，业务侧只需要关心"我要 Agent 做什么"，而不需要关心"Agent 怎么和模型通信"。

我在这个项目里负责的是交互链路这块，核心目标有两点。第一，让大模型的回答能像打字机一样逐字出现，而不是让用户对着空白页面等三五秒；第二，让 Agent 的状态能被追踪，也就是多轮对话里它"记得"什么、不记得什么，这个数学模型必须清楚。这两点分别指向了 SSE 和记忆体系，也是这次标题里两个关键词的由来。

技术选型上，我们前后对比过几条路。最原始的方案是前端轮询，也就是发起请求后每隔 500ms 去查一次结果，但这样 WebSocket 的优势体现不出来，反而增加服务端压力。第二条路是 WebSocket 全双工通信，好处是消息可以双向推，但 Agent 场景里用户发消息的频率远低于模型回消息的频率，全双工的价值并不大，而且 WebSocket 在网关层、鉴权层都要单独做适配，成本和复杂度不成比例。最终定了 SSE，单向流式推送，恰好匹配"用户发一次，模型流式回多次"的交互模型。

1.2 为什么 SSE 比 WebSocket 更适合 Agent 场景

很多人一听到"实时"就想到 WebSocket，但这里有个误区：Agent 交互的实时性需求是单向的。用户在输入框敲完回车那一刻，请求已经提交了，后续真正需要实时更新的只有模型的输出流。这叫"请求-流式响应"模型，SSE 天生就是干这个的。

SSE 基于纯 HTTP，意味着它可以复用现有的负载均衡、鉴权中间件、日志链路，不需要额外维护一个长连接网关。你可以在 Nginx 层面直接给它配缓冲关闭、超时调优，也可以在应用层用最普通的 WSGI/ASGI 服务器跑起来。对部署和运维来说，这比 WebSocket 少了一整个维度的麻烦。

还有一个工程上的隐性收益：SSE 天然支持断线重连。浏览器原生的 EventSource API 自带 reconnection 机制，而 WebSocket 挂了之后要自己实现心跳和重连策略。对 Agent 这种动辄十几秒的推理过程来说，中间网络抖一下是常有的事，SSE 的重连机制能省掉不少脏活。

2. SSE 流式输出实战拆解

2.1 服务端怎么把 LLM 的流式输出转发给前端

核心思路用一个代码块就能说清楚：大模型 SDK 返回的是一个生成器，我们把这个生成器里的每一段增量文本包装成 SSE 格式的data:事件，一次一个 event 推到客户端。

from fastapi import FastAPI from fastapi.responses import StreamingResponse from llm_sdk import chat_stream app = FastAPI() @app.post("/v1/agent/chat") async def agent_chat(request: ChatRequest): # 这里拿到的是一个异步生成器，每次 yield 一小段文本 generator = chat_stream( messages=request.messages, tools=request.tools, agent_id=request.agent_id, ) async def event_stream(): # 先发一个 session 开始事件，方便前端做 loading 收敛 yield "event: session_start\ndata: {}\n\n" async for chunk in generator: # chunk 可能是文本增量，也可能是工具调用的元信息 yield f"event: delta\ndata: {chunk.model_dump_json()}\n\n" # 流结束后发一个 done 标记 yield "event: done\ndata: {}\n\n" return StreamingResponse( event_stream(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no", }, )

这里有个细节值得注意：X-Accel-Buffering: no这个响应头。如果你们的服务前面挂了 Nginx，默认情况下 Nginx 会缓冲后端响应，攒够一定量才发给客户端，这样前端看到的就是"卡一下然后一整段出现"，流式效果直接没了。加上这个头就是为了告诉 Nginx 别缓冲，数据到了就往外吐。

2.2 前端怎么优雅处理 Abort 和异常中断

SSE 接入前端有两种方式。一种是浏览器原生 EventSource，但 EventSource 只支持 GET 请求，没法带 Authorization header，对需要鉴权的业务很不友好。另一种是用 fetch 加 ReadableStream 手动解析，这也是我在项目中采用的方式。

async function chatWithAgent(messages, { onDelta, onDone, onError }) { const controller = new AbortController(); // 把 controller 暴露出去，组件卸载或者用户点了“停止生成”时调用 pendingController = controller; try { const response = await fetch("/v1/agent/chat", { method: "POST", headers: { "Content-Type": "application/json", "Authorization": `Bearer ${token}`, }, body: JSON.stringify({ messages }), signal: controller.signal, }); if (!response.ok) throw new Error(`HTTP ${response.status}`); const reader = response.body.getReader(); const decoder = new TextDecoder("utf-8"); let buffer = ""; while (true) { const { done, value } = await reader.read(); if (done) break; buffer += decoder.decode(value, { stream: true }); // SSE 事件之间以空行分隔，按这个规则解析 const events = buffer.split("\n\n"); buffer = events.pop(); for (const rawEvent of events) { const lines = rawEvent.split("\n"); const eventType = lines.find(l => l.startsWith("event:"))?.slice(6).trim(); const data = lines.find(l => l.startsWith("data:"))?.slice(5).trim(); if (eventType === "delta") { onDelta(JSON.parse(data)); } else if (eventType === "done") { onDone(); } } } } catch (err) { if (err.name === "AbortError") { // 用户主动取消，不算错误 onAbort?.(); } else { onError(err); } } }

这段代码里最容易踩坑的是TextDecoder的{ stream: true }参数。SSE 数据流是字节级别的，一个中文字符可能被拆在两个 chunk 里，如果不加 stream 模式，解码时就会偶尔出现乱码。加上这个参数之后，解码器会把没凑完整的字符留在缓冲区，等下一段字节到了再拼起来。

关于 Abort 还有个很容易被忽略的点：客户端abort()之后，服务端其实不一定立刻感知到。Python 的异步生成器里，如果模型还在推理，StreamingResponse不会因为你断开了连接就自动取消模型调用。这意味着用户取消了生成，但后端还在占着模型资源继续算，多来几个这样的请求就能把算力池拖垮。解决方案是在生成器里监听客户端断开事件：

async def event_stream(): try: yield "event: session_start\ndata: {}\n\n" async for chunk in generator: yield f"event: delta\ndata: {chunk.model_dump_json()}\n\n" yield "event: done\ndata: {}\n\n" except asyncio.CancelledError: # 客户端断开时，StreamingResponse 会取消这个生成器 # 在这里手动去 cancel 底层模型调用 await generator.aclose() raise

这块我在初版实现里漏了，压测时发现前端点了停止，后端日志里模型调用还在跑，浪费了不少 token。后来加上aclose()才把这个资源泄漏问题解决。

2.3 流式输出的超时与空闲断开问题

热搜词里有一条 "stream disconnected before completion: idle timeout waiting for SSE"，这个报错我实测中遇到过。它的大致场景是：模型在某个工具调用节点上思考了比较久，中间有一段时间没有输出任何 SSE 事件，结果网关层的 idle timeout 触发了，连接被强制断开，前端收到一个不完整的流。

排查思路是这样：先看你的服务链路里有哪些组件有超时配置。通常有三个位置会影响 SSE：

Nginx 配置里有两个参数最要命：proxy_read_timeout和proxy_send_timeout。默认值是 60s，如果你的模型工具调用链超过这个时间，连接必断。实测中我把它调到了 300s：

location /v1/agent/ { proxy_pass http://agent_service; proxy_buffering off; proxy_cache off; proxy_read_timeout 300s; proxy_send_timeout 300s; chunked_transfer_encoding off; }

另外一个容易翻车的地方是：模型在思考期间真的一个字节都不发。为了让中间链路知道"这个连接还活着"，很多实现会用 SSE 的注释行做心跳。注释行形如: ping\n\n，浏览器和代理都会把它当成活跃信号，但不会触发表层事件。我建议在流式生成里搞一个后台心跳任务，每 15s 发一次注释行：

async def event_stream(): # 用一个 task 定期发心跳 async def heartbeat(): while True: await asyncio.sleep(15) yield ": heartbeat\n\n" # 把 heartbeat 和主生成器合并输出

这个方案不算复杂，但能显著减少因为网络静默导致的误杀。

3. 从需求到落地：记忆体系的演化路径

3.1 Agent 记忆体系的四层结构

热搜词里提到"agent 记忆体系中短期、长期、永久记忆如何实现"，这个问题我被问了很多次。说清楚这件事之前，先建立一个统一的框架。我把 Agent 记忆按时间尺度和持久化策略分成四层：

短期记忆：当前对话窗口内的上下文，直接拼在 Prompt 里送给模型。它的特点是容量小、变化快，模型每次调用都重新传一遍。对应到大模型的世界里，就是 context window 内的 messages 列表。短期记忆的实现最简单，但也是最容易失控的——上下文一旦超过模型的 context window，要么截断，要么报错。

中期记忆：跨对话但不过长时间的记忆，比如用户在一个工作会话里的历史操作、偏好变化。实践中通常用摘要的方式管理，也就是把一段对话压缩成几百字的摘要存下来，下次对话时作为系统提示的一部分注入。中期记忆的核心是"压缩"，因为原始的对话内容太多，不可能全量传给模型。

长期记忆：用户跨会话、跨天甚至跨月的稳定偏好、事实信息。比如用户是 Python 开发者、偏好简洁回答、上次讨论过某个项目。这类信息适合结构化存储，比如放 SQLite、PostgreSQL 或者向量数据库，在对话开始时按需检索注入。长期记忆的难点是"什么时候写入"和"什么时候读取"，写早了会记录噪音，读多了会污染上下文。

永久记忆：用户画像级别的数据，不随对话变化，比如账号信息、权限范围、不可修改的基础属性。这种一般直接落在业务数据库，跟 Agent 系统解耦，需要时通过工具调用查出来。

这个四层结构不是拍脑袋定的。你去看 LangGraph、AutoGen 这些框架的记忆模块，本质上都在做这四件事，只是叫法不同。理解了分层逻辑，再去看具体框架的 API 就会豁然开朗。

3.2 DeepAgent 当前记忆模块的实现现状

回到 DeepAgent 本身。我评估完它的记忆模块，给的结论是：短期和中期基本可用，长期是个半成品。为什么这么说？

短期记忆方面，DeepAgent 维护了一个 messages 数组，每次调用模型时完整传入。这个逻辑没毛病，问题在于它没有处理上下文超出窗口的情况。我实测发了几十轮对话之后，服务端直接报了 context length exceeded，而且错误信息里没有任何提示告诉你"该滚动了"或"该压缩了"。这说明框架把这个责任完全扔给了上层调用者。

中期记忆方面，DeepAgent 提供了会话摘要能力，基于一个简单的策略：对话轮数超过阈值时，让模型对已有内容做总结，用摘要替换掉最早的一部分消息。我测下来功能是能跑的，但触发的时机比较死板。它只按轮数触发，不看实际 token 占用，导致在长回复场景下，摘要还没触发，上下文就已经超限了。

长期记忆这块，标题里说它"半成品"，一点不冤枉。DeepAgent 目前的实现只有一个memory字段，挂在一个 agent 实例上，进程一重启，里面的内容全没了。没有持久化层，没有向量检索，没有独立的存取服务。如果你要做多实例部署，这个 memory 字段连分布式共享都做不到，每个 worker 各存各的，用户会话被路由到不同实例时记忆完全是错乱的。

我给它的定位是：提供了一个记忆接口的骨架，但工程上能用的长期记忆能力需要自己动手补全。

3.3 基于 LangGraph 的交互机制分析

热搜词里有一条很有意思："deepagent 和 langaph 通过什么方法交互"。这里"langaph"我理解是 LangGraph 的笔误。目前市面上确实有不少团队在用 LangGraph 编排复杂 Agent 流程，同时用 DeepAgent 这类框架做应用封装，两者的通信机制值得聊一聊。

LangGraph 的核心抽象是图：节点是处理步骤，边是状态流转路径。它本身不关心你用什么协议调用外部服务，所以 DeepAgent 和 LangGraph 的集成方式完全取决于你把它放在哪个位置。我实际采用的方式是：用 LangGraph 做顶层编排，把 DeepAgent 封装成一个 Tool 节点。LangGraph 根据用户的输入决定要不要调用 DeepAgent，调用时通过 HTTP 请求把参数传过去，DeepAgent 返回的结果作为 Tool 的响应回到 LangGraph 的 state 里。

还有一种反向集成：DeepAgent 作为入口，内部通过 LangGraph 的 API 定义子图。这种场景下，通信机制一般是函数调用而非 HTTP，因为两个框架在同一个进程内。DeepAgent 的execute_graph接口接收 LangGraph 的 state 对象，内部走完图之后把最终 state 返回。

不管哪种方式，关键点在于状态如何传递。LangGraph 的 state 是线程私有的，跨图传递时需要注意可序列化。我的经验是：在两个框架的边界处定义一个明确的 DTO（数据传输对象），只传必要字段，不要把整个 state 直接塞过去。曾经踩过坑，尝试把 LangGraph 的整个 state 对象 JSON 序列化后传给 DeepAgent，结果里面有不可序列化的自定义类，直接 500。

3.4 长期记忆落地的补全方案

既然 DeepAgent 的长期记忆是半成品，那就自己动手补。我的思路是把长期记忆拆成两层：事实型记忆和语义型记忆。

事实型记忆用一个标准的 KV 存储或者关系型表搞定。表结构大概是：用户 ID、key、value、更新时间、来源会话 ID。查询时直接按用户 ID 拉取全部记忆，注入系统提示。优点是简单、可控、可审计，缺点是没法做语义匹配——你没法问"用户之前有没有提过和部署相关的事情"然后模糊命中。

语义型记忆用向量数据库。对话过程中把关键信息切片、embedding、入库，新对话开始时用当前问题去检索 top-k 相关的记忆片段。这种做法对"用户提过类似问题"的场景特别有效，但需要额外的 embedding 服务和向量库运维成本。

我的建议是：预算有限时先用事实型 KV 记忆，把语义型记忆作为迭代计划。原因是事实型记忆的效果确定性更强，出现问题也好排查；向量召回如果质量不好，你说不清是 embedding 的问题、切分的问题还是检索参数的问题，排查成本会高得多。

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

4.1 SSE 断了、前端不渲染、数据乱码排查表

这段时间踩过的 SSE 相关坑整理成一张表，希望帮你少走弯路：

排查 SSE 问题时有个实用技巧：直接用 curl 不带-N参数看报错，用curl -N看流式输出。如果 curl 能正常一个 chunk 一个 chunk 地出，前端却不行，问题大概率在前端解析；如果 curl 也卡住或者断了，问题就在后端或中间链路。

4.2 记忆体系实测中的典型故障

记忆相关的坑比 SSE 更难定位，因为它的错误往往是"逻辑正确但结果不对"，而不是"程序直接报错"。举一个我实际遇到的例子。

用户问 Agent："我上次让你总结的那篇关于 RAG 的文章，核心观点是什么？" 理想情况下，Agent 应该从长期记忆里检索出"用户上次讨论过 RAG 文章"这个事实，然后基于记忆内容回答。但实际跑起来，Agent 直接回复"我没有找到相关记录"。

排查后发现原因在于：记忆写入的时机错了。DeepAgent 的实现是在会话结束时整体写入记忆，结果进程崩溃或者会话超时被清理时，记忆根本没落库。用户下次来问，自然什么都没有。这个问题的根因是"写入时机没有保证"，不是"检索逻辑不对"。

解决思路有两个层面。第一是写入时机下沉，把"提取记忆并写入"这个动作从会话结束挪到每一轮模型调用之后，这样即使会话中途挂掉，至少已经完成的对话内容会被存下来。第二是加写入确认机制，记忆写入后返回一个确认标识，在日志里能追踪"这条记忆是哪个会话、哪个时间点写入的"。

另一个高频问题是记忆注入位置。长期记忆如果一股脑全塞进系统提示的最前面，模型容易被大量历史事实干扰，影响对当前问题的注意力。我实践下来的做法是：把记忆按相关度排序，只注入 top 5 条事实，每条用一句话表达，放在系统提示的末尾，紧挨着用户消息。这样既给了模型参考，又不会让它淹没在陈年旧事里。

4.3 生产环境中的稳定性兜底策略

SSE 生产环境最大的敌人是"所有东西看起来都对，但就是偶尔断流"。这种随机性问题，没有银弹，只能用多层次的兜底来对冲。

我在服务端做了三个兜底。第一是流式响应的超时熔断，单次 SSE 连接超过 5 分钟强制断开，防止模型卡死导致连接被无限占用。第二是错误事件注入，生成过程中如果模型调用抛异常，不直接中断连接，而是发一个event: error事件，前端收到后可以给用户展示"这次回答生成失败，请重试"，而不是让人面对一个毫无反馈的加载转圈。第三是断点续传的简化版，前端在收到 done 事件后才把内容写入本地状态，在这之前如果断了，用户刷新后回看，看到的还是上一轮完整回答，不会出现一截截残文。

前端层面我加了一个指数退避的重试策略。前 5 秒内断开不重试，因为大概率是网络抖动，用户自己会刷新；超过 5 秒断开，每 5 秒重试一次，最多 3 次。重试时带上当前已渲染内容的长度，服务端如果解析到这个参数，可以选择跳过已经输出的部分，从断点继续生成。这个功能我还没完全实现，目前是重试时从头生成，但至少用户不用手动刷新页面了。

5. 后续演进方向与个人经验总结

5.1 记忆模块下一步的两个优先改造项

长期记忆要从不合格到能用，我的优先级排序是：先做持久化，再做召回质量，最后才是多模态扩展。

持久化是最急迫的。目前 memory 字段在内存里，进程重启就丢，这连"半成品"都算不上，只能算 demo。我的计划是引入 SQLite 做本地持久化，把 memory 结构化成表，以 user_id + agent_id 作为联合主键。这样至少做到进程重启不丢数据，多实例部署时再考虑引入 Redis 做共享存储。

召回质量方面，先用 BM25 这类传统的稀疏检索试试。不要一上来就上向量数据库，因为 embedding 的维护成本高，且对短期迭代不友好。BM25 基于关键词匹配，对大部分"用户之前提过 XXX"这类事实性召回已经够用。等数据量积累到一定程度，确实出现语义匹配需求了，再切换到向量检索方案。

5.2 从这次实战里沉淀下来的几条经验

第一，SSE 不是越底层越好。直接用 fetch + ReadableStream 是最灵活，但也最容易出 bug。如果你们的场景简单，不需要自定义 header 和 Abort 控制，那原生 EventSource 反而更靠谱，省得自己处理断线重连和事件解析。

第二，记忆体系的设计要奔着"可观测"去。每一层记忆的写入、读取、命中情况都要有日志，否则线上出了问题你根本无从下手。我在项目里给每次记忆查询都加了 trace_id，能顺着一条完整链路看到：当前问题是什么、检索到了哪些记忆、最终注入了几条、模型有没有引用这些记忆。这个投入大大缩短了排查问题的平均时间。

第三，框架封装得再好，核心逻辑还是得自己吃透。DeepAgent 把 SSE 和会话管理的骨架搭好了，但记忆这块基本是留白。接手这样的项目，不要想着"找个框架一步到位"，而是先把这个领域的基本原理摸清，然后针对自己的场景做定制。框架负责省时间，你负责正确性。

这次实战最大的感受是：AI Agent 的工程化，流的实时性只是入门，记忆才决定体验的上限。SSE 是管道，管道通了只是开始；记忆是大脑，大脑还没发育好之前，Agent 充其量是个带打字效果的聊天机器人。后续我会持续跟进长期记忆的迭代，等这块完善了再写一篇更深的拆解。