1. 从标题说起为什么“SSE 已上线长期记忆还是半成品”值得单独聊“DeepAgent 实战SSE 已上线长期记忆还是半成品”——这个标题本身就带着一股很真实的味道。做过 Agent 项目的人一眼就能看懂流式输出这条链路已经跑通了用户能实时看到模型一个字一个字往外蹦体验层面算是及格了但真正让 Agent 从“玩具”变成“工具”的长期记忆模块还处在一个能用但不敢说好用的状态。我最近几个月一直在折腾基于 LangChain 和 FastAPI 的 Agent 服务前端用 Vue 3 做交互层中间靠 SSE 把大模型的输出流推到浏览器。这套组合在社区里讨论度很高LangChain 负责编排 Agent 的思考与工具调用FastAPI 提供异步接口SSE 承担流式传输Vue 3 负责实时渲染。听起来是一条很顺的链路但实际落地的时候长期记忆这块才是真正让人头疼的地方。这篇文章想聊的就是这个“半成品”状态SSE 流式输出到底怎么和 LangChain 的 Agent 配合FastAPI 侧怎么封装 AI 交互逻辑Vue 3 前端怎么接住流并配合 abort 做中断以及长期记忆为什么难做、我目前踩到了哪些坑、有哪些还算靠谱的折中方案。适合正在做 Agent 应用、已经跑通基础对话但卡在记忆和流式细节上的朋友。如果你还在纠结 LangChain 和 LangGraph 的区别、FastAPI 项目目录怎么组织文中也会顺带把这些基础问题讲清楚。2. 整体架构设计与技术选型思路2.1 为什么是 SSE 而不是 WebSocket先说一个被问得最多的问题流式输出到底该用 SSE 还是 WebSocket。我一开始也纠结过后来实测下来对于“大模型回答实时渲染”这个场景SSE 是更省心的选择。SSE 本质上是基于 HTTP 的单向流服务端持续往客户端推文本客户端只管接收。WebSocket 是全双工能力更强但代价是要维护连接状态、处理心跳、做重连逻辑服务端还得为每个连接保留上下文。Agent 对话这个场景里绝大多数时候是“用户发一次请求服务端流式返回一大段内容”单向推送完全够用。用 WebSocket 属于杀鸡用牛刀反而增加了复杂度和出错面。还有一个很实际的原因SSE 走的是标准 HTTP天然兼容现有的鉴权、网关、日志体系。你不需要为它单独开端口、单独做负载均衡配置。FastAPI 里用StreamingResponse就能直接返回 SSE 流配合text/event-stream的 media type前端用EventSource或者 fetch 的 ReadableStream 都能接。不过 SSE 有个众所周知的限制浏览器原生的EventSource只支持 GET 请求没法带复杂的 POST body。Agent 对话往往需要传一大坨上下文、工具配置、会话 ID用 GET 拼 query 参数很不优雅。所以我的做法是放弃EventSource改用fetchReadableStream手动解析 SSE 格式。这样既能 POST又能配合AbortController做中断一举两得。2.2 LangChain 在链路里到底扮演什么角色很多人对 LangChain 的定位是模糊的。我自己的理解是LangChain 是 Agent 的“编排层”它不负责网络传输也不负责前端渲染它管的是“模型该调用哪个工具、工具返回后怎么继续推理、多轮对话的上下文怎么拼”。在这个项目里LangChain 的 Agent 负责几件事接收用户输入结合历史消息和记忆决定是否调用工具把工具结果喂回模型最终生成回答。而 FastAPI 负责把 Agent 的输出包装成 SSE 流Vue 3 负责消费这个流。这里要澄清一个高频疑问LangChain 和 LangGraph 的区别。简单说LangChain 的 Agent 是线性的、相对固定的执行流程适合大多数常规场景LangGraph 把执行过程建模成图节点和边可以自定义支持循环、条件分支、人工介入human in the loop。如果你要做复杂的多步推理、需要人在环里审批LangGraph 更合适。但如果只是“对话 工具调用”LangChain 的 Agent 足够别为了追新而上 LangGraph徒增学习成本。2.3 FastAPI 侧的项目目录结构FastAPI 项目最容易写乱尤其是 Agent 这种涉及模型、工具、记忆、流式输出的场景。我目前用的目录结构大致是这样app/ ├── main.py # 应用入口注册路由和中间件 ├── api/ │ ├── chat.py # SSE 流式对话接口 │ └── memory.py # 记忆相关接口 ├── core/ │ ├── config.py # 配置管理 │ └── agent.py # LangChain Agent 初始化 ├── services/ │ ├── llm.py # 模型调用封装 │ └── memory.py # 记忆读写逻辑 ├── models/ │ └── schemas.py # Pydantic 数据模型 └── db/ └── session.py # 数据库会话这个结构的好处是职责清晰api只管请求响应services管业务逻辑core管初始化和配置。Agent 的初始化放在core/agent.py避免每次请求都重新构建 Agent 对象——这个坑我踩过每次请求都 new 一个 Agent延迟直接翻倍。3. SSE 流式输出的核心实现细节3.1 FastAPI 侧怎么把 Agent 输出变成 SSE 流FastAPI 返回 SSE 流核心是用StreamingResponse包一个异步生成器。生成器里不断 yield 符合 SSE 格式的字符串。SSE 的格式很简单每条消息以data:开头以两个换行结尾from fastapi import FastAPI from fastapi.responses import StreamingResponse import json app FastAPI() async def event_generator(user_input: str): async for chunk in agent.astream({input: user_input}): # chunk 是 Agent 流式返回的片段 payload json.dumps({content: chunk}, ensure_asciiFalse) yield fdata: {payload}\n\n # 结束标记前端据此关闭连接 yield data: [DONE]\n\n app.post(/chat/stream) async def chat_stream(payload: ChatRequest): return StreamingResponse( event_generator(payload.input), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no, }, )这里有几个细节必须注意。第一X-Accel-Buffering: no这个头很关键如果你前面挂了 Nginx不加这个头Nginx 会缓冲整个响应再一次性吐给客户端流式效果直接没了。第二media_type必须是text/event-stream否则前端解析会出问题。第三结束标记[DONE]是我自己约定的方便前端知道流结束了你也可以用别的标记。3.2 LangChain Agent 的流式调用LangChain 的 Agent 支持astream方法能逐块返回。但要注意Agent 的流式返回和纯 LLM 的流式返回不一样。纯 LLM 流式返回的是 tokenAgent 流式返回的可能是中间步骤、工具调用、最终回答混在一起。你需要根据 chunk 的类型做过滤只把最终回答推给前端。async def stream_agent_response(agent, user_input, history): async for event in agent.astream_events( {input: user_input, chat_history: history}, versionv1, ): kind event[event] # 只处理 LLM 输出的 token if kind on_chat_model_stream: content event[data][chunk].content if content: yield content用astream_events比astream更细粒度能精确区分“模型在思考”“工具在调用”“模型在输出”。我一开始用astream结果工具调用的中间结果也被推到前端了用户看到一堆乱七八糟的东西。换成astream_events后只过滤on_chat_model_stream输出就干净了。3.3 Vue 3 前端怎么接住流并实时渲染前端这块我用的是 fetch ReadableStream而不是 EventSource。原因前面说了EventSource 不支持 POST。核心逻辑是读流、按 SSE 格式切分、逐条解析async function streamChat(input, onChunk, signal) { const response await fetch(/chat/stream, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ input }), signal, }); 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 }); // 按双换行切分完整消息 const parts buffer.split(\n\n); buffer parts.pop(); // 最后一段可能不完整留到下次 for (const part of parts) { if (!part.startsWith(data: )) continue; const data part.slice(6); if (data [DONE]) return; const parsed JSON.parse(data); onChunk(parsed.content); } } }这里最容易出错的是 buffer 的处理。流是分块到达的一个 SSE 消息可能被切成两半所以必须用 buffer 缓存不完整的部分等下一块数据到了再拼。我一开始没做这个处理偶尔会出现 JSON 解析失败排查了半天才发现是消息被截断了。3.4 配合 AbortController 做中断用户点了“停止生成”你得真的把请求掐掉不然模型还在后台跑浪费算力。前端用AbortControllerconst controller new AbortController(); streamChat(input, onChunk, controller.signal); // 用户点停止 function stopGeneration() { controller.abort(); }但光前端 abort 不够服务端也得感知到。FastAPI 里可以通过request.is_disconnected()检测客户端是否断开from fastapi import Request app.post(/chat/stream) async def chat_stream(payload: ChatRequest, request: Request): async def event_generator(): async for chunk in agent.astream(...): if await request.is_disconnected(): break # 客户端断开停止生成 yield fdata: {chunk}\n\n return StreamingResponse(event_generator(), media_typetext/event-stream)这个检测不是实时的但足够用。实测下来用户点停止后服务端大概几百毫秒内能感知到并中断不会造成明显的资源浪费。4. 长期记忆为什么它还是半成品4.1 长期记忆到底难在哪短期记忆好做把当前会话的消息列表塞进 prompt 就行。长期记忆难难在三个地方存什么、怎么存、怎么取。存什么不是所有对话都值得记。用户随口一句“今天天气不错”没必要进长期记忆但“我下个月要搬家到杭州”这种信息就得记住。判断哪些信息值得长期保存本身就需要模型参与这就引入了额外的调用成本和延迟。怎么存向量数据库是主流方案Chroma、Milvus、pgvector 都能用。但向量检索有个天然缺陷它擅长语义相似不擅长精确匹配。用户问“我上次说的那个项目叫什么来着”向量检索可能召回一堆相关但不精确的内容。怎么取每次对话都把所有长期记忆塞进 prompt 不现实token 成本扛不住。得做相关性筛选但筛选逻辑做不好要么漏掉关键信息要么塞进一堆噪音。4.2 我目前的折中方案我现在用的是“向量库 结构化字段”的混合方案。向量库存语义记忆比如用户的偏好、背景、历史讨论过的话题结构化字段存关键事实比如用户 ID、项目名、时间节点。检索的时候两路并行向量路召回语义相关的结构化路精确匹配关键字段最后合并去重。去重这块我踩过坑。LangChain 默认的 RRFReciprocal Rank Fusion实现在某些场景下去重逻辑是有缺陷的会把本该保留的不同来源的结果误判为重复。我的做法是自己写合并逻辑用内容哈希做去重而不是依赖框架默认实现。def merge_memories(vector_results, structured_results): seen set() merged [] for item in vector_results structured_results: key hash(item[content]) if key in seen: continue seen.add(key) merged.append(item) return merged这个逻辑简单粗暴但实测比框架默认的靠谱。代价是可能漏掉一些语义相同但表述不同的重复项不过对记忆场景来说宁可多留一点也别误删。4.3 记忆写入的时机问题另一个坑是记忆写入的时机。一开始我在每轮对话结束后同步写记忆结果发现两个问题一是写记忆本身要调模型做摘要和抽取增加了响应延迟二是有些对话还没结束用户可能还在补充信息过早写入会导致记忆碎片化。后来改成异步写入对话结束后丢到后台任务里处理。FastAPI 里用BackgroundTasks就能做from fastapi import BackgroundTasks app.post(/chat/stream) async def chat_stream(payload: ChatRequest, background_tasks: BackgroundTasks): # ... 流式返回逻辑 background_tasks.add_task(extract_and_save_memory, payload.session_id, full_response)这样用户感知不到写记忆的延迟体验好很多。但异步写入带来新问题如果服务重启还没处理完的后台任务就丢了。生产环境得用消息队列兜底我目前还在用内存队列属于“半成品”的一部分。5. 常见问题与排查技巧实录5.1 SSE 流式输出常见问题速查问题现象可能原因排查方向前端一次性收到全部内容Nginx 缓冲了响应加X-Accel-Buffering: no头流中途断开网关超时调大网关 timeout加心跳JSON 解析失败消息被分块截断前端加 buffer 拼接中文乱码编码问题用 TextDecoder 指定 utf-8停止按钮无效服务端未检测断开加request.is_disconnected()5.2 记忆检索召回不准怎么办召回不准通常有两个原因embedding 模型不适合中文或者 chunk 切分粒度不对。中文场景建议用专门的中文 embedding 模型别直接用英文模型硬套。chunk 切分上我试过按固定长度切、按句子切、按语义切最后发现按“对话轮次”切效果最好——一轮问答作为一个 chunk语义完整性最好。5.3 LangChain 版本升级导致的兼容问题LangChain 迭代很快小版本升级经常有 breaking change。我踩过一次升级后astream_events的返回结构变了前端解析直接崩。建议锁死版本升级前先在测试环境跑一遍完整流程。conda 环境里用pip freeze导出依赖别用这种模糊版本号。5.4 实操心得三条第一别在请求里初始化 Agent。Agent 初始化涉及模型加载、工具注册开销很大放在应用启动时做一次就行。第二SSE 的心跳不能省。长时间没有数据推送中间的网络设备可能把连接掐了。每隔 15 秒推一个注释行: heartbeat\n\n保持连接活跃。第三记忆模块先做“能用”别一上来追求“好用”。我一开始想做一个完美的记忆系统结果卡了两周没进展。后来退一步先用最简单的向量检索跑通再逐步优化反而推进得快。6. 后续可以继续打磨的方向长期记忆这块我接下来想试的是分层记忆把记忆分成“会话级”“用户级”“全局级”三层不同层级用不同的存储和检索策略。会话级的放内存用户级的放向量库全局级的做定期摘要。这样检索的时候可以按需加载避免一次性拉太多。另一个方向是记忆的“遗忘机制”。人脑会遗忘Agent 的记忆系统也应该有。长期不用的记忆降低权重甚至归档避免记忆库无限膨胀。这个目前还只是想法没动手实现。SSE 这条链路目前算是稳了但还有个优化点断线重连。用户网络抖动导致流断了能不能从断点续传SSE 原生支持Last-Event-ID但需要服务端配合记录已发送的消息 ID。这个我还没做属于下一个要填的坑。说实话Agent 这个方向流式输出只是门面记忆才是内核。门面好做内核难啃。我现在这个状态SSE 上线了记忆半成品但至少方向是清楚的。踩过的坑都记下来了后面慢慢填。