企业官网建设流程全解析

如果用一句话概括我最近折腾的东西，那就是：我终于把一个 AI Agent 用成了“几乎零存在感”的状态。以前每次等它响应，我都有种想把电脑合上的冲动，倒不是模型回答得不好，而是那个“卡顿感”实在太劝退：问一句话要等三五秒的空白，工具调用稍多就直接像死掉一样，团队里几个人同时一用，请求全排在后面吃灰。这哪里是智能助手，分明是智能压力测试。

“几乎零存在感”这个词，是我自己定的体验标准：打开就能用，问题发出去之后，内容像真人打字一样逐字出现；复杂任务执行到哪一步，界面上能看到哪一步；几个人同时用，也不会互相拖累。它不是一个功能点，而是一整套工程体验。这篇文章就围绕这个标准，把我踩过坑之后的选型思路、代码结构、并发改造和实测数据完整写出来。适合两类人：一是想从 0 到 1 搭 AI Agent 但不知道从哪下手的初学者，二是已经在跑 Agent 但总被“卡顿、超时、抢资源”折磨的开发者。

1. 先说结论：AI Agent 的“存在感”到底差在哪

1.1 卡顿不是模型问题，是工程问题

很多人一遇到 Agent 响应慢，第一反应是“模型不够强”“换个大参数模型”。我一开始也这样想，后来发现方向完全错了。模型 API 本身的响应速度，在一个完整 Agent 链路里往往只占不到一半时间。真正让用户感知到“卡”的，是请求在 Web 层、编排层、工具调用层、传输层之间的排队和阻塞。

我见过一个典型的慢 Agent：用户发一句话，后端先同步调一次模型识别意图，再同步调一次工具查询数据，最后再同步调一次模型生成回答。三个同步调用串起来，中间还有可能遇到工具接口偶发超时。一轮问答跑下来十几秒，用户盯着白屏，只能干等。这就是典型的“存在感过强”——不是 Agent 功能刷存在感，而是等待感和失控感刷存在感。

1.2 “零存在感”的三条硬指标

我把目标拆成三条很容易验证的指标。第一条是首字响应时间足够短，短到用户感觉到“系统已经收到输入并且正在处理”，而不是“页面是不是挂了”。在我的环境里，这个数字至少要做到 1 秒内。第二条是全程流式输出与状态上报，模型生成的 token 实时推给前端，工具调用阶段也不要静默，而是明确显示“正在查询资料”“正在计算数据”这类进度状态。第三条是并发场景下不互相拖累，多个用户同时请求时，不能让一个人把整个进程的事件循环堵死。

这三条指标一环扣一环。第一条要靠异步框架和流式协议，第二条要靠事件流设计，第三条要靠并发控制与资源隔离。它们不是调参能解决的，而是一整套架构选择。

2. 项目拆解：AI Agent 为什么会“卡”

2.1 同步调用阻塞了异步事件循环

先说最基础也最常见的坑。FastAPI 本身是异步框架，很多人也在用，但在async def接口里，却调用了同步的 LLM SDK。这在 Python 里是一个非常隐蔽的问题：同步网络请求一旦发起，就会阻塞当前线程，而 FastAPI 的事件循环也被卡住。表现在用户侧，就是并发一高，所有请求一起变慢，好像大家在排队抢一个资源。

我特意做过对比测试：同一个任务，用同步客户端写在异步接口里，压 20 个并发，所有请求几乎都以相同的慢速度完成；改用异步客户端之后，单个请求的响应速度没有明显变化，但并发场景下整体吞吐提升了好几个量级。所以“卡顿”的第一个责任主体，往往不是模型，而是写进了请求路径里的同步调用。

2.2 非流式响应制造了“空白地狱”

第二个典型卡顿源是“等完整结果”。默认情况下，LLM API 会把生成的整段文字收完再返回。这就意味着用户体验是：发出问题后，先保持五六秒的完全空白，然后突然蹦出一大段文字。这五秒的空白，就是“存在感”最强的时候，因为在用户眼里，系统不是在处理问题，而是疑似死机。

更坑的是，很多 Agent 项目自己也不知道这问题有多严重。有的项目声称做了流式，实际还是等完整结果返回，然后在服务端用split按字符切片一点点吐给前端。这种“假流式”只能骗过肉眼，骗不过网络层，用户等完整结果的时间一点没少。真正要紧的是从模型 API 到前端呈现的整条链路都是流式的，模型出第一个 token，前端立刻显示第一个 token。

2.3 工具编排链成了“毛线团”

当 Agent 从单轮问答扩展到多工具协作，第二个坑就来了：编排写得太乱。我最早用的也是最简单的链式调用：先识别意图，再调工具，再总结输出。等到工具一多，条件分支、循环执行、失败重试开始层层嵌套，代码很快就变成一团乱麻。

这种结构最大的问题还不是丑，而是不可观测。你根本说不清卡在某一步是因为工具慢了，还是因为模型调用太多，还是因为分支逻辑绕了一圈又回去了。我甚至遇到过工具 A 已经拿到结果，但因为条件写错，又跑了一轮多余的模型调用，最后把响应时间拖了一倍。这让我意识到：Agent 编排不该靠手写 if-else 硬怼，而应该交给有状态、有节点概念的执行框架。

2.4 没有超时与重试机制

最后一个容易被忽略的“卡点”：外部接口不是 100% 可靠的。模型服务会偶发超时，工具接口会断连，这些异常如果没有被处理，表现就是请求被无限期“挂住”。没有超时控制就一直等，没有重试就只能等死。

我给所有外部依赖加了三件套：超时时间、带退避的自动重试、失败时的降级文案。尤其是流式接口，断流是新常态。前后端都要处理“中途断掉但已生成的内容还能正常展示”的情况。这三件套看着不起眼，它往往比换一个大模型更能直接降低用户的卡顿体感。

3. 从 0 到 1 搭建“零存在感”Agent：技术选型与最小闭环

3.1 为什么选 FastAPI + LangChain + LangGraph，而不是 Spring AI

有人问过我：现在像 Spring AI 这套企业级方案也挺成熟，为什么不用？我的回答是：看场景。如果是 Java 技术栈的公司，想在企业内部快速落地 Agent 服务，Spring AI 确实值得考虑，尤其是和已有微服务架构、认证体系对接时，Java 生态有天然优势。

但我自己这个项目定位是“个人也能拿起来用、改得动”的轻量 Agent，我更看重三件事：异步原生支持、LLM 生态密度、调试迭代速度。在这一前提下，Python 路线显然更顺手。FastAPI 提供原生 async Web 层，LangChain 负责模型调用和工具封装，LangGraph 负责有状态编排。这套组合让我在一个周末内就能跑通从立项到出 Demo 的完整闭环。

如果你之前完全没有代码经验，也完全可以先在扣子这类可视化开发平台上，把提示词、知识库、工具流调通，验证业务想法。可视化平台适合快速验证，代码方案适合做低延迟和深度定制。两者不冲突，甚至可以先在可视化平台上设计流程，再下沉到代码实现，这条路我试过，效率很高。

3.2 项目骨架：一个可运行的最小结构

不整花活，直接给出我实际使用的目录结构。这个结构足够小，也能支撑后续长成中台：

agent_service/ ├── app/ │ ├── main.py # FastAPI 入口，注册路由和中间件 │ ├── api/ │ │ ├── chat.py # SSE 流式接口 │ │ └── models.py # 请求与响应的 Pydantic 模型 │ ├── agent/ │ │ ├── state.py # Agent 状态定义 │ │ ├── nodes.py # agent 节点、工具节点 │ │ ├── graph.py # LangGraph 状态图构建 │ │ └── tools.py # 工具注册与统一调用入口 │ ├── core/ │ │ ├── config.py # 环境变量与模型配置 │ │ └── llm.py # 模型客户端初始化 │ └── utils/ │ └── retry.py # 超时/重试/退避工具 ├── tests/ └── pyproject.toml

这样的好处是关注点清晰：Web 层不知道编排细节，编排层不知道 HTTP 细节。我后来做并发改造时，基本只动core和utils，agent和api的改动量很小。一个“零存在感”的 Agent，先得有存在感弱的模块边界。

3.3 LangGraph 状态图：把“毛线团”整理成铁路网

LangGraph 解决的核心问题是“可控的循环与状态”。传统的 LangChain 链式调用适合线性流程，但 Agent 特点是必须反复调用模型、判断是否需要工具、工具结果回来后再喂给模型，这本质上是一个循环。LangGraph 把它建模成一张图，由节点和边组成。

直接看核心代码。先是状态定义：

from typing import TypedDict, Annotated, Sequence from langchain_core.messages import AnyMessage from operator import add class AgentState(TypedDict): messages: Annotated[Sequence[AnyMessage], add] remaining_steps: int

这里最关键的是Annotated[Sequence[AnyMessage], add]。它告诉 LangGraph：多个节点返回的消息不是互相覆盖，而是追加合并。如果不做这个声明，后面你就会发现节点 A 返回一条消息，节点 B 返回一条消息，结果状态里只剩一条，前面的全丢了，那 Agent 就会“失忆”。

工具节点的实现也很直接：

from langchain_core.messages import ToolMessage TOOL_MAPPING = {} def tool_node(state: AgentState): last_message = state["messages"][-1] tool_results = [] for tool_call in last_message.tool_calls: tool = TOOL_MAPPING[tool_call["name"]] result = tool.invoke(tool_call["args"]) tool_results.append( ToolMessage( content=str(result)[:2000], tool_call_id=tool_call["id"], ) ) return {"messages": tool_results, "remaining_steps": state["remaining_steps"] - 1}

这里我故意给工具结果加了一个[:2000]的截断，很多人不注意这一点。工具返回的数据可能非常长，如果原封不动塞回模型上下文，不仅浪费 token，还会让模型在生成时被无关信息干扰。截断到关键摘要，效果反而更好。

然后是整张图的拼装：

from langgraph.graph import StateGraph, START, END def agent_node(state: AgentState): response = llm.invoke(state["messages"]) return {"messages": [response], "remaining_steps": state["remaining_steps"] - 1} def should_continue(state: AgentState): if state["remaining_steps"] <= 0: return "end" last_message = state["messages"][-1] if getattr(last_message, "tool_calls", None): return "tools" return "end" builder = StateGraph(AgentState) builder.add_node("agent", agent_node) builder.add_node("tools", tool_node) builder.add_edge(START, "agent") builder.add_conditional_edges( "agent", should_continue, {"tools": "tools", "end": END} ) builder.add_edge("tools", "agent") graph = builder.compile()

这个图跑起来之后，就能很清晰地看到 Agent 的执行轨迹：从 START 进入 agent 节点，模型判断需要工具就进入 tools 节点，工具执行完再回 agent，直到模型不再请求工具为止。整个循环什么时候结束、一共跑了几步，全部可以追踪。这就是从“手写毛线团”到“铁路网”的差别。

4. 让 Agent“能扛并发”的三个关键改造

4.1 全链路 Async：把同步库请出请求路径

并发改造的第一件事，就是把同步调用全部替换成异步调用。这件事说起来简单，做起来有不少细节。以 LangChain 为例，模型调用要优先使用ainvoke而不是invoke，工具调用也要优先写成async函数，节点本身也要定义成async def。任何一环漏掉同步，整个请求路径都会被拖进阻塞的泥潭。

async def async_agent_node(state: AgentState): response = await llm.ainvoke(state["messages"]) return {"messages": [response], "remaining_steps": state["remaining_steps"] - 1}

另外要提醒一个很多人忽略的坑：不要在全局变量里保存会话状态。每个请求进来，都应该基于自己的输入创建独立的图运行实例。LangGraph 本身支持并发运行多个图实例，但前提是你不能把一个全局agent对象当作有状态实例来反复调用。正确做法是把编译好的graph对象当作“模板”，每次请求时都从同一个模板创建新的执行上下文。状态隔离做到位，并发才不会乱。

4.2 SSE 真流式：让 Agent 一边干活一边上报状态

异步可以解决“进程卡死”，但用户感知的“流畅”还要靠流式协议。我这里选择的是 SSE（Server-Sent Events），而不是 WebSocket。原因是 Agent 场景信息流向很单一：主要是服务端往客户端推，不需要客户端频繁上行消息，SSE 更轻、更符合语义。

FastAPI 侧的流式接口长这样：

import json from fastapi.responses import StreamingResponse from langchain_core.messages import HumanMessage def sse_packet(data: dict) -> str: return f"data: {json.dumps(data, ensure_ascii=False)}\n\n" @app.post("/v1/chat/stream") async def chat_stream(req: ChatRequest): user_input = {"messages": [HumanMessage(req.query)], "remaining_steps": 8} async def event_generator(): async for mode, chunk in graph.astream( user_input, stream_mode=["updates", "messages"], ): if mode == "messages": # 这是模型生成的新 token message, _metadata = chunk token = message.content if token: yield sse_packet({"type": "token", "content": token}) elif mode == "updates": # 这是节点状态变化，比如进入工具调用 yield sse_packet({"type": "node", "data": chunk}) return StreamingResponse(event_generator(), media_type="text/event-stream")

这里有两个细节很重要。第一，SSE 的事件必须以data:开头，以\n\n结尾，少一个换行浏览器就不认。第二，stream_mode要同时用updates和messages，才能既拿到节点状态的更新，又拿到模型吐字的流式 token。很多现成教程只用了updates，结果模型生成阶段还是攒一大段才推出来，体验就大打折扣。

前端方面有个障眼法要留意：浏览器原生EventSource只支持 GET，不支持 POST。有人为了用EventSource，把长文本塞进 query string，这种方案既难看又容易触发长度限制。我的建议是直接用fetch读取流式响应：

const response = await fetch("/v1/chat/stream", { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ query: "帮我查询上个月的项目数据" }), }); const reader = response.body.getReader(); const decoder = new TextDecoder("utf-8"); let buffer = ""; while (true) { const { value, done } = 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: ")) { const event = JSON.parse(part.slice(6)); handleAgentEvent(event); } } }

这里把buffer按\n\n切分，是因为 SSE 的分包不一定对齐网络包的边界。如果不做缓冲处理，极容易出现事件被卡成半截甚至 JSON 解析失败。

工具调用阶段也要做“存在感”处理。模型在思考要不要调工具时，可能有好几秒不发新 token，前端如果只看 token 事件，就会显示一屏静止。解决方法是后端在进入工具节点时，主动发一个{"type": "status", "content": "正在调用搜索工具..."}事件，前端收到后展示成进度提示。这样用户就不会觉得 Agent 又卡死了。

4.3 缓存、信号量与有限的并发控制

把并发真正做稳，靠的不只是异步，还要有流量控制。我在服务里加了三层保护。

第一层是工具结果缓存。LLM 的生成结果通常不适合缓存，因为内容多变，但工具结果往往可以。比如查天气、查库存这类接口，短时间内的结果基本一样。我用请求参数的哈希做 key，把工具结果缓存到 Redis，设置 30 到 60 秒过期。这样遇到多人同时问同一个问题，流量只打在缓存上，不需要重复调用外部工具。

第二层是并发信号量。模型 API 有 QPS 上限，如果前端一次性打来 50 个请求，最脆弱的反而是模型服务。我在模型调用入口加了一个asyncio.Semaphore，限制并发数量。

import asyncio model_semaphore = asyncio.Semaphore(10) async def limited_llm_call(messages): async with model_semaphore: return await llm.ainvoke(messages)

这种方法适合单机单进程场景。如果你已经把服务部署成多 worker，甚至多节点，那就需要 Redis 分布式锁或者网关层限流，这属于中台化的范畴。但作为个人项目，先做进程级信号量是最简单可靠的做法。

第三层是超时与熔断。Agent 的执行链路可能涉及多个外部服务，任何一个服务慢，整个请求都会跟着慢。我给模型调用和工具调用都设置了超时时间，超过就把这个 Agent 分支降级为失败分支，而不是无休止地等待。配合指数退避的重试策略，能显著减少偶发超时对用户体验的影响。

实测下来，加上这三层保护之后，高并发场景的错误率明显下降。之前是“并发一高就有人请求超时”，现在是“偶尔个别请求变慢，但不会雪崩”。

5. 实测：首字延迟与并发压测数据

5.1 测试环境与方法

我用自己的开发机作为测试环境：8 核 16G 的普通机器，服务跑在 Docker 里，模型接口走的是一个通用在线大模型 API，Agent 场景包含三个工具调用的中间流程。测试工具用的是locust，从另一台机器发压，避免本机发压影响结果。

需要说明的是，不同模型、不同网络环境、不同工具接口的延迟差异巨大，以下数字不能代表所有环境，但它们反映的相对变化趋势是有参考价值的：同一个 Agent 能力，修改前后的差距，主要来自架构设计而不是机器性能。

5.2 改造前后的核心对比

我整理了一张对比表，左边是改造前的方案：同步调用、非流式、手写链式编排；右边是改造后的方案：全异步、SSE 流式、LangGraph 状态图。

首字响应时间的提升最直观，从近 5 秒压到 1 秒内，用户的体感从“页面卡死”变成了“消息已发送，正在输入”。并发场景下的变化更是惊人：改造前 20 个并发任务几乎是串行排队完成的，改造后并发吞吐能力直接拉开了一个量级。

5.3 主观体验与资源占用

除了数字，我还记录了一些主观体验。改动之前，我每次在群里分享 Agent 给同事用，都会很心虚，因为知道又要有人抱怨“怎么转圈了”“是不是挂了”。改动之后，我可以放心地把链接扔给别人，因为他们看到的是逐字出现的内容，以及工具调用阶段的清晰状态提示。

资源占用方面，异步化之后，同样一台机器能同时服务的请求数量翻了倍，内存占用没有明显增长。CPU 从“动不动满核”变成了“平稳波动”，这要归功于阻塞调用从请求路径中被清走。一个健康的 Agent 服务，CPU 本该有波动，但不该长时间 100% 满载还打不出结果。

6. 调试实录：那些折磨人的卡顿真相

6.1 SSE 一直连不上，浏览器长时间 pending

这是我调试时遇到的第一个大坑。后端接口明明运行了，浏览器却一直 pending，像是数据卡在半路。排查下来原因有两个：一是StreamingResponse的media_type没有设置成text/event-stream，浏览器不认这个响应是事件流；二是 Nginx 开了缓冲，把流式数据攒到一定大小才转发，前端当然看不到实时效果。

解决办法也很直接：media_type设置正确，并且在 Nginx 配置里关掉该路由的proxy_buffering，改成proxy_buffering off;。如果用的是本地开发环境没有 Nginx，那基本就是第一个原因。

6.2 LangGraph 状态被“吃掉了”

LangGraph 的 state 默认会 merge，但 merge 策略必须显式声明。我第一次写状态时没有用Annotated声明消息列表的追加语义，结果工具节点返回的消息总是覆盖掉之前的消息，Agent 起跑即“失忆”。

解决方式是给状态字段加上Annotated[Sequence[AnyMessage], add]。这也让我理解了一个规则：节点函数的返回值不是“单元格覆盖”，而是“增量更新”。想清楚这个模型，写节点的时候就很少再犯错了。

6.3 工具结果太长，模型开始胡言乱语

当我给 Agent 接入一个能返回表格数据的工具时，发现模型生成的最终回答质量急剧下降，会出现张冠李戴的内容。仔细看上下文才发现，工具返回了 50 多行的原始数据，模型在长文本里迷失了重点。

后来我养成了一个习惯：所有工具返回结果都经过一个摘要层，先截断到前 2000 个字符，再根据业务场景决定是否保留全部细节。这个改动直接提升了最终回答的准确性，也降低了 token 消耗，属于性价比很高的“便宜优化”。

6.4 多并发时工具被重复调用

并发压测时发现，同一个工具接口会被重复调用好几次。排查后有两个原因：一个是重试机制对非幂等接口产生了副作用，一个是工具调用失败后没有去重，同一轮 Agent 循环里同一个工具调用被再次发起。

解决办法是：对重试请求增加request_id去重逻辑，确保同一轮执行里同一个工具调用只执行一次；同时把工具函数尽可能设计成可重放的，至少要做到“同一个参数重复调用不会产生副作用”。这一点在中台化之后尤其重要，因为多个用户同时使用，工具调用的并发数和出错概率都会大幅上升。

7. 新手学习路线与后续扩展方向

7.1 从 0 到 1 的五个阶段

如果你现在想学 AI Agent 搭建，我建议按下面这条路线走，每一步都能自己动手验证，不会卡在概念上。

第一阶段，先用可视化平台跑通一个完整场景。比如在扣子这类平台上，创建一个带知识库和工具调用的问答 Agent，把提示词、流程、调试都体验一遍。这个阶段的目标不是写代码，而是理解 Agent 的基本能力边界和交互方式。

第二阶段，用 Python 写第一个“裸 Agent”。不引入任何重框架，直接用模型 API 的 tool calling 接口，写一个循环：调用模型，看是否请求工具，请求了就执行工具并把结果放回消息列表，直到模型不再请求工具。这一步能让你深刻理解 Agent 的核心机制。

第三阶段，引入 LangChain。把模型调用、消息封装、工具定义换成 LangChain 的标准方式，学会ChatPromptTemplate、Tool、create_react_agent这些基础概念。这个阶段你已经有能力做单工具 Agent 了。

第四阶段，上 LangGraph。把之前手写循环的逻辑迁移到状态图里，理解节点、边、状态、条件分支这四个核心概念。学会用astream拿到流式事件后，你就同时掌握了“编排”和“流式”两个关键能力。

第五阶段，工程化。把 FastAPI 接进来，加上 SSE、信号量、缓存、超时重试，最后做一个简单的并发压测。这一步完成，你搭出来的东西就不只是“能跑”，而是“能给别人用”了。

7.2 三个适合练手的小项目

练手项目不必贪大求全。我推荐三个方向，难度递增。

第一个是客服问答 Agent：一个模型加一个知识库检索工具，再配上简单的多轮对话管理，适合练提示词和工具调用。第二个是工作日报生成 Agent：提供几个数据源工具，让它自动汇总数据并生成模板化文档，适合练多工具编排和摘要能力。第三个是定时巡检 Agent：定时唤醒，调用多个检测工具，最后把结果推送出来，适合练异步任务和后台调度。

这三个项目分别对应了 Agent 学习路线中的关键词：工具调用、多步编排、异步与自动化。做完这三个，你就有足够底子去设计自己的 Agent 中台了。

7.3 从单体 Agent 到 Agent 中台

个人项目跑成熟之后，自然会想把它做成一个能被更多人使用的小平台。这时候要做的不是堆代码，而是抽象配置：模型配置、工具注册、提示词模板、权限控制都要变成可配置的项。

Agent 中台本质上就是把这套能力底座化：前端通过统一接口进，后端通过注册中心把所有 Agent 和工具统一管理。针对不同的业务团队，只需要配置不同的提示词和工具集，不需要每个团队都从零搭一套服务。到这一步，你之前做的并发控制、缓存、超时重试、可观测性全部都会派上用场。

如果再往前走，我建议关注 MCP 这类标准协议。它正在逐步成为工具接入的通用语言，让 Agent 有机会像浏览器加载插件一样，动态加载能力。到那时候，“中台”的概念可能会进一步被淡化，取而代之的是更开放、更标准化的 AI 工具生态。

最后再分享一个小技巧：如果你只是自己用这个 Agent，别一上来就想着把它做成中台，也别一上来就追求最复杂的多 Agent 编排。先把“流畅”做到位，把一个最少可用场景打磨到手感顺畅，再逐步加能力。我在这条路上最大的感受就是：卡顿不是玄学，它是一个又一个工程细节叠出来的，也是一层一层能被拆掉和修复的。“零存在感”不是让步，而是把一个 Agent 从“能跑”推向“好用”的必经门槛。