你是个外卖站长手下有一群骑手。每位骑手就是一个节点Node他们需要依次或按规则完成任务。而你给骑手下达的“下一步去哪儿”的指令就是边Edge。搞懂了边你就掌握了 LangGraph 工作流的全部交通规则。本文不讲高深数学只用骑手送餐的故事把普通边、条件边、入口点、循环控制讲得明明白白每个概念都配有完整可运行的代码和逐行注释。一、先认识“状态State”——骑手们共享的小白板在 LangGraph 中所有骑手节点共用一个“小白板”叫做State。它就像一个订单信息栏谁取了餐、金额多少、送到哪一步了所有人都能看、能改。我们的所有代码都会使用下面这个简单的小白板from typing_extensions import TypedDict # 定义状态小白板 class GraphState(TypedDict): value: int # 比如订单金额、剩余距离 step: str # 当前步骤说明如“取餐中”“送餐中”为什么需要 State如果没有 State每个骑手干活时都不知道之前发生了什么。有了 StateA 骑手把金额加 1 元B 骑手就能看到新金额并继续处理。State 是图执行过程中传递数据的唯一通道。二、普通边Normal Edge——一条路走到黑最听话的骑手2.1 故事版普通边就是固定不变的单行道。站长说“小王你先去 A 店取餐然后必须去 B 店换电瓶最后去 C 小区送餐送完回家。”骑手不会问为什么也不会看路上堵不堵就按这个顺序执行。这就是普通边从 A → B → C → END结束。2.2 什么时候用普通边工作流是直线流水线数据清洗 → 特征提取 → 模型预测没有分支不需要决策你想强制按顺序执行2.3 代码实战完整 保姆级注释# 导入必要模块 from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END # ---------- 1. 定义小白板 ---------- class GraphState(TypedDict): value: int step: str # ---------- 2. 定义节点函数骑手---------- def node_a(state: GraphState) - dict: 骑手 A把金额加 1 元并更新步骤说明 print(f骑手 A 开始工作当前金额: {state[value]}) new_value state[value] 1 print(f骑手 A 完成金额变为: {new_value}) return {value: new_value, step: A 执行完毕} def node_b(state: GraphState) - dict: 骑手 B把金额翻倍 print(f骑手 B 开始工作当前金额: {state[value]}) new_value state[value] * 2 print(f骑手 B 完成金额变为: {new_value}) return {value: new_value, step: B 执行完毕} def node_c(state: GraphState) - dict: 骑手 C金额减 1 元比如优惠 print(f骑手 C 开始工作当前金额: {state[value]}) new_value state[value] - 1 print(f骑手 C 完成金额变为: {new_value}) return {value: new_value, step: C 执行完毕} # ---------- 3. 创建图构建器 ---------- # StateGraph 需要一个状态类型参数告诉框架小白板长什么样 builder StateGraph(GraphState) # ---------- 4. 添加节点把骑手注册到地图上---------- builder.add_node(node_a, node_a) # 第一个参数是节点名自己起第二个是实际函数 builder.add_node(node_b, node_b) builder.add_node(node_c, node_c) # ---------- 5. 添加普通边固定路线---------- # START 是 LangGraph 内置的特殊节点代表“用户输入进入的点” builder.add_edge(START, node_a) # 从 START 到 node_a builder.add_edge(node_a, node_b) # 从 node_a 到 node_b builder.add_edge(node_b, node_c) # 从 node_b 到 node_c builder.add_edge(node_c, END) # 从 node_c 到 END结束 # ---------- 6. 编译图 ---------- # 编译后得到一个可运行的“应用” graph builder.compile() # ---------- 7. 执行图 ---------- print( 普通边演示骑手 A → B → C ) # invoke 传入初始状态必须与 GraphState 定义匹配 result graph.invoke({value: 1, step: }) print(f最终结果: {result}\n)运行结果 普通边演示骑手 A → B → C 骑手 A 开始工作当前金额: 1 骑手 A 完成金额变为: 2 骑手 B 开始工作当前金额: 2 骑手 B 完成金额变为: 4 骑手 C 开始工作当前金额: 4 骑手 C 完成金额变为: 3 最终结果: {value: 3, step: C 执行完毕}关键点解析每个节点函数必须接收一个 state 参数并返回一个字典要更新的状态字段。不需要返回全部 State只返回改变的部分即可框架会自动合并。add_edge 的两个参数都是节点名包括 START 和 END 这两个特殊名字。三、条件边Conditional Edge——会看情况拐弯的聪明骑手3.1 故事版条件边就是骑手到达一个点后要看看小白板上当前的信息再决定下一步去哪。例如骑手 A 取完餐后看一眼订单金额如果是偶数走 B 路线送高档小区价格翻倍如果是奇数走 C 路线送普通小区价格减 1 元这个“看金额决定走哪条路”的函数叫做路由函数Routing Function。3.2 路由函数怎么写路由函数接收当前 state返回一个字符串目标节点名或者返回多个字符串并行执行后面会讲。LangGraph 会根据返回值去查找路由映射表决定实际跳转到哪个节点。3.3 完整代码从 A 分支到 B 或 Cfrom typing import Literal # 用于限制返回值只能是固定的几个字符串 from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END class GraphState(TypedDict): value: int step: str def node_a(state: GraphState) - dict: print(f骑手 A取餐当前金额 {state[value]}加 1 元) new_value state[value] 1 return {value: new_value, step: A 执行完毕} def node_b(state: GraphState) - dict: print(f骑手 B高档小区金额翻倍当前 {state[value]} - {state[value]*2}) return {value: state[value] * 2, step: B 执行完毕} def node_c(state: GraphState) - dict: print(f骑手 C普通小区减 1 元当前 {state[value]} - {state[value]-1}) return {value: state[value] - 1, step: C 执行完毕} # ---------- 路由函数 ---------- def route_condition(state: GraphState) - Literal[node_b, node_c]: 根据 state[value] 决定去 node_b 还是 node_c # 注意此时 state 是经过 node_a 更新后的值 current_val state[value] if current_val % 2 0: print(f路由决策{current_val} 是偶数去 node_b) return node_b else: print(f路由决策{current_val} 是奇数去 node_c) return node_c # 创建图 builder StateGraph(GraphState) builder.add_node(node_a, node_a) builder.add_node(node_b, node_b) builder.add_node(node_c, node_c) # 固定入口总是从 A 开始 builder.add_edge(START, node_a) # ---------- 添加条件边 ---------- # 从 node_a 出发使用 route_condition 函数决定下一步 builder.add_conditional_edges( node_a, # 源节点 route_condition, # 路由函数 { # 路由映射函数返回值 - 实际节点名 node_b: node_b, node_c: node_c } ) # B 和 C 结束后都结束 builder.add_edge(node_b, END) builder.add_edge(node_c, END) graph builder.compile() # 测试偶数情况初始 value2 print( 条件边测试初始偶数 value2 ) result1 graph.invoke({value: 2, step: }) print(f最终结果: {result1}\n) # 测试奇数情况初始 value1 print( 条件边测试初始奇数 value1 ) result2 graph.invoke({value: 1, step: }) print(f最终结果: {result2}\n)运行结果分析输入 value2A 执行后 value 3奇数 → 路由函数返回 node_c → 执行 C → value 变成 2 → 输出 {value: 2, step: C 执行完毕}输入 value1A 执行后 value 2偶数 → 路由函数返回 node_b → 执行 B → value 变成 4 → 输出 {value: 4, step: B 执行完毕}3.4 多个出口并行执行如果一个路由函数返回多个节点名比如返回一个列表 [node_b, node_c]那么这些节点会并行执行在下一个超级步骤中同时运行。这是 LangGraph 的一个重要特性可以显著提高效率。def route_to_multiple(state) - list: if state[some_flag]: return [node_b, node_c] # 两个节点并行执行 return [node_b]四、固定入口点Entry Point——老板指定第一个骑手其实我们在前面的例子中已经用过了。固定入口点就是通过 add_edge(START, 某个节点) 来设置的它定义了工作流从哪个节点开始。没有入口点的图是无法执行的因为框架不知道第一个该调用谁。4.1 代码回顾builder.add_edge(START, node_a) # 固定从 node_a 开始就这么简单。你也可以把 START 直接连到任何节点不一定非叫“node_a”。五、条件入口点Conditional Entry Point——根据订单选第一个骑手5.1 故事版有些订单是大额订单需要先让“超级骑手 D”做预处理比如加 10 元附加费有些是普通订单直接让“普通骑手 A”开始即可。老板站在START路口看一眼订单金额大于 5 元的派 D 先上否则派 A 先上。这就是条件入口点START 后面接的不是固定边而是一个条件边。5.2 什么时候用多租户系统不同客户走不同的初始化流程根据输入参数选择不同的处理链A/B 测试随机决定从哪个分支开始5.3 完整代码条件入口点 后续合并from typing import Literal from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END class GraphState(TypedDict): value: int step: str def node_a(state: GraphState) - dict: 普通骑手 A加 1 元 print(f普通骑手 A金额 {state[value]} 1 {state[value]1}) return {value: state[value] 1, step: A 执行完毕} def node_d(state: GraphState) - dict: 超级骑手 D加 10 元大单专属 print(f超级骑手 D金额 {state[value]} 10 {state[value]10}) return {value: state[value] 10, step: D 执行完毕} def node_b(state: GraphState) - dict: 最终配送员 B金额翻倍 print(f最终配送 B金额 {state[value]} * 2 {state[value]*2}) return {value: state[value] * 2, step: B 执行完毕} # 条件入口路由函数注意它接收的是初始 state还没有经过任何节点 def entry_condition(state: GraphState) - Literal[node_a, node_d]: if state.get(value, 0) 5: print(f条件入口订单金额 {state[value]} 5派超级骑手 D) return node_d else: print(f条件入口订单金额 {state[value]} 5派普通骑手 A) return node_a builder StateGraph(GraphState) builder.add_node(node_a, node_a) builder.add_node(node_d, node_d) builder.add_node(node_b, node_b) # ----- 关键条件入口点 ----- # 从 START 出发使用 entry_condition 选择第一个节点 builder.add_conditional_edges( START, entry_condition, { node_a: node_a, node_d: node_d } ) # 无论从 A 还是 D 出来都去 B然后结束 builder.add_edge(node_a, node_b) builder.add_edge(node_d, node_b) builder.add_edge(node_b, END) graph builder.compile() # 小订单测试 print( 条件入口点小订单 value3 ) r1 graph.invoke({value: 3, step: }) print(f结果: {r1}\n) # 大订单测试 print( 条件入口点大订单 value10 ) r2 graph.invoke({value: 10, step: }) print(f结果: {r2}\n)运行输出 条件入口点小订单 value3 条件入口订单金额 3 5派普通骑手 A 普通骑手 A金额 3 1 4 最终配送 B金额 4 * 2 8 结果: {value: 8, step: B 执行完毕} 条件入口点大订单 value10 条件入口订单金额 10 5派超级骑手 D 超级骑手 D金额 10 10 20 最终配送 B金额 20 * 2 40 结果: {value: 40, step: B 执行完毕}注意条件入口点与普通条件边的区别仅在于源节点是 START 而不是一个普通节点。六、循环控制Cycle——让骑手来回跑直到老板喊停6.1 故事版有些任务需要反复执行比如“检查餐做好了没”骑手 A 去厨房看一眼如果没做好就绕一圈经过 B 点再回来重新看直到做好为止。这种A → B → A的回路就是循环。但循环必须要有终止条件否则骑手会跑到天荒地老死循环。我们通过一个条件边来判断如果做好达到最大次数就去 END否则继续循环。此外还要设置递归限制recursion_limit相当于给骑手一个“最多跑几圈”的安全绳。6.2 为什么需要循环重试逻辑调用外部 API 失败后重试迭代计算牛顿法、梯度下降多轮对话AI 需要反复思考直到得出最终答案状态机等待某个外部条件满足6.3 完整代码计数器循环 递归限制from typing import Literal from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.errors import GraphRecursionError # 用于捕获超限异常 # 定义循环专用的状态增加 max_count 字段 class LoopState(TypedDict): count: int # 当前循环次数 result: str max_count: int # 最大允许循环次数 def node_a(state: LoopState) - dict: 骑手 A执行一次核心任务并将 count 加 1 current_count state[count] new_count current_count 1 print(f【循环 {new_count}】骑手 A 开始工作当前计数: {current_count} - {new_count}) return { count: new_count, result: f已完成第 {new_count} 次处理 } def node_b(state: LoopState) - dict: 骑手 B辅助处理比如记录日志、延迟等不改变 count print(f【循环 {state[count]}】骑手 B 辅助记录: {state[result]}) # 注意这里没有返回 count所以 count 保持不变框架会自动合并 return { result: f辅助后 - {state[result]} } def route(state: LoopState) - Literal[b, END]: 路由函数判断是继续循环还是结束。 当 count max_count 时结束否则继续去 node_b然后 node_b 会连回 node_a if state[count] state[max_count]: print(f✅ 终止条件满足当前计数 {state[count]} 最大 {state[max_count]}结束循环) return END else: print(f 继续循环当前计数 {state[count]} 最大 {state[max_count]}) return b # 创建图 builder StateGraph(LoopState) builder.add_node(a, node_a) builder.add_node(b, node_b) # 入口从 START 到 a builder.add_edge(START, a) # 条件边从 a 出发根据 route 决定去 END 还是去 b builder.add_conditional_edges(a, route) # 关键回路b 执行完后无条件回到 a形成循环 builder.add_edge(b, a) graph builder.compile() print( 循环控制演示最多循环 3 次 ) try: result graph.invoke( input{ count: 0, result: , max_count: 3 }, config{ recursion_limit: 10 # 安全限制最多执行 10 个超级步骤每个节点调用算一步 } ) print(\n最终结果) print(result) except GraphRecursionError as e: print(f递归错误超出限制: {e})运行结果解释第 1 次a (count0→1) → route 判断 13 → 去 b → b → 回到 a第 2 次a (1→2) → route 判断 23 → 去 b → b → 回到 a第 3 次a (2→3) → route 判断 33 → 去 END结束。最终 count 3。关于递归限制recursion_limit 限制了图中节点被调用的总次数包括重复调用。如果循环次数超过这个值框架会抛出 GraphRecursionError防止程序卡死。在上面的例子中最大循环 3 次实际节点调用次数约为 3 次 a 2 次 b 5 次远小于 10所以安全。七、总结所有“边”的类型对比表类型决策依据边数常见应用场景比喻普通边无固定 1 条出边线性 pipeline直路必须走条件边状态值多条可选分支逻辑、错误处理看路牌拐弯固定入口点无1 条出边START → 节点单入口工作流老板指定第一个人条件入口点初始状态多条可选多模式入口、A/B测根据订单选第一个骑手循环状态值 回路条件边 回路边迭代、重试、轮询绕圈直到满意几个关键原则状态是唯一通信方式节点之间不要用全局变量所有需要共享的数据都放在 State 里。每个节点返回字典只更新变化的字段不要返回整个 State框架会自动合并。循环必须带终止条件并且在 config 中设置合理的 recursion_limit。善用条件入口点可以让同一个图服务不同场景减少重复代码。并行执行如果一个节点有多个出边比如条件边返回列表目标节点会并行执行提高效率。