☰
LangGraph从零搭建ReAct代理:状态、节点与工具调用实践
2026/10/22 21:43:28 网站建设 项目流程

写这篇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( model="your-model-name", api_key="your-key", base_url="your-endpoint/v1", temperature=0, ) 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 算术表达式,例如 '(3+5)*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, mode="eval") 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 -> agent:ToolNode 执行完工具后,会把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 算术表达式,例如 '(3+5)*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(checkpointer=memory) 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( checkpointer=memory, 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(resume=True), 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 的设计刚好把这个问题留了一个优雅的口子。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询