我在用LangChain写Agent那段时间，心里一直有个别扭的地方：明明想让模型先判断用户的意图，再决定走哪条处理流程，但Chain的线性结构非要我一条路走到黑，中途想岔开一条分支，只能在外层写一堆if/else硬编码，把流程切开、再拼起来。直到换到LangGraph，我才真正体会到“图”这个抽象的价值——分支执行不再是流程的例外情况，而是图结构里的一等公民。
这篇文章想聊透的，就是LangGraph的分支执行逻辑：它如何基于状态（State）做路径选择，条件边和普通边怎么配合，路由函数（Router）的返回值到底怎么被解析，以及我在实际项目里踩过的那些坑。适合两类人看：一是已经把LangChain玩熟、但还没搞懂Graph控制流的开发者；二是刚接触LangGraph、被add_conditional_edges这类API绕晕的初学者。看完之后，你能自己设计带意图路由、循环反思、多分支并发的Agent工作流，而不是只会照着官方示例抄。
1. 先搞清楚：LangGraph为什么会“分叉”
1.1 从Chain到Graph：为什么线性流程不够用
LangChain的LCEL（LangChain Expression Language）把可执行的单元连成一条链，比如prompt | model | parser。这种流水线对“固定步骤”很有效，比如翻译、摘要、提取结构化数据。但它有个先天限制：链是单一路径，A做完之后只能去B，不能根据A的结果决定去B还是去C。
你可能说“我可以在外面写if/else啊”——对，很多项目确实是这么干的。但问题在于：当分支数量一多，if/else嵌套开始爆炸；当分支里还要嵌套分支，控制流就完全被拆散到业务代码里，图上的每个节点倒是很干净，但整个执行流程“看不见”。LangGraph的解法很朴素：把控制流画成一张有向图，节点就是执行单元，边就是跳转关系。分支不再是你代码里的判断语句，而是图结构中的一条条件边。
我打过一个比方：LCEL是机场里的动线，安检完了必然去登机口，顺序是死的；LangGraph是地图导航，到了路口根据路况、里程、时间选择走哪条路，而每次选择都可能改变后续路线。
1.2 分支执行的三种形态：条件路由、并行扇出、循环回边
LangGraph分支执行看起来花样多，但本质上逃不出三种形态：
- 条件路由（Conditional Routing）：一个节点执行完毕后，根据当前State的值，从多个后继节点中“挑一个”继续走。这是最常见的形式，比如意图识别后进入不同处理分支。
- 并行扇出（Fan-out）：一个节点后面同时挂多个节点，单个节点执行完后，多个分支可以并行执行，最后再汇聚。这在LangGraph里是通过普通边实现的——从一个节点连出多条边，就是扇出。
- 循环回边（Cycle）：条件边指向它上游的节点，形成回路。Agent里的“工具调用循环”、反思再生成，本质都是图的回路。
理解这三者的区别，是读懂分支逻辑的第一步。条件路由是“多选一”，并行扇出是“多选多”，循环回边是“走回头路”。很多业务场景是这三种的组合：先条件路由到某分支，分支内部再扇出多个并行子任务，子任务不满足要求时又通过回边绕回去。
2. 分支执行的核心机制拆解
2.1 StateGraph：状态是分支决策的唯一依据
LangGraph里的图是StateGraph，它接收一个State类型定义。你可以用TypedDict、Pydantic BaseModel或者dataclass来定义。所有节点函数签名都是(state) -> dict：读全局状态，返回对状态的部分修改。
这点对理解分支非常重要：边的跳转不靠节点间的返回值传递，而是靠共享的State。节点A做了判断，要把判断结果写进State（比如写intent字段），条件边再根据intent的值做路由。节点函数返回的字典会被合并到全局State里。
看一个最小定义：
from typing_extensions import TypedDict class AgentState(TypedDict): messages: list intent: str order_id: str节点返回{"intent": "query_order"}，这个字段就写进了State。后续所有节点、所有边都能读得到。共享状态即通信。这是LangGraph控制流设计的基石，也是很多人一开始不习惯的地方：节点本身不带返回值给“下一个节点”，而是把结果写到一个公共区域，下一个节点从公共区域里取。
更要留心的是状态合并规则。默认情况下，新写入的字典值会覆盖旧值；但如果某个字段需要累加、追加（比如对话消息列表要不断接长），就得用Annotated告诉LangGraph用自定义Reducer合并。最典型的写法：
from typing import Annotated import operator class AgentState(TypedDict): messages: Annotated[list, operator.add] intent: stroperator.add在这里不是加法的意思，而是“把新返回的list拼到旧list末尾”。如果不加Annotated，每次返回只保留最后一次的消息，前面的消息全丢。分支越多、状态更新越频繁，这个合并规则越关键。
2.2 条件边（conditional_edges）：分支跳转的开关
条件边是LangGraph分支执行的核心API。用法是：
graph.add_conditional_edges( source_node, # 从哪个节点出发 router_function, # 决定走向的函数 path_map # 路由函数返回值 -> 目标节点的映射 )router_function接收当前State，返回一个字符串。这个字符串是path_map里的一个key，path_map里存的是目标节点名。举例：
def route_by_intent(state: AgentState) -> str: return state["intent"] path_map = { "query_order": "query_node", "after_sales": "after_sales_node", }执行到source_node后，LangGraph会调用route_by_intent(state)，拿到返回值"query_order"，接着查path_map，锁定目标节点"query_node"，然后把执行权交给它。
这里有个版本差异值得提：LangGraph 0.2.x之后，add_conditional_edges的第三个参数可以省略不传，路由函数直接返回目标节点名字符串。比如上面可以简写为：
def route_by_intent(state): if state["intent"] == "query_order": return "query_node" return "after_sales_node" graph.add_conditional_edges("classify", route_by_intent)省略path_map虽然省事，但我个人建议项目初期仍然显式写映射字典。原因后面在“踩坑”部分会讲：显式path_map能帮你把所有可能路径列出来，漏一条都编译都不报错，但运行时会直接报错，显式字典更容易做路径清单审查。
还有个隐藏细节：路由器不只是能返回节点名，还能返回END。在LangGraph里，你可以把条件边的一条路径指向终点。比如：
path_map = { "stop": END, # 这个分支走完就结束 "continue": "next_node" }这样你不需要专门建一个“空节点”来结束流程，直接在路由层就宣判流程终止。
2.3 普通边与条件边的配合套路
普通边（add_edge）是无条件转移：A执行完，必然去B。条件边是按State动态选择目标。两者经常配在一起形成“主线+分流”格局。
我常用的套路是：主线负责“必然要走的步骤”，条件边负责“决策点”。比如一个客服Agent：
START -> 意图识别节点 意图识别节点 -> 条件边： 查订单 -> 查询节点 -> END 售后 -> 售后节点 -> END 人工 -> 人工节点 -> ENDSTART到意图识别节点是普通边，意图识别到后面各分支是条件边。这样整张图读起来一目了然：哪一步必须做，哪一步看情况做，哪个分支做完就结束。
一旦分支增多，普通边和条件边的边界容易模糊。我的经验是：能不用条件边走到的节点，都用普通边。普通边执行顺序固定，没有额外调用开销，心智负担也小；只有“目的地不唯一”的时候，才动用条件边。
3. 手把手实战：一个带意图路由的客服Agent
理论讲完，上实操。我做一个简化但完整的客服Agent：用户输入一句消息，先做意图分类，再分流到不同处理节点。以此演示条件边分支执行的全过程。
3.1 设计消息状态与意图类型
先定义状态。messages要用Annotated累加，intent存路由依据，order_id可以留空，展示节点之间如何通过State传数据：
from typing import Annotated, TypedDict import operator class AgentState(TypedDict): messages: Annotated[list, operator.add] intent: str order_id: str START_INTENT = "pending"这里我额外初始化了intent的默认值。LangGraph本身没有“默认值”机制，但你在首次调用invoke时把intent传成"pending"即可。注意：如果State里没有对应key，节点函数读取时就会抛KeyError，所以最好是显式传初值。
3.2 自定义Router函数怎么写
意图分类节点要做的事：读取用户最后一条消息，做关键词匹配，把意图写回State：
def classify_node(state: AgentState) -> dict: last_message = state["messages"][-1] text = last_message if isinstance(last_message, str) else last_message.content if "订单" in text and ("查" in text or "哪里" in text): intent = "query_order" elif ("退" in text) or ("换" in text): intent = "after_sales" elif "人工" in text or "投诉" in text: intent = "human" else: intent = "human" # 兜底走人工 return {"intent": intent}然后是路由函数。它是条件边的决策器，只读取State里的intent，不做业务处理：
def route_by_intent(state: AgentState) -> str: return state.get("intent", "human")这个函数越薄越好，因为它聚焦一件事：把State值翻译成路径名。
3.3 注册节点、建图与条件边绑定
定义三个下游节点，分别模拟查订单、处理售后、转人工：
def query_order_node(state: AgentState) -> dict: return {"messages": ["订单查询中，你的订单号已收到。"]} def after_sales_node(state: AgentState) -> dict: return {"messages": ["已登记退换货申请，售后专员将跟进。"]} def human_node(state: AgentState) -> dict: return {"messages": ["已转接人工客服，请稍等。"]}现在把它们组装进图：
from langgraph.graph import StateGraph, START, END graph = StateGraph(AgentState) graph.add_node("classify", classify_node) graph.add_node("query", query_order_node) graph.add_node("after_sales", after_sales_node) graph.add_node("human", human_node) graph.add_edge(START, "classify") graph.add_conditional_edges( "classify", route_by_intent, { "query_order": "query", "after_sales": "after_sales", "human": "human", } ) graph.add_edge("query", END) graph.add_edge("after_sales", END) graph.add_edge("human", END) compiled_graph = graph.compile()这里最核心的就三行：add_edge(START, "classify")把入口连到意图分类节点；add_conditional_edges注册决策点，列出三种路径；最后三个add_edge把分支节点收口到END。
3.4 编译，跑通三种分支路径
编译之后就能直接调用了。验证一下三种输入分别走到哪条分支：
def run_agent(user_input: str): result = compiled_graph.invoke({ "messages": [user_input], "intent": "pending", "order_id": "" }) return result["messages"], result["intent"] print(run_agent("我想查一下订单到哪里了")) print(run_agent("我要退货")) print(run_agent("转人工！"))输出对应应该是google订单查询中...、已登记退换货...、已转接人工客服...。
我实际调试的时候偏好开着stream模式看执行路径，而不是只盯着最终结果。invoke只给你终态，中间走了哪些节点看不到；用stream能看到每个节点的产出：
for event in compiled_graph.stream( {"messages": ["我要退货"], "intent": "pending", "order_id": ""}, stream_mode="updates" ): print(event)输出里会有每个节点返回的增量更新，比如{'classify': {'intent': 'after_sales'}}然后是{'after_sales': {'messages': ['已登记退换货申请，售后专员将跟进。']}}。这一行日志能确认路由函数有没有选中正确的下一个节点，排查分支问题比看最终结果直观得多。
4. 分支+循环：更灵活的“游走式”执行
4.1 用条件边实现图上的循环
分支不一定总是“选一条路走下去”，还可以让路径兜圈子。这就是LangGraph的循环能力，也是它区别于普通DAG工作流引擎的关键。
循环的实现方式很简单：条件边的一条路径，指向它自己的上游节点。比如流程是“生成回答 -> 质量检查 -> 不行就重新生成”。
def generate(state: AgentState) -> dict: # 模拟一次生成 return {"messages": ["生成了一版回答"]} def quality_check(state: AgentState) -> str: # 根据业务规则判断是否通过 import random passed = random.random() > 0.5 return "pass" if passed else "retry" builder = StateGraph(AgentState) builder.add_node("generate", generate) builder.add_node("check", quality_check) builder.add_edge(START, "generate") builder.add_edge("generate", "check") builder.add_conditional_edges( "check", lambda state: "retry", # 演示简化，实际根据检查结果动态判断 { "pass": END, "retry": "generate" } )当路由结果指向generate时，执行流又回到了上游节点，形成回路。这就是“反思-再生成”循环的雏形。
这样的循环能不能终止，完全取决于状态变化。所以循环节点里一定要修改State——比如累加循环计数、记录已尝试过的方案——否则路由函数永远返回retry，就死循环了。
4.2 recursion_limit：循环的刹车片
LangGraph对递归深度有兜底限制，通过config里的recursion_limit控制。默认值一般是25。注意它统计的是“执行步数”而不是“循环次数”，包括顺序执行的节点都算。
result = compiled_graph.invoke( initial_state, config={"recursion_limit": 10} )如果执行步数超过限制，会抛GraphRecursionError。踩过这坑的人都懂：不是你的逻辑挂了，是你的图里循环退不出来，被安全机制拦住了。
我给一个实操建议：调试任何带循环的图，先主动把recursion_limit设成很小的值（比如5），跑通再调大。这样如果真写成了死循环，几秒钟内就会暴露，而不是卡到默认限制才崩。生产环境里，给用户可交互的Agent设recursion_limit更要用保守值，防止LLM在工具调用循环里空转。
5. 分支执行的高频翻车现场与排查手册
这块内容是我个人的血泪账。LangGraph的报错不算友好，很多问题藏得深，我按照从“最常见”到“最隐蔽”的顺序列出来。
5.1 Router返回了不存在的路径名
这是最典型的错误。比如route_by_intent里写的是return state.get("intent", "unknown")，但path_map里没有"unknown"这个key。运行时LangGraph会直接抛出类似ValueError: Unknown node: unknown之类的错误。
我总结出三条防护规则：
- 路由函数里永远给一个兜底返回值，并且兜底值必须在
path_map里。 - 显式
path_map不要省略。虽然新版本允许省略，但显式字典强制你列出所有可达路径，漏一条运行时能看出来。 - 在编译前用
compiled_graph.get_graph().print_ascii()打印文本版图结构，检查每个连出去的边是否都有归宿。在控制台里直接看，不用启动任何可视化服务。
5.2 状态被覆盖、并行分支写冲突
并行扇出场景里，多个分支同时执行后，状态更新不是同步的，而是“先后合并”。如果两个节点都往同一个字段写值，默认Reducer是后者覆盖前者，你看到的最终结果是某个分支的值，另一个分支的值彻底消失。
这个问题刚开始很难察觉，因为单次运行看不出异常，多跑几次发现结果不稳定。
解法分两种：
- 如果字段含义是“累加型”（比如多个分支各产出一条消息），用
Annotated[list, operator.add]做追加合并。 - 如果字段是“独立型”（比如每个分支要写入各自的标志位），就别共用同一字段，拆成
query_result、after_sales_result这种不同字段。
并行分支改共享状态时，我要求团队遵守惯例：只追加，不覆盖；永远把“新增内容”返回成list元素，不直接改已有字段。
5.3 条件边不生效/节点执行顺序错乱
很多人以为add_conditional_edges("A", router, {...})之后，A的所有后继都按某种顺序执行。实际上条件边只会选一条路径，不存在“A之后同时执行B和C”的说法。如果希望A之后并行执行多个节点，那要用多条普通边扇出，而不是一条条件边。
我遇到过明明是“扇出”却放条件边的情况。结果就是每次只走一个分支，业务上漏了另一个并行任务，查了半天。排查技巧依旧是用stream_mode="updates"看实际输出事件，哪个节点被选中一目了然。
5.4 编译后还想改图
graph.compile()之后，返回的对象是CompiledStateGraph。很多人这时候想加节点、加边，直接compiled_graph.add_edge(...)，结果报错。
图一旦编译就是只读的。要改结构，必须回到StateGraph对象，修改完重新compile()。
我一开始也犯这个错，后来形成习惯：整个逻辑封装成一个build_graph()函数，每次返回新编译的图。调试时想改路由就直接改函数再重进，省去在REPL里保留StateGraph实例的负担。
5.5 checkpointer与分支暂停恢复
如果你给图加了checkpointer（内存版或持久化版），分支执行可以被“中断”和“恢复”。比如执行到某个节点后，等着人工审批，之后再从断点继续。
这里有个隐藏坑：从断点恢复时，LangGraph会重放历史事件。如果分支路由依赖外部随机性或时间，重放后可能走了不同的路径，导致状态和预期不一致。
解决办法是：路由函数保持纯函数，只依赖State里的字段，不做随机采样，不读外部时钟。让路由可重放且结果稳定，是分支执行可恢复的前提。我见过有人把random.random()写进路由，一恢复就走到别的分支去，整个流程当场变脸。
6. 我对分支执行的一些补充经验
调试LangGraph分支最顺手的工具其实不是IDE断点，而是stream输出。它能把每个节点的事件序列吐出来，你能看到决策点在什么时刻选择了下一条路。建议刚开始时，每个节点的返回字典都带一个可辨识的字段，比如节点名、意图值，这样事件流里一眼看出走到哪了。
还有个习惯：条件边的路由函数不要写复杂业务逻辑，尽量只做一个“翻译动作”。业务判断放节点里，节点把判断结果写进State，路由函数只是return state["xxx"]。好处是，以后想换判断逻辑，只改节点，不动图结构；想调整走向，只改path_map，不动业务代码。图和逻辑解耦，分支策略才能灵活调整。
如果你是从LangChain迁移过来，第一周很容易把“节点”理解成“链上的步骤”，然后下意识想在节点里写if/else控制下一个节点。这个惯性要刻意改掉：节点只负责加工State，通往哪个节点，是条件边的事。把这个心法刻进脑子，LangGraph的分支执行逻辑才算真正上手。
