最近一直在折腾AI Agent也就是智能体。用LangGraph做底层编排、FastAPI做对外服务层把思考、调工具、回话这一整套流程真正跑通之后我最大的感受是Agent开发的核心其实不是某个模型有多强而是你围绕模型搭的那套执行循环、状态管理和路由逻辑是否经得起真实请求的考验。这篇文章不说概念PPT直接把我在实际搭建过程中用到的架构思路、关键算法和能直接改来用的源码整理出来适合正准备从“调API demo”跨到“做完整Agent服务”的开发者参考。1. AI Agent 架构先搞清楚为什么需要一套“组合拳”1.1 Agent 不是“调用一次接口”而是一个可循环的执行过程很多刚接触Agent的开发者容易把Agent理解成“把用户的提示词扔给大模型然后返回一个结果”。如果只是这样那它和普通的聊天机器人接口没有本质区别。真正的Agent核心在于“代理”这个词它要代替用户去完成一个任务而不是只给出一段建议。这就意味着你需要的不是一次请求而是一个循环过程接收任务 → 模型推理 → 决定下一步是“调用某个工具”还是“直接给出最终答复” → 如果是调用工具就执行工具并把结果反馈给模型 → 模型基于新信息再次推理 → 直到模型给出最终答复。这个循环通常被称为“模型-工具-模型”的闭环。你可以把它类比成餐厅后厨大模型是主厨工具是砧板、灶台和食材供应商而后厨能够顺利运转靠的是传菜员执行循环和后厨动线状态流转把这些环节串起来。很多Agent项目翻车不是主厨不行而是动线和传菜环节设计得过于简陋——没有保存中间状态、没有条件路由、没有终止机制一遇到工具调用失败就原地卡死。我遇到过的真实情况是第一版Agent只把大模型返回的工具调用解析出来、执行完再把结果塞回对话里完全没有考虑“如果工具超时”“如果模型连续两次想调用同一个工具”“如果用户中途切换了话题”这些岔路。结果就是明明模型能力很强整体效果却很不稳定。所以后来我把重点从“调哪个模型”转移到“整个循环怎么设计”效果提升反而更明显。1.2 三种主流的 Agent 架构模式对比聊Agent设计绕不开几种经典架构模式。业界讨论比较多的是ReAct、Plan-and-Execute和Reflection三种。它们解决的问题侧重点不同适用场景也不同。ReActReasoning Acting是我个人最常优先考虑的方案。它的核心思想是让模型交替进行推理和行动每次先思考当前情况再决定是调用工具还是结束。直观来说它是“边想边做”执行路径是动态的模型可以根据中间结果调整下一步动作。比如用户问“帮我对比A和B两个产品的价格”Agent会先搜索A的信息再搜索B的信息最后综合对比。这个模式对单任务的灵活处理能力很强实现复杂度也相对可控。Plan-and-Execute则先把任务拆成一个计划再逐步执行。它分两个阶段第一个阶段让模型生成一个包含若干个步骤的计划第二个阶段按计划逐项执行。好处是逻辑清晰、便于人类审查中间步骤适合报表生成、批量数据处理这类流程相对固定的任务。缺点是如果计划制定阶段考虑不周全后续拆解执行时容易死板很难临时调整。Reflection模式强调的是“先做一遍再回头审视一遍”。它让Agent在完成初步结果后自己或借助另一个模型对结果进行评价和修正。这种模式适合内容创作、代码生成这类对结果质量要求较高的场景。缺点是开销会翻倍因为一次任务可能要跑两三轮完整的推理。从我实际项目的角度说一个生产级的Agent往往不只用单一模式。更多情况是用ReAct作为主干循环在特定步骤内套Plan-and-Execute的子流程再在关键输出前加入一次Reflection。这套“组合拳”听起来复杂但放到LangGraph里做其实就是多加几个节点和几条边的问题。架构模式核心思路优点缺点最适合的场景ReAct推理与行动交替灵活、动态决策、实现简单多步骤复杂任务可能遗忘计划问答、工具调用、信息检索Plan-and-Execute先计划后执行步骤清晰、可审查计划阶段可能固化和死板报表生成、批量处理、固定流程Reflection执行后自省修正结果质量高推理成本翻倍内容创作、代码生成、结果评审1.3 为什么是 LangGraph FastAPI 这套组合选型的问题本质上是在回答“用什么来承载循环逻辑”和“用什么方式对外提供服务”。如果你只是写个脚本做实验那用什么都可以。但要做成能够被网页、小程序、工作流平台调用的服务就得考虑状态管理、并发、流式输出、超时控制这些问题。LangGraph的价值在于把Agent流程变成了可编程的图。节点是你定义好的函数边是数据流动的方向条件边则让“下一步去哪”可以由模型或代码动态决定。这和以前的LangChain链式编排不一样。链式编排是固定顺序A完了B、B完了C很难表达“根据模型决定是调用工具还是结束”这种环形结构。LangGraph相当于把“流程控制权”还给了开发者你可以清晰地画出一条带分支、带循环的执行路径并且每一个状态都可以被持久化。选择FastAPI的原因更直接它是目前Python里做异步API比较顺手的一层。LangGraph编译出来的图对象可以同步运行也可以异步运行FastAPI原生支持async/await这让Agent接口可以做到真正的非阻塞。再加上FastAPI的自动文档、参数校验、依赖注入封装一个Agent服务不需要额外引入臃肿的框架。我在选型时也对比过其他方案。自己手动写状态机和循环调度听起来简单但一旦涉及“断点续跑”“持久化历史”“并发会话”代码会迅速膨胀到没法维护。而常被提到的AgentExecutor或者一些高层次的Agent封装封装程度太高出了问题很难定位。LangGraph则在灵活和易用之间平衡得比较好底层是图执行器但提供了一套小而清晰的原语允许你像搭积木一样搭建自己的Agent行为。2. LangGraph 的核心机制与算法拆解2.1 状态图三大件State、Node、EdgeLangGraph里的StateGraph遵循经典的有状态图模型。要理解它需要先搞清楚三个东西State状态、Node节点、Edge边。State是贯穿整个图的数据结构。它通常用一个TypedDict或者Pydantic模型来表示图里所有节点共享同一个State每个节点执行完返回的字典更新会合并进全局State。这里有个容易踩坑的点默认的更新方式是“覆盖”也就是说如果你返回的字段是列表它会直接替换掉旧列表而不是追加。想要追加需要自定义reducer。我第一次处理messages字段时就没注意到这一点结果多轮对话中的历史消息被不断覆盖模型彻底失忆。后来我在状态定义里为messages指定了自定义合并逻辑问题才解决。Node就是普通的Python函数。它接收当前的State作为参数执行完毕后返回一个字典表示要更新的字段。从执行引擎的角度看节点内部做什么都行可以调大模型、可以执行Python函数、可以调用外部API甚至可以是另一个SubGraph。节点之间通过State传递数据不需要额外定义参数这让代码解耦得很彻底。Edge则分两种。普通边add_edge表示无条件从节点A流向节点B条件边add_conditional_edges则由一个路由函数决定具体流向哪个节点。路由函数接收State返回一个节点名字符串或者一个节点名字符串列表多路径并行。这个条件边就是Agent所有动态行为的根源模型下一步做什么是由当前状态决定的而不是预先写死的。这三个元素的组合表达能力远超我的预期。你可以构建一个普通的顺序流水线A → B → C也可以构建一个带反馈回路的循环A → B → A还可以构建一个路由器根据模型输出在多个工具节点之间做选择。Agent需要的动态执行、循环回放、分支跳转图结构天然就能表达。2.2 ReAct 循环的图化实现现在把ReAct算法翻译成LangGraph图。我最常用的一套结构是这样的graph.add_node(agent, call_model) # 模型节点负责推理并决定下一步 graph.add_node(tools, execute_tools) # 工具节点执行模型选择的工具 graph.set_entry_point(agent) # 入口指向模型节点 graph.add_conditional_edges(agent, should_continue, {continue: tools, end: END}) graph.add_edge(tools, agent) # 工具执行完回到模型节点逻辑很清楚模型节点收到最新状态后调用大模型如果模型返回的是普通文本路路由函数返回“end”流程结束直接输出最终答案如果模型返回的是工具调用信息路由函数返回“continue”进入tools节点执行具体工具工具执行完成后把结果写回State边又把流程带回agent节点开始下一轮推理。这个结构的算法本质就是一个带终止条件的while循环。每执行一轮State里的消息列表就会增加三条新消息模型的思考/调用请求、工具的执行结果、以及下一轮模型的最终回应。通过检查消息列表里是否出现了“模型认为可以结束”的信号循环就能在合适的时候停下来。写路由函数的时候我建议不要只判断模型返回的文本内容要直接看模型输出的结构。如果你使用的是支持Function Calling或Tool Calling的模型模型返回的通常是一个列表而不是一段自然语言。此时路由逻辑可以是tools为空 → 返回“end”tools非空 → 返回“continue”。用结构判断比用文本判断可靠得多有效避免了模型用自然语言说“我要调用工具”但实际没有给出工具参数的情况。2.3 路由算法的设计细节与边界控制路由算法是整个Agent的“大脑”它直接决定循环的质量。我强烈建议在你自己的实现里加入以下三个层面的边界控制第一层最大迭代轮数。不管模型怎么想循环次数必须有一个上限。一个真实用户问题最多允许模型调用工具多少次这是需要通过压测来评估的。我常见的一个配置是5到10轮少于5轮往往不够因为复杂任务要多次查资料多于10轮只要一次工具调用卡顿用户就能感受到明显延迟。设置次数的方式很简单给State加一个计数器字段每次循环累加超过上限后路由函数强制返回“end”并追加一条提示文本说明“当前任务太复杂建议拆分成多个子问题”。第二层工具解析容错。模型输出的工具调用不一定总是格式正确的。有可能参数不是合法JSON有可能工具名不存在有可能参数缺了必填项。以前我会把解析错误直接抛给上层结果是整个执行画面红屏。后来我在tools节点里做了二次校验先把模型给的参数用JSON解析解析失败就构造一条“参数解析失败请按照XX格式重新调用”的工具反馈消息把它作为工具结果写回State。这样一来模型就知道自己刚才格式不对下一轮会自己修正。这个技巧在真实项目中帮了我很多强烈推荐。第三层并行工具调用的合并策略。如果模型一次请求里返回多个工具调用比如同时查天气和查机票可以把它们批量执行后一次性返回。LangGraph支持在一个节点里循环执行多个工具再统一把结果合并回数据。要注意的是并行工具的结果顺序必须和请求顺序保持一致否则模型看到的结果错位推理会乱。我通常用字典以工具名或调用ID为键保存结果再按原始顺序拼接。另外还要考虑一个现实问题工具执行是有副作用的。比如发送邮件、创建订单这类的工具一旦被模型错误地调用代价很高。我的经验是把工具列表分成“只读类”查询、搜索、计算和“写操作类”发送、创建、删除对写操作增加一个前置确认机制——要么在执行前要求用户二次确认要么在工具描述里显著标注副作用。这样的护栏能让你的Agent在生产环境里少出大事故。3. 实操基于 LangGraph FastAPI 搭建一个可运行的 Agent3.1 项目结构与依赖清单这部分直接进入能跑的代码。我的项目结构长这样agent_service/ ├── pyproject.toml ├── .env ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 入口 │ ├── agent_graph.py # LangGraph 图定义 │ ├── tools.py # 工具函数 │ ├── schemas.py # 请求/响应模型 │ └── settings.py # 配置项依赖相对精简就几样核心的langgraph图编排langchain-core模型接口抽象如果你不想绑定某个厂商SDKfastapiAPI服务uvicornASGI服务器pydantic数据校验python-dotenv读取环境变量如果你需要持久化历史会话可以额外加一个langgraph-checkpoint的SQLite或Redis实现。但先不加把主流程跑通再考虑记忆功能是我比较推荐的做法。一上来就把存储、缓存、消息队列全接上只会让第一个Demo的调试难度陡增。3.2 用 LangGraph 定义 Agent 工作流先定义State。我倾向于把State拆成更细的字段而不是把整个对话历史都塞在一个messages列表里。这样路由函数和节点函数取用数据时更清晰。from typing_extensions import TypedDict from langgraph.graph import StateGraph, END class AgentState(TypedDict): messages: list # 完整的消息历史 user_query: str # 当前用户问题 tool_result: str # 最近一次工具结果 step_count: int # 当前迭代次数 max_steps: int # 最大迭代次数接着是模型节点。这里需要注意一个容易被忽略的点传给模型的messages要包含全部历史但模型的服务端上下文有限历史过长时要做裁剪。简单做法是只保留最近N轮对话或者用摘要压缩更早的内容。from langchain_core.messages import HumanMessage, SystemMessage def call_model(state: AgentState) - dict: 调用大模型并决定是否调用工具 messages [ SystemMessage(content你是一个智能助手。当需要信息时你可以调用可用工具。), ] state[messages][-10:] response llm_with_tools.invoke(messages) return {messages: [response]}工具节点负责执行模型请求的工具调用。它的返回值要按工具调用ID组织这样每条工具结果才能正确对应到模型的请求。def execute_tools(state: AgentState) - dict: last_message state[messages][-1] tool_messages [] for tool_call in last_message.tool_calls: tool_name tool_call[name] tool_args tool_call[args] # 根据工具名分发执行 result dispatch_tool(tool_name, tool_args) tool_messages.append(ToolMessage(contentjson.dumps(result, ensure_asciiFalse), tool_call_idtool_call[id])) return {messages: tool_messages, step_count: state[step_count] 1}路由函数的写法至关重要。我给出的实现方案是def route_after_model(state: AgentState) - str: last_message state[messages][-1] # 1. 超过最大迭代轮数强制结束 if state[step_count] state[max_steps]: return end # 2. 如果模型带了工具调用进入工具节点 if getattr(last_message, tool_calls, None): return continue # 3. 否则直接返回最终答复 return end最后把图组装起来builder StateGraph(AgentState) builder.add_node(agent, call_model) builder.add_node(tools, execute_tools) builder.set_entry_point(agent) builder.add_conditional_edges(agent, route_after_model, {continue: tools, end: END}) builder.add_edge(tools, agent) agent_app builder.compile()到这里最核心的Agent循环就已经能跑了。你可以在本地直接用agent_app.invoke({messages: [HumanMessage(content...)], step_count: 0, max_steps: 6})测试。3.3 FastAPI 接口层封装图已经编译好了下面要把它暴露成HTTP接口。FastAPI这边的设计重点是每个API请求对应一个独立的会话用session_id隔离上下文。我采用的做法是把会话ID透传给Agent由Agent把历史消息加载到State里。from fastapi import FastAPI, HTTPException from pydantic import BaseModel app FastAPI(titleAgent Service) class AgentRequest(BaseModel): session_id: str query: str max_steps: int 6 class AgentResponse(BaseModel): answer: str trace: list app.post(/agent, response_modelAgentResponse) async def run_agent(req: AgentRequest): try: result await agent_app.ainvoke({ messages: load_history(req.session_id) [HumanMessage(contentreq.query)], user_query: req.query, step_count: 0, max_steps: req.max_steps, }) save_history(req.session_id, result[messages]) return AgentResponse(answerresult[messages][-1].content, traceresult[messages]) except Exception as e: raise HTTPException(status_code500, detailstr(e))这段代码里load_history和save_history可以是简单的内存字典也可以是数据库操作。如果你只是想先跑通Demo用一个dict做内存存储就够了上生产再替换成Redis或数据库。有一个细节很重要FastAPI接口里的await agent_app.ainvoke(...)必须配合支持异步的大模型客户端使用。如果你用的是某同步SDK比如requests-based的ainvoke会阻塞事件循环导致并发能力下降。这种情况下建议用asyncio.to_thread把它放到线程池里执行或者直接换成异步客户端。3.4 流式输出与异步改造真实用户体验Agent时等一个十秒的完整结果和边生成边看到文字感受完全不同。把Agent改成流式输出是我认为上线前必须做的一步。LangGraph提供了astream_events接口可以逐事件地获取执行过程中的中间信息。结合FastAPI的StreamingResponse实现SSEServer-Sent Events流式输出非常方便from fastapi.responses import StreamingResponse app.post(/agent/stream) async def agent_stream(req: AgentRequest): async def event_generator(): config {configurable: {thread_id: req.session_id}} async for event in agent_app.astream_events( {messages: [HumanMessage(contentreq.query)], step_count: 0, max_steps: req.max_steps}, versionv2, configconfig, ): node_name event.get(name, ) # 当模型生成 token 时逐块返回 if event[event] on_chat_model_stream: token event[data][chunk].content if token: yield fdata: {token}\n\n # 当工具被调用时发送一个标记事件 if event[event] on_tool_start: yield fdata: [TOOL_START] {event[data][input]}\n\n return StreamingResponse(event_generator(), media_typetext/event-stream)这里必须提一下SSE的一个坑如果前置有网关或代理一定要关闭缓冲否则流式内容会积压一段时间才批量到达体验和普通请求没区别。同时客户端要正确解析SSE事件不能把它当作普通JSON响应。4. 部署与运行时调优的关键细节4.1 状态持久化与消息记忆Agent做得越久越会发现记忆机制的重要性。用户今天问了一个问题明天接着问“上次那个方案的第三点具体怎么实施”如果服务端没有保存历史上下文模型根本无法回答。我的做法分两层。短期记忆用Redis以session_id为key存储最近的消息列表TTL设置为2小时。这样当一个会话内连续互动时消息读取走内存级缓存速度快会话结束后一段时间自动过期避免堆积。长期记忆用一个简单的表或向量库存储用户画像、关键事实、历史偏好在每次启动会话时筛选相关记忆注入System Prompt。注意不要把所有历史无脑塞给模型上下文越杂模型越容易跑偏。消息裁剪也是记忆管理的一部分。我常用两个策略一是按条数裁剪只保留最近20条消息二是按字符数裁剪超长时让模型对更早的历史做摘要追加到消息列表头部。第二种效果更好但会增加一次额外的模型调用成本可以在会话比较空闲时执行。4.2 并发与性能参数生产环境里FastAPI本身可以承载大量并发请求但瓶颈几乎总是集中在模型调用和工具调用上。模型API的速率限制是第一个要面对的坎我一般做一个简单的令牌桶限流器确保并发窗口内的请求数不超过模型服务的配额。不需要引入重型消息队列一个异步信号量加一个计数就能解决大部分问题from asyncio import Semaphore sem Semaphore(10) # 同一时间最多10个并发模型调用 async def call_llm_safe(messages): async with sem: return await llm.ainvoke(messages)工具调用的超时设置同样重要。由于外部HTTP服务可能很慢我建议为工具请求设置整体超时和TCP连接超时两个参数。整体超时控制在10秒以内是比较合理的。如果你调用的是搜索类接口还可以加一层结果缓存相同关键词在5分钟内不重复请求外部服务能显著降低延迟和成本。模型参数层面temperature的取值会影响Agent的“听话程度”。在工具调用为主的场景我建议把temperature调低到0.2以下甚至直接设为0让模型的输出更可控。如果是创意写作类的Agent再调高到0.7以上。这是我在实际调优中感受非常明显的一个参数温度高时模型更容易在多个工具之间反复横跳自己给自己加戏。5. 常见问题与排查技巧实录5.1 工具调用死循环怎么打断现象模型反复调用同一个工具即使工具已经明确返回“没有结果”它还要再试一次。这是ReAct模式最常见的问题之一。排查思路看路由函数的判断条件和State里的step_count是否在正确更新。很多情况下是模型觉得“不够确定”想通过更多轮工具调用收集更多信息。解决方案有三种一是收紧max_steps比如从10改成5二是在模型节点里显式追加一条系统提示如果工具已经返回足够信息或连续两次返回相同结果必须停止调用工具三是借助工具结果里返回的结构化状态码比如加一个complete: true/false字段让路由函数直接据此判断终止。5.2 状态污染导致上下文错乱现象会话A的历史消息串到了会话B里。这通常是State数据在并发请求下被共享了或者存储历史时key没有按session_id隔离。排查方法检查State里是否保存了不该保存的全局对象如果使用内存缓存确认每个会话进入时都在调用load_history(session_id)并且图调用结束后再调用save_history(session_id, ...)。在多线程场景下还要注意消息列表这类可变对象被多个请求同时修改。最好的做法是进入图前做一次深拷贝返回结果确认无误再持久化。5.3 FastAPI 流式输出被网关缓冲现象流式接口返回正常但前端每隔很久才收到一整段内容。十有八九是前置代理开启了缓冲。解决方法是在网关或反向代理层的配置里关闭对SSE路径的缓冲或者动态响应压缩。如果这一层你无法控制可以尝试改用WebSocket替代SSEWebSocket天然没有缓冲问题但实现复杂度更高。5.4 模型返回的 JSON 参数格式不稳定现象工具的参数解析时好时坏尤其当模型返回较长的工具调用时JSON容易出现截断或者合并不合法。应对办法给语言模型使用结构化输出约束比如Pydantic输出解析器让模型只输出符合Schema的JSON而不是自由文本里夹带JSON代码块。另一个补充手段是容错解析先用标准库解析失败后用正则抽取大括号内容再解析再失败则把原始字符串返回给模型让它重试。我个人在实际操作中的体会是Agent开发本质上是一个“把不确定性锁进受控格子”的过程。模型本身是概率性的但你的编排层、路由逻辑和状态管理必须是确定性的。上面这套基于LangGraph FastAPI的搭建方案谈不上完美但在我自己的项目中经受住了不少真实流量的考验稳定性比我最早用纯Chain实现的那版好了太多。最后再分享一个小技巧在调试Agent时把每一次工具调用的入参和出参都记录下来因为大部分诡异问题最后都能从工具调用日志里找到根源。
