AI Agent实战：从ReAct到LangGraph的主流架构演进与并发优化

如果只看2025年的AI圈热门词，可能至少有一半项目都自称“AI Agent”。从自动整理邮件的助手，到能接管一整条数据处理管线的智能体框架，名字都叫Agent，底子却完全不一样。我做了十多年系统开发，近两年几乎把所有业余时间都砸在LLM应用落地、尤其是Agent这一块。从最早拿LangChain写带记忆的聊天机器人，到后来用LangGraph编排多智能体状态流，再用FastAPI把整套东西包成服务扛真实请求，中间踩过的坑比写过的demo还多。如果你最近也在搜索“ai agent怎么扛并发”、“ai agent主流架构”、“基于rut语言ai agent”、“基于fastapi + langchain + langgraph的ai agent服务”这类关键词，说明你跟我一样，已经不满足于做一个能聊天的壳子，而是想让Agent真正下场干活。

这篇文章不打算做概念科普，我会从演进路线讲起，把主流架构、工具选型、代码实现、并发与稳定性问题，以及几张实际处理过的排错记录，全部摊开讲明白。你只需要有一点Python基础，懂一点HTTP服务，剩下的我用大白话解释。

1. AI Agent演进路线：从规则机器人到图执行智能体

1.1 萌芽期：规则脚本和预训练时代的“假智能”

提到Agent的演进，不能跳过2022年之前的阶段。早年做对话机器人，本质就是一棵决策树。用户输入进来，系统先分词，再匹配关键词，命中就返回预设话术，没命中就走兜底。整个过程没有任何“理解”，只有规则匹配。做RPA（机器人流程自动化）的团队更直接，录制一套鼠标键盘操作流程，定时或者按条件触发，模拟人点按钮、填表格。这套方案的维护成本高到离谱，因为规则库是无限膨胀的：你加一个实体，就得补一组分支逻辑。实际业务里稍微来一个没见过的句子，机器人马上就哑火。当时我们内部有个段子：规则系统不是写出来的，是补丁摞出来的。

这个时期也有“Agent”这个词，比如 IBM 早期的会话代理、微软的小冰，但更多是厂商包装。底层技术栈是搜索引擎式的检索匹配，或者早期的深度学习分类模型，本质上是对意图做槽位填充（Slot Filling）。这一代系统解决了“能不能自动回复”的问题，但没有任何推理能力，更谈不上动态规划。现在回头看，它是整个演进过程的地基：大部分工程规范、会话管理思路、错误处理范式，都是从那个时代沉淀下来的。

1.2 爆发期：LLM让模型第一次拥有了推理与行动闭环

2022年底ChatGPT出现之后，整个思路彻底变了。LLM在理解上下文、生成计划、抽取意图这三件事上的能力，几乎把过去规则系统里最难啃的环节直接抹平。2023年起，研究者陆续把“Prompt循环调用工具”做成标准范式，最具代表性的就是ReAct论文提出的模式。它的思路很朴素：让模型在每轮循环里先“想”一步，想清楚接下来该做什么，再“做”一步，调用一个外部工具，然后观察工具返回的结果，再继续进入到下一轮“想”。这个模式替代了写死规则的旧路线，只要提示词里把工具定义描述清楚，模型能自己决定是查数据库、调天气API，还是去拉文档。

我把这个阶段形容成“从编码到编排”的转变。以前写功能是程序员把每个分支写死在代码里，现在是靠模型在推理过程中动态决定调用路径。demo阶段的惊艳源于此，生产环境的隐患也源于此：ReAct循环天然不可控。模型可能在一个工具上反复调用十几次，上下文越来越长，前面已经拿到过的结论后面又忘记，更麻烦的是没有全局规划，遇到分步骤的任务容易陷入局部最优。2023年下半年我接过一个项目，让模型帮忙查订单并生成退货运费单，模型经常在“查订单详情”和“查运费规则”两个工具之间来回横跳，最后token耗尽也没出结果。这就是ReAct的典型毛病。

1.3 工程化期：LangGraph和图状态编排成为主流

ReAct的失控促使大家重新思考一个问题：Agent的能力边界应该由模型决定，还是应该由流程决定？2024年前后，LangGraph这类专门做Agent状态编排的框架火起来，核心思路是把Agent从“单循环”升级为“图执行”。图中每个节点是一次动作，每条条件边是一道决策切换，整个执行过程的状态被显式地管理起来，可以中断、恢复、持久化。这是我眼中Agent演进最关键的分水岭：不再是三万行if-else，也不是盲目信任模型，而是把人类可理解的流程硬约束和模型自主决策的灵活性结合起来。

这种演进的意义，打个比方就是：ReAct像一个没有项目经理的团队，谁有空谁上，遇到复杂需求容易乱；图执行则像是给项目画了张带里程碑的甘特图，关键节点必须过，步骤可以动态调整，但核心链路不能跳。LangGraph之后，出现了大量基于图编排的多智能体平台，比如各厂推出的Agent工作流编辑器。普通用户拖拽节点就能搭Agent，内部实现依然是状态图那套逻辑。如果现在你还在单纯用“for循环+工具调用”写Agent，建议尽早升级到图编排思路，哪怕不引入LangGraph，也应该按这个理念设计自己的状态机。

2. 主流架构选型：ReAct、Plan-and-Execute与多智能体到底区别在哪

2.1 ReAct：简单直接，但容易在上下文里绕圈

ReAct的架构极其简洁：一个LLM，一组工具，一个循环。模型根据用户问题决定要不要调用工具、调用哪个工具，拿到结果后继续推理，直至认为可以回答用户。因为它把思考过程以自然语言写在上下文里，调试非常直观，你打开聊天记录就能看到模型一步步想了什么。实测下来，ReAct在单一目标、工具数量少于5个的场景里表现还不错，比如“查一下明天的天气并生成穿衣建议”。

但如果你把工具数量扩到十几个，task复杂度一上来，ReAct就露怯了。典型问题是工具选择的“犹豫”：模型不知道该用哪个工具，会反复尝试相近的工具；更烦的是它已经拿到关键结果，但自己不知道，还在继续查。解决手段有两个：一是给每个工具写极其精确的描述，告诉模型“什么情况下不要用这个工具”；二是设定循环上限，比如最多执行6步，超过就强制输出当前能给出的结论或者直接转人工。

2.2 Plan-and-Execute：先出一份计划，再按计划执行

Plan-and-Execute的思路有别于ReAct：模型先一次性生成完整任务计划，再按顺序执行每一项，执行过程中根据结果动态修订后续计划。这样做的好处是避免了ReAct的“局部最优陷阱”，模型对整体目标有了全局认知，适合目标明确、步骤清晰的任务，比如“采集最近一周的行业新闻，分类后生成摘要报告”。我在实际项目里用这个模式做数据聚合类Agent，效果比ReAct稳定很多。

代价也很明显：对模型规划能力要求高，初始计划如果就是错的，后面执行得再稳也是白干。所以用Plan-and-Execute时，我一般会在生成计划后强制做一个“自检节点”，让模型拿着计划重新读一遍原始需求，检查有没有漏项、有没有不合理的步骤顺序。这个自检在工程上就是图里的一个节点，成本不高，但对最终正确率提升非常明显。

2.3 图状态与多智能体协作：把流程变成一等公民

用LangGraph的StateGraph建模，把每个工具封装成节点，把节点之间的调度关系画成边，这是我现在最推崇的生产架构。关键区别在：ReAct是靠模型自由探索的工具循环，图则是把“绝对不能跳过的环节”变成强边。比如鉴权、落库、人工审批这些步骤，模型就算再聪明也不允许绕过去。这解决了Agent落地时最头疼的合规问题：不是所有决策都该交给模型。

多智能体协作则是在图之上演变出的下一层复杂度。几种常见模式我列个表，帮你快速选型：

选择建议很直接：如果任务流程稳定、分支固定，用图就足够，不要为了“多智能体”这个词强行拆多个Agent；如果任务里确实出现天然的角色差异，比如写稿、审稿、配图是三类不同专家，再考虑Supervisor模式。多智能体最大的敌人是状态同步和上下文爆炸，拆之前先掂量一下你的token预算。

3. 技术栈分工：LangChain、LangGraph、FastAPI和Rust各自该承担什么

3.1 LangChain与LangGraph不是二选一，而是上下层关系

我见过不少团队纠结“选LangChain还是LangGraph”，这其实是个伪命题。LangGraph本来就是LangChain生态的一部分，LangChain负责提供模型封装、Prompt模板、输出解析这些通用能力，LangGraph负责状态管理和执行编排。正确姿势是把LangChain当作工具库按需取用，而不是整个框架端进来。我现在写Agent，只用LangChain的ChatPromptTemplate、输出解析器还有模型封装，核心编排全在LangGraph里。如果你被LangChain的抽象层弄得头大，可以直接不依赖LangChain，用OpenAI SDK和LangGraph核心包也能跑得很稳。

3.2 FastAPI凭什么成为Agent服务层的事实标准

Agent要对外提供服务，本质就是一个API网关，需要处理并发请求、鉴权、限流、长连接。FastAPI在这件事上有天然优势：原生async支持，可以让多个Agent会话在同一个进程内并发执行，遇到LLM调用这种网络IO操作会自动让出事件循环。实测中我用FastAPI包LangGraph跑过几十路并发请求，只要LLM供应商不限速，FastAPI本身几乎没有压力。它内置的Pydantic可以做请求参数校验，避免脏数据进入Agent流程。生产部署时配合gunicorn+uvicorn多worker，再挂一层Nginx做负载均衡，结构就很完整了。

3.3 Rust在Agent生态里的真实定位：不是来抢Python饭碗的

热词里有“基于rust语言ai agent”，我得先说句实话：绝大多数业务场景不需要用Rust重写Agent框架，也不需要用Rust重新实现一遍推理循环。Rust的优势在高并发、低延迟、内存安全，更适合作Agent链路上的专用加速组件。比如你想自己实现一个并发托管层，或者对服务启动速度、包体积有变态要求，Rust才有发挥空间。我见过一个项目，把消费队列里的Agent跑批任务全部用Rust重写，单机吞吐翻了接近三倍，但代价是开发周期拉长了至少一倍，生态里缺很多东西要自己补。个人项目或者中小团队，Python是更理性的选择，Rust可以留到真正的性能瓶颈出现时，再抽出来单独做服务。

4. 生产三大关：并发、状态持久化和可观测性

4.1 Agent怎么扛并发：三层策略缺一不可

热词里“ai agent怎么扛并发”出现频率极高，说明很多人已经踩到生产门槛了。第一层是FastAPI的async能力，把每个Agent的执行栈切成协程任务，LLM调用和工具HTTP请求遇到IO就自动让出，这是最基础的并发来源。第二层是请求队列解耦，不要傻等用户同步发起Agent请求、同步等结果，而是把请求扔进Redis Streams或RabbitMQ，由后台Worker异步消费处理完再回写状态。第三层是资源复用：模型调用连接池复用、工具返回结果缓存、对LLM供应商做token限流。这三层缺一层，并发一上来就会现原形。

4.2 状态持久化：Agent会话不丢记忆的保命手段

Agent生产事故的高发区就是会话状态。很多demo应用把所有上下文一股脑塞给模型，看起来简单，一旦量上来就完蛋，token爆炸、上下文超长、回复开始驴唇不对马嘴。正确做法是用LangGraph的Checkpointer机制，把图的中间状态持久化到Redis或Postgres，节点重启、服务挂掉都能从最近一个检查点恢复。我踩过的坑是项目早期用默认的内存Checkpointer，路由一重启用户上下文全没，多轮问答直接答非所问。后来换成Postgres持久化，问题立刻消失。

4.3 可观测性和兜底逻辑：Agent不是黑盒

Agent必须可观测，每步推理都要记录：哪条边触发了、哪个工具被调用、耗时多少、token消耗多少、返回结果摘要。我发现排障时大多数问题不是模型幻觉，而是工具超时没被正确反馈到上下文里，导致Agent拿着残缺信息硬编答案。建议每轮循环都输出结构化日志，连暂停、恢复都要打点。兜底逻辑同样重要：设定最大步数，超过就停止循环给稳定回复；工具连续三次失败就不再重试；模型输出解析失败时给用户友好的提示，而不是抛异常挂掉。

5. 实操：用FastAPI + LangChain + LangGraph搭一个能对外服务的智能体

5.1 场景设定与项目结构

这里我用一个真实做过的场景演示：企业内部的智能客服助手，用户问“我要去上海出差，帮我查一下明天的高铁车次，如果有优惠券也是好”。Agent需要调用车次查询工具和优惠券匹配工具，通过图状态流转完成多步决策，最后返回车次、时间和优惠后的价格。项目结构大致如下：

agent_service/ ├── agent/ │ ├── graph.py # LangGraph状态图定义 │ ├── tools.py # 车次查询、优惠券匹配工具 │ ├── state.py # Agent状态模型 ├── api/ │ ├── main.py # FastAPI入口 │ └── schemas.py # 请求/响应模型 └── config.py # 模型、Redis等配置

5.2 定义工具：让模型能用的函数，必须把错误也变成正常返回

先看工具层。一个能被LangGraph调用的工具，关键在于函数注释和返回结构要足够清晰，让模型知道什么情况唤起什么、结果怎么解读。我习惯把错误也变成可解析的返回信息，而不是抛异常打到模型外面：

from langchain_core.tools import tool @tool def query_train_schedule(city_from: str, city_to: str, date: str) -> str: """根据出发城市、到达城市和日期查询高铁车次，返回车次列表和余票信息。""" try: result = railway_api.query(city_from, city_to, date) if not result.get("data"): return "未查到该日期的高铁车次，请提示用户换一个日期。" return format_schedule(result["data"]) except Exception as e: return f"车次查询接口异常：{str(e)}。请告知用户稍后重试。"

注意一点：工具返回的是字符串，不是结构化dict。这是为了让模型在推理时可以直接读取，不需要额外的解析步骤，减少一层出错概率。等模型判断执行完毕，再由输出解析器把最终答案提取出来。

5.3 构建LangGraph状态图：把流程画成代码

状态定义就用LangGraph自带的TypedDict，记录车次查询结果、优惠券匹配结果和最终答复。图里我设计了三个核心节点：查车次、查优惠、生成答复，条件边确保必须先查车次成功且用户要求优惠，才进入优惠节点，否则直接跳到生成答复：

from typing import TypedDict from langgraph.graph import StateGraph, END class AgentState(TypedDict): query_text: str train_info: str coupon_info: str final_answer: str def node_query_train(state: AgentState): # 这里从state解析出城市和日期，调用工具 return {"train_info": query_train_schedule(...)} def node_query_coupon(state: AgentState): return {"coupon_info": query_coupon(state["train_info"])} def condition_need_coupon(state: AgentState): # 用户明确提到优惠/价格，则进入优惠节点，否则直接答复 return "coupon" if "优惠" in state["query_text"] else "answer" graph = StateGraph(AgentState) graph.add_node("train", node_query_train) graph.add_node("coupon", node_query_coupon) graph.add_node("answer", node_generate_answer) graph.set_entry_point("train") graph.add_conditional_edges("train", condition_need_coupon, {"coupon": "coupon", "answer": "answer"}) graph.add_edge("coupon", "answer") graph.add_edge("answer", END)

LangGraph会自动把每个节点的返回合并回全局状态，这就是“图执行”的核心便利：节点只管自己的输出，状态由框架管理，不怕节点之间互相踩踏。等你跑通这个流程，再把Redis Checkpointer一挂，生产可用性立刻上一个台阶。

5.4 FastAPI接口：普通响应与流式响应两种模式

接口层最大的问题是Agent执行耗时长，HTTP请求很容易超时。我建议保留两类接口：短任务用同步返回；长任务或需要用户实时看到过程的场景用流式返回。下面是一个简化版，演示普通响应接口：

from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class ChatRequest(BaseModel): session_id: str query: str class ChatResponse(BaseModel): session_id: str answer: str graph_app = graph.compile(checkpointer=redis_checkpointer) @app.post("/chat", response_model=ChatResponse) async def chat(req: ChatRequest): config = {"configurable": {"thread_id": req.session_id}} # 注意：LangGraph默认是阻塞执行，放在线程池避免卡住事件循环 import anyio result = await anyio.to_thread.run_sync( lambda: graph_app.invoke( {"query_text": req.query, ...}, config=config ) ) return ChatResponse(session_id=req.session_id, answer=result["final_answer"])

为什么这里用anyio.to_thread.run_sync绕一下？因为LangGraph的invoke是同步阻塞调用，直接扔在async函数里会卡住FastAPI的事件循环，导致并发能力骤降。把它丢给线程池执行，事件循环就能继续接收其他请求，这是并发性能的关键细节之一。等模型逐步支持流式调用，你还可以用astream接口把每一步输出推给前端，体验会好很多。

5.5 部署与压测：别只跑通就以为完事

部署层面我建议用gunicorn -k uvicorn.workers.UvicornWorker -w 4匹配FastAPI的异步特性，每个Worker等于一个是独立进程，四路并发就是一个不小的并发池。再加一层限流中间件，比如用slowapi把每个session的请求压到每分钟20次，防止单用户打爆LLM配额。上线前用Locust压一遍，重点关注P95延迟，你会很直观地看到LLM调用耗时占了整个链路的90%以上，所以真正要做性能优化的不是HTTP层，而是模型调用层：加缓存、合并请求、必要时上更快的模型。

6. 高频故障排查：循环、超时、丢状态、工具异常

6.1 模型在工具调用里死循环

现象：同一个工具被触发十几遍，最后超时报错。原因多半是工具返回结果没有正确进入模型的下轮推理，或者工具描述写得含糊，让模型误以为没拿到结果。解法：一是给@tool的描述加上“什么情况下不要用”的提示；二是在LangGraph里给循环边增加步数计数器，超过预设值强制跳转到END节点。我一般把最大工具调用次数设为6次，用户问复杂问题时的成功率没有因为限步而下降，反而因为及时止损稳了不少。

6.2 并发一上来就大量504

现象：30路并发时请求开始排队，前端的请求超时率飙升。先看日志里是不是LLM供应商返回429限速错误，这几乎占了70%的原因。解法分三层：先在入口给相同参数的请求加结果缓存，复用之前已经算过的答案；再在工具调用层做并发控制，用一个信号量把同时打往同一接口的工具调用限制在5以内；最后把非关键链路上耗时长的请求改成异步队列，用后台任务消费结果。这样做完，通常不用加机器就能扛住原来的并发量。

6.3 多轮会话丢状态，回答牛头不对马嘴

现象：Agent第一轮查到车次，第二轮问价格时它完全“忘”了刚才的信息。原因基本锁定在Checkpointer没有持久化，或者服务重启后状态归零。解法：把checkpointer从内存版换成Redis或Postgres版，同时在每个请求的config里都带上thread_id，这是LangGraph找回会话状态的钥匙。老项目有一次状态没恢复，用户问“那上午那趟有票吗”，Agent回答“哪趟？”，当场被投诉，后来我在日志里加了会话级摘要，每次持久化时把前几轮结论压缩存储，问题才根治。

6.4 工具返回超时或格式异常

现象：第三方接口返回503，或者字段缺失导致后续解析报KeyError。重要认知：工具不可靠是常态，模型能不能从错误中恢复是Agent质量的试金石。我的固定范式是工具函数里用try/except捕获所有异常，将错误信息整理成文本返回给模型；模型看到“服务暂不可用，请告知用户”后就不会硬编一个假结果，会给出诚实回复。同时，工具返回的文本尽量模板化，尽量不返回自由格式JSON，让模型的解析负担减到最低。

7. 最后聊聊我的真实体会

从规则脚本到多智能体图执行，AI Agent不是一夜封神，它是一步步把不确定性往确定性框架里装。早期做对话机器人，怕的是规则写不全；现在做Agent，怕的反而是模型太“自由”。我自己经过这么多项目后的建议只有一个：先把架构模式搞清楚，再挑一个小场景从ReAct跑起，跑通后交给LangGraph重构流程，最后才考虑并发和持久化。这样踩的坑最少，每个阶段的经验都能沉淀下来。目前我把LangGraph的Checkpointer换成了Redis，把大模型调用环节单独拆出来做并发缓存，生产稳定性明显上了一个台阶，下一步准备尝试在Rust侧重写一个专用工具调用代理，专门解决超高频调用的性能瓶颈。Agent这条路还很长，但每个阶段做完后的回报都是实打实的。