ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

LangGraph图模型:用状态机思维编排复杂Agent流程

LangGraph图模型:用状态机思维编排复杂Agent流程 我最早用LangChain写Agent的时候总觉得哪里不得劲链Chain是一条直线Prompt进来、结果出去中间出个岔路就要拆分重写更别说工具调完还要回来再判断一次。后来换到LangGraph的图模型才算是把Agent的流程彻底理顺了。图模型这东西说白了就是让你把Agent的每一步都画成一张有向图——节点是干活的地方边是跳转的规则State则是贯穿全图的传话人。这篇文章写给两类人一类是已经在用LangChain、想搞复杂Agent又不知道从哪下手的朋友另一类是听说过LangGraph但一直没搞明白图模型到底解决什么问题的读者。我会先讲清楚为什么Agent编排需要一张图再把LangGraph的核心概念拆开揉碎最后带你把一个会调用工具的Agent完整跑起来。1. 为什么Agent编排非得用图模型——一个老LangChain用户的思考1.1 链式调用的天花板只能前进不能返工LangChain早期的编排方式是把Prompt、模型、输出解析器串成一条链形式大概是prompt | llm | parser。数据流是单向的就像工厂里的传送带原材料从一头进去从另一头出来中间环节再多也始终是一条直线。这种设计在处理单轮问答文档总结这类任务时很顺手可一旦涉及Agent问题就暴露了。真实的Agent是先判断要不要调工具——调完工具拿到结果——再根据结果决定下一步干什么。这个再判断就意味着流程要往回走回到LLM节点重新来一轮。传送带不会往回送零件链式结构也表达不了这种回环。我第一次尝试用Chain描述这个逻辑时最后写出来的东西是一堆if-else包着多个链外层判断走哪条链链内部又分支没写多少就开始头晕。更麻烦的是Agent的每一步都在累积对话上下文链式结构没有一个统一的地方存到目前为止发生了什么。每次都要手动把历史消息塞进下一条链的Prompt里稍不留神就漏掉几轮。1.2 图模型带来的四个核心能力分支、循环、状态、可控LangGraph用图模型解决了这些拧巴的地方。一个普通Ethiopian Chain是DAG有向无环图其实也就是长一点的直线小分叉。而LangGraph的图模型允许有环这让它可以自然表达LLM判断要不要调工具→调完回到LLM再判断这种循环动作这是第一个核心能力。第二个能力是分支。业务里经常出现消息分类成A/B/C三类各走各的处理流程这种需求图模型里的条件边conditional edge就是干这个的。路由函数读一下当前State返回一个key图引擎根据映射表决定跳到哪个节点这就是一个标准的意图路由。第三个能力是全局状态。图上所有节点共用一份State对象节点从State里取它想要的输入再把输出写回State。这就好比整个团队共享同一块白板谁走之前都往上添几笔后面的人永远不会缺上下文不需要手动传参。第四个能力是可控。LangGraph支持checkpointer持久化每一步状态也支持在任意节点前后暂停等人工确认后再继续。这一点对上线太重要了——我后面会专门演示怎么给Agent装一个人闸。说白了图模型不是把Agent变成什么黑科技而是让Agent的控制流变得跟普通后端程序一样清晰。那些说让AI真的下地干活的项目底层十有八九都是这种图化的编排方式。1.3 LangGraph图模型的整体架构节点、边、状态的三角关系LangGraph的整套API其实就是围绕三个概念转节点Node、边Edge、状态State。节点是函数入参是当前State返回值是一个dict表示要对State做的更新。边是连线决定节点执行完后下一步进哪个节点。条件边更高级一点它会调用一个路由函数根据函数的返回结果动态决定走向。状态就是这个框架的白板为了避免不同节点写同一个字段时互相覆盖LangGraph还设计了Reducer机制——同一个字段的新旧值怎么合并由定义字段时挂载的Reducer函数说了算。你做的事情只有四步用StateGraph新建一张图注册节点连边然后compile编译成可执行对象。编译完的app长得跟一个LangChain的chain差不多直接invoke就能跑。我第一次跑通的时候感慨这玩意儿从使用体验上很像在写一个非常简单的状态机框架但因为它把LLM、工具、路由全部抽象成普通节点所以能力上限完全不是一个量级。2. 核心概念拆解State、Node、Edge的正确打开方式2.1 State整个图的血液所有节点共用一套状态State在LangGraph里就是用TypedDict定义的一个数据模型。举个例子from typing import Annotated, TypedDict, Sequence from langchain_core.messages import BaseMessage from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[Sequence[BaseMessage], add_messages] current_step: str retry_count: int这里定义了两个字段messages是对话历史current_step记录当前步骤retry_count记录重试次数。每个节点执行完返回的dict只会更新它涉及的那几个字段其他字段保持不变不会把整个State捂手重来。有一个新手最常踩的坑如果你直接写messages: list没有挂add_messages这个Reducer那每次节点返回新的消息列表时会整体覆盖掉旧消息。等于每走一个节点历史对话就清空一次Agent直接失忆。加上Annotated[..., add_messages]之后框架会自动把旧消息和新消息拼成一个列表而不是覆盖。想保存完整的多轮对话这一步就是命根子。2.2 Node节点是唯一会干活的地方节点就是普通Python函数入参一个State的dict返回一个dict。函数里想干什么都行调LLM、查数据库、调外部API甚至sleep十秒再去执行没人拦你。从框架的角度看所有节点一视同仁——不区分什么LLM节点工具节点就是一个一个的执行单元。def agent_node(state: AgentState): messages state[messages] response llm_with_tools.invoke(messages) return {messages: [response], current_step: agent} def tool_node(state: AgentState): # 根据最后一条带tool_calls的消息执行对应工具 ... return {messages: tool_messages, current_step: tools}写节点的时候记住一个原则节点只从State取需要的数据只把自己要更新的字段return出去。不要试图在节点之间靠参数传值图模型的设计就是所有信息走白板。我之前因为贪方便把中间计算结果塞在了函数返回值里而不是State里结果另一边节点拿不到数据查了半天才发现问题出在自己绕开了State。2.3 Edge与条件边直线和岔路口边的API很少常用的固定边和条件边两种。固定边用add_edgeA节点执行完必须走B节点。启动入口用set_entry_point或者更推荐的add_edge(START, 节点名)。终点统一叫END图跑到END就算这一轮结束。条件边用add_conditional_edges它需要三个参数起点节点名、路由函数、映射表。路由函数接收当前State返回一个字符串key映射表负责把key转换成下一步节点名。def route_after_agent(state: AgentState): last_message state[messages][-1] if last_message.tool_calls: return continue return finish graph.add_conditional_edges( agent, route_after_agent, { continue: tools, finish: END } )这段逻辑看起来简单但它就是Agent决策循环的核心LLM说我要调工具就跳去工具节点LLM说答案齐了就结束。映射表的key必须跟路由函数的返回值完全对得上少一个直接运行时报错这个我后面在踩坑部分细说。2.4 内置MessagesState能少写就少写LangGraph在langgraph.graph.message里内置了一个MessagesState已经定义好了带add_messages的messages字段。如果你只关心对话消息可以直接继承它from langgraph.graph import MessagesState class AgentState(MessagesState): current_step: str这样连Annotated和add_messages都不用自己写messages字段天然就是自动累加的行为。适合快速跑Demo的场景。不过一旦Agent业务复杂我还是建议自己定义完整的State模型——字段名写清楚代码里搜起来方便别人接手也看得懂。3. 从零搭一个会调用工具的图模型Agent3.1 环境准备装什么、怎么验证先用pip把依赖装上pip install langgraph langchain langchain-openai我用Python 3.11验证过这段代码Python 3.9以上应该都没问题。如果你已经装了langchain相关包注意langgraph要比较新的版本旧版本里有些API名字不一样比如老版本用FINISH表示终点新版本叫END老版本用graph.set_entry_point(node)新版本也可以用graph.add_edge(START, node)。我本地跑的时候LangGraph还在0.2.x现在升到0.4了核心概念没变但个别导入路径会变。装完验证一下import langgraph print(langgraph.__version__)3.2 定义状态和工具给图准备血管和手我先建两个模拟工具一个是查天气一个是查订单状态。这里不做真实API请求函数返回假数据就够了重点是把工具注册进图里的链路跑通。from langchain_core.tools import tool tool def get_weather(city: str) - str: 查询指定城市当前天气 return f{city} 当前晴天气温 26℃微风 tool def get_order_status(order_id: str) - str: 查询订单物流状态 return f订单 {order_id} 已发货预计明天送达然后定义State和绑定工具的LLMfrom langchain_openai import ChatOpenAI from langgraph.graph import MessagesState, StateGraph, START, END from langgraph.prebuilt import ToolNode llm ChatOpenAI(modelgpt-4o-mini, temperature0) llm_with_tools llm.bind_tools([get_weather, get_order_status]) class AgentState(MessagesState): passbind_tools的作用是把工具的描述和参数schema合进模型的调用接口里这样模型才知道这批工具里有这么两把能用。你的工具函数写得越清晰docstring写得越完整模型判断该调哪个工具就越准确。3.3 搭节点和路由让图跑起来接着定义Agent节点和工具节点再写路由函数。工具节点直接使用prebuilt里的ToolNode它会自动从对话消息里找出模型返回的tool_calls执行对应工具把结果以ToolMessage形式追加回messages省去自己写解析逻辑的麻烦。from typing import Annotated def agent_node(state: AgentState): response llm_with_tools.invoke(state[messages]) return {messages: [response]} def should_continue(state: AgentState): last_message state[messages][-1] if hasattr(last_message, tool_calls) and last_message.tool_calls: return tools return end tool_node ToolNode(tools[get_weather, get_order_status])然后建图graph StateGraph(AgentState) graph.add_node(agent, agent_node) graph.add_node(tools, tool_node) graph.add_edge(START, agent) graph.add_conditional_edges( agent, should_continue, { tools: tools, end: END } ) graph.add_edge(tools, agent) app graph.compile()整个控制流长这样START - agentagent判断要不要调工具要就跳tools执行完再回agent不要就直接到END。这个tools回agent的环就是LangGraph相比普通Chain最核心的价值模型可以调一次、两次、三次工具直到它认为信息够了才结束。3.4 编译、执行与可视化看你的图长什么样compile()之后拿到的app就相当于一张装配好的流水线直接invoke就能跑from langchain_core.messages import HumanMessage result app.invoke({ messages: [HumanMessage(content帮我查一下北京天气和订单20250088的状态)] }) print(result[messages][-1].content)执行过程大概是agent先调用LLM模型返回我需要同时调用get_weather和get_order_status两条tool_calls图引擎跳到tools节点执行两个工具工具结果追加回messages然后回到agent节点LLM看到工具结果后生成最终答案。如果模型觉得信息还不够它会继续发起下一轮工具调用循环直至答案成型。调试时有个特别方便的方法直接打印图的结构print(app.get_graph().draw_ascii())它会输出一张ASCII结构图节点和边的流向一目了然出了路由问题看这张图比看日志直观得多。我在项目里几乎每改一次图结构就打印一次确认连接没粗错。3.5 加一个人闸用checkpointer和interrupt做人工审核让Agent全自动跑完所有流程在Demo里很爽但在真实业务里退款、发邮件、改库存这种操作总得有个人拍板。LangGraph给了一套断点续跑方案。先引入内存检查点from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() app graph.compile(checkpointermemory, interrupt_before[agent])interrupt_before[agent]的意思是每次进agent节点之前先暂停。配合线程ID使用config {configurable: {thread_id: order-20250088}} app.invoke({ messages: [HumanMessage(content查询北京天气)] }, config)这时候图会在进入agent之前暂停你可以用app.get_state(config)查看当前快照判断该不该放行。确认没问题之后再调一次app.invoke(None, config)图会从断点继续执行而不是从头开始。这个机制在业务里就是给Agent加了一道人工审批闸门比在外面用一个大if包住整个Agent要优雅得多。我实际用下来的体会是所谓让AI下地干活起码有一半的工作是设计好这些打断点和恢复机制不然AI跑得再快也不敢让它碰生产数据。4. 图模型落地实录常见坑、调试法和一点个人心得4.1 状态丢失和字段被覆盖Reducer不是可选项第一次跑带状态更新的图我在State里写的是messages: list结果每经过一个节点消息就只剩最新一条前面几轮到工具调用全丢了。当时一度以为是图执行顺序有问题查了一个多小时最后才发现是缺了Reducer。LangGraph里的State更新规则很简单字段没有挂Reducer新值直接覆盖旧值挂了Reducer由Reducer决定怎么合并。处理消息列表就必须挂add_messages其他像计数器、累加器同理。所以写State的时候多问自己一句这个字段被多个节点写过之后我是要合并还是覆盖想清楚再定义不然后面全是坑。4.2 条件边永不触发或路由报错先查映射表key条件边相关的报错最常见的是InvalidUpdateError或者Conditional edge returned a value that did not match any of the provided keys之类大意是路由函数返回了一个映射表里没有的key。我的排查顺序是这样的第一步看路由函数的返回值和映射表的key是不是逐字对应连空格和大小写都别放过第二步确认路由函数里读的字段在State里真实存在别把tool_calls读成tool_call这个单复数错误我犯过第三步看节点名有没有注册add_conditional_edges映射表里写的目标节点必须是已经add_node过的。映射表key一旦对不上LangGraph不会帮你猜直接抛异常所以报错反而是好事怕的是路由函数逻辑错误导致永远走同一条分支那才真的难查。4.3 工具调用死循环给Agent装上刹车模型在Tool calling模式下偶尔会反复返回同样的tool_calls进入调用工具→结果回来→再调用工具的死循环。这种情况在复杂业务里不是小概率尤其是工具结果比较长、模型没看明白的时候。我在项目里设了三层防线第一层graph.compile(recursion_limit20)限制整个图的节点执行次数超过直接终止第二层在路由函数里判断工具调用次数超过N次就强制走END第三层配合前面说的checkpointer把每轮工具调用的信息都落盘就算真出问题也能拿着快照复盘而不是整个流程报废。# 限制整体递归深度 app graph.compile(recursion_limit20)至于设多少次合适我一般先按业务链路长度估算比如模型最多需要调3个工具上限就定15~20留出余量但不至于让Agent无限空转。4.4 高效调试三板斧debug模式、状态快照、ASCII图调试图模型我最常用的三板斧。第一板斧是invoke的时候打开debug参数它会打印每一步进入哪个节点、State怎么变化信息量比看prompt日志大得多。第二板斧是用get_state和get_state_history看状态快照尤其是配了checkpointer之后整个图的执行轨迹都能回溯这对定位哪一步开始状态坏了价值极高。第三板斧就是前面说的draw_ascii()快速确认图结构。还有一个很土但很有效的习惯每次工具返回后单独打一层日志把ToolMessage的内容截断打印出来。很多Agent跑偏不是模型问题而是工具返回了一堆没用的噪音模型被带跑了。日志里把工具结果单独列出来一眼就能看出是不是工具那边的问题。说句实在话图模型解决的不只是技术问题更是思路问题。我以前写Agent总想着怎么把逻辑塞进一次调用里现在会先花时间画一张图哪个节点做什么判断哪条边循环哪个位置需要人介入。画清楚了代码基本就是一比一翻译。这个习惯救了我很多次每次项目上线前返工几乎都是因为流程没理清而不是代码写不出来。LangGraph的图模型本质上是把这个先想清楚流程的要求用框架僵化地固定了下来——你可以说它啰嗦但正是这种啰嗦让Agent变得真正可控。
返回列表