LangGraph开发-条件分支/循环图实战

发布时间：2026-04-08 21:50:04编辑：123阅读（15）

借助Pydantic对象创建图

Pydantic是一个用于创建"数据模型"的Python库，它可以自动校验数据类型，并将字典数据转换为结构化对象。字典加了一个"类型安全+自动验证"的外壳，是现代Python项目中最主流的"数据结构定义工具"。

代码如下：

from langgraph.graph import StateGraph
from langgraph.graph import START,END
from pydantic import BaseModel

class MyState(BaseModel):
    x:int

def addition(state:MyState):
    # 接收到的是初始状态
    print(f"init_state: {state}")
    return MyState(x=state.x + 1)

def subtraction(state:MyState):
    # 接收到的是上一个节点的状态
    print(f"addition_state: {state}")
    return MyState(x=state.x - 2)

# 构建图
builder = StateGraph(MyState)

# 向图中添加两个节点
builder.add_node('addition', addition)
builder.add_node('subtraction', subtraction)

# 构建节点之间的边
builder.add_edge(START, 'addition')
builder.add_edge('addition', 'subtraction')
builder.add_edge('subtraction', END)

# 执行图的编译，需要通过调用compile()方法将这些设置编译成一个可执行的图。
graph = builder.compile()

# 执行图
initial_state = MyState(x=10)
value = graph.invoke(initial_state)
print(f"[最终结果]:{value}")

运行结果:

init_state: x=10

addition_state: x=11

[最终结果]:{'x': 9}

需要注意的是，无论输入端输入什么结构的对象，最终图计算返回的结果是一个字典类型对象。




创建条件分支图

基于LangGraph框架构建了一个简单的有状态条件分支图，用于演示如何使用结构化状态(通过Pydantic定义)在多步骤的决策流程中进行状态传递与条件控制。

from pydantic import BaseModel
from langgraph.graph import StateGraph, START, END

# 定义结构化状态
class MyState(BaseModel):
    x: int
    result: str | None = None

# 定义各节点处理逻辑(接受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(f"[handle_even] x 是偶数")
    return MyState(x=state.x, result="偶数")

def handle_odd(state: MyState)-> MyState:
    print(f"[handle_odd] x 是奇数")
    return MyState(x=state.x, result="奇数")

# 构建图
builder = StateGraph(MyState)

builder.add_node("check_x", check_x)
builder.add_node("handle_even", handle_even)
builder.add_node("handle_odd", handle_odd)

# 添加条件分支
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)

# 编译图
graph = builder.compile()

# 执行测试
value1 = graph.invoke(MyState(x=4))
print(value1)

value2 = graph.invoke(MyState(x=3))
print(value2)

with open("graph1.png", "wb") as f:
    f.write(graph.get_graph(xray=True).draw_mermaid_png())

运行结果：

[check_x] Received state: x=4 result=None

[handle_even] x 是偶数

{'x': 4, 'result': '偶数'}

[check_x] Received state: x=3 result=None

[handle_odd] x 是奇数

{'x': 3, 'result': '奇数'}




结构图:









创建条件循环图

from pydantic import BaseModel
from langgraph.graph import StateGraph,START,END

# 定义结构化状态图
class MyState(BaseModel):
    x: int

# 定义节点逻辑
def increment(state: MyState)-> MyState:
    print(f"[increment] 当前 x = {state.x}")
    return MyState(x=state.x + 1)

def is_done(state: MyState)-> bool:
    return state.x > 10

# 构建图
builder = StateGraph(MyState)
builder.add_node("increment", increment)

# 设置循环控制：is_done 为True则结束，否则继续
builder.add_conditional_edges("increment", is_done,
        {True:END, False:"increment"})


builder.add_edge(START, 'increment')

# 编译图
graph = builder.compile()

# 执行测试
value1 = graph.invoke(MyState(x=5))
print(value1)

with open("graph2.png", "wb") as f:
    f.write(graph.get_graph(xray=True).draw_mermaid_png())

执行结果：


[increment] 当前 x = 5

[increment] 当前 x = 6

[increment] 当前 x = 7

[increment] 当前 x = 8

[increment] 当前 x = 9

[increment] 当前 x = 10

{'x': 11}

结构图：









创建条件分支加条件循环图

from pydantic import BaseModel
from typing import Optional
from langgraph.graph import StateGraph,START,END

# 定义状态模型
class MyState(BaseModel):
    x: int
    done: Optional[bool] = False

# 定义各节点逻辑
def check_x(state: MyState) -> MyState:
    print(f"[check_x] 当前 x = {state.x}")
    return state

def is_even(state: MyState) -> bool:
    return state.x % 2 == 0

def increment(state: MyState) -> MyState:
    print(f"[increment] x 是偶数, 执行+1 ->{state.x +1}")
    return MyState(x=state.x+1)

def done(state: MyState) -> MyState:
    print(f"[done] x是奇数，流程结束")
    return MyState(x=state.x, done=True)

# 构建图
builder = StateGraph(MyState)
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()

# 测试执行
value = graph.invoke(MyState(x=6))
print(value)

with open("graph3.png", "wb") as f:
    f.write(graph.get_graph(xray=True).draw_mermaid_png())

执行结果

[check_x] 当前 x = 6

[increment] x 是偶数, 执行+1 ->7

[check_x] 当前 x = 7

[done] x是奇数，流程结束

{'x': 7, 'done': True}

结构图：

关键字：