节点

普通节点

节点通常是 Python 函数（同步或异步）

  • 第一个参数是状态
  • 第二个参数（可选）是配置
from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph

builder = StateGraph(dict)


def my_node(state: dict, config: RunnableConfig):
    print("In node: ", config["configurable"]["user_id"])
    return {"results": f"Hello, {state['input']}!"}


# 第二个参数是可选的
def my_other_node(state: dict):
    return state


builder.add_node("my_node", my_node)
builder.add_node("other_node", my_other_node)
...

如果在不指定名称的情况下将节点添加到图中，它将被赋予一个与函数名相同的默认名称。

builder.add_node(my_node)

START 节点

START 节点是一个特殊节点，代表将用户输入发送到图的节点。引用此节点的主要目的是确定应该首先调用哪些节点。

from langgraph.graph import START

graph.add_edge(START, "node_a")

END 节点

END 节点是一个特殊节点，代表终端节点。当你想表示哪些边在完成后没有后续操作时，可以引用此节点。

from langgraph.graph import END

graph.add_edge("node_a", END)

节点重试

配置节点的重试策略，可以在 add_node 定义 retry 参数，并传入 RetryPolicy 对象

from langgraph.pregel import RetryPolicy

builder.add_node(
    "my_node",
    my_node,
    retry=RetryPolicy(max_attempts=5)
)

RetryPolicy 可传入参数

key类型描述例子
initial_intervalFloat初始重试间隔时间（秒）initial_interval=0.5
backoff_factorFloat重试间隔时间的指数退避因子（每次重试间隔 = 上次间隔 × backoff_factor）backoff_factor=2.0
max_intervalFloat最大重试间隔时间（秒）max_interval=128.0
max_attemptsInteger最大重试次数max_attempts=3
jitterBoolean是否在重试间隔中添加随机抖动（避免多个任务同时重试）jitter=True
retry_onFunction判断是否重试的函数（默认函数会捕获所有异常）retry_on=custom_function
  1. jitter 的随机抖动范围是 [0, current_interval]，用于避免惊群效应。
  2. retry_on 默认会重试所有异常，若需自定义逻辑可传入函数（如 retry_on=lambda e: isinstance(e, TimeoutError)）。
  3. 退避间隔计算公式：min(initial_interval * (backoff_factor ** n), max_interval)，其中 n 是已重试次数。