LangGraph控制流：条件边、循环与并行（二）

上篇文章中介绍的图是“一条直线走到底”，但真实业务往往需要分支、循环和并行。这一篇涵盖了 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. 路由函数可以返回 END

def 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") # 工具执行完 → 回到 agent

3. 循环流程图

这个循环就是 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 与普通边的区别

4. 并行分支的结果合并

多个分支同时更新 State 时，必须用 reducer 避免冲突：

class MyState(TypedDict): results: Annotated[list, add] # 多个分支的返回会追加到同一个列表

如果不用add，后执行的分支会覆盖先执行的结果。

四、三种控制流的组合

真实业务往往是三者组合使用：

• 条件边：意图识别分流

• 并行：多个子任务同时处理

• 循环：结果不满意时回到起点重试

注意