企业官网建设流程全解析

做了大半年 AI Agent 相关项目，从最早用 LangChain 写 demo，到后来搭 FastAPI + LangGraph 上生产，中间踩了不少坑，也总结出一些实在的经验。最近看到不少人在问 agent 怎么扛并发、怎么部署、token 怎么控，干脆把这段时间的实践整理成一篇，把我觉得真正有用的东西写出来。

这篇文章不聊那些花哨的概念，只讲实际干活时遇到的问题：架构怎么选、并发怎么处理、token 怎么省、上下文怎么管、线上出问题了怎么排查。适合已经在接触 AI Agent、正准备从 demo 往生产环境推的开发者，也适合刚入门但对"Agent 到底是什么"还没完全吃透的朋友。

1. 先说清楚：AI Agent 到底是什么

1.1 别把 Agent 当成 ChatBot

我见过很多项目，表面上叫 AI Agent，实际上就是一个 ChatBot 套了层壳——用户输入一句，模型回一句，最多加个 system prompt 说是"角色扮演"。这不算 Agent。

真正的 Agent 核心在于它具备"自主行动"的能力：拿到一个目标后，能自己拆分任务、调用工具、根据结果调整下一步。比如你让它"帮我查一下这周的销售数据并生成一份报告"，它需要先调用数据库查询接口，拿到数据后可能发现数据不完整，再调用另一个接口补充，最后调一个生成文档的工具。这个过程中的每一步，是 Agent 自己决策出来的，不是提前硬编码的 if-else。

我用一个简单类比来理解这件事：ChatBot 是客服，你问一句它答一句；Agent 是员工，你布置一个任务，它自己规划怎么做、找什么人配合、遇到问题怎么解决，最后给你交付结果。

这个区别决定了架构设计的走向。做 ChatBot 时，你只需要关心 prompt 和上下文窗口；做 Agent 时，你还得关心工具注册、状态管理、循环终止机制、错误恢复这些从来没遇到过的问题。

1.2 Agent 的核心三要素：规划、工具、记忆

拆开来看，一个能落地的 Agent 系统通常包含三块：

• 规划（Planning）：大模型根据用户目标，拆解出需要执行的步骤。这一步依赖模型的推理能力，不同模型的表现差异巨大。实操中我的体会是，规划能力强的模型能省掉大量重试和修正逻辑，不要太指望用小模型做复杂规划。
• 工具（Tools）：Agent 能调用的外部能力，比如搜索引擎、数据库查询、HTTP API、文件读写。工具定义的清晰程度，直接影响模型调用的成功率。
• 记忆（Memory）：短期记忆是当前任务上下文，长期记忆则是跨会话的用户偏好、历史结果。很多 Agent 翻车就翻在记忆管理上——上下文窗口越塞越满，最后模型连最初的目标都忘了。

这三块缺一块，Agent 就只能算"半成品"。我自己的项目中，规划能力靠模型本身，工具和记忆是代码侧可以重点优化的地方，后面我会详细展开这两块的实现细节。

2. 架构选型：为什么我用了 FastAPI + LangGraph

2.1 LangChain 的问题在哪里

最早我用的是 LangChain。说实话，LangChain 的生态确实全，各种链、各种模块都有，文档看起来也很完善。但真正做项目时我发现几个问题：

第一，抽象层太重。LangChain 把很多东西包装成了统一的接口，看起来很方便，但出了问题很难调试。比如某个环节报错，它的堆栈信息会跨好几个抽象层，排查起来特别费劲。

第二，流程控制不够直观。Agent 的循环、分支、条件判断，在 LangChain 的 Chain 模型里很难表达。你想实现"如果工具 A 返回异常，则走工具 B"这种逻辑，得嵌套好多层 LCEL，写出来的代码可读性很差。

第三，维护过山车式。LangChain 的 API 变更频繁，我遇到过好几次升级小版本后老代码不兼容的情况。对于生产项目，这很致命。

2.2 LangGraph 的图状态机制

后来我切到了 LangGraph。LangGraph 把 Agent 的执行流程建模成一张图，节点（Node）是具体的处理步骤，边（Edge）是节点之间的跳转条件。这种表达方式天然适合 Agent 的循环和分支逻辑。

举个例子，一个典型的 Agent 流程图里有这些节点：

• plan_node：模型接收用户输入，输出计划（或直接输出工具调用指令）。
• tool_node：执行工具调用，返回结果。
• decide_node：判断当前结果是最终答案，还是需要继续调用工具。

这三个节点之间形成循环，decide_node 决定是跳到 tool_node 继续执行，还是跳到输出节点结束。LangGraph 里用add_conditional_edges实现这种动态跳转，比 LangChain 的链式拼接清晰得多。

我实际体会最深的是它的状态机制。LangGraph 维护一个全局的 State 对象，各个节点都能读和写这个 State。这意味着你在任何节点都能拿到完整的执行历史，做上下文管理、做审计日志都非常方便。在 LangChain 里想拿到完整执行链路，要自己想办法往 chain 里塞变量，体验差很多。

2.3 FastAPI 承担的角色

FastAPI 在网络服务层做三件事：接收用户请求、把任务投递给 Agent 执行引擎、返回结果。它本身不包含 Agent 逻辑，只负责 API 层和并发管理。

我选择 FastAPI 的原因很实际：

• 原生 async/await 支持，协程并发处理高 IO 场景（调用 LLM API 就是典型的高 IO 场景）。
• Pydantic 做数据校验，用起来顺手，定义工具参数 schema 时比较方便。
• 自动生成 OpenAPI 文档，联调省事。
• 部署方便，一个 Uvicorn 进程就能跑起来。

整体架构简单画一下：FastAPI 接收请求 -> 创建任务放入队列 -> Worker 进程消费任务并执行 LangGraph 图 -> 结果写回 -> 前端轮询或 WebSocket 拉取。这套架构后面我会详细讲。

3. 并发与性能：Agent 怎么扛住真实流量

3.1 串行调用的致命问题

很多人第一次把 Agent 暴露给真实用户时都会懵：为什么单个请求挺快，并发一上来就全卡住了？

原因在于 Agent 的执行链路是多个模型调用 + 多个工具调用串联的。一次请求可能涉及 3~5 次 LLM 调用，每次调用 2~5 秒，再加上中间的工具执行时间，一个完整任务跑完基本要 10~20 秒。如果每个请求占用一个同步 Worker 线程，10 个并发请求就把线程池打满了，后续请求全部排队。

更麻烦的是，Agent 执行期间大部分时间都在等待——等 LLM 返回、等外部 API 返回。这段时间 CPU 完全空闲，但线程被占着，非常浪费。

3.2 异步改造：Agent 引擎要设计成可暂停的

解决并发问题的核心思路：别让 HTTP 请求线程和 Agent 执行线程绑死在同一个生命周期里。

我的做法是引入队列。FastAPI 收到请求后，立刻把任务信息放入队列并返回一个 task_id，前端拿着这个 task_id 去轮询结果。Agent 执行引擎作为独立的 Worker 进程，从队列里取任务，异步执行完把结果写入存储。

这样做的直接好处是：HTTP 层的并发能力不再受 Agent 执行时间限制。即使 Agent 任务平均耗时 20 秒，FastAPI 也能接受成百上千的并发请求，因为每个请求只花几毫秒就把任务写入队列了。

在 FastAPI 里，接收任务的接口长这样：

from fastapi import FastAPI from pydantic import BaseModel import asyncio app = FastAPI() class AgentTask(BaseModel): user_id: str session_id: str message: str @app.post("/agent/task") async def create_task(task: AgentTask): # 把任务投入队列，立即返回 task_id task_id = await task_queue.put(task) return {"task_id": task_id, "status": "pending"}

注意这个接口必须是 async 的，await task_queue.put(task)不阻塞事件循环。如果任务队列的写入本身很快，也可以不用 async，但保持统一用 async 更稳妥。

3.3 Worker 进程的并发控制

Worker 侧同样需要考虑并发。一个 Worker 进程可以同时跑多个 Agent 任务吗？答案是要限流。

LLM API 通常有每分钟请求次数限制（RPM）和每分钟 token 数限制（TPM），不同供应商限制不同。我在实际项目中踩过教训：一次性开 20 个并发任务调用模型接口，结果直接被限流，一堆 429 错误，重试之后反而把成本抬高了。

我的做法是给 Worker 加信号量控制并发数：

import asyncio # 控制同时执行的 Agent 任务数量 semaphore = asyncio.Semaphore(5) async def run_agent_task(task): async with semaphore: result = await execute_agent(task) return result

并发数设置多少合适？取决于你的 LLM API 限额和单任务平均耗时。一个粗略的估算公式：

最大并发数 ≈ LLM API 每分钟限额（请求数） / 单任务平均 LLM 调用次数 × 容错系数（建议 0.6）

比如 API 限额是 60 RPM，单任务平均调用 5 次 LLM，理论支持 12 个并发任务，乘上容错系数 0.6，就设 7 个左右。

3.4 Token 与成本控制：既要跑得快，又要有钱赚

Token 是 Agent 项目里绕不开的核心成本。LangGraph 每次循环都会向模型发送完整的 State，这意味着状态越长，每次调用的 token 消耗就越大。

我自己实测过：一个简单任务，初始状态只有几千 token，但经过 5 轮工具调用循环后，每次发送的 token 可能膨胀到 2~3 万。如果模型计价较高，一次任务的成本可能从几厘飙到几毛，放大到线上流量就是不小的数字。

控制 token 有几个实用手法：

• 消息裁剪：只保留最近 N 轮对话和工具调用结果，早期的历史信息做摘要后存入摘要节点，而不是完整带在上下文里。
• 工具结果精简：工具返回的数据不要原封不动塞进上下文，先做摘要处理，只保留模型决策需要的关键信息。
• 模型分级：规划和最终输出用更强的模型，中间的工具结果判断、简单分支决策用便宜快速的小模型。
• 流式输出：最终回答用流式传输，用户感知速度更快，也减少超时重试的概率。

其中工具结果精简是我觉得性价比最高的优化。比如一个 SQL 查询返回了 50 行数据，你不需要把 50 行全给模型，先让代码做统计聚合，把"总行数、最大值、最小值、趋势"这种摘要信息给模型，token 消耗直接砍掉一半以上。

4. 实操：搭建一个可复用的 Agent 服务

4.1 环境准备与依赖安装

你可以直接按照下面的步骤来搭建。我用的环境是 Python 3.11，依赖如下：

pip install fastapi uvicorn langgraph langchain langchain-openai redis

其中 Redis 用来做任务队列和结果存储。如果你不想引入 Redis，也可以用内存队列（比如 Python 的asyncio.Queue），但要注意进程重启会丢任务，生产环境建议至少用 Redis。

4.2 定义 Agent 的图结构

这是 LangGraph 的核心代码。我直接给出一个可运行的简化版本，包含三个节点：规划、工具执行、决策。

from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated, List import operator class AgentState(TypedDict): messages: Annotated[List[dict], operator.add] task: str tools_output: List[dict] final_answer: str # 节点1：调用模型，决定下一步动作（回答或调用工具） async def model_node(state: AgentState): messages = state["messages"] # 这里省略具体的 LLM 调用代码，返回模型决策 decision = await call_model(messages) if decision["type"] == "tool_call": return {"messages": [decision["message"]]} else: return {"final_answer": decision["answer"]} # 节点2：执行工具调用 async def tool_node(state: AgentState): last_message = state["messages"][-1] tool_name = last_message["tool_calls"][0]["name"] tool_args = last_message["tool_calls"][0]["args"] # 从注册表找到工具并执行 result = await execute_tool(tool_name, tool_args) return {"messages": [{"role": "tool", "content": result}]} # 节点3：判断继续循环还是结束 def decide_finish(state: AgentState) -> str: last_message = state["messages"][-1] if state.get("final_answer"): return "end" elif "tool_calls" in last_message: return "continue" else: return "end" # 构建图 def build_agent(): graph = StateGraph(AgentState) graph.add_node("model", model_node) graph.add_node("tools", tool_node) graph.set_entry_point("model") graph.add_conditional_edges( "model", decide_finish, {"continue": "tools", "end": END} ) graph.add_edge("tools", "model") return graph.compile()

这里有个关键点在decide_finish函数。我见过不少人在这地方写错：判断条件没有覆盖模型输出 JSON 解析失败、工具调用参数缺失这些异常情况，导致图进入死循环或者静默结束。

给个建议：所有分支判断都要有兜底。比如模型输出了非法格式的 tool_call，你宁可结束任务返回"无法处理"，也不要让它重新循环。因为现实场景里模型输出非法格式的概率比你想的高很多。

4.3 工具注册与外部系统对接

工具定义的质量直接决定 Agent 的可用性。我的经验是遵循三个原则：

• 一个工具只做一件事。不要写一个"万能处理"工具，模型会不知道传什么参数。粒度拆细一点，反而调用成功率更高。
• 参数 schema 要写清楚。字段命名要直观，description 写清楚参数含义和格式要求。模型不是人，它只能靠这些描述来理解你的工具。
• 返回结果要结构化。工具返回 JSON 而不是纯文本，方便模型处理，也方便你自己做日志。

工具注册的代码大致这样：

tools = [ { "name": "query_sales_data", "description": "查询指定时间段的销售数据，返回汇总统计", "parameters": { "type": "object", "properties": { "start_date": {"type": "string", "description": "开始日期，格式 YYYY-MM-DD"}, "end_date": {"type": "string", "description": "结束日期，格式 YYYY-MM-DD"} }, "required": ["start_date", "end_date"] } } ] async def execute_tool(name: str, args: dict) -> dict: if name == "query_sales_data": return await query_sales_data(args) raise ValueError(f"Unknown tool: {name}")

4.4 记忆与上下文管理

记忆模块我踩过的坑最多。初期为了方便，我把所有历史消息全部保留在 State 里，结果跑到第 10 轮对话时 token 开销肉眼可见地涨，而且模型开始表现"失忆"——一开始的任务目标被大量工具结果淹没了。

后来我改成两级记忆架构：

• 短期记忆：当前任务的完整执行轨迹，包括用户输入、模型决策、工具调用和结果。这个保留在 LangGraph 的 State 里。
• 长期记忆：跨任务的用户偏好、历史结论摘要。存到数据库或 Redis，每次新任务开始时注入到 system prompt。

长期记忆的写入时机很关键。我是在每个任务结束时，额外调用一次模型，把当前任务的要点摘要成 200 字以内的结构化文本，然后入库。下次同用户发起新任务时，把它作为背景信息注入。

async def summarize_and_store(user_id: str, state: AgentState): messages = state["messages"] summary = await call_model( system="请将这次对话的关键信息总结为结构化文本，不超过200字", messages=messages ) await redis.set(f"user_memory:{user_id}", summary, ex=7*24*3600)

这样做的代价是每次任务多一次模型调用，但换来的是长期记忆的精准性，我认为这个成本值得。

4.5 部署时的几个细节

部署方面我补充几个容易被忽略的点：

• Uvicorn 启动时要用--workers开启多进程，但 Worker 数量不是越多越好。每进程有自己的事件循环，进程间共享任务队列需要走 Redis，所以多进程的好处有限，我一般设 2~4 个进程就够。
• 给 Agent 执行设置超时。LangGraph 里可以给图编译传入recursion_limit，控制最大循环次数。我一般设 10~15 次循环上限，超过就强制结束，避免死循环烧钱。
• 日志要记录完整链路。每个任务的 task_id、每个节点的输入输出、每次调用的 token 数，全部结构化日志输出。线上出问题时，就靠这些日志排查。

5. 常见问题与排查技巧实录

5.1 模型开始"胡言乱语"？先查上下文

Agent 输出的内容莫名其妙，或者回答和用户问题完全对不上，这是最常见的线上问题。大多数情况下不是模型"傻了"，而是上下文被污染了。

我排查这个问题的顺序是：

1. 先看完整执行日志，检查 messages 里是不是混入了异常内容。比如上一个工具返回了一条错误堆栈，被当作正常内容喂给了模型。
2. 再看 State 里的系统提示词有没有被工具结果覆盖。我遇到过工具返回的键名和 system prompt 里的变量重名，导致 prompt 被意外替换。
3. 最后看循环次数。如果任务循环了 8 次以上，早期上下文里的大量重复工具结果会干扰模型判断。

针对第二个原因，我后来给 prompt 模板做了一层隔离：用户内容、系统内容、工具结果分别用不同的命名空间，避免变量冲突。

5.2 工具调用陷入死循环

Agent 反复调用同一个工具不停循环，这是另一个高频问题。通常有两种情况：

一种是工具本身执行成功但结果不符合模型预期，模型执意重试。比如查询天气返回"晴"，模型想要"温度"，于是反复查询。这种情况我在工具返回结果里加了一层"结果评估"：如果工具返回内容里包含明确结论（如"无数据""条件不满足"），就直接在结果末尾追加一句"已尽力获取，没有更多信息可提供"，引导模型停止重试。

另一种是模型生成的 tool_call 参数每次都有一点随机差异，导致新调用、新结果，但整体目标没有推进。这种就只能靠recursion_limit兜底，到上限强制终止，同时记录终止原因到日志，后面再针对具体场景调 prompt。

5.3 并发场景下的会话隔离

在线上的多用户场景里，我踩过一个很深很深的坑：Session 状态串了。

原因是当时我把 UserID 作为 Redis key 的一部分来存储会话状态，但 Agent 执行引擎单例复用时，某个中间变量没有从 State 里清除干净，导致用户 A 的任务残留到了用户 B 的 State 里。

排查非常费劲，因为错误不是每次都复现，只在并发高的时候偶发。最后通过给每个任务分配一个全局唯一的 session_id，并在 State 初始化时强制清空非必要字段，才彻底解决。

这里给个硬性建议：State 字段必须显式声明生命周期。哪些字段是任务级的（每个任务新建），哪些是用户级的（跨任务保留），哪些是全局只读的（共享配置），在代码里写清楚，不要隐式依赖。

5.4 Token 超限的处理策略

很多项目跑着跑着突然报错output_token_limit_exceeded或者context_length_exceeded。处理策略取决于错误发生在哪一层：

• 如果是输入上下文超限：说明 State 里的历史消息太多，触发消息裁剪逻辑，保留最近 N 轮 + 摘要，重新执行。
• 如果是输出超限：说明模型生成长文本时超出限制，改用分块生成，或者调高 max_tokens 参数，或者让模型生成大纲后逐段展开。
• 如果是工具结果太大：这是最需要提前预防的。我在工具执行入口统一做结果截断，超过 2000 字符自动摘要，从源头避免超限。

还有一个小技巧：LangGraph 支持在节点里捕获模型调用的 Token 消耗，我把它累加到日志里，每个任务完成后统计总 token。线上跑一段时间后，对照这个数据能看到哪些用户、哪些场景在烧钱，有利于后续优化。

5.5 前端轮询还是 WebSocket？

任务投递到队列后，结果要如何返回给用户？两种方案我都试过：

• 轮询：前端每 2 秒请求一次/agent/task/{task_id}查结果。实现简单，但实时性差，请求频繁。
• WebSocket：Agent 执行完，服务端主动推送结果。实时性好，但连接管理复杂，还要处理断线重连。

我的建议是：MVP 阶段用轮询，上线后如果用户反馈等待体验差，再升级 WebSocket。轮询接口做一下优化——查一次更新一次状态，状态未变化就返回 304，减少不必要的数据传输。

6. 根据我的经验，最后再唠叨几句

做 AI Agent 项目，最难的不是写代码，而是对整个执行链路的掌控。模型输出有随机性，工具调用有不确定性，状态管理有复杂性，这三者叠加在一起，线上问题会非常"玄学"。我的应对思路是：把所有能结构化控制的东西全部用代码控制住，模型只负责决策，不负责流程。流程的每一步校验、超时、重试、终止，都用确定性代码实现。

另外，我建议你先从一个极小的场景切入，比如只做一个"查库存 + 生成回复"的 Agent，跑通完整链路后再加复杂度。不要一上来就规划什么多智能体协作、知识库增强、复杂工作流编排，那些都是后话。先把底层的状态管理、并发控制、成本监控做扎实，再往上盖楼，不然后期返工的代价非常大。

上面这些经验和代码，我都是在一线项目里一个个踩出来、调出来的。AI Agent 这个方向还在快速演进，框架和模型隔几个月就换一轮，但底层的架构思维和排查方法论是通用的。希望这篇分享能帮你少走点弯路。