1. 为什么我选择在这个时间点做 DeepAgentDeepAgent 这个概念最近被讨论得很多但真正动手把它从零搭起来的人其实不多。我在过去三周里用 LangChain FastAPI Vue 3 做了一版可运行的 DeepAgent 原型SSE 流式输出已经跑通但长期记忆模块目前还是个半成品。这篇文章不打算讲什么宏大的架构愿景就是把整个搭建过程、踩过的坑、以及目前卡住的地方摊开来说清楚。先说清楚这个项目是什么。DeepAgent 在我的理解里是一个能自主规划任务、调用工具、维护上下文、并且通过流式方式把推理过程实时推给前端的智能体系统。它和普通的 ChatBot 最大的区别在于ChatBot 是一问一答DeepAgent 是接到一个目标后自己拆解步骤、自己决定调什么工具、自己判断什么时候该停下来。适合谁来参考如果你已经写过基础的 LangChain Chain想往 Agent 方向走一步或者你正在做 AI 交互类产品需要一套前后端打通的流式方案那这篇内容应该能帮你省掉不少查文档的时间。技术栈选型上后端用 FastAPIAgent 编排用 LangChain前端用 Vue 3流式传输用 SSE。为什么不用 WebSocket后面会详细说。长期记忆这块我尝试了向量库方案但效果不稳定目前还在迭代中我会把遇到的问题如实写出来。2. 整体架构设计与技术选型逻辑2.1 为什么是 FastAPI 而不是 Flask 或 DjangoFastAPI 在这个场景下的优势非常明显。DeepAgent 的核心交互是流式的需要长时间保持连接并持续推送数据这对异步支持要求很高。FastAPI 原生基于 ASGI配合async def可以轻松处理并发流式请求而 Flask 默认是 WSGI 同步模型做流式输出需要额外引入 gevent 之类的方案复杂度上去了。另一个原因是 FastAPI 的依赖注入系统。Agent 在执行过程中需要用到 LLM 客户端、工具注册表、记忆存储等多个组件用Depends可以把这些组件的初始化逻辑统一管理测试的时候也方便替换。我试过在 Flask 里手动管理这些全局对象代码很快就变得难以维护。Django 就更重了ORM、Admin、模板系统这些在这个项目里基本用不上引入进来只会增加启动时间和认知负担。FastAPI 配合 Pydantic 做请求校验配合 SQLAlchemy 做持久化这套组合在中小型 AI 服务里已经足够。2.2 SSE 和 WebSocket 的取舍这是我在项目初期纠结最久的一个点。SSEServer-Sent Events和 WebSocket 都能实现服务端推送但适用场景不同。SSE 的本质是 HTTP 长连接服务端单向推送文本流。它的优势在于协议简单浏览器原生EventSource支持自动重连走标准 HTTP 端口不需要额外协议升级。对于 DeepAgent 这种用户发一个请求服务端持续推送推理过程的场景SSE 的单项推送模型刚好匹配。WebSocket 是全双工适合需要频繁双向通信的场景比如多人协作编辑、实时游戏。但它的连接管理更复杂需要处理心跳、重连、消息分片等问题。我在这个项目里不需要客户端在流式过程中再发消息所以 SSE 是更轻的选择。不过 SSE 有一个硬限制浏览器对同一域名的 HTTP/1.1 连接数限制是 6 个。如果你的页面同时开多个 SSE 连接会很快耗尽。解决方案是升级到 HTTP/2或者用 HTTP/1.1 时注意控制并发连接数。我在开发阶段就遇到过这个问题调试的时候开了好几个标签页结果新连接一直挂起排查了半天才发现是连接数限制。2.3 LangChain 在 Agent 编排中的角色LangChain 在这个项目里承担的是 Agent 的大脑部分。具体来说我用到了这几个核心模块LLM 封装统一不同模型提供商的调用接口方便切换Tool 抽象把外部能力搜索、计算、数据库查询封装成 Agent 可调用的工具Agent Executor负责编排思考-行动-观察的循环Memory管理对话历史和长期记忆LangChain 和 LangGraph 的区别这里也提一下。LangChain 的 Agent Executor 是一个相对固定的循环结构适合线性推理任务。LangGraph 则是把 Agent 的执行过程建模成图节点和边可以自定义适合需要复杂分支、循环、人工介入的场景。我这个项目目前用 LangChain 的 Agent Executor 够用但如果后续要加 human-in-the-loop 或者复杂的条件分支可能会迁移到 LangGraph。2.4 Vue 3 前端的流式渲染方案前端用 Vue 3 的 Composition API核心是处理 SSE 流的接收和渲染。这里有个细节浏览器原生的EventSource只支持 GET 请求但 DeepAgent 的请求往往需要携带复杂的参数比如工具配置、上下文 ID用 GET 传参不合适。我的解决方案是用fetchReadableStream手动处理 SSE 流。这样可以用 POST 请求发送参数同时通过response.body.getReader()逐块读取数据。配合AbortController可以实现用户主动中断生成。这个方案比EventSource灵活得多代价是需要自己处理重连逻辑。3. 后端核心实现细节3.1 FastAPI 项目目录结构我用的目录结构是这样的deepagent/ ├── app/ │ ├── main.py │ ├── api/ │ │ ├── routes/ │ │ │ ├── agent.py │ │ │ └── health.py │ │ └── deps.py │ ├── core/ │ │ ├── config.py │ │ └── logging.py │ ├── agents/ │ │ ├── executor.py │ │ ├── tools.py │ │ └── prompts.py │ ├── memory/ │ │ ├── short_term.py │ │ └── long_term.py │ ├── models/ │ │ └── schemas.py │ └── db/ │ ├── session.py │ └── models.py ├── tests/ ├── requirements.txt └── .env这个结构的好处是职责清晰。agents/目录放 Agent 相关的逻辑memory/放记忆管理api/放路由和依赖注入。当项目变大时每个模块可以独立演进。3.2 SSE 流式接口的实现核心的流式接口大概长这样from fastapi import APIRouter, Depends from fastapi.responses import StreamingResponse from app.agents.executor import DeepAgentExecutor from app.models.schemas import AgentRequest router APIRouter() router.post(/agent/stream) async def stream_agent( request: AgentRequest, executor: DeepAgentExecutor Depends(get_executor) ): async def event_generator(): async for event in executor.astream(request.query, request.session_id): yield fevent: {event[type]}\n yield fdata: {json.dumps(event[data], ensure_asciiFalse)}\n\n return StreamingResponse( event_generator(), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no, } )这里有几个关键点。第一media_type必须是text/event-stream否则浏览器不会按 SSE 解析。第二X-Accel-Buffering: no这个头很重要如果你前面有 Nginx 反向代理不加这个头 Nginx 会缓冲响应导致流式效果失效。第三SSE 的消息格式必须是event: xxx\ndata: xxx\n\n最后两个换行是消息结束标志少一个都会导致前端解析异常。3.3 Agent Executor 的异步流式改造LangChain 的 Agent Executor 默认是同步的要接入 SSE 需要改成异步流式。我用的是astream_events方法它可以细粒度地暴露 Agent 执行过程中的各种事件class DeepAgentExecutor: def __init__(self, llm, tools, memory): self.agent create_react_agent(llm, tools) self.memory memory async def astream(self, query: str, session_id: str): history await self.memory.get_history(session_id) async for event in self.agent.astream_events( {input: query, chat_history: history}, versionv2 ): kind event[event] if kind on_chat_model_stream: chunk event[data][chunk] if chunk.content: yield {type: token, data: {text: chunk.content}} elif kind on_tool_start: yield {type: tool_start, data: { name: event[name], input: event[data].get(input) }} elif kind on_tool_end: yield {type: tool_end, data: { name: event[name], output: str(event[data].get(output)) }} await self.memory.save(session_id, query)astream_events的versionv2参数必须加v1 的事件格式和 v2 不兼容。事件类型里on_chat_model_stream是 LLM 逐 token 输出on_tool_start和on_tool_end是工具调用的开始和结束。前端可以根据这些事件类型渲染不同的 UI 组件。3.4 工具注册与调用工具这块我用 LangChain 的tool装饰器定义from langchain_core.tools import tool tool def search_knowledge(query: str) - str: 搜索内部知识库输入自然语言查询 results vector_store.similarity_search(query, k3) return \n.join([r.page_content for r in results]) tool def calculate(expression: str) - str: 计算数学表达式输入如 2 3 * 4 try: return str(eval(expression, {__builtins__: {}}, {})) except Exception as e: return f计算错误: {e}工具的 docstring 非常重要Agent 就是靠这个描述来判断什么时候该调用哪个工具。描述要写得具体说明输入格式和功能边界。我一开始写得太简略结果 Agent 经常选错工具。注意eval在生产环境有安全风险这里只是示例。实际项目中应该用ast.literal_eval或者专门的表达式解析库。4. 前端流式渲染与中断控制4.1 用 fetch 处理 SSE 流前面说了不用EventSource的原因这里给出fetch方案的实现async function streamAgent(query, sessionId, onEvent, signal) { const response await fetch(/api/agent/stream, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ query, session_id: sessionId }), 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 messages buffer.split(\n\n) buffer messages.pop() for (const msg of messages) { const lines msg.split(\n) let eventType message let data for (const line of lines) { if (line.startsWith(event: )) eventType line.slice(7) if (line.startsWith(data: )) data line.slice(6) } if (data) onEvent(eventType, JSON.parse(data)) } } }这里的关键是buffer的处理。SSE 消息以\n\n分隔但网络传输是分块的一个消息可能被拆到多个 chunk 里。所以要用 buffer 累积每次按\n\n分割最后一个不完整的部分留在 buffer 里等下一个 chunk。4.2 AbortController 实现中断用户点停止生成的时候需要中断流式请求。AbortController就是干这个的const controller new AbortController() // 开始流式请求 streamAgent(query, sessionId, handleEvent, controller.signal) // 用户点击停止 function stopGeneration() { controller.abort() }abort()调用后fetch的 promise 会 reject 一个AbortErrorreader.read()也会抛出异常。前端需要捕获这个异常并做清理。后端这边FastAPI 的StreamingResponse在客户端断开后会停止生成器但要注意 Agent 的执行可能不会立即停止需要在生成器里检查await request.is_disconnected()。4.3 Vue 3 中的状态管理前端的状态管理用reactive就够了不需要上 Piniaimport { reactive } from vue const state reactive({ messages: [], isStreaming: false, currentToolCall: null }) function handleEvent(type, data) { if (type token) { const last state.messages[state.messages.length - 1] if (last last.role assistant) { last.content data.text } else { state.messages.push({ role: assistant, content: data.text }) } } else if (type tool_start) { state.currentToolCall { name: data.name, input: data.input } } else if (type tool_end) { state.currentToolCall null } }渲染的时候messages数组直接v-for循环工具调用状态单独渲染一个指示器。Vue 3 的响应式系统会自动处理更新不需要手动触发重渲染。5. 长期记忆目前卡住的地方5.1 短期记忆的实现短期记忆就是对话历史我用 SQLAlchemy 存到数据库里class ConversationHistory(Base): __tablename__ conversation_history id Column(Integer, primary_keyTrue) session_id Column(String, indexTrue) role Column(String) content Column(Text) created_at Column(DateTime, defaultdatetime.utcnow)每次请求时取出最近 N 轮对话拼到 prompt 里。N 的大小需要权衡太大 token 消耗高太小上下文不够。我目前设的是 10 轮超过的部分做摘要压缩。5.2 长期记忆的尝试与问题长期记忆我尝试了向量库方案把对话内容 embedding 后存到 Chroma检索时用相似度搜索。但实际跑下来有几个问题。第一个问题是检索质量不稳定。用户问上次我们讨论的那个方案向量检索很难准确找到对应的历史片段因为那个方案本身没有语义信息。这个问题需要结合实体识别和指代消解才能解决超出了当前的范围。第二个问题是记忆的写入时机。如果每轮对话都写入向量库会迅速膨胀检索噪声变大。如果只写入重要信息又需要判断什么算重要这本身就需要一次 LLM 调用增加了延迟和成本。第三个问题是去重。相似的内容反复写入检索时返回一堆重复结果。我试过用相似度阈值去重但阈值设高了漏掉有用信息设低了去重效果差。热词里提到的去重逻辑存在缺陷确实是个真实问题。目前我的临时方案是只对用户显式标记为重要的对话做长期存储检索时结合时间衰减和相似度排序。但这不是一个通用方案还在继续迭代。5.3 记忆模块的后续方向接下来我打算尝试几个方向。一是用 LangGraph 的 checkpointer 机制它天然支持状态持久化和恢复可能比手动管理记忆更优雅。二是引入一个专门的记忆管理 Agent让它来决定什么信息值得长期存储、什么时候该检索记忆。三是尝试分层记忆结构把事实性记忆、偏好性记忆、情景性记忆分开存储和检索。6. 实操中踩过的坑与排查技巧6.1 SSE 连接建立后收不到数据这个问题我遇到过一次排查了很久。原因是 Nginx 的缓冲。默认情况下 Nginx 会缓冲上游响应等缓冲区满了或者连接关闭才发给客户端。对于 SSE 这种持续小数据量的场景表现就是连接建立了但一直没数据。解决方案是在 Nginx 配置里加location /api/agent/stream { proxy_pass http://backend; proxy_buffering off; proxy_cache off; proxy_set_header Connection ; proxy_http_version 1.1; chunked_transfer_encoding off; }同时在 FastAPI 响应头里加X-Accel-Buffering: no。两个地方都要改只改一个可能不生效。6.2 Agent 执行到一半卡住有时候 Agent 会卡在某个工具调用上不返回。排查下来发现是工具函数里有同步阻塞操作比如requests.get没设 timeout。在异步环境里一个同步阻塞调用会卡住整个事件循环。解决方案是把所有工具函数改成异步或者用run_in_executor包装同步调用import asyncio from concurrent.futures import ThreadPoolExecutor executor ThreadPoolExecutor(max_workers4) tool async def fetch_data(url: str) - str: loop asyncio.get_event_loop() return await loop.run_in_executor(executor, sync_fetch, url)6.3 常见问题速查表现象可能原因排查方向SSE 连接建立但无数据Nginx 缓冲检查 proxy_buffering 和 X-Accel-Buffering流式输出断断续续网络分块或代理缓冲检查 buffer 处理逻辑和代理配置Agent 卡住不返回同步阻塞调用检查工具函数是否有阻塞操作前端解析 SSE 报错消息格式不对确认\n\n分隔和 event/data 前缀中断后后端仍在执行未检查断开状态在生成器中加 is_disconnected 检查长期记忆检索不准向量语义局限考虑混合检索或实体增强6.4 几个实操心得第一个心得是关于 prompt 的。Agent 的系统 prompt 里一定要明确工具的使用边界和输出格式要求。我一开始写得太宽松Agent 经常在不该调工具的时候调工具或者输出格式不符合预期。后来加了 few-shot 示例稳定性明显提升。第二个心得是关于错误处理的。Agent 执行过程中任何一步出错都可能导致整个流程失败。我的做法是在工具层面捕获所有异常并返回错误信息字符串让 Agent 自己决定怎么处理而不是让异常向上抛导致流中断。第三个心得是关于日志的。流式场景下调试很麻烦因为你看不到完整的请求响应。我的做法是在每个事件生成时打一条日志记录事件类型和关键数据这样出问题的时候可以回溯整个执行链路。7. 关于这套方案的一些个人体会SSE 这块目前跑得比较稳前端渲染和中断控制都没什么问题。长期记忆确实是半成品向量检索的方案在简单场景下能用但离智能还差得远。我现在的判断是长期记忆不是一个纯技术问题更多是一个产品问题——你需要先定义清楚记住什么和什么时候用技术方案才有意义。LangChain 在这个项目里帮我省了很多事但它的抽象层也带来了一些调试上的困难。有时候出问题了你得一层层往下查最后发现是某个底层组件的行为不符合预期。我的建议是核心链路的关键环节最好自己写一遍不要完全依赖框架的黑盒。FastAPI 的异步流式支持确实好用但要注意所有环节都得是异步的一个同步阻塞就能毁掉整个流式体验。Vue 3 这边fetchReadableStream的方案比EventSource灵活代价是要自己处理分块和重连代码量多一些但可控性更强。如果后续要扩展我优先会做两件事一是把长期记忆迁移到 LangGraph 的 checkpointer 方案上试试二是加一个 human-in-the-loop 的机制让 Agent 在执行敏感操作前先请求确认。这两个方向都需要对现有的执行流程做比较大的改动等有进展了再写一篇。