LangGraph 入门：从 State、Node、Edge 到工具调用实战

1. 从零认识 LangGraph：它到底解决什么问题

如果你已经用过 LangChain，大概率经历过这样的场景：想让大模型按固定流程走几步，中间还要根据结果决定下一步做什么，用 Chain 拼来拼去，代码很快就变成了一团乱麻。链式调用是线性的，A 完了必须 B，B 完了必须 C，一旦需要"如果 B 的结果是 X 就走 D，否则回到 A 重试"，整个逻辑就开始失控。LangGraph 就是冲着这个痛点来的。

LangGraph 是 LangChain 生态里专门用来构建有状态、可循环的智能体工作流的框架。它的核心思路非常朴素：把整个流程建模成一张图（Graph），图里有节点（Node）和边（Edge），状态（State）在节点之间流动。你可以把它理解成一张流程图，但这张图是"活"的——它能根据当前状态动态决定下一步走向哪，也能循环回到之前的节点，直到满足退出条件。

这套模型特别适合几类场景：多轮对话管理、需要调用工具的 Agent、带条件分支的审批流、需要人工介入（Human-in-the-loop）的流程、以及任何"走一步看一步"的复杂任务。说白了，只要你的业务逻辑用 if-else 和循环能描述清楚，LangGraph 基本都能优雅地表达出来。

这篇文章我打算从最基础的概念讲起，把 StateGraph、State、Node、Edge 这几个核心概念掰开揉碎，然后带你从环境准备一路走到能跑通一个带条件分支和循环的最小示例。中间会穿插我自己踩过的坑，比如状态字段怎么设计、条件边为什么没触发、循环怎么防止死循环等等。不管你是刚接触 LangGraph 的新手，还是已经用过但总觉得没吃透的开发者，应该都能从里面捞到点东西。

提示：本文假设你有基本的 Python 基础，了解 LangChain 的基本用法会更好理解，但不是硬性要求。如果你完全没接触过 LangChain，也能跟着走，我会把必要的前置知识补上。

2. 核心概念拆解：State、Node、Edge 到底是什么

在动手写代码之前，必须先把这几个概念搞清楚。很多人学 LangGraph 卡壳，不是因为 API 难，而是因为脑子里没有建立起"图"的心智模型。我用一个生活化的类比帮你建立直觉：把整个 LangGraph 想象成一家餐厅的后厨流水线。

2.1 State：流动的"订单信息"

State 是整个图的共享数据，它在所有节点之间传递。你可以把它理解成一张订单小票，上面记录着当前这桌菜的所有信息：点了什么、做到哪一步了、有没有特殊要求。每个节点都能读到这张小票，也能往上面写新的信息。

在 LangGraph 里，State 通常用一个TypedDict或者 Pydantic 模型来定义。比如一个最简单的对话场景：

from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class State(TypedDict): messages: Annotated[list, add_messages] user_intent: str retry_count: int

这里有个关键点必须讲清楚：Annotated[list, add_messages]里的add_messages是一个reducer（归约函数）。它的作用是定义"当多个节点都想更新这个字段时，怎么合并"。默认情况下，新值会直接覆盖旧值；但用了add_messages之后，新消息会追加到列表里，而不是覆盖。这个设计非常关键，因为对话历史就是不断追加的，如果每次都被覆盖，历史就丢了。

我见过太多新手在这里翻车：定义了一个messages字段，节点里返回新消息，结果发现历史消息全没了。原因就是没加 reducer，默认覆盖了。所以记住一句话：需要累积的字段，一定要配 reducer。

2.2 Node：干活的"工位"

Node 就是图里的一个处理单元，本质上是一个函数。它接收当前的 State，做一些处理，然后返回一个字典，表示要更新 State 的哪些字段。每个 Node 就像后厨的一个工位：切菜的、炒菜的、装盘的，各司其职。

一个典型的 Node 长这样：

def call_model(state: State): response = llm.invoke(state["messages"]) return {"messages": [response]}

注意返回值是一个字典，键对应 State 里的字段名。LangGraph 会拿这个字典去更新全局 State。如果字段有 reducer，就按 reducer 的规则合并；没有的话就覆盖。

Node 的命名也有讲究。我建议用动词或者动宾短语，比如call_model、execute_tool、check_result，这样看图的时候一眼就知道每个节点在干嘛。别用node1、node2这种，图一复杂你自己都记不住。

2.3 Edge：决定"下一步去哪"

Edge 定义了节点之间的连接关系，也就是"这个节点干完活之后，接下来走哪个节点"。LangGraph 里有三种边：

• 普通边（Normal Edge）：从 A 直接到 B，无条件。用add_edge("A", "B")添加。
• 条件边（Conditional Edge）：从 A 出发，根据一个函数的返回值决定去 B 还是 C。用add_conditional_edges添加。
• 入口和出口：START和END是两个特殊节点，分别表示图的起点和终点。

条件边是 LangGraph 的灵魂所在。它让图有了"判断能力"。比如一个 Agent 循环：模型输出后，判断是要调用工具还是直接回复用户。如果是调用工具，就走到工具节点；如果是回复，就走到 END。这个判断逻辑就写在条件边的路由函数里。

def should_continue(state: State): last_message = state["messages"][-1] if last_message.tool_calls: return "tools" return END graph.add_conditional_edges("agent", should_continue, { "tools": "tools", END: END })

路由函数返回一个字符串，这个字符串会去映射表里找对应的目标节点。映射表的作用是让路由函数的返回值和实际节点名解耦，路由函数可以返回语义化的标签，映射表负责翻译成真实节点名。

2.4 三者的关系：一张图说清楚

把这三个概念串起来：State 是数据，Node 是计算，Edge 是控制流。数据在节点之间流动，每经过一个节点就被处理一次，边决定流向。整个图从 START 开始，沿着边一路走到 END，中间可能循环、可能分支，但最终一定会终止（前提是你设计得当）。

理解了这个模型，再看 LangGraph 的 API 就会觉得顺理成章。它没有魔法，就是把"状态机"这个古老的概念用现代的方式重新包装了一遍。状态机你肯定不陌生，LangGraph 无非是让状态机的定义更声明式、更容易和 LLM 结合而已。

3. 环境准备与依赖安装：别在第一步就卡住

概念清楚了，接下来动手。环境这块看似简单，但热词里"node安装"、"nvm管理node版本"、"node安装后npm不能用"这些搜索量很高，说明不少人在环境上栽过跟头。虽然 LangGraph 是 Python 生态的东西，但实际项目里经常和前端、Node 工具链混在一起，所以我把 Python 和 Node 两块都讲一下。

3.1 Python 环境：用虚拟环境隔离

LangGraph 对 Python 版本的要求是 3.9 及以上，我建议直接用 3.11 或 3.12，兼容性和性能都更好。第一步永远是建虚拟环境，别往全局环境里装：

python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate

建好之后装核心依赖：

pip install langgraph langchain langchain-openai

如果你用的是其他模型提供商，把langchain-openai换成对应的包就行，比如langchain-anthropic。装完之后验证一下：

python -c "import langgraph; print(langgraph.__version__)"

能打印出版本号就说明装好了。这里有个小坑：LangGraph 的版本迭代很快，API 偶尔会有变动。如果你照着老教程写代码报错，先检查版本。我建议在项目里固定版本，比如langgraph==0.2.x，避免某天自动升级后代码跑不起来。

3.2 Node 环境：什么时候需要，怎么装

LangGraph 本身不需要 Node，但如果你要做以下事情，Node 就跑不掉：用 LangGraph 的 JS/TS 版本、跑 LangGraph Studio 的本地开发界面、或者项目里有前端部分。热词里"nvm安装及全局配置node"、"升级node"、"node高版本兼容低版本吗"这些问题，本质都是版本管理没做好。

我的建议是用 nvm（Node Version Manager）来管理 Node 版本，别直接装一个全局的 Node。原因很简单：不同项目依赖不同 Node 版本，全局装一个迟早冲突。nvm 可以让你在多个版本之间秒切。

Linux/macOS 装 nvm：

curl -o- https://raw.githubusercontent.com/nvm-sh/nvm/v0.39.7/install.sh | bash

装完重开终端，然后：

nvm install 20 nvm use 20 nvm alias default 20

Windows 用户用 nvm-windows，去 GitHub 下载安装包即可。装完node -v和npm -v都能正常输出就 OK。

关于"node高版本兼容低版本吗"这个问题，答案是：大部分情况兼容，但不是绝对。Node 的 API 总体向后兼容做得不错，但有些废弃的 API 在高版本会被移除，有些原生模块需要重新编译。所以升级 Node 版本前，先在测试环境跑一遍，别直接在生产上升。

3.3 常见环境问题速查

问题现象	可能原因	解决办法
npm: command not found	Node 装了但 npm 没配好	检查 PATH，重装 Node 或用 nvm 重装
pip install报权限错误	用了全局 Python	建虚拟环境，别用 sudo pip
导入 langgraph 报 ModuleNotFoundError	装到了别的环境	确认which python指向虚拟环境
Node 版本切换后命令失效	nvm 没设默认版本	nvm alias default <version>
离线环境装不了包	没网	提前下载 whl 文件，用pip install xxx.whl

注意：如果你在服务器上跑 LangGraph 服务，用 SSH 连接时要注意——断开 SSH 后进程可能会被挂断。用nohup或者tmux/screen把进程托管起来，别直接前台跑。

4. 第一个 LangGraph 程序：从 Hello World 到能跑通

环境好了，写代码。我习惯从最小的例子开始，先让图跑起来，再逐步加复杂度。这样出问题容易定位。

4.1 最小可运行示例

先来一个最简单的：两个节点，线性连接，从 START 到 END。

from typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): text: str def node_a(state: State): print("节点 A 收到:", state["text"]) return {"text": state["text"] + " -> A"} def node_b(state: State): print("节点 B 收到:", state["text"]) return {"text": state["text"] + " -> B"} builder = StateGraph(State) builder.add_node("a", node_a) builder.add_node("b", node_b) builder.add_edge(START, "a") builder.add_edge("a", "b") builder.add_edge("b", END) graph = builder.compile() result = graph.invoke({"text": "start"}) print("最终结果:", result)

跑一下，输出应该是：

节点 A 收到: start 节点 B 收到: start -> A 最终结果: {'text': 'start -> A -> B'}

这个例子虽然简单，但把 LangGraph 的核心流程全走了一遍：定义 State、定义 Node、建图、加边、编译、调用。compile()这一步很关键，它会把图"冻结"成一个可执行对象，之后每次invoke都是基于这个编译好的图跑。

4.2 加入条件分支

线性图没意思，加点判断。假设我们要根据文本长度决定走哪条路：

def router(state: State): if len(state["text"]) > 10: return "long" return "short" builder.add_conditional_edges("a", router, { "long": "b", "short": END })

这样，如果文本长度超过 10，就走 b 节点；否则直接结束。条件边的路由函数返回的字符串，通过映射表找到真实节点。这个设计的好处是路由逻辑和目标节点解耦，改路由不用动图结构。

4.3 加入循环：让图"转起来"

LangGraph 真正强大的地方是支持循环。比如一个重试逻辑：某个节点处理失败就回到自己重试，最多重试 3 次。

class State(TypedDict): value: int retry: int def process(state: State): new_value = state["value"] + 1 return {"value": new_value, "retry": state["retry"] + 1} def check(state: State): if state["value"] >= 3 or state["retry"] >= 5: return "done" return "retry" builder.add_node("process", process) builder.add_edge(START, "process") builder.add_conditional_edges("process", check, { "retry": "process", "done": END })

这个图会不断调用 process，直到 value 达到 3 或者重试次数达到 5。注意这里有个防死循环的设计：retry字段作为计数器，达到上限就强制退出。这是写循环图必须养成的习惯，否则一旦条件永远不满足，程序就卡死了。

提示：LangGraph 编译时可以设置recursion_limit，默认是 25。如果你的图循环次数可能超过这个值，记得调大，否则会抛GraphRecursionError。

5. 状态设计进阶：reducer 与多字段协同

前面提过 reducer 的重要性，这里展开讲。State 的设计直接决定了图好不好写，我见过太多项目因为 State 设计得烂，导致节点之间互相打架。

5.1 默认覆盖 vs reducer 追加

默认情况下，节点返回的字段值会覆盖State 里的旧值。这在某些场景是对的，比如一个status字段，你就是要它反映最新状态。但对于需要累积的字段，比如消息列表、日志、中间结果，就必须用 reducer。

LangGraph 内置了几个常用的 reducer：

• add_messages：专门处理消息列表，支持追加、去重、按 ID 更新。
• operator.add：通用的加法，列表拼接、数字相加都能用。
• 自定义 reducer：任何(old, new) -> merged签名的函数都行。

自定义 reducer 的例子：

def keep_max(old: int, new: int) -> int: return max(old, new) class State(TypedDict): score: Annotated[int, keep_max]

这样无论哪个节点返回多大的 score，State 里永远保留最大值。这个技巧在"取最优结果"的场景特别有用。

5.2 多字段协同的坑

一个常见的坑是：多个节点同时更新同一个字段，但你没意识到 reducer 的合并顺序。LangGraph 在并行分支合并时，reducer 的调用顺序是不确定的。如果你的 reducer 不满足交换律（比如字符串拼接），结果可能每次都不一样。

解决办法有两个：要么让 reducer 满足交换律（比如用集合去重），要么避免并行更新同一字段。我个人的经验是，State 字段尽量设计得"正交"，每个字段只被一类节点更新，减少冲突。

5.3 消息状态的正确姿势

对话类应用里，messages字段几乎是标配。用add_messages的时候要注意：它会给每条消息分配或读取 ID，如果两条消息 ID 相同，后者会更新前者而不是追加。这个特性在"编辑历史消息"的场景很有用，但如果你不小心复用了 ID，就会出现消息"消失"的诡异现象。

from langchain_core.messages import HumanMessage, AIMessage def chat_node(state: State): response = llm.invoke(state["messages"]) return {"messages": [response]}

这里返回的response是一条 AIMessage，add_messages会把它追加到列表末尾。下次调用时，state["messages"]就包含了完整历史。这就是多轮对话的基础。

6. 工具调用实战：让 Agent 真的"下地干活"

热词里"langgraph 工具调用"、"让 ai 真的下地干活"出现频率很高，说明大家最关心的还是怎么让 Agent 调用外部工具。这块我用一个完整例子讲透。

6.1 定义工具

先用 LangChain 的@tool装饰器定义几个工具：

from langchain_core.tools import tool @tool def get_weather(city: str) -> str: """查询指定城市的天气""" # 实际项目里这里调真实 API return f"{city}今天晴，25度" @tool def calculate(expression: str) -> str: """计算数学表达式""" return str(eval(expression)) tools = [get_weather, calculate]

工具的 docstring 很重要，模型就是靠它判断什么时候该调用哪个工具。写得越清楚，模型选得越准。

6.2 绑定工具到模型

from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-4o-mini") llm_with_tools = llm.bind_tools(tools)

bind_tools会把工具的 schema 告诉模型，模型在需要的时候就会返回tool_calls。

6.3 构建 Agent 图

from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langgraph.prebuilt import ToolNode class State(TypedDict): messages: Annotated[list, add_messages] def agent_node(state: State): response = llm_with_tools.invoke(state["messages"]) return {"messages": [response]} def should_continue(state: State): last = state["messages"][-1] if last.tool_calls: return "tools" return END builder = StateGraph(State) builder.add_node("agent", agent_node) builder.add_node("tools", ToolNode(tools)) builder.add_edge(START, "agent") builder.add_conditional_edges("agent", should_continue, { "tools": "tools", END: END }) builder.add_edge("tools", "agent") graph = builder.compile()

这个图的结构是：agent 节点调用模型，如果模型要调工具，就走到 tools 节点执行工具，执行完再回到 agent 节点，让模型基于工具结果继续推理。这个循环会一直转，直到模型不再调工具，直接给出最终回复。

ToolNode是 LangGraph 预置的节点，它会自动读取最后一条消息里的tool_calls，执行对应工具，把结果包装成 ToolMessage 返回。省去了手写工具执行逻辑的麻烦。

6.4 跑起来看看

result = graph.invoke({ "messages": [HumanMessage(content="北京天气怎么样？")] }) for msg in result["messages"]: print(type(msg).__name__, ":", msg.content)

你会看到消息序列大概是：HumanMessage -> AIMessage（带 tool_calls）-> ToolMessage（工具结果）-> AIMessage（最终回复）。这个流程就是标准的 ReAct 模式，LangGraph 把它用图的方式表达出来了。

注意：工具执行可能失败，比如 API 超时、参数错误。生产环境里一定要给工具加异常处理，返回友好的错误信息给模型，让模型有机会重试或者换方案。别让一个工具报错就把整个图搞崩。

7. 常见问题与排查技巧实录

这部分是我踩坑踩出来的经验，网上教程很少讲，但实际开发中天天遇到。

7.1 条件边不触发

最常见的原因是路由函数的返回值没在映射表里。比如路由函数返回"tool"，映射表里写的是"tools"，LangGraph 找不到对应节点，就会报错或者静默走默认分支。排查方法：在路由函数里加 print，确认返回值；再检查映射表的键是否完全匹配。

另一个原因是路由函数读的字段是空的。比如你想根据state["intent"]判断，但前面的节点根本没设置这个字段，读出来是 None，判断逻辑自然不对。这种问题用graph.stream()逐步调试最有效，能看到每一步 State 的变化。

7.2 循环停不下来

要么是退出条件写错了，要么是计数器没更新。我建议每个循环图都加一个"最大迭代次数"的兜底逻辑，不管业务条件是否满足，超过次数就强制走 END。这个习惯能救你无数次。

def check(state: State): if state["iterations"] >= MAX_ITER: return "force_end" if state["done"]: return "end" return "continue"

7.3 状态字段丢失

前面说过，多半是 reducer 的问题。还有一个隐蔽的原因：节点返回的字典键名和 State 字段名不一致。比如 State 里叫messages，节点返回{"message": [...]}，LangGraph 会忽略这个未知字段，你以为更新了其实没有。这种错误不报错，特别难查。养成习惯：节点返回的键名和 State 定义严格对齐。

7.4 问题速查表

现象	排查方向	解决
图跑完但结果不对	检查每个节点的返回值	用 stream 逐步看 State
报 GraphRecursionError	循环次数超限	调大 recursion_limit 或修退出条件
工具没被调用	模型没返回 tool_calls	检查工具描述和 bind_tools
消息历史丢失	reducer 没配	用 add_messages
并行分支结果不稳定	reducer 不满足交换律	改 reducer 或避免并行写同字段
编译报错	节点名重复或边指向不存在的节点	检查 add_node 和 add_edge 的参数

7.5 调试利器：stream 模式

graph.invoke()是一次性跑完，出问题不好定位。用graph.stream()可以逐步输出每个节点的执行结果：

for chunk in graph.stream({"messages": [HumanMessage(content="你好")]}): print(chunk)

每个 chunk 是一个字典，键是节点名，值是那个节点返回的更新。这样你能清楚看到数据在每一步是怎么变的，定位问题快很多。我调试复杂图的时候基本都用 stream，比打断点还方便。

8. 一些实战心得与扩展方向

写到这里，核心内容基本讲完了。最后分享几个我自己的体会，以及这个基础能往哪些方向扩展。

关于 State 设计，我的经验是宁可字段多一点，也别让节点之间靠隐式约定传递数据。比如两个节点需要共享一个中间结果，就老老实实在 State 里加个字段，别想着通过消息内容去解析。显式的字段让图更可读，也更好调试。

关于图的粒度，节点别切得太碎。我见过有人把每个小操作都做成一个节点，结果图里有几十个节点，看着就头大。我的建议是：一个节点对应一个"有意义的处理步骤"，比如"调用模型"、"执行工具"、"校验结果"，而不是"拼接字符串"、"打印日志"这种。粒度合适，图才清晰。

关于错误处理，LangGraph 支持给节点加重试策略，也可以在图层面做兜底。生产环境里，我一般会在关键节点外面包一层 try-except，把异常转成状态字段，让后续节点能感知到并做相应处理。这比让异常直接抛出中断整个图要健壮得多。

扩展方向上，LangGraph 还能做很多事：接入持久化（checkpointer）实现断点续跑和会话记忆、用interrupt实现人工审批、用子图（subgraph）拆分复杂逻辑、用SendAPI 做动态并行。这些都是在掌握了 State、Node、Edge 基础之后自然能上手的东西。等你把基础图玩熟了，再去啃这些高级特性，会顺很多。

我个人在实际项目里的体会是，LangGraph 最大的价值不是它提供了多少 API，而是它逼着你把业务流程想清楚。画图的过程就是梳理逻辑的过程，哪些步骤是顺序的、哪些是分支的、哪些要循环、状态怎么流转，想明白了再写代码，基本一次就能跑通。反过来，如果你连流程图都画不出来，那说明业务逻辑本身还没理清，这时候硬写代码只会越写越乱。所以我的建议是：动手之前，先在纸上把图画出