说实话LangGraph 真正让人上头的地方不是它能把几个节点串成一条链而是它在“下一步该往哪走”这件事上给了你极大的自由度。链式结构谁都会写递归调用 LLM、if-else 判断也不难但一旦你开始做多智能体协作、带质量反馈的生成循环、意图驱动的任务分发你就会发现普通的流程编排根本扛不住这种复杂度。LangGraph 的分支执行逻辑就是专门用来解决这类问题的核心机制。这篇文章我想围绕 LangGraph 的分支执行展开讲清楚三个层面条件分支的 API 到底怎么用、它的运行时是怎么决定路径的、以及我在实际项目里踩过的那些分支相关的大坑。无论你是刚接触 LangGraph 还是已经用它做过几个 demo这篇文章都能给你一些可直接抄作业的经验。1. 为什么要单独聊“分支执行”1.1 图结构不是链式结构的“升级版”很多人第一次看 LangGraph 的示例第一反应是“这不就是 LangChain 的链换了个写法吗”。确实单看add_node加add_edge的写法它跟串行链很像。但分支执行出现之后两者的差距就彻底拉开了。链式结构写死的是一条固定路径A 执行完必然去 BB 执行完必然去 C。你可以在节点内部写 if-else 来控制逻辑但路径本身不会因为内容的差异而改变。LangGraph 的add_conditional_edges则把“下一步去哪”这个决策从节点内部抽出来变成图中一个显式的路由环节。这意味着流程的拓扑结构可以随运行时的状态动态变化一个节点可以通向不同的后续节点甚至可以通过循环回到自己前面的节点。我见过的比较典型的需求是这样用户输入一句 query先做意图识别识别出“闲聊”就走对话节点识别出“搜索”就走检索节点识别出“工具调用”就走函数调用节点。这个流程如果用链写你得在节点内部自己维护一个状态标志位然后在下一步节点里判断这个标志位非常别扭。LangGraph 的add_conditional_edges就是为这种场景设计的路由本身是图的一部分意图判断完成后图自己决定下一步。1.2 分支不是 if-else 那么简单这里必须先纠正一个直觉分支执行不等于 if-else。if-else 是单次判断而 LangGraph 里的分支有几个更复杂的变体单路条件分支根据路由函数返回值选择多个后续节点中的一个本质就是 if-else。多路并行分支路由函数返回多个目标节点图中同时执行多条路径最后再汇合。动态并行分支通过SendAPI根据状态动态创建多个任务并行执行。循环回边条件分支把路径指回之前的节点形成循环比如“生成内容 - 质量检查 - 不合格则重新生成”。这些变体组合起来能表达相当复杂的业务逻辑。多智能体协作里常见的“规划 - 委派 - 汇总”模式质量门槛里的“生成 - 评审 - 改写”循环本质上都是分支执行的组合结果。所以理解分支执行不能只停留在 API 层还要理解它的调度模型和状态管理机制否则一旦流程复杂起来你会被各种莫名其妙的行为折磨疯掉。2. 分支执行的核心机制拆解2.1 add_conditional_edgesLangGraph 的“路由中枢”先看 API 本身。add_conditional_edges是 StateGraph 上的一个方法签名大致是这样builder.add_conditional_edges( node_a, # 源节点 route_function, # 路由函数接收当前状态返回目标标识 path_map # 返回值到目标节点的映射 )这里的三个参数各司其职源节点表示“从这个节点出发时要重新规划路径”路由函数接收当前的完整状态经过判断后返回一个可哈希的标识通常是字符串path_map 把这个标识映射到具体的下一个节点。需要特别注意的点是路由函数不负责实际业务逻辑它只做决策。最理想的状态是它是一个轻量纯函数输入状态、输出字符串不调 LLM、不做重 IO。为什么强调这点因为 LangGraph 的运行时会多次调用路由函数如果你在里面塞了一段大模型调用每跑一次图就白花一次 token而且排查问题时也很难定位到底是哪次调用出了问题。另外add_conditional_edges必须在compile()之前调用。这是 LangGraph 构建期的静态约定所有边的信息在编译时就固定下来。运行时你改变不了拓扑能变的只有路由函数的返回结果。很多新手用反射或者动态修改图结构来“动态规划路径”这是走不通的LangGraph 的设计哲学就是图结构静态、路径决策动态。来看一个具体例子这是我在一个客服机器人项目里写的意图路由from typing import Literal from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END class TicketState(TypedDict, totalFalse): user_input: str intent: str reply: str def router_node(state: TicketState) - dict: # 实际项目中这里会调 LLM 或分类模型做意图识别 intent classify_intent(state[user_input]) return {intent: intent} def chat_reply(state: TicketState) - dict: return {reply: 闲聊模式 state[user_input]} def search_reply(state: TicketState) - dict: return {reply: 检索模式 state[user_input]} def ticket_reply(state: TicketState) - dict: return {reply: 工单模式 state[user_input]} def route_by_intent(state: TicketState) - Literal[chat, search, ticket]: mapping { 闲聊: chat, 咨询: search, 投诉: ticket, } return mapping.get(state[intent], chat) builder StateGraph(TicketState) builder.add_node(router, router_node) builder.add_node(chat, chat_reply) builder.add_node(search, search_reply) builder.add_node(ticket, ticket_reply) builder.add_edge(START, router) builder.add_conditional_edges( router, route_by_intent, { chat: chat, search: search, ticket: ticket, }, ) builder.add_edge(chat, END) builder.add_edge(search, END) builder.add_edge(ticket, END) graph builder.compile()这段代码最关键的一行是route_by_intent。它返回的是chat、search、ticket三个字符串之一然后 LangGraph 拿着这个字符串去 path_map 里找对应的目标节点。如果你返回的字符串没有在 path_map 里定义运行时会直接抛异常。这个机制看起来很直白但实际项目中大部分分支相关的 bug 都出在这个映射关系上后面我会单独展开。2.2 返回路径与映射表字符串键如何决定整个图的走向path_map 的键必须和路由函数的返回值严格匹配。这个匹配是精确的、不分大小写模糊匹配的你返回Chat映射表里写chat照样报错。path_map 的值有三种常见写法直接写节点名比如chat: chat_node意思是返回chat时去执行chat_node。用 END 常量比如end: END意思是返回end时结束整个图。用其他条件边的别名LangGraph 允许 path_map 的值指向另一个条件边的标签实现更复杂的嵌套跳转。前两种最常用。第三种是在大图里做层级路由时用的能把多个条件判断串联起来但代价是调试复杂度直线上升。我的建议是除非你的图已经大到无法用平铺节点表达否则尽量避免嵌套条件边宁可多用几个中间节点也不要让路由逻辑变成一团乱麻。路由函数本身有个硬性要求返回值必须是可哈希的。因为 LangGraph 内部要用返回值作为键去查找 path_map所以返回字符串是最稳妥的。你千万别返回一个动态生成的 dict 或者 list那不仅没法稳定匹配还会让状态缓存出现诡异的问题。再补充一个压箱底的细节add_conditional_edges也可以传入一个列表作为 path_map 的简写形式。如果你写的是builder.add_conditional_edges(node_a, route_fn, [node_b, node_c])那 LangGraph 会认为路由函数返回值是一个索引整数或者一个列表然后根据索引或者列表内容并行派发到多个节点。但很多教程没明说的是这种用法在路由函数返回列表时会触发多路并行分支。也就是说node_b和node_c会同时执行。如果你只是想做单选路由拉一个列表当映射表那就会惊喜地发现两个分支全都跑了一遍。这个坑我实际踩过后面会详细讲。2.3 执行时的实际调度深度优先还是并行分支执行在运行时到底是怎么跑的这个问题我研究了挺久LangGraph 的官方文档写得很含糊但通过观察执行 trace 能看出实际调度逻辑。当一个节点有多个出边且同时被激活时LangGraph 会并行调度这些分支。它内部用的是 asyncio 的任务调度所以如果你的节点函数里有阻塞调用比如 requests.post、openai.ChatCompletion.create它们会阻塞事件循环导致“所谓的并行”退化成串行。我的做法是节点函数里统一用 async 定义配合异步客户端才能拿到真正的并发效果。不过这里有个微妙的点LangGraph 的“并行”是分水岭式的。分支并行时每个分支有自己的执行上下文但它们共享同一个 State。也就是说如果两个并行分支同时往 State 里写同一个键后写的那一方会覆盖先写的。这个覆盖行为不是随机而是由调度顺序决定极难复现和定位。解决方案是用Annotated类型给状态字段声明 reducer让 LangGraph 知道这个字段是追加合并而不是覆盖替换。from typing import Annotated from operator import add from typing_extensions import TypedDict class ParallelState(TypedDict): results: Annotated[list[str], add]声明了Annotated[list[str], add]之后多个分支往results里写入时LangGraph 会把这些值按顺序拼接成列表而不是互相覆盖。这个机制是我觉得 LangGraph 设计得最出彩的地方但也是最容易被忽略的地方。如果你没声明 reducer并行分支的返回值收集基本就是碰运气。再说回调度策略。LangGraph 本身没有显式的深度优先或广度优先参数它更像是一个面向 Pregel 风格的消息传递模型。每个节点执行完它的 output 会作为消息发给下游节点下游节点满足所有上游输入后才会被调度。所以当你说“并行分支”时本质是上游节点同时向多个下游节点发送了消息多个下游节点各自独立执行。当你说“汇合”时本质是某个下游节点需要接收多个上游消息全部到齐后才会执行。理解这个模型之后你就能解释很多看起来匪夷所思的行为为什么一个节点明明只有两个入边却只执行了一次状态更新为什么并行分支的汇合节点迟迟不触发为什么循环图里同样一个节点会被反复执行。这些都是消息传递模型的自然结果不是 bug。3. 三张图带你落地分支场景3.1 场景一意图识别路由器这个场景算是分支入门必做。核心思路是入口节点做完意图识别路由函数根据意图把任务分发到不同处理节点。前面已经展示过完整代码了这里补充几个我在真实落地时加进去的细节。首先是路由函数里的 fallback。我的route_by_intent里最后一行return mapping.get(state[intent], chat)就是兜底逻辑。模型输出永远可能出意外分类模型给你返回一个训练集里根本不存在的标签这时候如果没有 fallback图就会因为映射不匹配直接崩溃。不要假设你的模型永远输出干净的分类把每一个分支都当作文本处理该清洗要清洗该兜底要兜底。其次是路由函数的位置。我习惯把路由函数定义成模块级函数而不是闭包或者 lambda。原因很简单路由函数需要被单元测试。我写过十几个路由函数之后发现它们其实是最值得单独测试的组件因为整个图的路径走向都由它们决定。用闭包定义在某个节点函数内部的话测试时需要导出整个图麻烦得不行。最后是路径常量。我在代码里不会直接写chat这种魔法字符串而是定义成常量class Route: CHAT chat SEARCH search TICKET ticket class NodeName: ROUTER router CHAT_REPLY chat_reply SEARCH_REPLY search_reply TICKET_REPLY ticket_reply这样写的好处是路由函数和 path_map 共用同一组常量永远不会出现字符串写错导致映射失败的问题。语言模型的输出里可能会出现Chat这种带大写字母的情况我一般在路由函数前做一个统一小写化和去空格清洗保证进入映射的键是稳定的。3.2 场景二质量门槛与重试循环分支执行最有杀伤力的场景不是单选路由而是循环。质量门槛就是一个标准例子生成结果评审打分分数不够就改写重来直到分数达标或者达到最大重试次数。这个流程如果用传统编程写你会陷入一层层的 while 循环嵌套状态变量散落各处一旦流程要加一个“重试三次后降级处理”的分支整个逻辑立刻面目全非。LangGraph 里这就是一个非常自然的结构from typing import Literal from langgraph.graph import StateGraph, START, END class QualityState(TypedDict): draft: str score: float revise_count: int def generate(state: QualityState) - dict: return {draft: generate_draft(state.get(draft))} def review(state: QualityState) - dict: score assess_quality(state[draft]) return {score: score} def revise(state: QualityState) - dict: count state.get(revise_count, 0) return { draft: rewrite_draft(state[draft]), revise_count: count 1, } def route_by_score(state: QualityState) - Literal[revise, end]: if state.get(revise_count, 0) 3: return end if state[score] 0.7: return revise return end builder StateGraph(QualityState) builder.add_node(generate, generate) builder.add_node(review, review) builder.add_node(revise, revise) builder.add_edge(START, generate) builder.add_edge(generate, review) builder.add_conditional_edges( review, route_by_score, {revise: revise, end: END}, ) builder.add_edge(revise, review) graph builder.compile()这里最棒的一点是route_by_score把“循环条件”和“业务节点”彻底解耦了。评审节点只负责打分改写节点只负责修改文本而“要不要继续改、改几次算完”完全由路由函数控制。以后需求变了比如改成“以相关性指标为标准”或者“重试次数从 3 次改成 5 次”只需要改路由函数不需要动任何业务节点。这个模式里有一个很容易被忽略的性能问题当分数不达标、路由到 revise 再回到 review 时generate节点会被跳过因为review只有一个入边来自generate一次循环后review的上游只有revise。这避免了重复生成是对的。但如果你在review节点里调用了大模型打分而大模型打分本身不稳定那第二次评审和第一次评审的结果就可能差异很大。我的经验是评审类节点尽量做成可复现的固定住模型温度参数禁用采样或者用确定性规则打分辅助验证。循环还有一个隐形坑recursion_limit。LangGraph 默认的递归限制是 25也就是单个图最多执行 25 个节点。如果你把重试上限设在 10而每个循环周期有 3 个节点那 30 个节点的执行量就会直接超过限制。所以有循环的图必须显式设置recursion_limitconfig {recursion_limit: 100} graph.invoke(initial_state, configconfig)很多人都不知道这个参数结果图跑一半突然报GraphRecursionError还以为是业务逻辑写错了排查半天。其实只是一行配置的问题。3.3 场景三并行分支Send API当你的分支数量不是静态的比如用户输入 5 个任务、你希望每个任务并行处理那么用普通的add_conditional_edges就不行了因为节点的数量是运行时才知道的。这时候要用SendAPI。from langgraph.types import Send from typing import Annotated, Literal from operator import add from typing_extensions import TypedDict class TaskState(TypedDict): task: str result: str class AggregateState(TypedDict, totalFalse): tasks: list[str] results: Annotated[list[str], add] def dispatcher(state: AggregateState) - list[Send]: return [Send(process, {task: t}) for t in state[tasks]] def process(state: TaskState) - dict: return {results: [execute_task(state[task])]} builder StateGraph(AggregateState) builder.add_node(dispatch, dispatcher) builder.add_node(process, process) builder.add_edge(START, dispatch) builder.add_conditional_edges(dispatch, dispatcher, [process]) builder.add_edge(process, END) graph builder.compile()Send的作用是动态创建一条从一个节点到另一个节点的并行执行路径。dispatcher函数返回一个Send列表每个Send明确指定去哪个节点、带什么状态。LangGraph 运行时收到这个列表后会为每个Send单独调度一个process的实例这些实例并行执行。这里有个关键点process节点的返回值得走 reducer否则并行执行的结果会把results字段互相覆盖掉。我在上面的代码里用了Annotated[list[str], add]这样每个process返回的值都会追加到results列表里最后你拿到的就是一个完整的任务结果合集。如果你想在所有并行任务完成后做汇总处理不能在process后面直接接 END而是应该加一个aggregate节点。因为process有多个并行实例它们的输出都会发给aggregateaggregate必须等接收到所有上游消息才会触发。这就是消息传递模型的汇合语义。汇合节点的代码里不要假设结果的顺序tasks 的输入顺序和 results 累积顺序不一定一一对应最好在Send时携带一个 index然后汇总节点手工排序。用 Send API 时还容易遇到一个问题每个 Send 携带的状态副本是独立的。你在process里修改的字段如果定义在AggregateState里必须通过 reducer 机制合并回去如果定义在TaskState里且AggregateState没声明该字段那修改会被直接丢弃。这个坑我曾经调了一个晚上才弄明白本质上还是对状态合并机制理解不够深。4. 调试分支逻辑的常见事故现场4.1 “Conditional edge returned a value that was not found” 报错这是分支执行里最常见的报错信息大概长这样Conditional edge returned a value xxx which was not found in path map出现这个报错大概率是路由函数的返回值没有在 path_map 的键集合里。常见原因有三个意图分类结果里有你没处理的标签比如模型返回了unknown而你的 path_map 里没有unknown。字符串大小写不一致比如路由函数返回Chatpath_map 里写的是chat。路由函数里出现了意外异常被上层框架包装成了这个报错。这种情况很隐蔽因为异常堆栈只显示条件边判断失败真正的异常源在路由函数内部被吞掉了。排查建议把路由函数单独拎出来喂几组典型的状态数据直接调用并打印返回值。因为这个函数是纯函数完全可以脱离图环境单测。4.2 路由函数返回了 None路由函数没有显式 return或者逻辑分支里漏了 return返回了None。LangGraph 的路由函数不能接受None作为合法的分支标识除非你的 path_map 里有None作为键否则直接抛异常。这个坑多数发生在你用一个复杂的 if-elif 链时漏掉了 else 分支。我前两年写的路由代码就吃过这个亏当时总觉得所有条件都覆盖了直到线上出现了一个意想不到的输入组合路由函数走进了死胡同。现在的我写路由函数一定会先写兜底 return然后再写具体分支。4.3 并行分支的写入覆盖问题并行分支同时写同一个状态键是最难排查的问题之一。表现是两个分支分别执行了但最终的输出只剩一个分支的结果另一个分支的结果凭空消失了。原因我在前面也说过就是没有给状态字段声明 reducer。排查这个问题时不要盯着节点逻辑看先检查 State 定义。如果字段类型是list就用Annotated[list, add]如果是dict大概率需要自定义 reducer如果只是单一值那就得考虑是不是业务上根本不该并行写同一个键。顺便说一句Annotated[list, add]这个写法虽然简单但add操作符在你需要做去重、排序、限定最大长度时会显得力不从心。这种场景我建议写自定义 reducer 函数用一下Annotated的类型别名把 reducer 本身也变成一个可测试的单元。4.4 可视化调试与日志追踪排查分支问题最快的方式永远是看全量执行路径。LangGraph 提供了两个手段一是graph.get_graph().draw_mermaid()输出图的文本表示二是通过LangGraph Studio或者自定义回调看每一步的实际路由结果。我更依赖后者。在回调里打印每个节点执行前后的状态快照和当前节点名能很快定位到“到底从哪一步开始走错了路”。注意不要在回调里打印完整的 prompt 或者 model response那会把日志冲爆。打印关键字段就好。另外有个小技巧给路由函数加一个自定义的日志包装直接输出它收到的 state 和返回的键。我在所有条件边的路由函数上都会加这种日志线上排查时这个日志几乎是唯一可靠的信息来源。顺带提一句draw_mermaid()输出的是 Mermaid 文本不要把它当成可交互的图。你要真正可视化可以用graph.get_graph().draw_mermaid_png()渲染成图片或者用draw_ascii()直接在终端里看 ASCII 图。5. 分支设计上的一点经验沉淀5.1 分支粒度不要抠太细很多初学者会把“判断一个字段是否存在”也设计成一条条件边。比如图跑到某个节点后检查state.get(lang)是中文还是英文然后分成两条路径。这种设计不是不行但分支粒度越细路径组合越多测试覆盖越难做全排查问题也就越费劲。我的习惯是把路由相关的逻辑尽量集中到一两个“决策节点”里让业务节点保持线性。一个路由节点管意图分发一个路由节点管质量循环把不同层级的决策拆开而不是让每个节点都顺手做一个分支判断。这样图的结构清晰路径数量可控出问题时也容易定位到唯一的路由函数。5.2 状态读写要有明确约定分支执行越复杂状态的读写就越需要纪律。我给自己定了几条规则状态字段尽量显式定义在 TypedDict 里用totalFalse放宽必填约束但绝不凭空往状态里塞一个 schema 之外的字段。每个节点只返回自己负责修改的字段不要试图在节点里读取并原样返回所有状态字段那容易引发多余的状态更新。涉及并行写入的字段一律在定义时就设计好 reducer不要等到数据被覆盖了再补救。跨节点共享的对象比如 database session、http client不要塞在状态里要放到 config 或依赖注入体系里。因为分支并行时状态会被复制你塞进去的非序列化对象会导致各种诡异错误。5.3 别在路由函数里做重活这是我最想强调的一条实操心得。路由函数只做一件事看状态返回字符串。不要在里面调用大模型不要做网络请求不要读数据库不要做复杂计算。我见过有的项目把意图识别整个逻辑都塞进了路由函数结果图跑到条件边时都会卡一两秒而且因为路由函数被 LangGraph 的运行时多次调用同样的识别逻辑重复执行了好几次白白浪费 token 和延迟。正确的做法是把重计算放到一个专门的节点里把计算结果写入状态路由函数只负责读取状态里的字段并匹配映射。前面客服机器人的例子就是这种模式router_node负责调分类模型route_by_intent只做查表。这样路由函数足够轻图执行稳定出问题也好定位。分支执行逻辑是 LangGraph 里最能拉开设计水平的部分。同样一个“生成-评审-改写”的循环新手可能写成三个节点加两个条件边每一步都在担心循环次数会不会超限而有经验的人会先把 Reviewer 做成纯逻辑节点把循环条件收敛到一个路由函数里用 recursion_limit 兜住执行深度再让关键路径上的每一步都能通过日志验证。我实际操作下来最大的感受就是要把条件路由当成一等公民来设计而不是节点逻辑的附属品。给路由函数做单测、给状态字段声明 reducer、为每一个条件边预留兜底路径——这些习惯比记住任何 API 细节都重要。如果你正在做一个稍微复杂一点的 LangGraph 项目我的建议是不要急着把每一个业务判断都变成条件边。先画一画图想清楚哪些决策是流程级的、哪些判断只影响节点内部实现然后把流程级决策抽出来做成路由函数。等你跑通了第一个“生成-评审-重试”的循环回头看那些只会串行的链式方案大概率就回不去了。