深入浅出LangGraph AI Agent智能体开发教程(七)—构建条件分支图和循环图
本期内容分享了利用LangGraph底层API构建条件分支图和循环图的基本技巧,学会本期内容大家将掌握了利用LangGraph构造复杂图的能力。
前言
本系列前六篇文章:
- 深入浅出LangGraph AI Agent智能体开发教程(一)—全面认识LangGraph
- 深入浅出LangGraph AI Agent智能体开发教程(二)—LangGraph预构建图API快速创建Agent
- 深入浅出LangGraph AI Agent智能体开发教程(三)—LangGraph工具调用实战
- 深入浅出LangGraph AI Agent智能体开发教程(四)—LangGraph全生态开发工具使用与智能体部署
- 深入浅出LangGraph AI Agent智能体开发教程(五)—LangGraph 数据分析助手智能体项目实战
- 深入浅出LangGraph AI Agent智能体开发教程(六)—LangGraph 底层API入门
上篇分享我们入门了LangGraph底层API构建图的基本流程,掌握了构建图三要素“点、边、状态”,并实战构建了一个线性的图结构。不过我们在日常使用LangGraph搭建智能体往往不会用到线型图,反而经常会用到条件分支图和循环结构图,本期分享我们就来学习如何使用LangGraph底层API搭建这两种复杂的图模式(线型图直接用LangChain,还用LangGraph干嘛~)
本系列分享是笔者结合自己学习工作中使用LangChain&LangGraph经验倾心编写的,力求帮助大家体系化快速掌握LangChain&LangGraph AI Agent智能体开发的技能!笔者的LangChain系列教程暂时更完,后面也会实时补充工作实践中的额外知识点,建议大家在学习LangGraph分享前先学习LangChain的基本使用技巧。大家感兴趣可以关注笔者CSDN账号和系列专栏。更可关注笔者同名微信公众号: 大模型真好玩, 每期分享涉及的代码均可在公众号私信: LangChain智能体开发获得。
一、LangGraph构建条件判断图
本示例我们将基于LangGraph 底层 API构建一个简单的有状态条件分支图,用于向大家演示如何使用结构化状态(通过Pydantic
模型定义)在多步骤决策流程中进行状态传递与条件控制。
- 定义名为
MyState
的Pydantic
结构化状态并构建图。MyState
用于描述图中每个节点共享的上下文状态信息。该状态包含两个字段:x
表示输入数值,result
表示最终处理结果。result
可以是字符串类型也可以是None。
from typing import Optional
from pydantic import BaseModel
from langgraph.graph import StateGraph, START, END
# 定义结构化状态
class MyState(BaseModel):
x: int
result: Optional[str] = None
# 构建图
builder = StateGraph(MyState)
- 定义节点处理逻辑并将其加入到图结构中。图的结构由三个核心节点组成:
check_x
节点接收初始状态并进行输出转发,不修改任何字段。handle_even
处理偶数情况,标记result=even
。hancle_odd
处理奇数情况,标记result=odd
。
# 定义各节点处理逻辑(接受 MyState,返回 MyState)
def check_x(state: MyState) -> MyState:
print(f"[check_x] Received state: {state}")
return state
def is_even(state: MyState) -> bool:
return state.x % 2 == 0
def handle_even(state: MyState) -> MyState:
print("[handle_even] x 是偶数")
return MyState(x=state.x, result="even")
def handle_odd(state: MyState) -> MyState:
print("[handle_odd] x 是奇数")
return MyState(x=state.x, result="odd")
builder.add_node("check_x", check_x)
builder.add_node("handle_even", handle_even)
builder.add_node("handle_odd", handle_odd)
- 定义边处理逻辑,引入基于
is_even
判断函数的条件分支控制逻辑。LangGraph提供的add_conditional_edges
方法允许我们根据状态中的值跳转至不同的执行路径,从而实现类似于传统编程语言的if-else
分支控制。
def is_even(state: MyState) -> bool:
return state.x % 2 == 0
# 添加条件分支
builder.add_conditional_edges("check_x", is_even, {
True: "handle_even",
False: "handle_odd"
})
# 衔接起始和结束
builder.add_edge(START, "check_x")
builder.add_edge("handle_even", END)
builder.add_edge("handle_odd", END)
构建好的图结构如下:
- 编译图并进行测试, 可以看到图输出正确结果。这正是LangGraph优势之一:在状态流转的基础上,灵活地控制流程分支,并通过结构化模型传递和对上下文信息管理能力,为构建复杂控制流智能体打下基础。
# 编译图
graph = builder.compile()
# 执行测试
print("\n测试 x=4(偶数)")
graph.invoke(MyState(x=4))
print("\n测试 x=3(奇数)")
graph.invoke(MyState(x=3))
二、LangGraph构建条件循环图
LangGraph不仅可以构建条件判断图,还可以构建循环图。接下来我们分享LangGraph构建条件循环图的示例,该示例实现如下功能,对初始状态中的x
值不断+1直到x
值大于10。图结构如下:
- 定义结构数据并构建图
from pydantic import BaseModel
from langgraph.graph import StateGraph, START, END
# 定义结构化状态模型
class LoopState(BaseModel):
x: int
# 构建图
builder = StateGraph(LoopState)
- 定义相关节点,该节点可以将
state
中的x
值加1
# 定义节点逻辑
def increment(state: LoopState) -> LoopState:
print(f"[increment] 当前 x = {state.x}")
return LoopState(x=state.x + 1)
builder.add_node("increment", increment)
- 定义边,同样我们需要使用
add_conditional_edges
方法进行判断,如果x
的值小于10,我们连接一条increment
节点自己到自己的通路,循环+1操作。
def is_done(state: LoopState) -> bool:
return state.x > 10
# 设置循环控制:is_done 为 True 则结束,否则继续
builder.add_conditional_edges("increment", is_done, {
True: END,
False: "increment"
})
builder.add_edge(START, "increment")
- 编译图并进行测试, 可以看到输入的6一直加1直到10才退出图执行,符合预期。
print("\n 执行循环直到 x > 10")
final_state = graph.invoke(LoopState(x=6))
print(f"[最终结果] -> x = {final_state['x']}")
三、LangGraph构建条件判断与条件循环复合图
本期分享的最后我们将所学的条件判断与条件循环图相结合解决如下需求:对状态中的x
判断是偶数还是奇数,如果是偶数将进入循环将x
的值加1,直到x
的值为奇数为止。根据该需求构建的图结构如下:
具体代码如下,相信大家经过以上的学习已经了解了图构建的基本技巧,理解下面代码不成问题,这里就不加赘述了。
from pydantic import BaseModel
from typing import Optional
from langgraph.graph import StateGraph, START, END
# 定义状态模型
class BranchLoopState(BaseModel):
x: int
done: Optional[bool] = False
# 定义各节点逻辑
def check_x(state: BranchLoopState) -> BranchLoopState:
print(f"[check_x] 当前 x = {state.x}")
return state
def is_even(state: BranchLoopState) -> bool:
return state.x % 2 == 0
def increment(state: BranchLoopState) -> BranchLoopState:
print(f"[increment] x 是偶数,执行 +1 → {state.x + 1}")
return BranchLoopState(x=state.x + 1)
def done(state: BranchLoopState) -> BranchLoopState:
print(f"[done] x 是奇数,流程结束")
return BranchLoopState(x=state.x, done=True)
# 构建图
builder = StateGraph(BranchLoopState)
builder.add_node("check_x", check_x)
builder.add_node("increment", increment)
builder.add_node("done_node", done)
builder.add_conditional_edges("check_x", is_even, {
True: "increment",
False: "done_node"
})
# 循环逻辑:偶数 → increment → check_x
builder.add_edge("increment", "check_x")
# 起始与终点
builder.add_edge(START, "check_x")
builder.add_edge("done_node", END)
graph = builder.compile()
# 测试执行
print("\n初始 x=6(偶数,进入循环)")
final_state1 = graph.invoke(BranchLoopState(x=6))
print("[最终结果1] ->", final_state1)
print("\n初始 x=3(奇数,直接 done)")
final_state2 = graph.invoke(BranchLoopState(x=3))
print("[最终结果2] ->", final_state2)
【特别注意: LangGraph 会把所有节点名、状态字段、通道名放在一个命名空间中处理,为了避免歧义,它会严格检查有没有冲突,最保险的做法是:节点名不要与字段名重复,既如果使用 state.result =“done”,也不要有“result”这个节点。】
四、 总结
本期内容分享了利用LangGraph底层API构建条件分支图和循环图的基本技巧,学会本期内容大家就掌握了利用LangGraph构造复杂图的能力,下期分享我们将利用LangGraph底层API构建ReACT图并接入工具,从而复现create_react_agent
方法,大家敬请期待~
本系列分享预计会有20节左右的规模,保证大家看完一定能够掌握LangChain&LangGraph的开发能力,大家感兴趣可关注笔者CSDN账号和专栏,更可关注笔者的同名微信公众号:大模型真好玩, 本系列分享的全部代码均可在微信公众号私信笔者: LangChain智能体开发 免费获得。
更多推荐
所有评论(0)