1. 为什么我要把 DeepAgent 的交互层整个重写一遍去年年底我开始做一个内部用的智能体平台代号 DeepAgent目标很明确让团队里不懂代码的运营同学也能通过自然语言驱动一套自动化流程比如自动整理竞品动态、自动生成周报草稿、自动从一堆测试用例里抽出 UI 自动化脚本骨架。最开始我用的是最省事的方案——前端发一个 POST 请求后端跑完整个 Agent 链路再把结果一次性返回。这个方案在 demo 阶段跑得挺欢但一上真实场景就露馅了一个稍微复杂点的任务LangChain 那边要串好几步工具调用用户盯着转圈的加载图标能等二十多秒中途没有任何反馈体验极差。后来我把交互层换成了 SSE 流式输出前端用 Vue 3 做实时渲染后端用 FastAPI 挂 SSE 端点LangChain 负责 Agent 编排。这一套组合拳打下来首字节时间从二十多秒压到了几百毫秒用户能看到 Agent 一步步在干什么中途还能点“停止”按钮中断。但问题也随之而来——长期记忆这块我到现在都只能算个半成品。标题里说的“SSE 已上线长期记忆还是半成品”就是我此刻的真实状态。这篇东西不是教程也不是什么最佳实践宣言就是我踩完坑之后的一份复盘。如果你也在用 LangChain FastAPI Vue 3 这套技术栈做 Agent 类产品尤其是正在纠结流式输出怎么接、记忆怎么存、中断怎么处理那这篇应该能帮你少走点弯路。我会把每个技术选型背后的“为什么”讲清楚把能直接抄的代码和配置贴出来也会老实交代哪些地方我还没搞定。2. 整体架构设计与技术选型拆解2.1 为什么是 SSE 而不是 WebSocket这是被问得最多的一个问题。热词里“web socket 和 sse”的搜索量一直不低说明很多人卡在这个选择上。我的结论很直接对于“用户发一句话服务端持续推流用户偶尔中断”这种单向为主的场景SSE 是更优解。原因有三。第一SSE 基于标准 HTTPFastAPI 里用StreamingResponse就能实现不需要额外维护 WebSocket 连接状态部署时 Nginx 也不用特殊配置只要关掉缓冲。第二SSE 自带断线重连机制浏览器EventSource会自动处理虽然我在实际项目里因为要支持 POST 传参和自定义 header最终用的是fetchReadableStream手动解析但协议本身的简单性让调试成本低很多。第三WebSocket 是全双工能力更强但强出来的那部分我根本用不上反而要处理心跳、连接池、粘性会话这些额外复杂度。注意SSE 在 HTTP/1.1 下每个域名有 6 个并发连接限制如果你的页面同时开多个 Agent 会话要留意这个坑。HTTP/2 下这个限制会好很多。具体到 DeepAgent我的数据流向是这样的Vue 3 前端通过fetch发一个 POST 请求body 里带用户输入和会话 IDFastAPI 收到后创建一个StreamingResponse用text/event-stream作为 media typeLangChain 的 Agent 在astream_events模式下逐事件产出我筛选出需要展示给用户的事件类型包装成 SSE 格式推给前端前端边收边解析把 token 追加到消息气泡里。2.2 LangChain 在链路里到底扮演什么角色热词里“langchain和langgraph的区别”被搜了很多次我一开始也纠结过。简单说LangChain 更像一套积木提供了模型、工具、提示词模板、输出解析器这些组件以及最基础的 Agent 执行循环LangGraph 则是在 LangChain 之上做的一层状态机编排适合那种有明确节点、边、条件分支的复杂工作流尤其是需要 human-in-the-loop 的场景。DeepAgent 目前的定位是“通用型助手”任务路径相对灵活不需要预先画一张状态图所以我用的是 LangChain 的AgentExecutor配合astream_events。但我要说句实话如果你的 Agent 需要多轮人工确认、需要在某个节点暂停等外部输入LangGraph 会更合适LangChain 原生的 AgentExecutor 在这块比较别扭。我现在的“半成品”长期记忆有一部分原因就是想在 AgentExecutor 上硬做跨会话状态做得很难受。如果重来一次我可能会在记忆这块直接上 LangGraph。2.3 FastAPI 项目目录结构怎么摆才不乱FastAPI 项目目录结构这个搜索词很实在因为项目一复杂文件乱放是灾难。我现在的结构是这样的deepagent/ ├── app/ │ ├── main.py # 应用入口挂路由和中间件 │ ├── api/ │ │ ├── chat.py # SSE 流式对话端点 │ │ └── session.py # 会话管理端点 │ ├── core/ │ │ ├── config.py # 配置用 pydantic-settings │ │ └── agent.py # LangChain Agent 构建逻辑 │ ├── memory/ │ │ ├── short_term.py # 短期记忆基于内存 │ │ └── long_term.py # 长期记忆半成品 │ ├── models/ │ │ └── schemas.py # Pydantic 模型 │ └── db/ │ └── session.py # SQLAlchemy 会话 ├── requirements.txt └── .env这个结构的好处是职责清晰api只管 HTTP 层core管 Agent 构建memory单独拎出来因为这块最需要迭代。用pydantic-settings管理配置环境变量和默认值都在config.py里避免到处os.getenv。3. SSE 流式输出的核心实现细节3.1 后端 FastAPI 端点怎么写才不踩坑先上代码这是api/chat.py的核心部分from fastapi import APIRouter, Request from fastapi.responses import StreamingResponse import json router APIRouter() async def event_generator(agent, user_input: str, session_id: str, request: Request): try: async for event in agent.astream_events( {input: user_input}, versionv2, config{configurable: {session_id: session_id}}, ): # 客户端断开时及时退出避免资源泄漏 if await request.is_disconnected(): break kind event[event] if kind on_chat_model_stream: chunk event[data][chunk] if chunk.content: yield fdata: {json.dumps({type: token, content: chunk.content})}\n\n elif kind on_tool_start: yield fdata: {json.dumps({type: tool_start, name: event[name]})}\n\n elif kind on_tool_end: yield fdata: {json.dumps({type: tool_end, name: event[name]})}\n\n yield fdata: {json.dumps({type: done})}\n\n except Exception as e: yield fdata: {json.dumps({type: error, message: str(e)})}\n\n router.post(/chat/stream) async def chat_stream(request: Request): body await request.json() agent build_agent() return StreamingResponse( event_generator(agent, body[input], body[session_id], request), media_typetext/event-stream, headers{ Cache-Control: no-cache, X-Accel-Buffering: no, # 关键告诉 Nginx 别缓冲 }, )这里有几个细节值得展开。第一X-Accel-Buffering: no这个 header 必须加否则 Nginx 默认会缓冲响应你的流式输出会变成“攒一批再发”用户看到的还是一次性蹦出来。我当初排查这个问题花了整整一个下午一直以为是 LangChain 那边没流式结果是 Nginx 在中间捣乱。第二request.is_disconnected()的检查不能省。用户点了停止按钮或者直接关页面如果不主动退出生成器后端会继续跑完整个 Agent 链路白白消耗 token 和算力。这个检查放在循环里每次事件产出前查一下成本很低。第三SSE 的消息格式必须是data: xxx\n\n两个换行符是消息分隔符少一个前端就解析不出来。我一开始只写了一个\n前端死活收不到完整消息debug 了半天。3.2 前端 Vue 3 怎么接流并实时渲染前端这块我用的是fetchReadableStream没用EventSource原因是EventSource只支持 GET 请求没法带复杂的 POST body也没法自定义 header。核心逻辑封装成一个 composable// composables/useAgentStream.js import { ref } from vue export function useAgentStream() { const messages ref([]) const isStreaming ref(false) let controller null async function send(input, sessionId) { isStreaming.value true controller new AbortController() const assistantMsg { role: assistant, content: , tools: [] } messages.value.push(assistantMsg) try { const resp await fetch(/api/chat/stream, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ input, session_id: sessionId }), signal: controller.signal, }) const reader resp.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 parts buffer.split(\n\n) buffer parts.pop() // 最后一段可能不完整留到下次 for (const part of parts) { if (!part.startsWith(data: )) continue const payload JSON.parse(part.slice(6)) if (payload.type token) { assistantMsg.content payload.content } else if (payload.type tool_start) { assistantMsg.tools.push({ name: payload.name, status: running }) } else if (payload.type tool_end) { const t assistantMsg.tools.find(t t.name payload.name t.status running) if (t) t.status done } } } } catch (e) { if (e.name ! AbortError) { assistantMsg.content \n[连接异常] } } finally { isStreaming.value false controller null } } function abort() { if (controller) controller.abort() } return { messages, isStreaming, send, abort } }这里最关键的坑是 buffer 的处理。SSE 的数据是分块到达的一个data:消息可能被 TCP 拆成两半你不能假设每次reader.read()拿到的都是完整消息。我的做法是维护一个 buffer按\n\n切分最后一段不完整的留在 buffer 里等下一块。这个逻辑如果写错表现就是偶尔丢字或者 JSON 解析报错而且很难复现。AbortController 是中断的核心。用户点停止按钮时调controller.abort()fetch 会抛一个AbortError后端那边request.is_disconnected()也会返回 true两边同时收手。实测下来这个中断响应在 100ms 以内体验很干脆。3.3 流式渲染的性能优化Vue 3 的响应式很香但流式场景下有个性能陷阱如果每个 token 都触发一次组件更新高频输出时页面会卡。我的做法是用requestAnimationFrame做节流把 token 累积到一个临时变量每帧最多更新一次 DOM。另外消息列表用v-memo或者把已完成的消息冻结避免 Vue 反复 diff 整个列表。还有一个细节是自动滚动。流式输出时用户希望看到最新内容但如果用户手动往上滚了就不该强制拉回底部。我监听滚动事件判断用户是否在底部附近只有在底部时才自动滚。4. 长期记忆为什么我说它还是半成品4.1 短期记忆和长期记忆的边界在哪先把概念理清楚。短期记忆是单次会话内的上下文用户这一轮说的话、Agent 调过的工具、中间结果这些需要传给模型让它保持连贯。长期记忆是跨会话的用户上周问过什么、偏好是什么、哪些结论已经确认过这些要持久化下次开新会话还能用上。短期记忆我用的是 LangChain 的ConversationBufferWindowMemory只保留最近 N 轮防止 token 爆炸。这块跑得挺稳没什么好说的。真正麻烦的是长期记忆。4.2 我试过的三种长期记忆方案方案一全量存数据库每次检索最近 N 条塞进 prompt。这是最土的办法用 SQLAlchemy 建一张messages表按user_id和created_at查。问题是它只有“最近”没有“相关”。用户三个月前说过一句关键偏好早就被淹没了。方案二向量检索。把历史对话切片、embedding、存进向量库我用的是 Chroma热词里“ollama langchain chroma 如何搭建本地知识库”也是这个路子每次新会话时用当前输入去检索最相关的 K 条历史。这个方案理论上更聪明但实际用下来有几个问题切片粒度很难定切太细丢上下文切太粗检索不准embedding 模型对中文口语的语义捕捉一般检索出来的历史片段塞进 prompt 后模型有时候会“串戏”把旧话题的结论套到新问题上。方案三让 Agent 自己决定记什么。这是我现在在试的方向给 Agent 一个save_memory工具让它判断哪些信息值得长期保留主动写入。好处是记忆质量高坏处是模型经常该记的不记、不该记的乱记而且多一次工具调用就多一次延迟。4.3 半成品的具体表现和根因说它半成品是因为现在这套东西有三个明显缺陷。第一记忆写入没有去重。热词里提到“去重逻辑存在缺陷”我深有同感。用户重复说同一件事向量库里就存了多条几乎一样的记录检索时全被捞出来浪费 token 还干扰判断。我试过用余弦相似度做去重阈值定在 0.92但中文短句的相似度计算很不稳定经常误杀。第二记忆没有时效性管理。用户上个月说“我最近在学 Rust”这个月可能已经学完了但这条记忆还在Agent 还会拿它当当前状态。理想情况应该给记忆加时间衰减或者过期机制但我还没想清楚怎么设计。第三检索和生成的耦合太紧。现在是“检索 → 塞 prompt → 生成”一条直线检索不准就直接污染生成。更合理的做法可能是让 Agent 先判断“这个问题需不需要查长期记忆”需要再查但这样又多了一步推理开销。实操心得如果你现在也要做长期记忆我的建议是先别上向量库。用最简单的“结构化字段 规则”撑过第一阶段比如只记用户的显式偏好语言、时区、常用工具这些用键值对存就够了。等真实数据积累起来你才知道哪些记忆真正有价值再针对性地上向量检索。我一开始就上 Chroma结果发现 80% 的检索都是噪音。5. 实操过程中踩过的坑与排查实录5.1 SSE 连接建立但收不到数据这是最经典的问题。表现是前端 fetch 成功resp.ok为 true但reader.read()一直 pending。排查顺序先看后端有没有真的在 yield加日志确认再看 Nginx 配置proxy_buffering off和X-Accel-Buffering: no都要有最后看中间有没有别的代理层。我遇到过一次是公司网关做了响应缓冲改网关配置才解决。5.2 LangChain astream_events 事件类型对不上astream_events的 version 参数很关键v1 和 v2 的事件结构不一样。我一开始照着旧文档写事件名对不上什么都筛不出来。确认版本后用on_chat_model_stream拿 tokenon_tool_start/on_tool_end拿工具状态基本够用。如果用的是自定义工具记得工具内部如果有 LLM 调用也会产生嵌套事件需要按event[parent_ids]过滤。5.3 中断后后端还在跑前面提过request.is_disconnected()但有个细节这个检查在生成器里是异步的如果 Agent 某一步是同步阻塞的比如某个工具函数是同步的检查会失效。解决办法是把所有工具都写成 async或者在关键节点手动插入检查点。5.4 常见问题速查表现象可能原因排查方向前端收不到流Nginx 缓冲检查X-Accel-Buffering和proxy_buffering消息被截断buffer 切分逻辑错确认按\n\n切分且保留残段中断无效同步阻塞工具工具改 async插入断开检查记忆检索串戏切片粒度或阈值问题调切片大小加时间过滤token 消耗异常高记忆全量塞 prompt限制检索条数加去重页面卡顿每 token 触发渲染用 rAF 节流冻结历史消息5.5 几个不那么明显但很要命的细节CORS 配置。SSE 跨域时Access-Control-Allow-Origin不能是*如果带了 credentials而且预检请求要放行。我因为 CORS 问题折腾过一轮最后用 FastAPI 的CORSMiddleware显式配置才稳。超时设置。SSE 连接可能持续几分钟Nginx 的proxy_read_timeout默认 60 秒长任务会被掐断。我把它调到 600 秒同时在应用层加了心跳每 15 秒发一个注释行: ping\n\n双保险。内存泄漏。每个 SSE 连接都会创建一个 Agent 实例如果 Agent 里持有大对象比如加载了本地模型连接多了内存会爆。我的做法是 Agent 构建逻辑做成轻量的重资源模型、向量库客户端用单例或者连接池复用。6. 后续可以怎么扩展长期记忆这块我打算下一步试试 LangGraph 的 checkpointer 机制它原生支持把状态持久化到数据库配合 human-in-the-loop 应该能做出更自然的“记忆确认”流程——Agent 觉得某条信息值得记先问用户一句“这个要记住吗”用户确认了再写。这样既解决了乱记的问题也让用户对记忆有掌控感。另外热词里提到的“基于 langchain 开发一个能读取测试用例自动生成 UI 自动化测试脚本的 agent”其实和我这个平台是能结合的。如果长期记忆做扎实了Agent 就能记住这个项目常用的页面元素定位方式、常用的断言风格生成脚本时直接套用不用每次重新问。这大概就是长期记忆真正的价值所在——不是记住对话而是记住“这个用户、这个团队是怎么干活的”。我现在这套东西SSE 那部分你可以直接抄代码贴出来就是能跑的。长期记忆那部分当个反面教材看避开我踩的坑然后按你自己的场景重新设计。毕竟记忆这东西没有通用解只有贴着业务长出来的解。