LangGraph从零搭建ReAct代理:状态、节点与工具调用实践
写这篇LangGraph实践教程之前我又把之前收藏夹里那堆ReAct大模型学习资料翻了一遍。说实话网上讲ReAct原理的文章已经很多了视频也一个比一个热闹但从我理解了思考-行动-观察到我能亲手写出一个会调工具的代理中间这道断层特别大。很多人看完收藏夹关掉页面打开IDE还是一脸懵。这篇文章就是来填这个断层的。我会用 LangGraph 从零搭一个最小但完整的 ReAct 代理把状态、节点、条件路由这些核心概念拆开讲清楚并且把我自己踩过的坑一并交代在第一线。适合第一次接触 LangGraph 的小白也适合已经被 AgentExecutor 黑盒搞到头疼、想换一种更可控方式的进阶读者。1. ReAct 的“思考-行动-观察”循环从论文概念到工程实践1.1 为什么普通 Chain 做不到“边想边干”先聊一个被问烂但真的很关键的问题大模型本身只是文本生成器你给它一串输入它给你一串输出仅此而已。它没有眼睛没有手无法主动去查数据库、看天气、查订单状态。传统思路下的 Chain 是把任务拆成线性几步第一步让模型生成一个中间结果第二步把中间结果拼进下一个 Prompt第三步继续……每一步都依赖于上一步的文本输出模型全程看不见外部世界的真实反馈。这就像是一个没出过门的人背地图你问他从 A 到 B 怎么走他能背得头头是道可一旦路上临时封路、天气突变他就只能开始编。Chain 更惨一点它连编完之后重新校正的机会都没有。ReAct 的思路完全不同模型每走一步不只是输出一段话而是输出一个下一步该做什么的动作这个动作被真实执行执行完的真实结果再被拿回来塞进模型的上下文里让它根据结果继续推理。这个过程循环往复直到模型觉得信息够了才输出最终答案。大模型从背答案变成了查证据、再下结论这就是 ReAct 大模型应用里最核心的价值。1.2 ReAct 如何用一段格式化文本指挥工具ReAct 这个名字本身是 Reasoning Acting 的组合论文里给出的典型运行轨迹长这样Thought: 我需要知道北京的当前天气先调用天气工具 Action: get_weather[Beijing] Observation: 20°C小雨东南风3级 Thought: 温度不高且有雨建议提醒用户带伞 Final Answer: 建议带伞今天北京20°C有小雨。注意一个关键点在最朴素的 ReAct 实现里模型并不会真的去执行get_weather。它只是在输出一段符合约定的文本你的执行循环负责解析这段文本找到Action和Action Input执行对应的函数再把真实的返回结果作为新的Observation填入下一轮 Prompt。这套格式化协议就是 ReAct 的工程接口。现在很多大模型已经原生支持结构化的函数调用能力模型可以直接输出一个tool_calls字段里面包含工具名和参数原来的文本解析工作被大大简化了。LangGraph 的做法也是建立在后者之上的它不让你手写正则去解析 Action 文本而是帮你把协议、执行、回填这些事情全部封装好。你可以把 ReAct 理解为一种思考模式LangGraph 则是把这种模式工程化的容器。下面的表格能帮你快速区分几个常见概念方案是否调用外部工具是否有循环典型短板单次 Prompt否否只适合简单生成容易一本正经地胡说普通 Chain否否线性流程无法根据中间结果修正方向ReAct Agent是是需要设计好循环不然容易失控2. 我为什么推荐用 LangGraph 实现 ReAct状态、节点与图2.1 LangGraph 解决了 AgentExecutor 时代的什么问题如果你以前接触过 LangChain 的 AgentExecutor应该有这种感受它像一个魔法黑盒。你丢给它工具列表和 Prompt它自己内部去循环调用模型和工具最后给你一个结果。正常情况下是好用的但一旦出问题就非常难受——你不知道它内部循环到哪一步了不知道哪条消息让模型产生了误判也没法在工具执行之前插进一道人工审核。LangGraph 把这一切从隐式循环变成了显式图。循环不再藏在某个 Agent 类的内部实现里而是由你自己用节点和边拼出来一个节点就是一个函数一条边就是一次状态转移。如果你会画流程图你就已经会 LangGraph 的基本设计了。这个改变对小白尤其友好因为调试一个图比调试一堆隐式循环要直观得多。2.2 State、Node、Edge、Conditional Edge一张图讲清核心概念我用一个生活化的类比来把这些概念一次说清。想象你走进一家流程化办事大厅State是挂在墙上的白板笔记。所有办事员共用这块白板每个人读到自己需要的信息然后把自己产出的新信息补上去。在 LangGraph 里State 通常是一个 TypedDict多个节点共享它。Node是一位办事员。它接收白板内容做自己的事情然后返回一段新内容更新白板。Node 本质就是普通 Python 函数。Edge是白板上画好的流程箭头标记办完这一步下一步找谁。Conditional Edge是站在分岔口的引导员。他会先看一眼白板上的结果再决定送到哪个窗口。ReAct 循环里到底继续调用工具还是直接输出最终答案这个判断就是由一个条件边完成的。之所以说这么多基础概念是因为很多新手一上来就急着写代码结果看到StateGraph.add_conditional_edges就懵了。你先知道它只是一个带判断的路由后面代码就很好读。2.3 Checkpointer 和可观测性带来的额外收益LangGraph 还有一个杀手锏叫 Checkpointer通俗说就是给白板定期拍照存档。有了存档你可以随时恢复某一轮的状态、回溯每一步的消息记录甚至把流程停在一个节点前等人来确认。这个能力在企业落地场景里非常关键后面第 6 部分我会演示。对于学习阶段它还有一个附加好处你把每一轮 State 打印出来相当于看到一个 ReAct 代理的完整大脑活动记录比任何日志都直观。3. 动手第一步搭建最小 ReAct 代理代码骨架3.1 环境准备与依赖版本先准备一个干净的 Python 环境建议用 3.10 或更高版本。我个人习惯用 venv 隔离项目避免把依赖装乱。python -m venv .venv source .venv/bin/activate # Windows 下用 .venv\Scripts\activate pip install -U langgraph langchain-core langchain-openai装完之后建议确认一下版本因为 LangGraph 的 API 在 0.2 之后有一些变化。如果你在网上看到比较早的教程里面的写法可能跟我这里不太一样问题多半出在版本上。LangGraph 本身不绑定某个大模型服务商。模型封装用哪一个 SDK 都可以下面示例用langchain-openai里的 ChatOpenAI 类是因为它能对接很多兼容 chat/completions 协议的服务端你把base_url换成自己的模型服务端点就能用。如果你用的是其他模型同样没问题换成对应的 LangChain 模型封装类就行。export API_KEY你的密钥 export API_BASE你的服务端点3.2 定义 AgentState 和模型节点这里的核心代码非常短先定义一个 AgentState它只有一个字段messages用来保存整段对话消息。关键是第二行Annotated[list, add_messages]意思是新消息不是覆盖旧消息而是自动追加到历史里。没有这个 reducer多轮对话的历史就维持不住。from typing import Annotated, TypedDict from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langchain_core.messages import HumanMessage from langchain_openai import ChatOpenAI class AgentState(TypedDict): messages: Annotated[list, add_messages] llm ChatOpenAI( modelyour-model-name, api_keyyour-key, base_urlyour-endpoint/v1, temperature0, ) def agent(state: AgentState): response llm.invoke(state[messages]) # add_messages 会自动追加所以这里只需要返回新消息 return {messages: [response]} graph StateGraph(AgentState) graph.add_node(agent, agent) graph.add_edge(START, agent) graph.add_edge(agent, END) compiled_graph graph.compile()你可以先跑一下这个最小图输入一句你好。它现在只会调用模型生成一段回复没有任何工具的参与。这个骨架存在的意义是让你确认State、Node、Graph 这三样基本组件是怎么组合起来的。3.3 绑定工具让模型具备“产生动作”的能力要让模型在回答过程中提出我想调用某个工具就得把工具的信息告诉模型。这个动作叫bind_tools。它做的事不是简单往 Prompt 里塞一句你可以调用工具而是把工具的函数签名、参数说明、docstring 全部转成结构化 Schema 送给模型。tool def get_current_time() - str: 获取当前系统时间适合回答“现在几点”这类问题。 from datetime import datetime return datetime.now().strftime(%Y-%m-%d %H:%M:%S) llm_with_tools llm.bind_tools([get_current_time])你拿这个llm_with_tools去调用一句现在几点观察返回对象里的tool_calls字段就会看到类似这样的结构它列出了工具名、参数、调用 ID但模型本身并不会真的去执行工具。执行工具是执行引擎的事这个引擎稍后会用ToolNode来做。理解模型只生成调用意图引擎去执行这一层是理解整套 ReAct 循环的关键。4. 把骨架补全为真正的 ReAct 循环工具节点和条件路由4.1 定义两个实用工具并讲清 docstring 的秘密为了让示例既干净又有点实用价值我用两个工具一个get_current_time一个calculate。它们都不依赖外部网络适合做本地实验。import ast import operator from datetime import datetime from langchain_core.tools import tool tool def get_current_time() - str: 获取当前系统时间适合回答“现在几点”这类问题。 return datetime.now().strftime(%Y-%m-%d %H:%M:%S) tool def calculate(expression: str) - str: 计算简单的数学表达式输入需要是合法的 Python 算术表达式例如 (35)*2/4。 # 教学演示场景下做一层轻量安全校验 allowed_nodes ast.Expression, ast.BinOp, ast.UnaryOp, ast.Constant, ast.Add, ast.Sub, ast.Mult, ast.Div try: tree ast.parse(expression, modeeval) for node in ast.walk(tree): if not isinstance(node, allowed_nodes): raise ValueError(仅支持基础算术运算) result eval(compile(tree, string, eval), {__builtins__: {}}, {}) return str(result) except Exception as e: return f计算失败{e}这里想强调一件事工具的 docstring 是给模型看的说明书不是给人看的注释。模型靠它判断这个工具什么时候该用、参数怎么填。写得越具体模型猜错的概率越低。我在实际项目里见过太多工具 docstring 写得很敷衍结果模型反复调错参数最后还被误认为是模型能力不行。4.2 构建 agent → tools → agent 回环现在把三个关键部分组装在一起一个模型节点负责思考一个工具节点负责动手一条条件边负责判断下一步。from langgraph.prebuilt import ToolNode tools [get_current_time, calculate] tool_node ToolNode(tools) llm_with_tools llm.bind_tools(tools) def agent(state: AgentState): return {messages: [llm_with_tools.invoke(state[messages])]} def route_after_agent(state: AgentState): last_message state[messages][-1] # 如果模型最后一条消息里带着工具调用意图就转去执行工具 if getattr(last_message, tool_calls, None): return tools return END graph StateGraph(AgentState) graph.add_node(agent, agent) graph.add_node(tools, tool_node) graph.add_edge(START, agent) graph.add_conditional_edges( agent, route_after_agent, {tools: tools, END: END}, ) graph.add_edge(tools, agent) compiled_graph graph.compile()逐条理解这几条边START - agent流程从模型节点开始。agent - tools/cool如果模型生成了tool_calls说明它想调用工具此时路由到tools节点。tools - agentToolNode 执行完工具后会把ToolMessage追加到 State 里再回到agent节点。模型这时能读到工具的返回结果于是继续推理要么再产生新的工具调用要么直接输出最终答案。agent - END模型没有产生工具调用说明它认为信息足够流程结束。这就是完整的 ReAct 闭环。agent和tools两个节点之间来回跳外层看起来就是一个循环但这个循环是能被你观察、控制和修改的。4.3 实测盯着每一步消息看 ReAct 是怎么循环的跑起来看结果。我建议你第一次执行时不要只关心最终回答而是把整段messages打出来看看消息序列长什么样。response compiled_graph.invoke( {messages: [HumanMessage(现在几点两个小时后是几点)]}, config{recursion_limit: 20}, ) for m in response[messages]: tc getattr(m, tool_calls, None) print( 消息类型:, m.type) print(内容:, m.content) print(工具调用:, tc) print()正常情况下你会看到这样的轨迹AI 消息里带着一个tool_calls内容是调用get_current_time接着是ToolMessage内容是工具返回的真实时间然后是 AI 基于这个时间继续计算、推理可能再调用一次calculate或者直接给出最终文本。这个真实的执行轨迹比任何原理图都直观我强烈建议你花五分钟盯着它看一遍。5. 小白最容易踩的五个坑从循环失控到状态污染理论讲完代码也跑通了接下来进入我最想写的一部分。以下五个问题是我在实际使用时踩过的也是群里新手问得最多的。5.1 循环失控RecursionLimit exceeded现象程序运行到一半抛出一个类似RecursionLimit reached的异常告诉你超过了recursion_limit。原因模型在 agent 和 tools 之间反复横跳停不下来。最常见的是两种场景一种是模型一直调用同一个工具比如反复获取当前时间每次都因为某种原因觉得信息还不够另一种是你的条件边写错了或者工具永远返回一个让模型想继续查的结果。排查思路把上面那段打印 messages 的代码改成可以逐步看最后几步尤其留意最后一个 AI 消息还在不在产生tool_calls。如果是模型自己陷入循环一个土办法是给循环加个安全阀在 agent 节点里统计历史中已经出现过的工具调用次数超过阈值就强制让模型收尾。from langchain_core.messages import SystemMessage MAX_TOOL_CALLS 5 def agent_with_safety(state: AgentState): history state[messages] tool_calls_count sum( 1 for m in history if getattr(m, tool_calls, None) ) if tool_calls_count MAX_TOOL_CALLS: forced_message llm.invoke( history [SystemMessage(请不要再调用工具了直接基于已有信息输出最终结论。)] ) return {messages: [forced_message]} return {messages: [llm_with_tools.invoke(history)]}5.2 工具参数格式错误ToolNode 解析失败现象异常信息像Missing required argument或KeyError经常出现在多参数工具上。原因模型输出的tool_calls.args里缺了某个必填参数。有时候是模型理解错了有时候是你的工具函数签名没写清楚比如参数名很别扭、docstring 没说参数含义。解决三个层面。第一给可能有歧义的参数加默认值第二在 docstring 里写清楚参数它应该长什么样第三如果工具很复杂用 Pydantic 的 Field 给参数写详细描述。我自己最常用的是第二层效果立竿见影。同一个工具docstring 里写一句expression 应是一个合法的 Python 算术表达式例如 (35)*2模型犯错的概率会低一个数量级。5.3 模型不调用工具你看到的只是纯文本现象模型输出了一长串我现在需要调用get_current_time工具来获取时间……之类的话但tool_calls字段是空的条件边直接走到了 END。原因这是两种 ReAct 实现范式被搞混导致的。ReAct 论文原本的思路是让模型输出Thought/Action/Action Input这种固定格式文本再用解析器去提取动作。而现代大模型的原生function calling思路是模型直接输出结构化的tool_calls。如果你自己在 Prompt 里手工写了一套 ReAct 格式文本又同时用了bind_tools模型很可能被带偏选择用文本回答而不是结构化的工具调用。解决既然用了 LangGraph就统一走bind_tools这条路。不要在系统 Prompt 里强行要求模型输出Action:前缀除非你打算自己写解析器。另外把 temperature 设成 0减少模型发挥空间。还有一个调试技巧把llm_with_tools.invoke(state[messages])的原始返回打出来看看里面到底有没有tool_calls。每次都靠猜是低效的。5.4 工具内部异常把错误吞掉还是让图崩掉现象工具函数本身抛了一个异常结果整个图立刻失败。比如calculate里遇到不合法的表达式抛了ValueError。原因ToolNode 执行的就是你的 Python 函数函数抛错就相当于这一节点抛错图会直接终止而不是把错误信息交给模型处理。解决工具内部的异常最好在工具函数内部就接住把错误描述作为字符串返回。这样模型在下一次思考时能看到计算失败xxx它可能会换一种写法再试一次或者直接告诉用户这个表达式有问题。这比让整个流程崩溃要优雅得多。如果你希望某个工具失败是不可恢复的那也应该返回一个专门的失败标记而不是直接抛异常。5.5 add_messages 的“追加陷阱”与 ToolMessage 错位现象消息历史变得越来越长还出现大量重复或者看到类似Can not add ToolMessage without a valid tool_call_id的报错。原因这是对Annotated[list, add_messages]理解不够造成的。add_messages会自动把返回的消息追加到 State 的现有消息列表后面所以你根本不需要在返回时手动把整段历史拼上。很多人习惯写成return {messages: state[messages] [response]}这就等于把旧消息又追加了一遍两条重复消息同时出现在新状态里。正确写法是只返回新增的那条return {messages: [response]}至于ToolMessage错位是因为每条工具消息必须带一个tool_call_id用来关联它是对应哪一次 AI 工具调用。你手工构造ToolMessage时很容易漏掉或者写错这个 ID。用 LangGraph 自带的ToolNode执行工具能自动处理好这些关联所以我的建议是能用 ToolNode 就不要手写工具执行的链路除非你很清楚自己在做什么。6. 进阶改造多轮记忆、人工确认和一点架构思考6.1 用 Checkpointer 实现跨会话上下文到目前为止每次调用invoke都是从一个全新状态开始的。上一轮你说的我叫小明下一轮模型早就忘了。要让它记住需要给图挂上 Checkpointer。from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() app_with_memory graph.compile(checkpointermemory) session_config {configurable: {thread_id: conversation-001}} app_with_memory.invoke( {messages: [HumanMessage(你好我叫小明)]}, session_config, ) response app_with_memory.invoke( {messages: [HumanMessage(我叫什么名字)]}, session_config, ) print(response[messages][-1].content)这里的关键是thread_id。它相当于一个会话编号同一个编号的多次调用会在同一个 State 上继续追加所以它能记住上下文。要注意的是这种记忆本质上还是把历史消息继续喂给模型并不是模型被你微调过了。从工程上看它解决的是状态持久化问题而不是模型学习问题。不同会话之间互不干扰这也是企业落地 Agent 时要搞清楚的边界。6.2 用 interrupt_before 实现人工审核环节有些操作不能交出去就直接执行比如写数据库、发邮件、转账、删除资源。LangGraph 允许你在工具节点执行前主动中断流程等待人类确认。app_with_audit graph.compile( checkpointermemory, interrupt_before[tools], ) result app_with_audit.invoke( {messages: [HumanMessage(请帮我计算一下这笔订单的税额)]}, session_config, ) # 此时流程会停在 tools 节点前查看状态 state app_with_audit.get_state(session_config) print(当前是否中断:, state.next) # 人工审核通过后恢复执行。较新版本用 Command旧版本可以直接传 None from langgraph.types import Command resumed app_with_audit.invoke(Command(resumeTrue), session_config)这个能力在真实系统里非常有用。ReAct 代理可以自由地思考和规划但真正有副作用的动作必须过我这一关。我的建议是一开始做原型时就可以把interrupt_before[tools]挂上让每一步都先人审一遍观察模型有没有频繁做错误决策确认模型足够稳了再考虑基于工具类型做分级控制比如只拦截高风险的写操作。6.3 从单个 ReAct 代理到多节点协作的扩展思路当你只有一个 ReAct 循环、手里只有两三个工具时一切很美好。但工作里往往会发现工具越来越多十个、二十个类型五花八门。模型在每个节点面对这么多工具时会出现选择困难误调用的概率随之上升。这时候不要再往同一个 agent 节点里堆工具了更常见的做法是拆成多个子代理每个子代理负责一个领域只绑定自己领域的工具外部再放一个协调节点由它决定任务该派给谁等子代理返回结果后再汇总决策。这种多图嵌套结构在 LangGraph 里通过子图来实现本质上就是把一个已经编译好的图当成大图中的一个节点来用。我不建议你第一天就学这个先把单循环跑熟有需要再扩展。架构这东西提前一步是优雅提前十步就是灾难了。7. 我的几点实操体会先把图跑通再谈优化最后分享几条我的真实经验不灌鸡汤。第一调试永远优先于炫技。我第一次写 ReAct 代理时一个tool_calls判断写反了模型每次都会先去调用工具再回来聊天当时完全没意识到。后来把每一步state[messages]打出来才发现问题在路由函数上。所以遇到任何诡异行为第一反应应该是打印整段消息序列而不是修改 Prompt 碰运气。第二图优于魔法。很多同学喜欢直接拉一个高级 Agent 库几行代码就搞定了。一旦出了问题他们连排查入口都找不到只能重启服务碰运气。亲手把三节点图写一遍之后你再看任何 Agent 框架的文档都会觉得顺畅很多因为你知道底层无非就是模型节点、工具节点、路由判断这三件事。第三不要一上来就追求完美。先两个工具跑通再慢慢加记忆、加人工确认、加流式输出。我记得自己在第一次跑通完整闭环那天反复调了十几条消息看着模型一步步调用工具、读取结果、修正结论才真正理解了 ReAct 大模型时代的智能是怎么被工程组装出来的。这套思路并不复杂复杂的是每一个细节是否都被你亲手验证过。最后再分享一个小技巧如果你要做一个长期在用的代理记得给每个会话分配有意义的thread_id而不是随机字符串。排查问题时能顺着会话 ID 把用户从第一句到最后一句的完整 State 拉出来看一眼比什么日志都好使。调试 ReAct 循环最怕的就是状态被冲掉、历史对不上号而 LangGraph 的设计刚好把这个问题留了一个优雅的口子。