LangGraph+FastAPI实战：AI Agent核心机制与工程化部署

最近一直在折腾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三种。它们解决的问题侧重点不同，适用场景也不同。

ReAct（Reasoning + Acting）是我个人最常优先考虑的方案。它的核心思想是让模型交替进行推理和行动：每次先思考当前情况，再决定是调用工具还是结束。直观来说，它是“边想边做”，执行路径是动态的，模型可以根据中间结果调整下一步动作。比如用户问“帮我对比A和B两个产品的价格”，Agent会先搜索A的信息，再搜索B的信息，最后综合对比。这个模式对单任务的灵活处理能力很强，实现复杂度也相对可控。

Plan-and-Execute则先把任务拆成一个计划，再逐步执行。它分两个阶段：第一个阶段让模型生成一个包含若干个步骤的计划，第二个阶段按计划逐项执行。好处是逻辑清晰、便于人类审查中间步骤，适合报表生成、批量数据处理这类流程相对固定的任务。缺点是如果计划制定阶段考虑不周全，后续拆解执行时容易死板，很难临时调整。

Reflection模式强调的是“先做一遍，再回头审视一遍”。它让Agent在完成初步结果后，自己或借助另一个模型对结果进行评价和修正。这种模式适合内容创作、代码生成这类对结果质量要求较高的场景。缺点是开销会翻倍，因为一次任务可能要跑两三轮完整的推理。

从我实际项目的角度说，一个生产级的Agent往往不只用单一模式。更多情况是：用ReAct作为主干循环，在特定步骤内套Plan-and-Execute的子流程，再在关键输出前加入一次Reflection。这套“组合拳”听起来复杂，但放到LangGraph里做，其实就是多加几个节点和几条边的问题。

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、Edge

LangGraph里的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 搭建一个可运行的 Agent

3.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：模型接口抽象（如果你不想绑定某个厂商SDK）
• fastapi：API服务
• uvicorn：ASGI服务器
• 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(content=json.dumps(result, ensure_ascii=False), tool_call_id=tool_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(title="Agent Service") class AgentRequest(BaseModel): session_id: str query: str max_steps: int = 6 class AgentResponse(BaseModel): answer: str trace: list @app.post("/agent", response_model=AgentResponse) async def run_agent(req: AgentRequest): try: result = await agent_app.ainvoke({ "messages": load_history(req.session_id) + [HumanMessage(content=req.query)], "user_query": req.query, "step_count": 0, "max_steps": req.max_steps, }) save_history(req.session_id, result["messages"]) return AgentResponse(answer=result["messages"][-1].content, trace=result["messages"]) except Exception as e: raise HTTPException(status_code=500, detail=str(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，实现SSE（Server-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(content=req.query)], "step_count": 0, "max_steps": req.max_steps}, version="v2", config=config, ): node_name = event.get("name", "") # 当模型生成 token 时，逐块返回 if event["event"] == "on_chat_model_stream": token = event["data"]["chunk"].content if token: yield f"data: {token}\n\n" # 当工具被调用时，发送一个标记事件 if event["event"] == "on_tool_start": yield f"data: [TOOL_START] {event['data']['input']}\n\n" return StreamingResponse(event_generator(), media_type="text/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替代SSE，WebSocket天然没有缓冲问题，但实现复杂度更高。

5.4 模型返回的 JSON 参数格式不稳定

现象：工具的参数解析时好时坏，尤其当模型返回较长的工具调用时，JSON容易出现截断或者合并不合法。

应对办法：给语言模型使用结构化输出约束（比如Pydantic输出解析器），让模型只输出符合Schema的JSON，而不是自由文本里夹带JSON代码块。另一个补充手段是容错解析：先用标准库解析，失败后用正则抽取大括号内容再解析，再失败则把原始字符串返回给模型让它重试。

我个人在实际操作中的体会是，Agent开发本质上是一个“把不确定性锁进受控格子”的过程。模型本身是概率性的，但你的编排层、路由逻辑和状态管理必须是确定性的。上面这套基于LangGraph + FastAPI的搭建方案，谈不上完美，但在我自己的项目中经受住了不少真实流量的考验，稳定性比我最早用纯Chain实现的那版好了太多。最后再分享一个小技巧：在调试Agent时，把每一次工具调用的入参和出参都记录下来，因为大部分诡异问题，最后都能从工具调用日志里找到根源。