LangGraph Node与Edge权威学习指南

一.Node基础笔记

在构建LangGraph应用时，Node（节点）和Edge（边）是最核心的两个元素。Node负责承载具体的业务逻辑，而Edge则决定了节点间的执行顺序和路由方式。接下来的内容会从实战角度，把这些基础概念讲清楚。

1.给节点添加retry_policy和cache_policy

在实际开发中，节点往往需要处理网络请求、数据库操作这类不可靠的任务。LangGraph为此提供了两个关键能力：重试策略（retry_policy）和缓存策略（cache_policy）。 先看一个完整的示例，它会展示如何定义带参数节点，并为其配置重试和缓存。

from functools import partial
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import RetryPolicy
from requests import RequestException, Timeout
from langgraph.types import CachePolicy

# 定义状态
class GraphState(TypedDict):
    process_data: dict

# 定义一个节点，入参为state
def input_node(state: GraphState) -> GraphState:
    print(f'input_node收到的初始值:{state}')
    return {"process_data": {"input": "input_value"}}

# 定义带参数的node节点
def process_node(state: dict, param1: int, param2: str) -> dict:
    print("-"*100)
    print(state, param1, param2)
    print("-"*100)
    return {"process_data": {"process": "process_value"}}

# 重试策略,add_node方法时可选
retry_policy = RetryPolicy(
    max_attempts=3, # 最大重试次数
    initial_interval=1, # 初始间隔
    jitter=True, # 抖动（添加随机性避免重试风暴）
    backoff_factor=2, # 退避乘数（每次重试间隔时间的增长倍数）
    retry_on=[RequestException, Timeout] # 只重试这些异常
)

stateGraph = StateGraph(GraphState)

# 添加inpu节点
stateGraph.add_node("input", input_node)

# 给process_node节点绑定参数
process_with_params = partial(process_node, param1=100, param2="test")

# 添加带参数的node节点
stateGraph.add_node(
    node="process",
    action= process_with_params,
    retry_policy=retry_policy,
    cache_policy= CachePolicy(ttl=8)
)

# 定义节点之间的执行顺序 edges
# 设置节点间的依赖关系，形成执行流程图
stateGraph.add_edge(START, "input")
stateGraph.add_edge("input", "process")
stateGraph.add_edge("process", END)

# 编译图构建器生成计算图
graph = stateGraph.compile()

# # 打印图的边和节点信息
print(stateGraph.edges)
print(stateGraph.nodes)

# 打印图的可视化结构
print(graph.get_graph().print_ascii())
print()

# 定义一个初始状态字典，包含键值对"x": 5
initial_state={"process_data": 5}

# 调用graph对象的invoke方法，传入初始状态，执行图计算流程
result= graph.invoke(initial_state)
print(f"最后的结果是:{result}")

需要注意的是，RetryPolicy的jitter（抖动）参数是个很实用的设计——它可以有效避免多个节点在失败后同时重试导致的“重试风暴”。而cache_policy配合ttl（缓存过期时间）参数，则能避免重复执行相同输入的节点。

二.Edge基础笔记

Edge是连接节点的桥梁。它定义了执行流程中有哪些路径可以走，以及如何走。

1.普通边

普通边是最直接、最无脑的路径选择——它表示从一个节点执行完毕后，无条件地跳转到另一个指定节点。

"""LangGraph普通边演示
普通边是直接连接两个节点的边，表示无条件地从一个节点跳转到另一个节点。
"""
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

# 定义状态
class AtguiguState(TypedDict):
    value: int
    step: str

# 定义节点函数
def node_a(state: AtguiguState) -> dict:
    """节点A"""
    print("执行节点A")
    return {"value": state["value"] + 1, "step": "A执行完毕"}

def node_b(state: AtguiguState) -> dict:
    """节点B"""
    print("执行节点B")
    return {"value": state["value"] * 2, "step": "B执行完毕"}

def node_c(state: AtguiguState) -> dict:
    """节点C"""
    print("执行节点C")
    return {"value": state["value"] - 1, "step": "C执行完毕"}

def main():
    """演示普通边"""
    print("=== 普通边演示 ===")
    # 创建图
    builder = StateGraph(AtguiguState)
    # 添加节点
    builder.add_node("node_a", node_a)
    builder.add_node("node_b", node_b)
    builder.add_node("node_c", node_c)
    # 添加普通边
    builder.add_edge(START, "node_a") # 从开始到A
    builder.add_edge("node_a", "node_b") # 从A到B
    builder.add_edge("node_b", "node_c") # 从B到C
    builder.add_edge("node_c", END) # 从C到结束
    # 编译图
    app = builder.compile()
    # 执行图
    result = app.invoke({"value": 1})
    print(f"执行结果: {result}n")
    # 打印图的边和节点信息
    print(builder.edges)
    #print(builder.nodes)
    # 打印图的ascii可视化结构
    print(app.get_graph().print_ascii())
    print("=================================")
    print()
    # 打印图的可视化结构，生成更加美观的Mermaid 代码，通过processon 编辑器查看
    print(app.get_graph().draw_mermaid())

if __name__ == "__main__":
    main()

执行结果：

看这段代码，流程就是典型的线性执行：A→B→C→END。每个节点依次计算，中间没有任何分支或选择。

2.条件边

实际业务中，更多的是需要根据条件决定下一步该走哪条路。条件边（conditional edges）就是用来干这个的。

"""
LangGraph 条件边分支流程控制
语句分支路由（Router → Weather / Chat）
使用langgraph构建了一个状态图，根据输入数值的奇偶性执行不同节点。
check_x接收并传递状态，is_even判断奇偶，handle_even和handle_odd分别处理偶数和奇数情况，最终输出结果。
"""
from typing import Optional
from langgraph.constants import START, END
from langgraph.graph import StateGraph
from loguru import logger
from pydantic import BaseModel

class MyState(BaseModel):
    """
    定义状态模型，用于在图节点之间传递数据
    Attributes:
        x (int): 输入的整数
        result (Optional[str]): 处理结果，可为"even"或"odd"
    """
    x: int
    result: Optional[str] = None

# 检查输入状态的节点函数
def check_x(state: MyState) -> MyState:
    """
    检查输入状态的节点函数
    Args:
        state (MyState): 包含输入数据的状态对象
    Returns:
        MyState: 返回原始状态对象，未做修改
    """
    logger.info(f"[check_x] Received state: {state}")
    return state

# 判断状态中x值是否为偶数的条件函数
def is_even(state: MyState) -> bool:
    """
    判断状态中x值是否为偶数的条件函数
    Args:
        state (MyState): 包含待判断数值的状态对象
    Returns:
        bool: 如果x是偶数返回True，否则返回False
    """
    logger.info(f"[is_even] :{state.x % 2 == 0} ")
    return state.x % 2 == 0

# 处理偶数情况的节点函数
def handle_even(state: MyState) -> MyState:
    """
    处理偶数情况的节点函数
    Args:
        state (MyState): 包含偶数输入的状态对象
    Returns:
        MyState: 返回更新后的状态对象，result设置为"even"
    """
    logger.info("[handle_even] x 是偶数")
    return MyState(x=state.x, result="even")

#处理奇数情况的节点函数
def handle_odd(state: MyState) -> MyState:
    """
    处理奇数情况的节点函数
    Args:
        state (MyState): 包含奇数输入的状态对象
    Returns:
        MyState: 返回更新后的状态对象，result设置为"odd"
    """
    logger.info("[handle_odd] x 是奇数")
    return MyState(x=state.x, result="odd")

builder = StateGraph(MyState)
# 添加节点
builder.add_node("check_x", check_x)
builder.add_node("handle_even", handle_even)
builder.add_node("handle_odd", handle_odd)
# 添加条件边，根据is_even函数的返回值决定流向哪个节点
builder.add_conditional_edges("check_x", is_even, {
    True: "handle_even",
    False: "handle_odd"
})
# 添加起始边，从START节点流向check_x节点
builder.add_edge(START, "check_x")
# 添加结束边，从处理节点流向END节点
builder.add_edge("handle_even", END)
builder.add_edge("handle_odd", END)
# 编译图结构
graph = builder.compile()
# 打印图的可视化结构
print(graph.get_graph().print_ascii())

# 测试用例：输入偶数4
logger.info("输入 x=4（偶数）")
graph.invoke(MyState(x=3))
# # 测试用例：输入奇数3
# logger.info("输入 x=3（奇数）")
# graph.invoke(MyState(x=3))

执行结果：

这段代码的核心在于 `add_conditional_edges` 方法：它接受一个判断函数（`is_even`）和一个映射字典，把True/False的返回值映射到对应的目标节点。这种方式可以用来实现类似if/else的分支逻辑。

3.多个条件边

当条件不止两个，而是多个分支时怎么办？答案是：可以定义返回字符串的路由函数，然后通过路径映射指到不同节点。

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, List, Annotated

# 定义状态
class AtguiguState(TypedDict):
    x: int

def addition1(state):
    """
    执行加法运算的节点函数
    参数:
        state (dict): 包含输入数据的状态字典，必须包含键"x"
    返回:
        dict: 返回更新后的状态字典，其中"x"的值增加1
    """
    print(f'加法节点addition1收到的初始值:{state}')
    return {"x": state["x"] + 1}

def addition2(state):
    print(f'加法节点addition2收到的初始值:{state}')
    return {"x": state["x"] + 2}

def addition3(state):
    print(f'加法节点addition3收到的初始值:{state}')
    return {"x": state["x"] + 3}

def route_by_sentiment(state: AtguiguState) -> str:
    # 路由逻辑...返回最终的条件
    flag = state["x"]
    if flag == 1:
        return "condition_1"
    elif flag == 2:
        return "condition_2"
    else:
        return "condition_3"

graph = StateGraph(AtguiguState)
graph.add_node("node1", addition1)
graph.add_node("node2", addition2)
graph.add_node("node3", addition3)
# 添加路由函数，参数：当前节点，路由函数，路由函数返回的条件与node的映射
graph.add_conditional_edges(
    source=START,
    path= route_by_sentiment,
    path_map={
        "condition_1": "node1",
        "condition_2": "node2",
        "condition_3": "node3"
    }
)
# 所有处理节点都连接到END
graph.add_edge("node1", END)
graph.add_edge("node2", END)
graph.add_edge("node3", END)

app = graph.compile()
# 定义一个初始状态字典，包含键值对"x": 具体数字
initial_state ={"x": 3}
# 调用graph对象的invoke方法，传入初始状态，执行图计算流程
result= app.invoke(initial_state)
print(f"最后的结果是:{result}")
# 打印图的边和节点信息
#print(graph.edges)
#print(graph.nodes)
# 打印图的ascii可视化结构
print(app.get_graph().print_ascii())
print("=================================")
print()
# 打印图的可视化结构，生成更加美观的Mermaid 代码，通过processon 编辑器查看
print(app.get_graph().draw_mermaid())

执行结果：

这个示例中，`route_by_sentiment` 函数根据 `x` 的值返回一个字符串标记（`condition_1`、`condition_2` 或 `condition_3`），然后 `path_map` 负责把每个标记映射到具体的处理节点。这种模式非常实用，比如可以用于事件分类、优先级路由等场景。 最后，通过 `print_ascii()` 或 `draw_mermaid()` 方法，可以直观地看到整个图的结构，便于调试和理解。这对排查节点执行链路的问题特别有帮助。

来源：互联网