上篇文章中介绍的图是“一条直线走到底”但真实业务往往需要分支、循环和并行。这一篇涵盖了 LangGraph 怎么实现这三种控制流。一、条件边让图学会“分岔”固定边只能走一条路但很多场景需要根据数据决定下一步用户意图是“退款”还是“咨询”走不同分支工具调用成功还是失败走不同处理内容审核通过还是拒绝走不同出口2. 基本用法条件边由三部分组成源节点 路由函数 映射表。from langgraph.graph import StateGraph, START, END def router(state) - str: 路由函数读取 state返回一个字符串 if state[score] 60: return pass else: return fail builder.add_conditional_edges( check_node, # 从哪个节点出发 router, # 用哪个函数决定去向 { pass: pass_node, # 返回值 → 真实节点 fail: fail_node } )3. 执行流程4. 简化写法如果路由函数直接返回真实节点名映射表可以省略def router(state): return pass_node if state[score] 60 else fail_node builder.add_conditional_edges(check_node, router)5. 路由函数可以返回列表并行分发如果返回的是节点名列表LangGraph 会并行执行这些节点def router(state): return [branch_a, branch_b] # 两个分支并行执行6. 路由函数可以返回 ENDdef router(state): if state[done]: return END # 直接结束 return continue_node二、循环让图“转起来”LangGraph 的图是允许循环的这是它比普通工作流引擎强大的地方。1. 最简单的循环节点连回自己builder.add_node(agent, agent_node) builder.add_edge(agent, agent) # 自己连自己但这样会无限循环必须配合条件边来终止。2. 用条件边实现循环最经典的 Agent 循环模式def should_continue(state) - str: 判断是否继续循环 messages state[messages] last_message messages[-1] # 如果还有工具调用继续 if last_message.tool_calls: return continue # 否则结束 return end builder.add_node(agent, agent_node) builder.add_node(tools, tool_node) builder.add_edge(START, agent) builder.add_conditional_edges( agent, should_continue, { continue: tools, # 有工具调用 → 去执行工具 end: END # 没有 → 结束 } ) builder.add_edge(tools, agent) # 工具执行完 → 回到 agent3. 循环流程图这个循环就是 ReAct 模式的核心Agent 思考 → 调用工具 → 观察结果 → 继续思考直到不再需要工具。4. 防止死循环LangGraph 提供了recursion_limit参数来限制最大迭代次数graph.invoke( {messages: []}, config{recursion_limit: 25} # 最多执行 25 步 )超过限制会抛出GraphRecursionError。三、并行让图“分头行动”1. 静态并行多个边指向多个节点最直接的方式从一个节点出发连多条边到不同节点。builder.add_edge(start_node, branch_a) builder.add_edge(start_node, branch_b) builder.add_edge(start_node, branch_c)三个分支会并行执行都完成后才进入merge_node。2. 用 Send 实现动态并行Map-Reduce静态并行需要提前知道有几个分支。如果分支数量是动态的就要用Send。from langgraph.types import Send def dispatch(state): 根据状态动态生成子任务 return [ Send(process_item, {item: item}) for item in state[items] ] builder.add_conditional_edges(split_node, dispatch, [process_item])执行流程3. Send 与普通边的区别特性普通边Send分支数量固定动态子任务状态共享全局 State每个子任务独立 State典型场景固定多分支Map-Reduce4. 并行分支的结果合并多个分支同时更新 State 时必须用 reducer 避免冲突class MyState(TypedDict): results: Annotated[list, add] # 多个分支的返回会追加到同一个列表如果不用add后执行的分支会覆盖先执行的结果。四、三种控制流的组合真实业务往往是三者组合使用条件边意图识别分流并行多个子任务同时处理循环结果不满意时回到起点重试注意坑点说明解决死循环循环没有终止条件设置recursion_limit并行覆盖多个分支同时写同一个字段用Annotated reducer条件边不生效路由函数返回值与映射表不匹配检查返回值是否在映射表中Send 子任务丢失上下文Send 只传 state不继承 runtime在 state 中显式复制所需字段