如果你跟我一样前几篇照着教程把 LangGraph 的 State 和 Graph 对象搭了起来多半会有同一个感觉图和状态都好理解真正卡住人的是 Node。状态只是一张共享的表格边只是路线图所有实际的活都要落到节点上。这次我就把 Node 节点好好拆开讲透重点放在功能节点、路由节点和工具节点这三类最常见的实现方式上。这篇内容适合正在写 Agent、准备接工具调用、或者已经用 LangChain 写了单链 chain 但被分支逻辑折磨过的开发者。看完你会发现LangGraph 所谓节点没有那么多神秘规则它就是一套传入 state、返回更新的函数契约剩下的全是职责划分和工程习惯问题。1. 为什么单独讲 NodeLangGraph 的执行单元到底怎么做职责划分先说一个容易误解的地方LangGraph 的节点并不是三种特殊的类它只有一种统一形式——一个可调用对象。功能节点、路由节点、工具节点这些说法是从实践角度按职责给人方便记忆的划分框架本身并不关心你写的这个节点到底算哪种。理解这一点后面看官方示例和别人项目时就不会晕。1.1 一个节点函数的契约用 add_node 注册进图时节点函数的基本契约是输入当前整个状态字典输出一个更新字典。from langgraph.graph import StateGraph, START, END def sample_node(state: dict) - dict: # state 是当前累积的状态 return {some_field: 新值}这里最容易忽略的细节是返回值不是新状态而是对状态的部分更新。LangGraph 会把返回的字典合并进原有状态没有返回的字段保持不变。如果你写的节点不需要改状态就直接返回 None 或者不写 return图也会正常往下走。我用一个生活化的比喻把状态想象成公司内部流转的工单每个节点是负责一个环节的工位。工位拿到工单把自己的处理结果填到对应栏位然后传给下一个工位。节点函数本身不知道下一个工位是谁路由和边才知道。1.2 三种节点的职责差异虽然框架不强制分类但实践中按下面这个表去规划代码会舒服很多节点种类核心动作典型输出类比功能节点执行业务逻辑更新状态字段修改后的字段字典生产工位把原材料加工成半成品路由节点判断当前走向决定下一个执行谁目标节点名或 Command 对象分流台按工单类型分到不同产线工具节点接收模型发出的工具请求执行真实函数工具执行结果消息外包车间真正干活的第三方为什么要单独把路由节点拎出来因为在最初写纯 LangChain chain 时分支逻辑通常埋在函数内部的一堆 if-else 里。逻辑一多整个流程就成了一个黑盒。LangGraph 的价值正是把这些 if-else 从函数内部抽到图结构上让每一步跳转都看得见、可调试、可回放。路由节点就是这套思路的落地手段。2. 功能节点用 State 进出约定吃尽所有普通逻辑绝大多数节点其实都是功能节点。它们做的事情可以抽象成一句话读状态里的若干字段算出一个结果把结果写回状态。不要把它想复杂但有几个细节值得好好掌握。2.1 状态类型和节点返回值的配合先定义一个典型的状态类型。很多新手在最开始不写 TypeDict所有字段都用 dict 硬扛短期没问题节点一多就会因为字段名拼写错误踩坑。所以我建议从一开始就显式声明from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] user_intent: str order_id: str | None attempts: int这里 messages 字段用了 add_messages 这个 reducer。它的含义是每次节点返回 messages 时不是在替换原有列表而是在原有列表上追加新消息。如果你不写 Annotated 直接定义 list那每次返回都会把之前所有消息覆盖成新的一条。这是 LangGraph 里最经典的为什么我的对话历史只剩一条事故原因。有了这个状态类型功能节点写起来非常直观def analyze_intent(state: AgentState) - dict: last_message state[messages][-1] text last_message.content if hasattr(last_message, content) else str(last_message) text text.lower() if 退款 in text: return {user_intent: refund} if 物流 in text: return {user_intent: logistics} return {user_intent: chat, attempts: state.get(attempts, 0) 1}这个节点做的事情很简单读最后一条消息判断用户意图把结果写入 user_intent 字段。它没有修改原始 state 对象而是返回了一个部分更新的字典。LangGraph 内部会把这个字段合并进状态里下个节点拿到的 state 就带上 user_intent 了。2.2 用闭包和类给节点注入外部依赖实际项目里节点几乎都要访问数据库、调用外部 API、读写缓存。最差的写法是把连接对象塞进 state图会越来越臃肿checkpoint 序列化时还会报错。正确做法是依赖注入把外部依赖通过闭包或类的方式绑定到节点函数上。def build_fetch_order_repo(get_conn): def fetch_order(state: AgentState) - dict: order_id state.get(order_id) if not order_id: return {order_detail: None} conn get_conn() detail conn.fetch_one(SELECT * FROM orders WHERE id ?, order_id) return {order_detail: detail} return fetch_order graph.add_node(fetch_order, build_fetch_order_repo(get_conn))这样每个节点都能独立测试单元测试里传一个假 get_conn 进去不用拉起整个图。我第一次用这种方式重构时感受很明显——之前为了 mock 一个藏在图深处的数据库连接我得跑通全流程现在只需要测一个函数。不过闭包写法有个小问题节点函数的签名会变复杂。如果你的团队更习惯面向对象可以定义一个类把依赖放在init实现call方法效果一样。至于选哪种纯粹看项目风格。2.3 功能节点应该有多胖这是我在多个项目里被问最多的问题一个节点里写多少逻辑算合理我的经验是一个节点解决一个问题整个函数控制在 80 到 100 行以内。超过这个量就该考虑拆节点。但反过来也不要为了纯净把每个小操作都拆成节点。节点之间的切换意味着状态读写、边缘检查、甚至检查点操作拆得太碎性能开销大流程图也会变成蜘蛛网。拿客服系统举例分析意图是一个合理的节点粒度。把分词、关键词匹配、调用意图分类模型、落库、返回意图全部揉进一个节点里虽然也能跑但你想在中间加一个人工复审分支时就得改这个胖节点。如果当时把模型分类和落库拆成两个节点插分支就只是一条边的事。3. 路由节点从 if-else 地狱到图的自动分流路由节点是我个人认为 LangGraph 比纯 LangChain 最有体感优势的地方。在 chain 里要加分支你得写 conditions逻辑一多很难读。在 LangGraph 里分支本身就是图的一部分路由节点的返回值决定了下一步走向。3.1 条件边、路由节点与 Command先看最基础的条件边用法。假设前面 analyze_intent 已经把 user_intent 写入状态现在要根据这个字段走不同分支def route_by_intent(state: AgentState) - str: return state.get(user_intent, chat) graph.add_conditional_edges( analyze_intent, route_by_intent, { refund: refund_node, logistics: logistics_node, chat: chat_node, }, )route_by_intent 就是典型的路由函数。它读状态里的字段返回一个字符串第三参的数字映射表把这个字符串翻译成目标节点名。这种写法很直观但它有个限制路由函数只能决定走向不能在决定走向的同时顺手更新状态。这时候就该用 Command 了。Command 是 LangGraph 提供的特殊返回对象可以在一个路由节点里同时完成改状态和指路两件事from langgraph.types import Command def smart_route(state: AgentState) - Command[str]: if state.get(must_human): return Command( gotohuman_review_node, update{handoff: True, user_intent: human}, ) if state.get(user_intent) refund: return Command(gotorefund_tool_node, update{attempts: state.get(attempts, 0) 1}) return Command(gotochat_node)用 Command 之后你就不需要先写一个路由节点返回特定字段、再用条件边读字段两步并一步图少跳一次逻辑也更聚合。这在多轮 Agent 里尤其好用因为路由节点经常要做一些记账工作。3.2 三种常见的路由判定逻辑路由节点的核心是怎么判断。我总结了三种最常见的模式按复杂度递增第一种是规则路由。基于关键词、字段值、权限标记等做确定性判断速度快、可解释性好。上面 route_by_intent 就是这种。适合业务规则明确的场景。第二种是 LLM 路由。用模型做意图分类或者内容判断适合规则模糊、长尾表达多的场景。关键在于让模型输出结构化结果而不是直接解析自然语言from typing import Literal from pydantic import BaseModel class Intent(BaseModel): intent: Literal[refund, logistics, chat] def llm_route(state: AgentState) - dict: intent llm.with_structured_output(Intent).invoke( [msg for msg in state[messages] if msg.type ! tool] ) return {user_intent: intent.intent}用一个只包含 single string 字段的 pydantic 模型比自己写提示词解析稳定得多。这也是我反复强调的习惯只要让模型决定事情就优先考虑结构化输出不要让它返回一大段话再去正则匹配。第三种是基于工具调用意图的路由。这个跟第四节的工具节点强相关。模型在回复中可能生成 tool_calls这时候路由节点要判断当前最后一条消息是否是带工具调用的 AIMessage如果是就走工具节点如果不是就走最终回复节点。这个模式是 ReAct Agent 的骨架也是后面要讲的工具节点配套部分。3.3 路由怎么写才能不无限循环无限循环是路由节点最容易犯的错。典型场景LLM 判断用户意图为 refund路由到 refund 处理节点处理节点发现信息不足又让 LLM 重新判断LLM 又判断成 refund于是原地打转。我实践下来有三个有效手段第一状态里加尝试次数上限。像我前面定义的 attempts 字段每次走进某个路由分支就加一路由函数里检查次数超过阈值就强制转到转人工节点。第二记录已经走过分支的标记。用 list 字段加上 reducer把每次路由结果追加进去下一次路由先看历史标记同一意图连续出现多次就不再重复跳转。第三调用器层面给 recursion_limit 兜底。如果不小心把图画出了环LangGraph 会在超过递归上限时抛出异常提醒你但别把这个当常规容错它只是最后的保护网。4. 工具节点把模型决定和代码执行正确缝合第四节是很多同学从 LangChain 单 chain 转到 LangGraph 后最困惑的地方。LangChain 里用 bind_tools 就把工具调用串起来了怎么到了 LangGraph 还要单独讲工具节点因为 LangGraph 的核心理念是显式流程——模型只负责决定调用什么工具、传什么参数真正执行工具这件事必须靠图里的节点完成。4.1 LLM 只负责决定不负责执行想象一个客服机器人接到查询订单的请求。模型在收到用户问题后会在回复中输出一个结构化的工具调用请求工具名 query_order参数 order_id12345。但模型自己并不会连数据库。它只是给了一个指令请执行这个函数。在 LangGraph 里这个指令是存储在消息对象里的 tool_calls 字段。接下来轮到工具节点干活解析这个指令找到对应的真实函数执行它把结果包装成一条 ToolMessage并追加回消息列表。这样模型在下一轮就能看到工具的执行结果并基于结果生成最终回复。4.2 用 ToolNode 接住 tool_calls最省事的实现是直接用 LangGraph 预构建的 ToolNode。假设你已经用 tool 装饰器定义了真实工具from langchain_core.tools import tool tool def query_order(order_id: str) - str: 查询订单当前状态返回 JSON 字符串。 # 这里写真实的数据查询逻辑 return {order_id: 12345, status: shipped} tool def refund_order(order_id: str, reason: str) - str: 发起订单退款。 return f退款申请已提交订单 {order_id}然后把这些工具交给 ToolNodefrom langgraph.prebuilt import ToolNode tools_node ToolNode([query_order, refund_order])在图中一切就变得很规则化LLM 节点处理完消息后条件边检查最后一条消息有没有 tool_calls有就 goto 到 tools_node然后 tools_node 自动执行对应工具把 ToolMessage 追加到 messages 里再回到 LLM 节点继续。整个过程可以无限循环直到模型不再生成工具调用。ToolNode 厉害的地方在于它把大量细节都封装好了按 tool_call_id 关联结果、处理并行多个工具请求、处理工具执行异常。对于大多数只需要模型调用工具的场景直接用它就够了。4.3 不经过 ToolNode 的手写工具调用流程但遇到需要定制错误处理、分批执行、按权限过滤工具的场景就得理解手写流程。工具节点本质上也是一个功能节点只不过它读取的是 tool_callsfrom langchain_core.messages import AIMessage, ToolMessage tool_map { query_order: query_order, refund_order: refund_order, } def run_tools_node(state: AgentState) - dict: last_message state[messages][-1] if not isinstance(last_message, AIMessage) or not last_message.tool_calls: return {messages: []} results [] for tc in last_message.tool_calls: tool_func tool_map.get(tc[name]) if tool_func is None: results.append(ToolMessage( contentf未找到工具: {tc[name]}, tool_call_idtc[id], nametc[name], )) continue try: output tool_func(**tc[args]) results.append(ToolMessage(contentstr(output), tool_call_idtc[id], nametc[name])) except Exception as exc: results.append(ToolMessage(contentf工具执行失败: {exc}, tool_call_idtc[id], nametc[name])) return {messages: results}这段代码的核心逻辑就三件事找到工具、执行工具、把结果作为 ToolMessage 追加回消息列表。其中 tool_call_id 与 AIMessage 里的 id 必须一一对应否则图里消息关联会错乱。自己写一遍的好处是你可以清楚地看到工具节点到底在做什么出问题时也能定位到具体是哪个环节。我建议每个人都手写一次工具节点再回到 ToolNode你会更信任预构建组件。4.4 多个工具是否要拆成多个工具节点所有工具塞进同一个 ToolNode 最简单但项目变大后我逐渐发现这不是好主意。想象一下机器人既有查询库存的只读工具又有直接删除订单的危险工具全放一个节点一旦路由失误模型可能直接执行危险操作。我的习惯是按权限等级或副作用强弱拆分工具节点。查询类工具一个节点写操作类工具一个节点不可逆操作单独一个节点路由节点根据意图把请求分流到不同的工具节点。结构上几乎不增加复杂度但安全性提升很多而且某个工具出错时影响范围也被隔离了。工具节点拆法适用场景优点所有工具合并个人 Demo、低风险工具代码最短逻辑最少按只读/写入拆客服、办公自动化写操作可单独加审核按业务域拆多部门系统每个节点对应独立服务职责清晰5. 三种节点混用的一个可运行 Agent 骨架前面讲了不少概念这里把它们组合成一个完整的可运行结构。以订单客服场景为例图结构是这样的入口节点 analyze_intent解析用户意图功能节点路由节点 llm_route生成工具调用决策或最终回复路径工具节点 run_tools_node执行工具并把结果回填收尾节点 final_response把工具结果整理成用户可读的回答from langgraph.graph import StateGraph, START, END def final_response(state: AgentState) - dict: # 这里根据最后一条消息生成最终回答交给模型或者模板处理 return {messages: [(assistant, 你的订单已发货预计明天送达。)]} def should_continue(state: AgentState) - str: last_message state[messages][-1] if isinstance(last_message, AIMessage) and last_message.tool_calls: return tools return final graph StateGraph(AgentState) graph.add_node(analyze_intent, analyze_intent) graph.add_node(llm_route, llm_route) graph.add_node(run_tools_node, run_tools_node) graph.add_node(final_response, final_response) graph.add_edge(START, analyze_intent) graph.add_edge(analyze_intent, llm_route) graph.add_conditional_edges(llm_route, should_continue, {tools: run_tools_node, final: final_response}) graph.add_edge(run_tools_node, llm_route) graph.add_edge(final_response, END) app graph.compile()这个结构最核心的地方是 run_tools_node 和 llm_route 之间形成的一个环模型决定调工具工具执行结果回去给模型模型再决定是否继续调工具或者收尾。只要 messages 字段正确使用 add_messages每一轮的工具结果都会保留模型有完整的上下文。5.1 状态里放什么最合理状态设计是整个 Agent 的另一个重点。我的原则是状态里只放所有节点都需要的共享信息和控制整个流程的开关字段。messages 是必须有的它是模型对话和工具调用的事实来源。user_intent 是路由的控制开关必须放。attempts 是防循环的计数器必须放。订单详情、业务结果这些放不放要看后续节点是否需要深度处理如果最后的 final_response 要根据订单详情生成回答那就要放如果只是把消息丢给模型生成回答那模型可以自己从 ToolMessage 里读不需要额外字段。有一个反面教训值得分享我之前在一个数据处理项目里试图把整个 Excel 文件对象塞进 state想着省得各节点重复读取。结果图一跑序列化直接出问题调试半天才发现是文件对象过大。正确的做法是在节点里重新读取文件句柄或者把文件路径和 sheet 名称放在状态里而不是数据本身。5.2 什么时候用预构建的 Agent 结构很多教程最后都会提到 LangGraph 预构建的 create_react_agent。它内部就是把模型节点 工具节点 路由边打包好了对标准 ReAct 场景来说是开箱即用的。那还要不要自己搭我的看法是如果只是让模型能调用几个工具直接用预构建没问题一旦你需要自定义路由逻辑比如按用户权限过滤工具、需要额外的控制字段比如防循环标记、需要插入人工审核节点手写骨架的优势立刻体现出来。理解了这个骨架你再看预构建节点的源码思路会非常清晰因为它们做的事情就是这节里展示的这些。6. 调试节点和路由时的常见崩坏现场最后分享几个我实际调试中踩过、也帮别人排查过的典型问题。每个都是真实报错现场路由节点和工具节点出问题的概率远高于预期。6.1 路由返回了不存在的节点名这是个非常经典的报错ValueError提示 node xxx is not present in the graph。多半是路由函数返回了某个字符串但条件边映射表里漏写了映射或者映射到的节点没有通过 add_node 注册。排查方法很简单先把图里的节点全部打印出来print(list(graph.get_graph().nodes.keys()))再检查路由函数所有可能的返回值和映射表的 key 对一遍。我后来养成了一个习惯把所有节点名在文件顶部用常量定义而不是散落在各处直接写字面量。这样一旦拼写错误IDE 和类型检查会第一时间发现。6.2 状态字段被并发写覆盖LangGraph 支持并行边多个节点可以同时处理同一个状态的不同分支。但如果两个并行节点同时返回同一个字段的更新最终状态只会保留其中一个结果。这是最后一个写入覆盖前面写入的经典问题。解决方法有三种一是把相关字段的 reducer 改成 operator.add 之类的合并策略二是把并发写统一收敛到一个节点里串行处理三是尽量避免在并行分支里写同一个业务字段。我实际项目中用得最多的是第二种因为并行节点本身数量通常不多收敛之后逻辑更可控。6.3 tool_call_id 配对失败手写工具节点时最常见的错误是构造 ToolMessage 忘了传 tool_call_id或者传了一个跟 AIMessage.tool_calls 对不上的 id。结果就是模型收到一堆工具结果却不知道它们对应哪次调用轻则回答混乱重则直接报错。排查套路是打印消息对象检查 AIMessage.tool_calls 里的 id 和 ToolMessage.tool_call_id 是否一一对应。用 ToolNode 时一般不会出这个问题因为它内部处理好了一旦你开始手写工具节点pair 逻辑必须自己保证。6.4 递归深度超限Agent 死循环的报错信息通常是 recursion limit 相关。这个错误最气人的地方在于它只是超时保护真正的问题是你图里的环没有正常退出条件。排查思路跟我前文讲防循环的三板斧一样先看路由函数有没有重复跳转再看状态里的 attempts 有没有递增最后临时把 recursion_limit 调大一点看看是不是真的还没跑到退出分支。注意不要依赖调大 recursion_limit 来修复问题它只是诊断工具。6.5 一次完整的快速定位链路分享一个最近的排错过程。现象是用户说我要退款图跑完却直接进入了闲聊回复没有调用退款工具。我用 stream 模式观察节点执行过程config {recursion_limit: 20} for event in app.stream({messages: [(user, 我要退款)]}, config): for node_name, node_output in event.items(): print(节点:, node_name, 输出:, node_output)很快就发现analyze_intent 的输出是正确的user_intent 确实被设置成了 refund但 llm_route 节点读到的消息列表里最后一条仍然是用户消息说明 should_continue 路由根本没有触发工具分支。再往下看发现 route_by_intent 的映射表里漏掉了 refund: refund_tool_node 这一行导致条件边返回的 refund 找不到目标。整个过程没超过十分钟。这正是把判断逻辑放到图上而不是塞进 if-else 的最大红利每个决策点都是图里的一个节点执行过程能被 stream 吐出每一层的输出问题定位就变成了对照输出找断点。最后再分享一个我个人的习惯每当我对一个复杂的 Agent 行为没把握时会先用最小图把它跑通再加分支。比如工具节点先只接一个工具跑通决定调用、执行、回填、总结的完整环路再加第二个工具。这种增量式构建方式比一次性把所有工具都接上再改 Bug 轻松得多。LangGraph 的节点编排本质上是一个工程问题理清职责、控制粒度、规划好状态字段比记任何 API 都重要。