企业官网建设流程全解析

1. 为什么我选择在这个时间点做 DeepAgent

DeepAgent 这个概念最近被讨论得很多，但真正动手把它从零搭起来的人其实不多。我在过去三周里用 LangChain + FastAPI + Vue 3 做了一版可运行的 DeepAgent 原型，SSE 流式输出已经跑通，但长期记忆模块目前还是个半成品。这篇文章不打算讲什么宏大的架构愿景，就是把整个搭建过程、踩过的坑、以及目前卡住的地方摊开来说清楚。

先说清楚这个项目是什么。DeepAgent 在我的理解里，是一个能自主规划任务、调用工具、维护上下文、并且通过流式方式把推理过程实时推给前端的智能体系统。它和普通的 ChatBot 最大的区别在于：ChatBot 是一问一答，DeepAgent 是接到一个目标后自己拆解步骤、自己决定调什么工具、自己判断什么时候该停下来。适合谁来参考？如果你已经写过基础的 LangChain Chain，想往 Agent 方向走一步，或者你正在做 AI 交互类产品，需要一套前后端打通的流式方案，那这篇内容应该能帮你省掉不少查文档的时间。

技术栈选型上，后端用 FastAPI，Agent 编排用 LangChain，前端用 Vue 3，流式传输用 SSE。为什么不用 WebSocket？后面会详细说。长期记忆这块我尝试了向量库方案，但效果不稳定，目前还在迭代中，我会把遇到的问题如实写出来。

2. 整体架构设计与技术选型逻辑

2.1 为什么是 FastAPI 而不是 Flask 或 Django

FastAPI 在这个场景下的优势非常明显。DeepAgent 的核心交互是流式的，需要长时间保持连接并持续推送数据，这对异步支持要求很高。FastAPI 原生基于 ASGI，配合async def可以轻松处理并发流式请求，而 Flask 默认是 WSGI 同步模型，做流式输出需要额外引入 gevent 之类的方案，复杂度上去了。

另一个原因是 FastAPI 的依赖注入系统。Agent 在执行过程中需要用到 LLM 客户端、工具注册表、记忆存储等多个组件，用Depends可以把这些组件的初始化逻辑统一管理，测试的时候也方便替换。我试过在 Flask 里手动管理这些全局对象，代码很快就变得难以维护。

Django 就更重了，ORM、Admin、模板系统这些在这个项目里基本用不上，引入进来只会增加启动时间和认知负担。FastAPI 配合 Pydantic 做请求校验，配合 SQLAlchemy 做持久化，这套组合在中小型 AI 服务里已经足够。

2.2 SSE 和 WebSocket 的取舍

这是我在项目初期纠结最久的一个点。SSE（Server-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 传参不合适。

我的解决方案是用fetch+ReadableStream手动处理 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 f"event: {event['type']}\n" yield f"data: {json.dumps(event['data'], ensure_ascii=False)}\n\n" return StreamingResponse( event_generator(), media_type="text/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}, version="v2" ): 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的version="v2"参数必须加，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, k=3) 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 一个AbortError，reader.read()也会抛出异常。前端需要捕获这个异常并做清理。后端这边，FastAPI 的StreamingResponse在客户端断开后会停止生成器，但要注意 Agent 的执行可能不会立即停止，需要在生成器里检查await request.is_disconnected()。

4.3 Vue 3 中的状态管理

前端的状态管理用reactive就够了，不需要上 Pinia：

import { 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_key=True) session_id = Column(String, index=True) role = Column(String) content = Column(Text) created_at = Column(DateTime, default=datetime.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_workers=4) @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 常见问题速查表

6.4 几个实操心得

第一个心得是关于 prompt 的。Agent 的系统 prompt 里一定要明确工具的使用边界和输出格式要求。我一开始写得太宽松，Agent 经常在不该调工具的时候调工具，或者输出格式不符合预期。后来加了 few-shot 示例，稳定性明显提升。

第二个心得是关于错误处理的。Agent 执行过程中任何一步出错都可能导致整个流程失败。我的做法是在工具层面捕获所有异常并返回错误信息字符串，让 Agent 自己决定怎么处理，而不是让异常向上抛导致流中断。

第三个心得是关于日志的。流式场景下调试很麻烦，因为你看不到完整的请求响应。我的做法是在每个事件生成时打一条日志，记录事件类型和关键数据，这样出问题的时候可以回溯整个执行链路。

7. 关于这套方案的一些个人体会

SSE 这块目前跑得比较稳，前端渲染和中断控制都没什么问题。长期记忆确实是半成品，向量检索的方案在简单场景下能用，但离"智能"还差得远。我现在的判断是，长期记忆不是一个纯技术问题，更多是一个产品问题——你需要先定义清楚"记住什么"和"什么时候用"，技术方案才有意义。

LangChain 在这个项目里帮我省了很多事，但它的抽象层也带来了一些调试上的困难。有时候出问题了，你得一层层往下查，最后发现是某个底层组件的行为不符合预期。我的建议是，核心链路的关键环节最好自己写一遍，不要完全依赖框架的黑盒。

FastAPI 的异步流式支持确实好用，但要注意所有环节都得是异步的，一个同步阻塞就能毁掉整个流式体验。Vue 3 这边，fetch+ReadableStream的方案比EventSource灵活，代价是要自己处理分块和重连，代码量多一些但可控性更强。

如果后续要扩展，我优先会做两件事：一是把长期记忆迁移到 LangGraph 的 checkpointer 方案上试试，二是加一个 human-in-the-loop 的机制，让 Agent 在执行敏感操作前先请求确认。这两个方向都需要对现有的执行流程做比较大的改动，等有进展了再写一篇。