1. 从Chain到Graph:为什么LangGraph是Agent开发的分水岭
先聊个真实感受。热搜词里那句“让AI真的下地干活”,几乎是所有做过Agent项目的人心里最痒的一句话。ChatGPT刚火那会儿,大家拿LangChain写链式调用,一个输入进去,经过几个Prompt模板,出来一段结果。但真做起Agent来,你会发现事情没那么简单:Agent要能观察环境、决定行动、调用工具、看到结果再继续思考,这是一个循环往复的过程。传统Chain是线性的,一次跑完就结束,根本没法表达“先查数据、再写SQL、发现数据不对、重新查一遍”这样的逻辑。
LangGraph就是为了解决这个核心痛点出现的。它是一个基于图结构的Agent编排框架,把AI流程建模成一张有向图:节点(Node)是你要执行的动作,边(Edge)是状态流转的路径,图的状态(State)则携带所有上下文在节点之间传递。说白了,它不是把Prompt串成一条直线,而是让你像画流程图一样编排AI的工作过程,节点之间可以跳转、循环、回退,完全由代码和AI的判断决定。
我第一次用LangGraph时,最强烈的感受是:这不就是给AI加了一张流程图吗?但正是这张流程图,解决了Agent开发里最恶心的两个问题——状态管理混乱和执行路径不可控。以前写Agent,循环逻辑要靠while循环硬写,每次迭代的结果要自己拼到一个大字典里,哪个环节出错了也很难回溯。用LangGraph,整个状态就是全局共享的一个数据对象,每个节点读取它、更新它,图框架负责传递和保存,你要做的就是定义好节点和边。
这篇文章我打算按自己的学习路径来写:先讲清楚LangGraph的核心设计思想,再逐个拆解State、Node、Edge这几个基础概念,然后从零手写一个带工具调用的小Agent,最后把服务用FastAPI包起来跑在线上去。内容覆盖LangGraph基础和工具调用落地,适合刚接触LangGraph、想搞懂它到底怎么用的朋友。要是你已经在链式调用里写了一堆if...else...,那这篇正好帮你从“链”跳到“图”。
1.1 传统链式调用覆盖不了的场景
咱们先把场景铺开。假设你要做一个售后客服Agent,用户说“我上周买的耳机充不进电,帮我查一下订单”。这个需求拆开来看,Agent至少要经历这么几步:
- 判断用户的意图——是退换货、维修,还是单纯咨询
- 从订单系统里查出订单状态和商品信息
- 根据售后规则判断下一步行动——是发退货链接,还是转人工
- 生成对用户的最终回复
这里面有个关键点:第二步的结果会影响第三步的走向。如果查出来订单已过退货期,Agent就要走“维修”分支;如果还能退,就走“退换货”分支。再细一步,调用订单API可能超时、可能查不到数据,那Agent还得自动换个策略,比如用用户ID再查一次。
这种有分支、有循环、有依赖的场景,用LangChain的Chain结构是非常痛苦的。Chain的RunnableSequence本质上是固定的管道,输入从一端流到另一端,中间不能停下来、不能跳转。你当然可以把if...else...写在自定义函数里,但那等于把流程控制权从框架手里抢回来自己维护,代码一多就变成一团乱麻。
LangGraph的解法是把这种流程直观地建模成图。节点代表“调用LLM”“调用工具”“运行Python函数”,边代表“下一步去哪儿”,条件边则让AI决定走哪条路。图天然支持分支和循环,而且状态是显式传递的,每一步都看得见摸得着。这才是Agent真正需要的运行时。
1.2 LangGraph的核心:状态机遇上AI流程
LangGraph本身借鉴了状态机(State Machine)的思想。你对状态机不熟也没关系,想象一个电梯控制系统:电梯在“运行”状态、在“静止”状态,按楼层按钮触发状态切换,每一步都有明确的规则。LangGraph把AI流程也看成这样的状态机:
- 系统的当前状况全部保存在
State里,像一个实时更新的中央数据仓库 Node是被触发执行的操作,执行完后会更新State- 根据State当前的值,条件边决定下一跳是哪个节点
- 整个过程在图(Graph)里循环,直到走到
END节点
这种设计让AI流程变得可控。传统Agent开发最怕的就是模型“天马行空”,一个循环能跑几十轮不收敛。LangGraph允许你显式设置最大递归次数、定义停止条件、甚至分支出去做多个并行任务再合并结果。这些能力一层层垒下来,LangGraph就不只是LangChain的“升级版”,而是一个独立的Agent编排层。
我在写第一个图的时候,心里只有一个感慨:流程不再是藏在代码里的隐式逻辑,而是像画架构图一样摆在了桌面上。这个变化带来的调试体验是质的飞跃——出问题不用打日志猜流程走到哪,直接打印State截图就能看出来。
2. 五个核心概念一次讲透
说实话,LangGraph的API设计得很有章法,但也因此劝退了不少人。初看文档时,满屏的StateGraph、add_node、add_edge、END,配合几个抽象的名词,很多人第一反应就是“这和LangChain不是一个套路吗,怎么那么绕”。其实它的核心概念只有五个,搞懂这五个,剩下的全是组合使用。
2.1 State:贯穿全流程的共享数据仓库
State是整个图运行时唯一的数据载体。你可以把它理解成一个不断被更新的大字典,图里的每个节点都能读它、改它。LangGraph官方文档里最常出现的State定义方式是用TypedDict:
from typing import TypedDict class AgentState(TypedDict): messages: list # 对话历史 order_info: dict # 查到的订单信息 intent: str # 用户意图分类结果 final_answer: str # 最终回复TypedDict的好处是给字典加上了类型约束,IDE能自动补全,运行时会校验报错,对于复杂Agent来说这个约束能少踩很多坑。当你调用StateGraph(AgentState)初始化图时,这个类型就成了整张图的“全局变量声明”。
实际操作中我发现一个设计State的关键点:不要图省事把所有东西塞进一个字段,要有意地区分“短期工作变量”和“长期上下文”。比如对话历史可能很长,但你可以只在最后一步汇总时用它;订单原始JSON很大,但下游节点只需要提取过的几个字段。把State设计得过胖,不仅每次传递都浪费token,而且会让排查问题变得费劲——因为你根本不知道是哪个节点改了哪个字段。
LangGraph还允许你通过Annotated配合operator.add来定义字段的更新方式。比如消息列表用追加而不是覆盖:
from typing import Annotated from typing_extensions import TypedDict import operator class AgentState(TypedDict): messages: Annotated[list, operator.add] # 新消息追加到旧消息后面 order_info: dict这样就不用写state["messages"] = state["messages"] + new_messages这种手动拼接代码了。定义State合并规则其实是LangGraph一个容易被忽略但极其重要的能力,它直接决定了多节点协作时数据怎么流转。
2.2 Node:真正“干活”的地方
Node就是图上的一个操作单元,本质上是一个普通Python函数。函数签名很固定:接收一个state参数(整个State字典),返回一个dict,返回的字典会被合并回State。来看一个最基础的节点函数:
def analyze_intent(state: AgentState) -> dict: # 假设这里已经调用了一个意图识别模型 user_input = state["messages"][-1]["content"] intent = "after_sales" # 简化判断 return {"intent": intent}这个函数读到了state里最后一条用户消息,做了处理后返回一个{"intent": ...},LangGraph就会用返回值更新State里的intent字段。需要注意的是,节点的返回值不需要覆盖全部State,只需要返回你改了的那部分。这个设计让每个节点可以只关心自己负责的领域。
说白了,Node是图里唯一能触碰外部世界的地方。你要查数据库、调API、运行重型计算,都写在Node里。LLM调用本身也可以封装成一个Node——把Prompt模板、模型调用、输出解析放在函数里,只暴露state进、dict出的接口。这样做有个额外好处:测试单个Node的时候你根本不需要起一张图,直接给个假State就能单测。
写Node的时候我踩过一个坑:不要在节点函数内部再去直接修改传入的state参数。LangGraph的State是不可变快照(immutable updates),你把state["order_info"] = xxx写在函数里,它确实能改这个局部引用,但不会真正影响图的State流转。正确做法永远是返回一个字典,让框架去合并。一开始不习惯,觉得多此一举,但调试几次后才会体会到这个约束的价值——每一步状态变化都有明确的“提交记录”。
2.3 Edge与条件边:控制流程走向的两把钥匙
只有节点没有边,图就只是一堆散落的函数。Edge的作用就是告诉LangGraph:这个节点跑完之后,下一步去哪个节点。最简单的添加方式是这样:
from langgraph.graph import StateGraph, END graph = StateGraph(AgentState) graph.add_node("analyze_intent", analyze_intent) graph.add_node("check_order", check_order) graph.add_node("generate_answer", generate_answer) graph.set_entry_point("analyze_intent") # 入口:先分析意图 graph.add_edge("analyze_intent", "check_order") # 分析完后查订单 graph.add_edge("check_order", "generate_answer") # 查完订单生成回复 graph.add_edge("generate_answer", END) # 生成完结束这种固定路径适合流水线场景,但Agent的核心价值恰恰在于不走固定路径。所以LangGraph提供了add_conditional_edges,让“下一步去哪”由节点函数的返回值动态决定:
def route_after_check(state: AgentState) -> str: # 根据查单结果决定走退货流程还是维修流程 if state["order_info"]["can_refund"]: return "refund" else: return "repair" graph.add_conditional_edges( "check_order", route_after_check, { "refund": "refund_node", "repair": "repair_node", } )这个条件和字典映射的组合,写起来特别像路由表:函数负责返回一个字符串标签,字典负责把标签映射到实际节点。LangGraph拿到返回值后,就去字典里查对应的节点名,跳到那个节点继续跑。有条件边的加持,一张图就能写出一棵完整的决策树。
2.4 图的编译与执行:把设计变成可运行的Agent
图设计好之后,还必须经过编译这一步才能执行:
app = graph.compile() result = app.invoke({"messages": [{"role": "user", "content": "耳机坏了怎么办"}]})compile()会做一次内部结构解析,把节点、边、条件检查一遍,有问题会立刻报错。比如你引用了一个不存在的节点名,编译阶段就能被抓出来,而不是等到运行到那一步才出异常。从这个角度说,compile()像是一个图结构的“静态检查器”。
invoke()是同步执行接口。数据进去后会从入口节点出发,沿着边和条件一路跑到END,最终返回完整的State(包含所有节点更新的字段)。如果图里有循环——比如Agent反复调用工具直到结果满意——invoke()会一直循环到满足退出条件为止。如果你希望拿到中间态,比如每跑完一个节点就拿到一次状态快照,可以用stream()接口:
for event in app.stream({"messages": [...]}, stream_mode="updates"): print(event) # 每个节点执行后都会输出一步调试新图的时候,我强烈建议先用stream()把每一步输出都打出来,确认每个节点的返回值符合预期,再切回invoke()做生产调用。这个习惯能让你把一个复杂的Agent调试时间从半天缩短到一小时。
2.5 循环不是Bug:Agent的“再想想”机制
讲了这么多基础概念,必须把Agent循环单独拿出来说一说。传统编程里循环要小心翼翼,但在Agent场景里,循环恰恰是智能的体现。一个Agent收到用户请求后,可能要用工具查一遍资料、发现资料不够、再调整查询词再查一遍,这个“查了又查”的过程本质上就是图上的一个环。
LangGraph对循环的支持是天然自带的:只要有一条边从后面的节点指向前面的节点,图就跑成了环。最常见的场景是“调用工具”节点结束后,把工具返回的结果放回State的messages,然后跳到“LLM决策”节点,让模型看了工具结果后再决定下一步动作。这就是ReAct模式的雏形——模型用一次推理决定要调哪个工具,工具返回后模型再推理下一步,直到模型认为问题已经解决。
写循环时最怕的是无限循环。LangGraph提供了两个保护措施:一是编译图时传recursion_limit参数限制最大步数,二是在条件边里写显式的“已完成”分支跳到END。我的习惯是:条件边里永远写一个终止分支,即使这个分支当时看起来永远不会走到。模型的行为没法100%预测,这条退路是给意外情况兜底的。
3. 从零构建第一个LangGraph应用
概念说再多,不如亲手跑一个。这一节我带你搭一个完整的LangGraph应用,它做的事情很简单:收到用户的问题后,先判断意图,再决定是直接回答还是调用一个工具。工具这里我用“查天气”来演示(纯模拟),但你完全可以把工具换成查订单、查数据库、调用业务API。
3.1 环境准备与工程结构
先装依赖。我推荐单独建一个虚拟环境,避免污染其他项目的依赖:
python -m venv .venv source .venv/bin/activate # Windows用 .venv\Scripts\activate pip install langgraph langchain-openai python-dotenv注意这里我用了langchain-openai,这是LangChain新版的OpenAI适配包。如果你用的是langchain旧版的langchain.llms.OpenAI,那大概率会碰到导入路径不兼容的问题,建议统一用新版。最后把OpenAI的API Key配到环境变量里,或者写在.env文件里启动时加载。
工程结构我习惯这样组织,便于后面扩展:
agent/ ├── main.py # 图组装与执行入口 ├── state.py # State定义 ├── nodes/ # 各节点的实现 │ ├── __init__.py │ ├── analyze.py │ ├── tools.py │ └── answer.py ├── tools/ # 工具函数 │ ├── __init__.py │ └── weather.py └── requirements.txt小项目不用分这么细,但当图里节点数量超过四五个,没有按职责拆文件的话,改起来会非常痛苦。LangGraph的节点本质上是纯函数,模块化本来就自然,没必要都堆在一个文件里。
3.2 定义State、工具和节点
State按上一节的思路定义,为了演示追加消息的合并规则,我用operator.add处理消息列表:
# state.py import operator from typing import Annotated, TypedDict class AgentState(TypedDict): messages: Annotated[list, operator.add] need_tool: bool # LLM判断是否需要调用工具 tool_result: str工具这里我用一个带延迟的模拟函数,模拟真实API调用:
# tools/weather.py import random def get_weather(city: str) -> str: """模拟查询天气,实际项目里替换成真实API调用""" temp = random.randint(15, 30) return f"{city} 当前气温 {temp} 摄氏度,天气晴转多云"节点部分,意图判断节点让LLM决定“要不要工具”——为了让行为可解释,我让模型用结构化的方式输出:
# nodes/analyze.py from langchain_openai import ChatOpenAI from state import AgentState model = ChatOpenAI(model="gpt-4o-mini", temperature=0) def analyze_intent(state: AgentState) -> dict: last_message = state["messages"][-1]["content"] # 让模型输出JSON,解析后作为判断结果 resp = model.invoke( f"用户说:'{last_message}'。请判断是否需要查询实时信息(比如天气、订单、库存)。" f"只需要回答是或否。" ) need_tool = resp.content.strip().startswith("是") return {"need_tool": need_tool, "messages": []}等一下,这里有个注意事项:不要随意往State里塞空消息列表占位。因为messages字段用了operator.add合并,如果你返回一个空列表,合并时它不会追加任何消息,这没问题;但如果你图省事返回{"messages": [...]},就会把一条空消息存进去,进而污染后面的对话上下文。LangGraph的更新是增量式的,你只需要返回真正想更新的字段。
如果need_tool为True,就进入工具调用节点:
# nodes/tools.py from state import AgentState from tools.weather import get_weather def call_tool(state: AgentState) -> dict: user_request = state["messages"][-1]["content"] # 这里简化处理:从消息里提取城市名,实际项目里让模型先做参数抽取 city = "北京" result = get_weather(city) return {"tool_result": result, "messages": [ {"role": "tool", "content": f"查询结果:{result}"} ]}最后是回答节点,它把工具结果和用户原始问题合并,用LLM生成最终回复:
# nodes/answer.py from langchain_openai import ChatOpenAI from state import AgentState model = ChatOpenAI(model="gpt-4o-mini", temperature=0.3) def generate_answer(state: AgentState) -> dict: last_message = state["messages"][-1] if state["need_tool"] and state["tool_result"]: prompt = f"工具查询结果:{state['tool_result']}\n请基于这个结果回答用户。" else: prompt = "直接回答用户的问题。" resp = model.invoke([ {"role": "user", "content": last_message["content"]}, {"role": "assistant", "content": prompt} ]) return {"messages": [{"role": "assistant", "content": resp.content}]}3.3 组装图并执行验证
现在把节点和边拼到一起。这一版我设计了三条路径:不需要工具就直连回答;需要工具就先去工具节点再生成回复;工具节点执行后也可以选择再走一次判断(演示循环能力,虽然这里用不上,但结构上留好了):
# main.py from langgraph.graph import StateGraph, END from state import AgentState from nodes.analyze import analyze_intent from nodes.tools import call_tool from nodes.answer import generate_answer def route_after_analyze(state: AgentState) -> str: if state["need_tool"]: return "call_tool" return "generate_answer" graph = StateGraph(AgentState) graph.add_node("analyze_intent", analyze_intent) graph.add_node("call_tool", call_tool) graph.add_node("generate_answer", generate_answer) graph.set_entry_point("analyze_intent") graph.add_conditional_edges( "analyze_intent", route_after_analyze, {"call_tool": "call_tool", "generate_answer": "generate_answer"} ) graph.add_edge("call_tool", "generate_answer") graph.add_edge("generate_answer", END) app = graph.compile() result = app.invoke({"messages": [{"role": "user", "content": "北京今天天气怎么样?"}]}) print(result["messages"][-1]["content"])这个流程跑起来之后,关键词“LangGraph 工具调用”的整个闭环就通了:用户输入被分析、LLM判断需要工具、工具被调用获得结果、结果被合成为最终回复。而且每一次状态流转都被LangGraph记录在案,出问题可以直接翻中间态。
我实测调试时最爱用stream模式,把每步状态变化打在终端上,基本一眼就能看出哪个节点出了问题。比如:
for chunk in app.stream( {"messages": [{"role": "user", "content": "北京今天天气怎么样?"}]}, stream_mode="updates" ): print(chunk)输出里能看到“分析节点”先执行、返回了need_tool=True,然后“工具节点”执行、把查询结果写入State,最后“回答节点”基于工具结果生成回复。这种透明度是传统链式调用完全给不了的。
4. 让Agent真正“下地干活”:工具调用与FastAPI实战
基础图能跑通之后,就该聊落地了。热搜词那半句话特别戳人——“让AI真的下地干活”。企业里的Agent不会只停留在玩玩具的阶段,它要去查数据库、写工单、调第三方API、在网页上操作。做到这些,核心就是工具调用(Function Calling / Tool Calling)的设计。
4.1 工具调用的本质:把函数说明书给模型
工具调用在技术本质上并不神秘:你写一批函数,把它们用@tool装饰器包装起来,连同函数的名称、参数描述、返回值说明一起发给LLM。模型在收到用户请求后,从这些“工具说明书”里选一个合适的函数和参数,然后以结构化的形式(JSON对象)返回“我想调用这个函数,参数是这样”。你的程序拿到这个JSON后,实际执行对应函数,再把结果作为新消息发回给模型,让模型基于函数输出继续回答。整个过程可以循环多次。
用LangChain写一个工具非常简单:
from langchain_core.tools import tool @tool def get_weather(city: str) -> str: """根据城市名查询当前的天气情况,参数city是城市名,如'北京'。""" return f"{city} 今天的天气是晴,气温26度"注意,get_weather函数体本身可以不重要,真正给模型看的是三样东西:函数名get_weather、函数签名参数city、以及docstring里的自然语言描述。我在实际项目里发现,docstring写得好不好,直接影响模型选工具的准确率。你要写“查询城市天气”,不能写“内部天气服务接口”这种模糊描述。参数说明也一样,最好带上示例值和边界条件,比如city要说明是中文城市名,避免模型传成拼音。
4.2 用FastAPI把Agent包成HTTP服务
工具定了,Agent图也定了,最后一步是让它以服务的形式常驻运行。这时候FastAPI就派上用场了。FastAPI的异步支持配合LangGraph的ainvoke,可以很自然地实现并发请求处理。
给你一份可以直接抄作业的服务代码:
# server.py from fastapi import FastAPI from pydantic import BaseModel from main import app as graph_app # 把编译好的图导入进来 app = FastAPI(title="AI Agent Service") class UserRequest(BaseModel): message: str session_id: str = "default" class AgentResponse(BaseModel): reply: str session_id: str @app.post("/api/agent", response_model=AgentResponse) async def run_agent(req: UserRequest): # 实际项目里,session_id可以从数据库或缓存里恢复历史状态 result = await graph_app.ainvoke({ "messages": [{"role": "user", "content": req.message}] }) return AgentResponse( reply=result["messages"][-1]["content"], session_id=req.session_id )启动服务后,你就能用curl测试整个链路:
curl -X POST http://localhost:8000/api/agent \ -H "Content-Type: application/json" \ -d '{"message": "北京现在多少度?"}'这一套下来就是热搜词里说的“基于FastAPI + LangChain + LangGraph的AI Agent”的标准雏形。之前我在博客里看过不少项目把这三样组合当作“全家桶”来用,实话实说,这个搭配确实顺——FastAPI负责Web层、LangChain负责LLM调用和工具抽象、LangGraph负责流程控制,各司其职,边界清楚。
4.3 三个让Agent更“顶用”的工程习惯
光把服务跑起来不算完,真正“下地干活”还需要把工程细节打磨到位。分享几个我在项目中反复打磨过的习惯,每个都踩过坑。
第一,工具结果必须“结构化回传”模型。工具函数返回的不一定要是自然语言字符串,也可以是一个结构化字典。但发回给模型时,要么转成可读文本,要么保留JSON结构并让模型明确知道这是工具输出。我在一个项目里遇到过模型持续误读工具结果的情况,排查半天发现是工具返回了一个纯数字,模型把它当成了最终答案而非参考数据。解决方案是每次都把工具结果包一层“工具执行完成,返回结果如下”的说明再放回消息列表。
第二,给工具加“失败兜底”路径。真实世界里API会超时、数据库会连接失败、第三方服务会返回脏数据。你在设计条件边时,一定要考虑到工具节点可能抛异常的情况。我习惯在工具节点里捕获所有异常,并把错误信息写进tool_result,让模型看到错误后自己决定是重试还是换方案。这比直接让Agent崩溃优雅得多。
第三,会话状态持久化。上面例子中每次请求都从零开始,实际用户不会接受这种“失忆”对话框。LangGraph提供了checkpointer机制,可以把每一步的State保存下来,后续用同一个thread_id恢复上下文。FastAPI层只需要在请求里带上session_id并传给调用入口:
from langgraph.checkpoint.memory import MemorySaver checkpointer = MemorySaver() app = graph.compile(checkpointer=checkpointer) result = await graph_app.ainvoke( {"messages": [{"role": "user", "content": req.message}]}, config={"configurable": {"thread_id": req.session_id}} )这样就把多轮对话、回溯历史、状态恢复全交给LangGraph框架处理,省掉大量自己写状态管理的代码。
5. 常见问题与排查技巧实录
用LangGraph写了几个月,遇到的坑不说上百也有几十个。这一节我挑最有代表性的几个,做一份“实测速查表”,帮助后来者少走弯路。
5.1 状态不更新的诡异现象
现象:节点函数里明明修改了State的值,但下一个节点读到的还是旧值。
原因:在节点函数内部直接改传入的state字典,而不是通过返回值更新。LangGraph的状态流转是基于返回值的增量合并,原地修改不会生效。这个问题新手最容易犯,因为Python里字典本来就是可变对象,改了好像也没报错。
排查:先在节点函数末尾加一个print看返回值,再在下一个节点开头打印整个State,比对差异。基本一眼就能定位。
5.2 Agent无限循环停不下来
现象:图在有“工具调用 → 模型分析 → 再调用工具”的循环边时,一直执行不停,直到触发recursion_limit报错终止。
先说结论:原因:条件边里没有写“结束分支”,或者模型每次判断都坚持要再调一次工具,形成了死循环。
排查与解决:
- 编译图时设置
recursion_limit,比如app = graph.compile() # 默认25步,可配置 recursion_limit=10,作为一种兜底保护 - 条件边必须包含“任务完成、直接END”的分支。我见过不少实现把
route_after_tool只写了“继续调用工具”和“生成答案”两条路,但漏了“任务其实已经完成,直接结束”这种判定,导致模型反复纠结 - 在工具结果消息里明确提示模型“如果已有足够信息,请直接给出最终答案”,这类Prompt工程微调真的管用
5.3 工具消息格式不对导致LLM调用报错
现象:调用模型时报错,提示消息序列中role=tool的消息必须紧跟在对应的assistant消息之后。
原因:LangGraph允许任意修State,但LLM对消息序列的格式有严格要求。如果你在messages里追加了一条工具结果,却没把它放在合适的对话位置——比如夹在两条user消息之间——模型API就会直接拒绝。
排查:把传给模型的messages列表完整打印出来,检查顺序。通常正确的序列是:user提问 →assistant说“我要调用工具” →tool返回结果 →assistant最终回答。如果在图里跳过了“assistant要调用工具”这条消息,就会出问题。
5.4 并发请求串号的坑
现象:线上服务并发高了以后,用户A的请求拿到了用户B的上下文。
原因:早期我在FastAPI里把State对象定义成了模块级全局变量,多个请求共享了同一个State实例。LangGraph本身是无状态的,它的State是每次调用的参数,但如果你在外面用了全局字典保存“会话状态”,并发场景就会互相覆盖。
排查:把State和Graph实例彻底分开,Graph是只读的、可复用的,State是每次调用重新创建的。会话级状态一律走checkpointer或者外部存储(Redis、数据库),不要放在模块级变量里。
5.5 我的避坑速查表
| 问题类别 | 典型表现 | 最快解法 |
|---|---|---|
| 状态不更新 | 下个节点读到旧值 | 检查是否在节点内直接改字典,改为return新字段 |
| 死循环 | 反复调用工具不停 | 加recursion_limit,条件边增加终止分支,优化终止Prompt |
| 消息顺序错乱 | LLM API拒绝请求 | 打印messages顺序,确保tool消息跟在assistant消息后 |
| 并发串数据 | 多用户上下文交叉 | 禁用模块级可变State,改用checkpointer或外部存储 |
| 节点异常吞没 | 图静默结束没结果 | 在节点内捕获异常并写入State字段,让模型看到错误信息 |
| 工具参数错误 | 模型传错参数值 | 优化工具函数docstring,加参数格式说明与示例 |
最后一点个人体会
从LangChain链式调用转到LangGraph,最直观的改变是思维方式的转换:不要再想“这个流程按什么顺序跑”,而是想“整个系统有哪些状态、哪些动作、状态之间如何流转”。这种建模方式更接近真实世界的业务逻辑,也因此更抗折腾。
我个人的建议是:刚开始不要追求复杂,从一个只有三个节点、一条条件边的图开始,把工具调用循环跑通,再逐步加持久化、加并行节点、加人工审批介入。LangGraph的复杂度是按需累加的,你要做的只是在每个阶段守住状态的清晰边界。把这套基础设施搭好,AI Agent就不再是演示台上的玩具,而是真正能在业务流程里稳定运转的“劳动力”。