LangGraph图中条件边的示例简析
先来看一个LangGraph图中条件边的用法示例:
# -*- coding: utf-8 -*-
"""
Created on Fri Jul 25 09:46:52 2025
@author: liguo
"""
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
# 定义状态结构
class State(TypedDict):
value: int
def start(state: State) -> State:
print("启动节点")
return {"value": 5}
def process_a(state: State) -> State:
print("执行路径A")
return state
def process_b(state: State) -> State:
print("执行路径B")
return state
def decide_next(state: State) -> str:
if state["value"] > 3:
return "process_a"
return "process_b"
# 创建 StateGraph 实例
workflow = StateGraph(State)
# 添加节点
workflow.add_node("start", start)
workflow.add_node("process_a", process_a)
workflow.add_node("process_b", process_b)
# 设置入口
workflow.set_entry_point("start")
# 添加条件边(注意:现在直接返回目标节点名)
workflow.add_conditional_edges(
"start",
decide_next,
{ # 可选:显式映射,但如果你的 decide_next 直接返回节点名,可以省略这个字典
"process_a": "process_a",
"process_b": "process_b"
}
)
# 添加普通边
workflow.add_edge("process_a", "process_b")
workflow.add_edge("process_b", END) # 设置结束点
# 编译并运行
app = workflow.compile()
result=app.invoke({"value": "your_output_value"})
print(result)
代码执行结果
C:\Users\xiayu\miniconda3\envs\langchain03\python.exe C:\Users\xiayu\PyCharmMiscProject\AI-Agent-Dev-Practices-Code\第8章代码\8.19-条件边示例.py
启动节点
执行路径A
执行路径B
{'value': 5}
Process finished with exit code 0
示例代码中条件边简析
条件边是实现分支逻辑的关键部分。
代码回顾:
workflow.add_conditional_edges(
"start", # 来源节点
decide_next, # 决策函数
{ # 映射字典
"process_a": "process_a",
"process_b": "process_b"
}
)
条件边的三个组成部分
1. 来源节点 ("start")
- 表示从哪个节点出发做决策
- 执行完 start节点后,会调用决策函数
2. 决策函数 (decide_next)
这是最核心的部分:
def decide_next(state: State) -> str:
if state["value"] > 3:
return "process_a" # 返回下一个节点的名称
return "process_b" # 返回下一个节点的名称
- 接收当前状态作为输入
- 根据业务逻辑返回下一个节点的名称
- 返回值必须是映射字典中的键
3. 映射字典
{
"process_a": "process_a", # 键: 决策函数的返回值
"process_b": "process_b" # 值: 实际的目标节点名
}
- 键:决策函数可能返回的字符串
- 值:对应的下一个节点名称
- 这个映射是可选的,如果省略,决策函数需要直接返回节点名
执行流程
graph LR
A[start节点] --> B{条件判断 decide_next}
B -- "返回'process_a'" --> C[process_a]
B -- "返回'process_b'" --> D[process_b]
执行 start节点
调用 decide_next(state)函数
根据 state["value"]的值决定:
如果 > 3:返回 "process_a"
如果 ≤ 3:返回 "process_b"
根据返回值选择下一个节点
条件边的核心作用
1. 动态路由
根据运行时数据决定执行路径,实现类似 if-else的逻辑。
2. 决策与执行分离
决策逻辑封装在 decide_next函数中,节点只负责执行操作。
3. 可配置的映射
通过映射字典,可以灵活改变决策结果和目标节点的对应关系。
简化写法(省略映射字典)
如果决策函数直接返回节点名,可以省略映射字典:
# 决策函数直接返回节点名
def decide_next_simple(state: State) -> str:
if state["value"] > 3:
return "process_a" # 直接返回节点名
return "process_b" # 直接返回节点名
# 条件边简化写法
workflow.add_conditional_edges(
"start",
decide_next_simple
# 省略映射字典,决策函数必须直接返回有效的节点名
)
条件边总结
条件边是 LangGraph 中实现工作流分支的核心机制:
- 让工作流能够根据运行时状态做出决策
- 将决策逻辑和执行逻辑分离
- 支持复杂的业务流程控制
- 是构建智能代理、决策系统等应用的基础
在你的代码中,条件边实现了根据输入值的大小选择不同处理路径的逻辑,这是构建灵活工作流的关键。


更多推荐


所有评论(0)