说实话，LangGraph 真正让人上头的地方，不是它能把几个节点串成一条链，而是它在“下一步该往哪走”这件事上给了你极大的自由度。链式结构谁都会写，递归调用 LLM、if-else 判断也不难，但一旦你开始做多智能体协作、带质量反馈的生成循环、意图驱动的任务分发，你就会发现普通的流程编排根本扛不住这种复杂度。LangGraph 的分支执行逻辑，就是专门用来解决这类问题的核心机制。

这篇文章我想围绕 LangGraph 的分支执行展开，讲清楚三个层面：条件分支的 API 到底怎么用、它的运行时是怎么决定路径的、以及我在实际项目里踩过的那些分支相关的大坑。无论你是刚接触 LangGraph 还是已经用它做过几个 demo，这篇文章都能给你一些可直接抄作业的经验。

1. 为什么要单独聊“分支执行”

1.1 图结构不是链式结构的“升级版”

很多人第一次看 LangGraph 的示例，第一反应是“这不就是 LangChain 的链换了个写法吗”。确实，单看add_node加add_edge的写法，它跟串行链很像。但分支执行出现之后，两者的差距就彻底拉开了。

链式结构写死的是一条固定路径：A 执行完必然去 B，B 执行完必然去 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_edges：LangGraph 的“路由中枢”

先看 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, total=False): 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_limit：

config = {"recursion_limit": 100} graph.invoke(initial_state, config=config)

很多人都不知道这个参数，结果图跑一半突然报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, total=False): 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有多个并行实例，它们的输出都会发给aggregate，aggregate必须等接收到所有上游消息才会触发。这就是消息传递模型的汇合语义。汇合节点的代码里不要假设结果的顺序，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"。
• 字符串大小写不一致，比如路由函数返回"Chat"，path_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 里，用total=False放宽必填约束，但绝不凭空往状态里塞一个 schema 之外的字段。
• 每个节点只返回自己负责修改的字段，不要试图在节点里读取并原样返回所有状态字段，那容易引发多余的状态更新。
• 涉及并行写入的字段，一律在定义时就设计好 reducer，不要等到数据被覆盖了再补救。
• 跨节点共享的对象，比如 database session、http client，不要塞在状态里，要放到 config 或依赖注入体系里。因为分支并行时状态会被复制，你塞进去的非序列化对象会导致各种诡异错误。

5.3 别在路由函数里做重活

这是我最想强调的一条实操心得。

路由函数只做一件事：看状态，返回字符串。不要在里面调用大模型，不要做网络请求，不要读数据库，不要做复杂计算。我见过有的项目把意图识别整个逻辑都塞进了路由函数，结果图跑到条件边时都会卡一两秒，而且因为路由函数被 LangGraph 的运行时多次调用，同样的识别逻辑重复执行了好几次，白白浪费 token 和延迟。

正确的做法是：把重计算放到一个专门的节点里，把计算结果写入状态，路由函数只负责读取状态里的字段并匹配映射。前面客服机器人的例子就是这种模式，router_node负责调分类模型，route_by_intent只做查表。这样路由函数足够轻，图执行稳定，出问题也好定位。

分支执行逻辑是 LangGraph 里最能拉开设计水平的部分。同样一个“生成-评审-改写”的循环，新手可能写成三个节点加两个条件边，每一步都在担心循环次数会不会超限；而有经验的人会先把 Reviewer 做成纯逻辑节点，把循环条件收敛到一个路由函数里，用 recursion_limit 兜住执行深度，再让关键路径上的每一步都能通过日志验证。我实际操作下来最大的感受，就是要把条件路由当成一等公民来设计，而不是节点逻辑的附属品。给路由函数做单测、给状态字段声明 reducer、为每一个条件边预留兜底路径——这些习惯比记住任何 API 细节都重要。

如果你正在做一个稍微复杂一点的 LangGraph 项目，我的建议是不要急着把每一个业务判断都变成条件边。先画一画图，想清楚哪些决策是流程级的、哪些判断只影响节点内部实现，然后把流程级决策抽出来做成路由函数。等你跑通了第一个“生成-评审-重试”的循环，回头看那些只会串行的链式方案，大概率就回不去了。