LangGraphadd_conditional_edges完整详解
一、基础语法
python
运行
workflow.add_conditional_edges( source: str, # 起始节点名 path_fn: Callable, # 路由判断函数 path_map: Dict[str, str] # 返回值 → 目标节点映射 )作用:条件分支跳转,同一个节点执行完后,根据函数返回值走不同分支,类似 if/else。
二、参数拆解
source条件边的起点节点,字符串,如
"analyze"。path_fn(路由函数 router)接收完整 state,返回一个字符串(分支标识):
python
运行
def router(state): if state["is_quick"]: return "quick" else: return "detailed"path_map 映射字典
{返回值: 目标节点}- key:router 函数返回的字符串
- value:要跳转的节点名称 所有 router 可能返回的值,都必须在字典里定义,否则报错。
你上一段代码补全示例
python
运行
def router(state): if state["quick_mode"]: return "quick" return "detailed" workflow.add_conditional_edges( "analyze", router, { "quick": "quick_node", "detailed": "detailed_node" } )三、两种简写模式
模式 1:显式映射(推荐,可读性高)
python
运行
workflow.add_conditional_edges( "start", branch_func, {"a": "node_a", "b": "node_b"} )模式 2:隐式映射(返回值 = 节点名,省略字典)
如果路由返回的字符串刚好等于目标节点名,可以不传 path_map:
python
运行
workflow.add_conditional_edges("analyze", router) # 等价于 {"quick":"quick", "detailed":"detailed"}对应你最初的代码场景,最简写法:
python
运行
workflow.add_conditional_edges("analyze", router)四、完整可运行最小示例
python
运行
from langgraph.graph import StateGraph from typing import TypedDict # 1. 定义状态 class State(TypedDict): content: str quick_mode: bool # 2. 节点函数 def analyze(state: State): return state def quick(state: State): print("执行快速流程") return state def detailed(state: State): print("执行详细流程") return state # 3. 路由函数 def route_analyze(state: State): if state["quick_mode"]: return "quick" return "detailed" # 4. 构建图 builder = StateGraph(State) builder.add_node("analyze", analyze) builder.add_node("quick", quick) builder.add_node("detailed", detailed) # 条件分支核心代码 builder.add_conditional_edges( source="analyze", path_fn=route_analyze, path_map={ "quick": "quick", "detailed": "detailed" } ) # 入口 builder.set_entry_point("analyze") graph = builder.compile() # 测试 graph.invoke({"content": "test", "quick_mode": True}) graph.invoke({"content": "test", "quick_mode": False})五、常见用法拓展
1. 多分支(3 个及以上分支)
python
运行
def multi_router(state): if state["type"] == "search": return "search" elif state["type"] == "write": return "write" return "review" builder.add_conditional_edges( "plan", multi_router, { "search": "search_node", "write": "write_node", "review": "review_node" } )2. 条件边指向结束节点 END
python
运行
from langgraph.graph import END builder.add_conditional_edges( "check", exit_router, {"finish": END, "continue": "loop_node"} )3. 循环逻辑(自环条件分支)
适合 Agent 多轮思考,判断是否还要继续调用工具:
python
运行
def tool_router(state): if state["need_tool"]: return "tool" return "end" builder.add_conditional_edges( "agent", tool_router, {"tool": "tool", "end": END} ) # tool执行完再回到agent builder.add_edge("tool", "agent")六、常见踩坑
- router 返回值不在 path_map 的 key 里 → KeyError所有可能返回的字符串必须全部写入映射字典。
- 节点名写错path_map 的 value 必须是已经
add_node注册过的节点。 - 路由函数必须接收完整 state 参数,不能无参。
- 条件边只能有一个起点,一个
add_conditional_edges对应一个 source。