使用图 API - Docs by LangChain中文

本指南演示 LangGraph 的 Graph API 基础。它会介绍状态，以及如何组合常见的图结构，例如序列、分支和循环。它还涵盖 LangGraph 的控制功能，包括用于 map-reduce 工作流的 Send API，以及用于将状态更新与跨节点“跳转”结合起来的 Command API。

设置

安装 langgraph：

pip install -U langgraph

设置 LangSmith 以便更好地调试注册 LangSmith，可以快速发现问题并提升 LangGraph 项目的性能。LangSmith 允许你使用 trace 数据来调试、测试和监控使用 LangGraph 构建的 LLM 应用——更多入门信息请阅读文档。

定义和更新状态

这里我们展示如何在 LangGraph 中定义和更新状态。我们将演示：

如何使用状态来定义图的 schema
如何使用 reducers 控制状态更新的处理方式。

定义状态

LangGraph 中的状态可以是 TypedDict、Pydantic 模型或 dataclass。下面我们将使用 TypedDict。有关使用 Pydantic 的详细信息，请参阅使用 Pydantic 模型作为图状态。默认情况下，图会具有相同的输入和输出 schema，而状态决定了该 schema。有关如何定义不同的输入和输出 schema，请参阅定义输入和输出 schema。我们来看一个使用 messages 的简单示例。对于许多 LLM 应用来说，这是一种通用的状态表达方式。更多细节请参阅我们的概念页面。

from langchain.messages import AnyMessage
from typing_extensions import TypedDict

class State(TypedDict):
    messages: list[AnyMessage]
    extra_field: int

这个状态会跟踪一个消息对象列表，以及一个额外的整数字段。

更新状态

我们来构建一个只有单个节点的示例图。我们的节点只是一个 Python 函数，它读取图的状态并对其进行更新。这个函数的第一个参数始终是状态：

from langchain.messages import AIMessage

def node(state: State):
    messages = state["messages"]
    new_message = AIMessage("你好！")
    return {"messages": messages + [new_message], "extra_field": 10}

这个节点只是向消息列表追加一条消息，并填充一个额外字段。

节点应该直接返回对状态的更新，而不是修改状态本身。

接下来我们定义一个包含此节点的简单图。我们使用 StateGraph 来定义一个对该状态进行操作的图。然后使用 add_node 填充图。

from langgraph.graph import StateGraph

builder = StateGraph(State)
builder.add_node(node)
builder.set_entry_point("node")
graph = builder.compile()

LangGraph 提供了内置工具来可视化你的图。我们来查看这个图。有关可视化的详细信息，请参阅可视化你的图。

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

在这个例子中，我们的图只执行单个节点。我们继续做一次简单调用：

from langchain.messages import HumanMessage

result = graph.invoke({"messages": [HumanMessage("你好")]})
result

{'messages': [HumanMessage(content='你好'), AIMessage(content='你好！')], 'extra_field': 10}

请注意：

我们通过更新状态中的单个 key 来启动调用。
我们会在调用结果中收到完整状态。

为方便起见，我们经常通过 pretty-print 查看消息对象的内容：

for message in result["messages"]:
    message.pretty_print()

================================ Human Message ================================

你好
================================== Ai Message ==================================

你好！

使用 reducer 处理状态更新

状态中的每个 key 都可以拥有自己独立的 reducer 函数，用于控制如何应用来自节点的更新。如果没有显式指定 reducer 函数，则默认认为对该 key 的所有更新都会覆盖原值。对于 TypedDict 状态 schema，我们可以通过使用 reducer 函数注解状态中的对应字段来定义 reducer。在前面的示例中，我们的节点通过追加一条消息来更新状态中的 "messages" key。下面，我们为这个 key 添加一个 reducer，使更新会自动追加：

from typing_extensions import Annotated

def add(left, right):
    """也可以从内置的 `operator` 导入 `add`。"""
    return left + right

class State(TypedDict):
    messages: Annotated[list[AnyMessage], add]
    extra_field: int

现在我们的节点可以简化为：

def node(state: State):
    new_message = AIMessage("你好！")
    return {"messages": [new_message], "extra_field": 10}

from langgraph.graph import START

graph = StateGraph(State).add_node(node).add_edge(START, "node").compile()

result = graph.invoke({"messages": [HumanMessage("你好")]})

for message in result["messages"]:
    message.pretty_print()

================================ Human Message ================================

你好
================================== Ai Message ==================================

你好！

MessagesState

在实践中，更新消息列表时还有一些额外注意事项：

我们可能希望更新状态中已有的消息。
我们可能希望接受消息格式的简写形式，例如 OpenAI 格式。

LangGraph 包含一个内置 reducer add_messages，可以处理这些情况：

from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    extra_field: int

def node(state: State):
    new_message = AIMessage("你好！")
    return {"messages": [new_message], "extra_field": 10}

graph = StateGraph(State).add_node(node).set_entry_point("node").compile()

input_message = {"role": "user", "content": "你好"}

result = graph.invoke({"messages": [input_message]})

for message in result["messages"]:
    message.pretty_print()

================================ Human Message ================================

你好
================================== Ai Message ==================================

你好！

对于涉及聊天模型的应用来说，这是一种通用的状态表示方式。为了方便使用，LangGraph 内置了 MessagesState，因此我们可以这样写：

from langgraph.graph import MessagesState

class State(MessagesState):
    extra_field: int

使用 `Overwrite` 绕过 reducer

在某些情况下，你可能希望绕过 reducer，直接覆盖某个状态值。LangGraph 为此提供了 Overwrite 类型。当节点返回用 Overwrite 包裹的值时，会绕过 reducer，并将通道直接设置为该值。当你想重置或替换累积状态，而不是将其与现有值合并时，这会很有用。

from langgraph.graph import StateGraph, START, END
from langgraph.types import Overwrite
from typing_extensions import Annotated, TypedDict
import operator

class State(TypedDict):
    messages: Annotated[list, operator.add]

def add_message(state: State):
    return {"messages": ["第一条消息"]}

def replace_messages(state: State):
    # 绕过 reducer，并替换整个 messages 列表
    return {"messages": Overwrite(["替换消息"])}

builder = StateGraph(State)
builder.add_node("add_message", add_message)
builder.add_node("replace_messages", replace_messages)
builder.add_edge(START, "add_message")
builder.add_edge("add_message", "replace_messages")
builder.add_edge("replace_messages", END)

graph = builder.compile()

result = graph.invoke({"messages": ["初始"]})
print(result["messages"])

['替换消息']

你也可以使用带有特殊 key "__overwrite__" 的 JSON 格式：

def replace_messages(state: State):
    return {"messages": {"__overwrite__": ["替换消息"]}}

当节点并行执行时，在给定的 super-step 中，同一个状态 key 只能有一个节点使用 Overwrite。如果多个节点尝试在同一个 super-step 中覆盖同一个 key，将会抛出 InvalidUpdateError。

定义输入和输出 schema

默认情况下，StateGraph 使用单个 schema 运行，并且期望所有节点都使用该 schema 进行通信。不过，也可以为图定义不同的输入和输出 schema。当指定了不同的 schema 时，节点之间的通信仍会使用内部 schema。输入 schema 确保提供的输入符合预期结构，而输出 schema 会过滤内部数据，只返回根据已定义输出 schema 相关的信息。下面我们将了解如何定义不同的输入和输出 schema。

from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict

# 定义输入的 schema
class InputState(TypedDict):
    question: str

# 定义输出的 schema
class OutputState(TypedDict):
    answer: str

# 定义整体 schema，将输入和输出合并
class OverallState(InputState, OutputState):
    pass

# 定义处理输入并生成答案的节点
def answer_node(state: InputState):
    # 示例答案和一个额外 key
    return {"answer": "再见", "question": state["question"]}

# 构建图，并指定输入和输出 schema
builder = StateGraph(OverallState, input_schema=InputState, output_schema=OutputState)
builder.add_node(answer_node)  # 添加 answer 节点
builder.add_edge(START, "answer_node")  # 定义起始边
builder.add_edge("answer_node", END)  # 定义结束边
graph = builder.compile()  # 编译图

# 使用输入调用图并打印结果
print(graph.invoke({"question": "你好"}))

{'answer': '再见'}

请注意，invoke 的输出只包含输出 schema。

在节点之间传递私有状态

在某些情况下，你可能希望节点交换一些对中间逻辑至关重要、但不需要成为图主 schema 一部分的信息。这些私有数据与图的整体输入/输出无关，并且应该只在特定节点之间共享。下面，我们将创建一个由三个节点（node_1、node_2 和 node_3）组成的顺序图示例，其中私有数据在前两个步骤（node_1 和 node_2）之间传递，而第三个步骤（node_3）只能访问公开的整体状态。

from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict

# 图的整体状态（这是跨节点共享的公共状态）
class OverallState(TypedDict):
    a: str

# node_1 的输出包含不属于整体状态的私有数据
class Node1Output(TypedDict):
    private_data: str

# 私有数据只在 node_1 和 node_2 之间共享
def node_1(state: OverallState) -> Node1Output:
    output = {"private_data": "由 node_1 设置"}
    print(f"进入节点 `node_1`：\n\t输入：{state}。\n\t返回：{output}")
    return output

# 节点 2 的输入只请求 node_1 之后可用的私有数据
class Node2Input(TypedDict):
    private_data: str

def node_2(state: Node2Input) -> OverallState:
    output = {"a": "由 node_2 设置"}
    print(f"进入节点 `node_2`：\n\t输入：{state}。\n\t返回：{output}")
    return output

# 节点 3 只能访问整体状态（不能访问来自 node_1 的私有数据）
def node_3(state: OverallState) -> OverallState:
    output = {"a": "由 node_3 设置"}
    print(f"进入节点 `node_3`：\n\t输入：{state}。\n\t返回：{output}")
    return output

# 按顺序连接节点
# node_2 接收来自 node_1 的私有数据，而
# node_3 看不到这些私有数据。
builder = StateGraph(OverallState).add_sequence([node_1, node_2, node_3])
builder.add_edge(START, "node_1")
graph = builder.compile()

# 使用初始状态调用图
response = graph.invoke(
    {
        "a": "在开始时设置",
    }
)

print()
print(f"图调用的输出：{response}")

进入节点 `node_1`：
    输入：{'a': '在开始时设置'}。
    返回：{'private_data': '由 node_1 设置'}
进入节点 `node_2`：
    输入：{'private_data': '由 node_1 设置'}。
    返回：{'a': '由 node_2 设置'}
进入节点 `node_3`：
    输入：{'a': '由 node_2 设置'}。
    返回：{'a': '由 node_3 设置'}

图调用的输出：{'a': '由 node_3 设置'}

使用 pydantic 模型作为图状态

StateGraph 在初始化时接受一个 state_schema 参数，用于指定图中节点可以访问和更新的状态“形状”。在我们的示例中，state_schema 通常使用 Python 原生的 TypedDict 或 dataclass，但 state_schema 可以是任何类型。这里，我们将了解如何使用 Pydantic BaseModel 作为 state_schema，以便对输入进行运行时校验。

已知限制

目前，图的输出不会是 pydantic 模型实例。
运行时校验只发生在图中第一个节点的输入上，不会发生在后续节点或输出上。
pydantic 的校验错误 trace 不会显示错误发生在哪个节点。
Pydantic 的递归校验可能较慢。对于性能敏感的应用，你可能需要考虑改用 dataclass。

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

# 图的整体状态（这是跨节点共享的公共状态）
class OverallState(BaseModel):
    a: str

def node(state: OverallState):
    return {"a": "再见"}

# 构建状态图
builder = StateGraph(OverallState)
builder.add_node(node)  # node_1 是第一个节点
builder.add_edge(START, "node")  # 使用 node_1 启动图
builder.add_edge("node", END)  # 在 node_1 之后结束图
graph = builder.compile()

# 使用有效输入测试图
graph.invoke({"a": "你好"})

使用无效输入调用图

try:
    graph.invoke({"a": 123})  # 应该是字符串
except Exception as e:
    print("由于 `a` 是整数而不是字符串，因此抛出了异常。")
    print(e)

由于 `a` 是整数而不是字符串，因此抛出了异常。
1 validation error for OverallState
a
  Input should be a valid string [type=string_type, input_value=123, input_type=int]
    For further information visit https://errors.pydantic.dev/2.9/v/string_type

请参阅下方了解 Pydantic 模型状态的其他特性：

序列化行为

使用 Pydantic 模型作为状态 schema 时，理解序列化的工作方式很重要，尤其是在以下场景中：

将 Pydantic 对象作为输入传递
从图中接收输出
使用嵌套的 Pydantic 模型

我们来看这些行为的实际效果。

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

class NestedModel(BaseModel):
    value: str

class ComplexState(BaseModel):
    text: str
    count: int
    nested: NestedModel

def process_node(state: ComplexState):
    # 节点接收经过校验的 Pydantic 对象
    print(f"输入状态类型：{type(state)}")
    print(f"嵌套类型：{type(state.nested)}")
    # 返回一个字典更新
    return {"text": state.text + " 已处理", "count": state.count + 1}

# 构建图
builder = StateGraph(ComplexState)
builder.add_node("process", process_node)
builder.add_edge(START, "process")
builder.add_edge("process", END)
graph = builder.compile()

# 创建一个 Pydantic 实例作为输入
input_state = ComplexState(text="你好", count=0, nested=NestedModel(value="测试"))
print(f"输入对象类型：{type(input_state)}")

# 使用 Pydantic 实例调用图
result = graph.invoke(input_state)
print(f"输出类型：{type(result)}")
print(f"输出内容：{result}")

# 如果需要，转换回 Pydantic 模型
output_model = ComplexState(**result)
print(f"已转换回 Pydantic：{type(output_model)}")

运行时类型强制转换

Pydantic 会对某些数据类型执行运行时类型强制转换。这可能很有帮助，但如果你不了解它，也可能导致意外行为。

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

class CoercionExample(BaseModel):
    # Pydantic 会将数字字符串强制转换为整数
    number: int
    # Pydantic 会将字符串布尔值解析为 bool
    flag: bool

def inspect_node(state: CoercionExample):
    print(f"number: {state.number} (type: {type(state.number)})")
    print(f"flag: {state.flag} (type: {type(state.flag)})")
    return {}

builder = StateGraph(CoercionExample)
builder.add_node("inspect", inspect_node)
builder.add_edge(START, "inspect")
builder.add_edge("inspect", END)
graph = builder.compile()

# 演示会被转换的字符串输入的强制转换
result = graph.invoke({"number": "42", "flag": "true"})

# 这会因校验错误而失败
try:
    graph.invoke({"number": "not-a-number", "flag": "true"})
except Exception as e:
    print(f"\n预期的校验错误：{e}")

使用消息模型

在状态 schema 中使用 LangChain 消息类型时，需要特别注意序列化。通过网络传输消息对象时，你应该使用 AnyMessage（而不是 BaseMessage）以确保正确的序列化/反序列化。

from langgraph.graph import StateGraph, START, END
from pydantic import BaseModel
from langchain.messages import HumanMessage, AIMessage, AnyMessage
from typing import List

class ChatState(BaseModel):
    messages: List[AnyMessage]
    context: str

def add_message(state: ChatState):
    return {"messages": state.messages + [AIMessage(content="你好！")]}

builder = StateGraph(ChatState)
builder.add_node("add_message", add_message)
builder.add_edge(START, "add_message")
builder.add_edge("add_message", END)
graph = builder.compile()

# 创建包含一条消息的输入
initial_state = ChatState(
    messages=[HumanMessage(content="你好")], context="客户支持聊天"
)

result = graph.invoke(initial_state)
print(f"输出：{result}")

# 转换回 Pydantic 模型以查看消息类型
output_model = ChatState(**result)
for i, msg in enumerate(output_model.messages):
    print(f"消息 {i}：{type(msg).__name__} - {msg.content}")

添加运行时配置

有时你希望在调用图时能够配置它。例如，你可能希望能够在运行时指定要使用哪个 LLM 或系统提示，而不把这些参数污染到图状态中。要添加运行时配置：

为配置指定 schema
将配置添加到节点或条件边的函数签名中
将配置传入图。

下面是一个简单示例：

from langgraph.graph import END, StateGraph, START
from langgraph.runtime import Runtime
from typing_extensions import TypedDict

# 1. 指定配置 schema
class ContextSchema(TypedDict):
    my_runtime_value: str

# 2. 定义一个在节点中访问配置的图
class State(TypedDict):
    my_state_value: str

def node(state: State, runtime: Runtime[ContextSchema]):
    if runtime.context["my_runtime_value"] == "a":
        return {"my_state_value": 1}
    elif runtime.context["my_runtime_value"] == "b":
        return {"my_state_value": 2}
    else:
        raise ValueError("未知值。")

builder = StateGraph(State, context_schema=ContextSchema)
builder.add_node(node)
builder.add_edge(START, "node")
builder.add_edge("node", END)

graph = builder.compile()

# 3. 在运行时传入配置：
print(graph.invoke({}, context={"my_runtime_value": "a"}))
print(graph.invoke({}, context={"my_runtime_value": "b"}))

{'my_state_value': 1}
{'my_state_value': 2}

扩展示例：在运行时指定 LLM

下面我们演示一个实用示例：在运行时配置要使用的 LLM。我们将同时使用 OpenAI 和 Anthropic 模型。

from dataclasses import dataclass

from langchain.chat_models import init_chat_model
from langgraph.graph import MessagesState, END, StateGraph, START
from langgraph.runtime import Runtime
from typing_extensions import TypedDict

@dataclass
class ContextSchema:
    model_provider: str = "anthropic"

MODELS = {
    "anthropic": init_chat_model("claude-haiku-4-5-20251001"),
    "openai": init_chat_model("gpt-5.4-mini"),
}

def call_model(state: MessagesState, runtime: Runtime[ContextSchema]):
    model = MODELS[runtime.context.model_provider]
    response = model.invoke(state["messages"])
    return {"messages": [response]}

builder = StateGraph(MessagesState, context_schema=ContextSchema)
builder.add_node("model", call_model)
builder.add_edge(START, "model")
builder.add_edge("model", END)

graph = builder.compile()

# 用法
input_message = {"role": "user", "content": "你好"}
# 不传配置时，使用默认值（Anthropic）
response_1 = graph.invoke({"messages": [input_message]}, context=ContextSchema())["messages"][-1]
# 或者，可以设置为 OpenAI
response_2 = graph.invoke({"messages": [input_message]}, context={"model_provider": "openai"})["messages"][-1]

print(response_1.response_metadata["model_name"])
print(response_2.response_metadata["model_name"])

claude-haiku-4-5-20251001
gpt-5.4-mini

扩展示例：在运行时指定模型和系统消息

下面我们演示一个实用示例：在运行时配置两个参数：要使用的 LLM 和系统消息。

from dataclasses import dataclass
from langchain.chat_models import init_chat_model
from langchain.messages import SystemMessage
from langgraph.graph import END, MessagesState, StateGraph, START
from langgraph.runtime import Runtime
from typing_extensions import TypedDict

@dataclass
class ContextSchema:
    model_provider: str = "anthropic"
    system_message: str | None = None

MODELS = {
    "anthropic": init_chat_model("claude-haiku-4-5-20251001"),
    "openai": init_chat_model("gpt-5.4-mini"),
}

def call_model(state: MessagesState, runtime: Runtime[ContextSchema]):
    model = MODELS[runtime.context.model_provider]
    messages = state["messages"]
    if (system_message := runtime.context.system_message):
        messages = [SystemMessage(system_message)] + messages
    response = model.invoke(messages)
    return {"messages": [response]}

builder = StateGraph(MessagesState, context_schema=ContextSchema)
builder.add_node("model", call_model)
builder.add_edge(START, "model")
builder.add_edge("model", END)

graph = builder.compile()

# 用法
input_message = {"role": "user", "content": "你好"}
response = graph.invoke({"messages": [input_message]}, context={"model_provider": "openai", "system_message": "请用意大利语回答。"})
for message in response["messages"]:
    message.pretty_print()

================================ Human Message ================================

你好
================================== Ai Message ==================================

Ciao! Come posso aiutarti oggi?

添加重试策略

在很多使用场景中，你可能希望节点拥有自定义重试策略，例如调用 API、查询数据库或调用 LLM 等。LangGraph 允许你为节点添加重试策略。要配置重试策略，请将 retry_policy 参数传递给 add_node。retry_policy 参数接收一个 RetryPolicy 命名元组对象。下面我们使用默认参数实例化一个 RetryPolicy 对象，并将其与一个节点关联：

from langgraph.types import RetryPolicy

builder.add_node(
    "node_name",
    node_function,
    retry_policy=RetryPolicy(),
)

默认情况下，retry_on 参数使用 default_retry_on 函数，该函数会对除以下情况之外的任何异常进行重试：* ValueError

TypeError
ArithmeticError
ImportError
LookupError
NameError
SyntaxError
RuntimeError
ReferenceError
StopIteration
StopAsyncIteration
OSError

此外，对于来自常用 HTTP 请求库（如 requests 和 httpx）的异常，它只会在 5xx 状态码时重试。

扩展示例：自定义重试策略

考虑一个从 SQL 数据库读取数据的示例。下面我们向节点传入两个不同的重试策略：

import sqlite3
from typing_extensions import TypedDict
from langchain.chat_models import init_chat_model
from langgraph.graph import END, MessagesState, StateGraph, START
from langgraph.types import RetryPolicy
from langchain.messages import AIMessage

con = sqlite3.connect(":memory:")
model = init_chat_model("claude-haiku-4-5-20251001")

def query_database(state: MessagesState):
    cursor = con.cursor()
    cursor.execute("SELECT * FROM Artist LIMIT 10;")
    query_result = str(cursor.fetchall())
    return {"messages": [AIMessage(content=query_result)]}

def call_model(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": [response]}

# 定义一个新的图
builder = StateGraph(MessagesState)
builder.add_node(
    "query_database",
    query_database,
    retry_policy=RetryPolicy(retry_on=sqlite3.OperationalError),
)
builder.add_node("model", call_model, retry_policy=RetryPolicy(max_attempts=5))
builder.add_edge(START, "model")
builder.add_edge("model", "query_database")
builder.add_edge("query_database", END)
graph = builder.compile()

设置节点超时

在 add_node 中使用 timeout 参数来限制单次异步节点调用可以运行的时长。以秒为单位提供超时时间，或使用 datetime.timedelta。

import asyncio
from typing_extensions import TypedDict

from langgraph.errors import NodeTimeoutError
from langgraph.graph import END, START, StateGraph


class State(TypedDict):
    value: str


async def call_model(state: State) -> State:
    await asyncio.sleep(2)
    return {"value": "done"}


builder = StateGraph(State)
builder.add_node("model", call_model, timeout=1.0)
builder.add_edge(START, "model")
builder.add_edge("model", END)
graph = builder.compile()

try:
    await graph.ainvoke({"value": "start"})
except NodeTimeoutError:
    print("节点超时")

节点超时仅支持异步节点。如果你在同步节点上设置 timeout，LangGraph 会在编译图时抛出错误，因为同步 Python 执行无法在进程内被安全取消。当节点超过其超时时间时，LangGraph 会抛出 NodeTimeoutError，它是 Python 内置 TimeoutError 的子类。如果该节点有一个会重试 TimeoutError 或 NodeTimeoutError 的 retry_policy，则会重试这次超时的尝试。超时会独立应用于每次尝试，因此每次重试都会重置计时器。超时的尝试不会提交其缓冲的写入。这可以防止状态更新或子任务调度在超时边界之后泄漏出去。

配置节点超时

add_node 上的 timeout= 参数用于限制单次异步节点尝试可运行的最长时间。传入一个数字（秒）、一个 timedelta，或一个 TimeoutPolicy 以更细粒度地控制运行超时和空闲超时。当超过限制时，LangGraph 会抛出 NodeTimeoutError，并让重试策略决定是否重试。

按节点设置超时需要 langgraph>=1.2。

from langgraph.types import TimeoutPolicy

builder.add_node(
    "call_model",
    call_model,
    timeout=TimeoutPolicy(run_timeout=120, idle_timeout=30),
)

请参阅容错，了解完整的超时生命周期、空闲超时刷新来源以及 runtime.heartbeat()。

处理节点错误

add_node 上的 error_handler= 参数会注册一个函数，该函数会在节点失败且所有重试都用尽后运行。处理器接收当前状态以及一个带有失败上下文的类型化 NodeError，并可以通过 Command 路由到恢复分支：

节点级错误处理器需要 langgraph>=1.2。

from langgraph.errors import NodeError
from langgraph.types import Command, RetryPolicy

def payment_error_handler(state: State, error: NodeError) -> Command:
    return Command(
        update={"status": f"compensated: {error.error}"},
        goto="finalize",
    )

builder.add_node(
    "charge_payment",
    charge_payment,
    retry_policy=RetryPolicy(max_attempts=3, retry_on=ConnectionError),
    error_handler=payment_error_handler,
)

请参阅容错，了解补偿模式和 Command 路由。

在节点内部访问执行信息

你可以通过 runtime.execution_info 访问执行身份和重试信息。它会暴露线程、运行和检查点标识符以及重试状态，而无需直接从 config 中读取。

属性	类型	描述
`thread_id`	`str \| None`	当前执行的线程 ID。没有 checkpointer 时为 `None`。
`run_id`	`str \| None`	当前执行的运行 ID。未在配置中提供时为 `None`。
`checkpoint_id`	`str`	当前执行的检查点 ID。
`checkpoint_ns`	`str`	当前执行的检查点命名空间。
`task_id`	`str`	当前执行的任务 ID。
`node_attempt`	`int`	当前执行尝试次数（从 1 开始计数）。第一次尝试为 `1`，第一次重试为 `2`，依此类推。
`node_first_attempt_time`	`float \| None`	第一次尝试开始时的 Unix 时间戳（秒）。在多次重试之间保持不变。

访问线程 ID 和运行 ID

使用 execution_info 在节点内部访问线程 ID、运行 ID 和其他身份字段：

from langgraph.graph import StateGraph, START, END
from langgraph.runtime import Runtime
from typing_extensions import TypedDict

class State(TypedDict):
    result: str

def my_node(state: State, runtime: Runtime):
    info = runtime.execution_info
    print(f"Thread: {info.thread_id}, Run: {info.run_id}")
    return {"result": "done"}

builder = StateGraph(State)
builder.add_node("my_node", my_node)
builder.add_edge(START, "my_node")
builder.add_edge("my_node", END)
graph = builder.compile()

根据重试状态调整行为

当节点有重试策略时，使用 execution_info 检查当前尝试次数，并在第一次尝试失败后切换到备用方案：

from langgraph.graph import StateGraph, START, END
from langgraph.runtime import Runtime
from langgraph.types import RetryPolicy
from typing_extensions import TypedDict

class State(TypedDict):
    result: str

def my_node(state: State, runtime: Runtime):
    info = runtime.execution_info
    if info.node_attempt > 1:
        # 在重试时使用备用方案
        return {"result": call_fallback_api()}
    return {"result": call_primary_api()}

builder = StateGraph(State)
builder.add_node("my_node", my_node, retry_policy=RetryPolicy(max_attempts=3))
builder.add_edge(START, "my_node")
builder.add_edge("my_node", END)
graph = builder.compile()

即使没有重试策略，Runtime 对象上也可以使用 execution_info —— node_attempt 默认值为 1，node_first_attempt_time 会设置为节点开始执行的时间。

在节点内部访问服务器信息

当你的图运行在 LangGraph Server 上时，可以通过 runtime.server_info 访问服务器特定的元数据。它会暴露助手 ID、图 ID 和已认证用户，而无需直接从配置元数据或可配置键中读取。

属性	类型	描述
`assistant_id`	`str`	当前部署的助手 ID。
`graph_id`	`str`	当前部署的图 ID。
`user`	`BaseUser \| None`	已认证用户（如果配置了 custom auth）。

from langgraph.graph import StateGraph, START, END
from langgraph.runtime import Runtime
from typing_extensions import TypedDict

class State(TypedDict):
    result: str

def my_node(state: State, runtime: Runtime):
    server = runtime.server_info
    if server is not None:
        print(f"Assistant: {server.assistant_id}, Graph: {server.graph_id}")
        if server.user is not None:
            print(f"User: {server.user.identity}")
    return {"result": "done"}

builder = StateGraph(State)
builder.add_node("my_node", my_node)
builder.add_edge(START, "my_node")
builder.add_edge("my_node", END)
graph = builder.compile()

当图未运行在 LangGraph Server 上时（例如本地开发或测试期间），server_info 为 None。

runtime.execution_info 和 runtime.server_info 需要 deepagents>=0.5.0（或 langgraph>=1.1.5）。

在节点内部访问排空状态

当请求了优雅关闭时，runtime.drain_requested 为 True。在节点内部读取它，可以在下一个 superstep 边界之前跳过昂贵的工作：

from langgraph.runtime import Runtime

def my_node(state: State, runtime: Runtime) -> State:
    if runtime.drain_requested:
        return {"status": "skipped", "reason": runtime.drain_reason}
    return {"status": do_work()}

属性	类型	描述
`drain_requested`	`bool`	如果已为本次运行调用 `RunControl.request_drain()`，则为 `True`。
`drain_reason`	`str \| None`	传给 `request_drain()` 的原因字符串；如果未请求排空，则为 `None`。

需要 langgraph>=1.2。请参阅优雅关闭，了解完整的 RunControl API。

添加节点缓存

当你想避免重复操作时，节点缓存非常有用，例如执行耗时或成本较高的操作。LangGraph 允许你为图中的节点添加单独的缓存策略。要配置缓存策略，请将 cache_policy 参数传给 add_node 函数。在下面的示例中，使用 120 秒的存活时间和默认的 key_func 生成器实例化了一个 CachePolicy 对象。然后将它关联到一个节点：

from langgraph.types import CachePolicy

builder.add_node(
    "node_name",
    node_function,
    cache_policy=CachePolicy(ttl=120),
)

然后，要为图启用节点级缓存，请在编译图时设置 cache 参数。下面的示例使用 InMemoryCache 设置一个带内存缓存的图，但也可以使用 SqliteCache。

from langgraph.cache.memory import InMemoryCache

graph = builder.compile(cache=InMemoryCache())

创建一系列步骤

前置条件 本指南假设你熟悉上文关于 state 的部分。

这里我们演示如何构建一个简单的步骤序列。我们将展示：

如何构建顺序图
用于构建类似图的内置简写方式。

要添加一系列节点，我们使用 graph 的 add_node 和 add_edge 方法：

from langgraph.graph import START, StateGraph

builder = StateGraph(State)

# 添加节点
builder.add_node(step_1)
builder.add_node(step_2)
builder.add_node(step_3)

# 添加边
builder.add_edge(START, "step_1")
builder.add_edge("step_1", "step_2")
builder.add_edge("step_2", "step_3")

我们也可以使用内置简写 .add_sequence：

builder = StateGraph(State).add_sequence([step_1, step_2, step_3])
builder.add_edge(START, "step_1")

为什么要用 LangGraph 将应用步骤拆分成一个序列？

LangGraph 让你可以轻松地为应用添加底层持久化层。这允许在节点执行之间对状态进行检查点保存，因此你的 LangGraph 节点会决定：

状态更新如何被 checkpointed
在 human-in-the-loop 工作流中，中断如何恢复
如何使用 LangGraph 的 time travel 功能“回退”并从执行中分支出来

它们还决定执行步骤如何被 streamed，以及如何使用 Studio 可视化和调试你的应用。我们来演示一个端到端示例。我们将创建一个包含三个步骤的序列：

在状态的某个键中填充一个值
更新同一个值
填充另一个不同的值

首先定义我们的 state。它决定了图的 schema，也可以指定如何应用更新。更多细节请参阅使用 reducer 处理状态更新。在本例中，我们只跟踪两个值：

from typing_extensions import TypedDict

class State(TypedDict):
    value_1: str
    value_2: int

我们的 nodes 只是一些 Python 函数，它们读取图的状态并对其进行更新。这个函数的第一个参数始终是状态：

def step_1(state: State):
    return {"value_1": "a"}

def step_2(state: State):
    current_value_1 = state["value_1"]
    return {"value_1": f"{current_value_1} b"}

def step_3(state: State):
    return {"value_2": 10}

请注意，在向状态发出更新时，每个节点只需要指定它希望更新的键的值。默认情况下，这会覆盖对应键的值。你也可以使用 reducers 来控制更新的处理方式——例如，你可以将后续更新追加到某个键上。更多细节请参阅使用 reducer 处理状态更新。

最后，我们定义图。我们使用 StateGraph 来定义一个在该状态上运行的图。然后我们将使用 add_node 和 add_edge 来填充图并定义其控制流。

from langgraph.graph import START, StateGraph

builder = StateGraph(State)

# 添加节点
builder.add_node(step_1)
builder.add_node(step_2)
builder.add_node(step_3)

# 添加边
builder.add_edge(START, "step_1")
builder.add_edge("step_1", "step_2")
builder.add_edge("step_2", "step_3")

指定自定义名称 你可以使用 add_node 为节点指定自定义名称：

builder.add_node("my_node", step_1)

请注意：

add_edge 接收节点名称；对于函数，默认是 node.__name__。
我们必须指定图的入口点。为此，我们添加一条带有 START node 的边。
当没有更多节点可执行时，图会停止。

接下来，我们 compile 图。这会对图结构进行一些基本检查（例如识别孤立节点）。如果我们通过 checkpointer 为应用添加持久化，也会在这里传入它。

graph = builder.compile()

LangGraph 提供了用于可视化图的内置工具。我们来检查一下这个序列。有关可视化的详细信息，请参阅可视化你的图。

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

接下来进行一次简单调用：

graph.invoke({"value_1": "c"})

{'value_1': 'a b', 'value_2': 10}

请注意：

我们通过为单个状态键提供一个值来启动调用。我们必须始终为至少一个键提供值。
我们传入的值被第一个节点覆盖了。
第二个节点更新了该值。
第三个节点填充了另一个不同的值。

内置简写 langgraph>=0.2.46 包含一个用于添加节点序列的内置简写 add_sequence。你可以按如下方式编译相同的图：

builder = StateGraph(State).add_sequence([step_1, step_2, step_3])
builder.add_edge(START, "step_1")

graph = builder.compile()

graph.invoke({"value_1": "c"})

创建分支

节点的并行执行对于加快整体图运行至关重要。LangGraph 原生支持节点的并行执行，可以显著提升基于图的工作流性能。这种并行化通过 fan-out 和 fan-in 机制实现，并同时利用标准边和 conditional_edges。下面是一些示例，展示如何创建适合你的分支数据流。

并行运行图节点

在这个示例中，我们从 Node A 扇出到 B and C，然后扇入到 D。在状态中，我们指定 reducer add 操作。它会对 State 中特定键的值进行合并或累积，而不是简单地覆盖现有值。对于列表，这意味着将新列表与现有列表连接起来。有关使用 reducer 更新状态的更多细节，请参阅上文 state reducers 部分。

import operator
from typing import Annotated, Any
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    # operator.add reducer 函数使其只能追加
    aggregate: Annotated[list, operator.add]

def a(state: State):
    print(f'Adding "A" to {state["aggregate"]}')
    return {"aggregate": ["A"]}

def b(state: State):
    print(f'Adding "B" to {state["aggregate"]}')
    return {"aggregate": ["B"]}

def c(state: State):
    print(f'Adding "C" to {state["aggregate"]}')
    return {"aggregate": ["C"]}

def d(state: State):
    print(f'Adding "D" to {state["aggregate"]}')
    return {"aggregate": ["D"]}

builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)
builder.add_node(c)
builder.add_node(d)
builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("a", "c")
builder.add_edge("b", "d")
builder.add_edge("c", "d")
builder.add_edge("d", END)
graph = builder.compile()

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

通过 reducer，你可以看到每个节点中添加的值会被累积起来。

graph.invoke({"aggregate": []}, {"configurable": {"thread_id": "foo"}})

Adding "A" to []
Adding "B" to ['A']
Adding "C" to ['A']
Adding "D" to ['A', 'B', 'C']

在上面的示例中，节点 "b" 和 "c" 会在同一个 superstep 中并发执行。因为它们位于同一步中，所以节点 "d" 会在 "b" 和 "c" 都完成之后执行。重要的是，来自并行 superstep 的更新顺序可能并不总是一致。如果你需要并行 superstep 中更新具有一致且预先确定的顺序，应将输出写入状态中的单独字段，并附带一个可用于排序的值。

异常处理？

LangGraph 在 supersteps 中执行节点，这意味着虽然并行分支是并行执行的，但整个 superstep 是事务性的。如果其中任何一个分支抛出异常，不会将任何更新应用到状态（整个 superstep 会报错）。重要的是，在使用 checkpointer 时，superstep 中成功节点的结果会被保存，并且在恢复时不会重复执行。如果你有容易出错的操作（比如想处理不稳定的 API 调用），LangGraph 提供了两种方式来解决：

你可以在节点中编写普通的 python 代码来捕获并处理异常。
你可以设置一个 retry_policy，指示图对抛出特定类型异常的节点进行重试。只有失败的分支会被重试，因此你无需担心执行冗余工作。

这两者结合起来，可以让你执行并行流程，并完全控制异常处理。

设置最大并发数 你可以在调用图时，通过在 configuration 中设置 max_concurrency 来控制最大并发任务数。

graph.invoke({"value_1": "c"}, {"configurable": {"max_concurrency": 10}})

延迟节点执行

当你希望将某个节点的执行推迟到所有其他待处理任务完成之后时，延迟节点执行非常有用。当分支长度不同（这在 map-reduce 这类工作流中很常见）时，这一点尤其相关。上面的示例展示了当每条路径都只有一步时如何 fan-out 和 fan-in。但如果其中一个分支有多个步骤呢？我们在 "b" 分支中添加一个节点 "b_2"：

import operator
from typing import Annotated, Any
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    # operator.add reducer 函数使其只能追加
    aggregate: Annotated[list, operator.add]

def a(state: State):
    print(f'Adding "A" to {state["aggregate"]}')
    return {"aggregate": ["A"]}

def b(state: State):
    print(f'Adding "B" to {state["aggregate"]}')
    return {"aggregate": ["B"]}

def b_2(state: State):
    print(f'Adding "B_2" to {state["aggregate"]}')
    return {"aggregate": ["B_2"]}

def c(state: State):
    print(f'Adding "C" to {state["aggregate"]}')
    return {"aggregate": ["C"]}

def d(state: State):
    print(f'Adding "D" to {state["aggregate"]}')
    return {"aggregate": ["D"]}

builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)
builder.add_node(b_2)
builder.add_node(c)
builder.add_node(d, defer=True)
builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("a", "c")
builder.add_edge("b", "b_2")
builder.add_edge("b_2", "d")
builder.add_edge("c", "d")
builder.add_edge("d", END)
graph = builder.compile()

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

graph.invoke({"aggregate": []})

Adding "A" to []
Adding "B" to ['A']
Adding "C" to ['A']
Adding "B_2" to ['A', 'B', 'C']
Adding "D" to ['A', 'B', 'C', 'B_2']

在上面的示例中，节点 "b" 和 "c" 会在同一个 superstep 中并发执行。我们在节点 d 上设置 defer=True，因此它不会执行，直到所有待处理任务都完成。在这个例子中，这意味着 "d" 会等待整个 "b" 分支完成后再执行。

条件分支

如果你的 fan-out 需要在运行时根据状态变化，可以使用 add_conditional_edges 通过图状态选择一条或多条路径。请看下面的示例，其中节点 a 生成一个状态更新，用来决定后续节点。

import operator
from typing import Annotated, Literal, Sequence
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    aggregate: Annotated[list, operator.add]
    # 向状态添加一个键。我们将设置这个键来决定
    # 如何分支。
    which: str

def a(state: State):
    print(f'Adding "A" to {state["aggregate"]}')
    return {"aggregate": ["A"], "which": "c"}

def b(state: State):
    print(f'Adding "B" to {state["aggregate"]}')
    return {"aggregate": ["B"]}

def c(state: State):
    print(f'Adding "C" to {state["aggregate"]}')
    return {"aggregate": ["C"]}

builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)
builder.add_node(c)
builder.add_edge(START, "a")
builder.add_edge("b", END)
builder.add_edge("c", END)

def conditional_edge(state: State) -> Literal["b", "c"]:
    # 在这里填写使用状态的任意逻辑
    # 来决定下一个节点
    return state["which"]

builder.add_conditional_edges("a", conditional_edge)

graph = builder.compile()

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

result = graph.invoke({"aggregate": []})
print(result)

Adding "A" to []
Adding "C" to ['A']
{'aggregate': ['A', 'C'], 'which': 'c'}

你的条件边可以路由到多个目标节点。例如：

def route_bc_or_cd(state: State) -> Sequence[str]:
    if state["which"] == "cd":
        return ["c", "d"]
    return ["b", "c"]

Map-Reduce 和 send API

LangGraph 支持使用 Send API 实现 map-reduce 和其他高级分支模式。下面是一个使用示例：

from langgraph.graph import StateGraph, START, END
from langgraph.types import Send
from typing_extensions import TypedDict, Annotated
import operator

class OverallState(TypedDict):
    topic: str
    subjects: list[str]
    jokes: Annotated[list[str], operator.add]
    best_selected_joke: str

def generate_topics(state: OverallState):
    return {"subjects": ["lions", "elephants", "penguins"]}

def generate_joke(state: OverallState):
    joke_map = {
        "lions": "Why don't lions like fast food? Because they can't catch it!",
        "elephants": "Why don't elephants use computers? They're afraid of the mouse!",
        "penguins": "Why don't penguins like talking to strangers at parties? Because they find it hard to break the ice."
    }
    return {"jokes": [joke_map[state["subject"]]]}

def continue_to_jokes(state: OverallState):
    return [Send("generate_joke", {"subject": s}) for s in state["subjects"]]

def best_joke(state: OverallState):
    return {"best_selected_joke": "penguins"}

builder = StateGraph(OverallState)
builder.add_node("generate_topics", generate_topics)
builder.add_node("generate_joke", generate_joke)
builder.add_node("best_joke", best_joke)
builder.add_edge(START, "generate_topics")
builder.add_conditional_edges("generate_topics", continue_to_jokes, ["generate_joke"])
builder.add_edge("generate_joke", "best_joke")
builder.add_edge("best_joke", END)
graph = builder.compile()

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

# 调用图：这里我们调用它来生成一个笑话列表
for step in graph.stream({"topic": "animals"}):
    print(step)

{'generate_topics': {'subjects': ['lions', 'elephants', 'penguins']}}
{'generate_joke': {'jokes': ["Why don't lions like fast food? Because they can't catch it!"]}}
{'generate_joke': {'jokes': ["Why don't elephants use computers? They're afraid of the mouse!"]}}
{'generate_joke': {'jokes': ['Why don't penguins like talking to strangers at parties? Because they find it hard to break the ice.']}}
{'best_joke': {'best_selected_joke': 'penguins'}}

创建和控制循环

创建带循环的图时，我们需要一种终止执行的机制。最常见的做法是添加一个 conditional edge，当达到某个终止条件时路由到 END 节点。你也可以在调用或流式处理图时设置图的递归限制。递归限制设置的是图在抛出错误之前允许执行的 super-steps 数量。请阅读更多关于 recursion limit concept 的内容。我们来看一个带循环的简单图，以便更好地理解这些机制如何工作。

如果想返回状态的最后一个值，而不是收到递归限制错误，请参阅下一节。

创建循环时，你可以包含一条指定终止条件的条件边：

builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)

def route(state: State) -> Literal["b", END]:
    if termination_condition(state):
        return END
    else:
        return "b"

builder.add_edge(START, "a")
builder.add_conditional_edges("a", route)
builder.add_edge("b", "a")
graph = builder.compile()

要控制递归限制，请在 config 中指定 "recursion_limit"。这会抛出一个 GraphRecursionError，你可以捕获并处理它：

from langgraph.errors import GraphRecursionError

try:
    graph.invoke(inputs, {"recursion_limit": 3})
except GraphRecursionError:
    print("递归错误")

让我们定义一个带有简单循环的图。请注意，我们使用条件边来实现终止条件。

import operator
from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    # operator.add 这个 reducer 函数会让它只能追加
    aggregate: Annotated[list, operator.add]

def a(state: State):
    print(f'节点 A 看到 {state["aggregate"]}')
    return {"aggregate": ["A"]}

def b(state: State):
    print(f'节点 B 看到 {state["aggregate"]}')
    return {"aggregate": ["B"]}

# 定义节点
builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)

# 定义边
def route(state: State) -> Literal["b", END]:
    if len(state["aggregate"]) < 7:
        return "b"
    else:
        return END

builder.add_edge(START, "a")
builder.add_conditional_edges("a", route)
builder.add_edge("b", "a")
graph = builder.compile()

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

这个架构类似于一个 ReAct agent，其中节点 "a" 是一个调用工具的模型，节点 "b" 表示这些工具。在我们的 route 条件边中，我们指定当状态中的 "aggregate" 列表超过某个长度阈值后就应该结束。调用该图时，我们可以看到在达到终止条件之前，会在节点 "a" 和 "b" 之间交替执行。

graph.invoke({"aggregate": []})

节点 A 看到 []
节点 B 看到 ['A']
节点 A 看到 ['A', 'B']
节点 B 看到 ['A', 'B', 'A']
节点 A 看到 ['A', 'B', 'A', 'B']
节点 B 看到 ['A', 'B', 'A', 'B', 'A']
节点 A 看到 ['A', 'B', 'A', 'B', 'A', 'B']

施加递归限制

在某些应用中，我们可能无法保证一定会达到给定的终止条件。在这些情况下，我们可以设置图的递归限制。在经过给定数量的超级步后，这会抛出一个 GraphRecursionError。然后我们可以捕获并处理这个异常：

from langgraph.errors import GraphRecursionError

try:
    graph.invoke({"aggregate": []}, {"recursion_limit": 4})
except GraphRecursionError:
    print("递归错误")

节点 A 看到 []
节点 B 看到 ['A']
节点 C 看到 ['A', 'B']
节点 D 看到 ['A', 'B']
节点 A 看到 ['A', 'B', 'C', 'D']
递归错误

Extended example: return state on hitting recursion limit

我们可以不抛出 GraphRecursionError，而是在状态中引入一个新的键，用来跟踪距离达到递归限制还剩多少步。然后我们可以使用这个键来判断是否应该结束本次运行。LangGraph 实现了一个特殊的 RemainingSteps 注解。在底层，它会创建一个 ManagedValue 通道——这是一个状态通道，只会在我们的图运行期间存在，运行结束后就不再存在。

import operator
from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.managed.is_last_step import RemainingSteps

class State(TypedDict):
    aggregate: Annotated[list, operator.add]
    remaining_steps: RemainingSteps

def a(state: State):
    print(f'节点 A 看到 {state["aggregate"]}')
    return {"aggregate": ["A"]}

def b(state: State):
    print(f'节点 B 看到 {state["aggregate"]}')
    return {"aggregate": ["B"]}

# 定义节点
builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)

# 定义边
def route(state: State) -> Literal["b", END]:
    if state["remaining_steps"] <= 2:
        return END
    else:
        return "b"

builder.add_edge(START, "a")
builder.add_conditional_edges("a", route)
builder.add_edge("b", "a")
graph = builder.compile()

# 试试看
result = graph.invoke({"aggregate": []}, {"recursion_limit": 4})
print(result)

节点 A 看到 []
节点 B 看到 ['A']
节点 A 看到 ['A', 'B']
{'aggregate': ['A', 'B', 'A']}

Extended example: loops with branches

为了更好地理解递归限制的工作方式，我们来看一个更复杂的例子。下面我们实现了一个循环，但其中一步会分叉到两个节点：

import operator
from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    aggregate: Annotated[list, operator.add]

def a(state: State):
    print(f'节点 A 看到 {state["aggregate"]}')
    return {"aggregate": ["A"]}

def b(state: State):
    print(f'节点 B 看到 {state["aggregate"]}')
    return {"aggregate": ["B"]}

def c(state: State):
    print(f'节点 C 看到 {state["aggregate"]}')
    return {"aggregate": ["C"]}

def d(state: State):
    print(f'节点 D 看到 {state["aggregate"]}')
    return {"aggregate": ["D"]}

# 定义节点
builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)
builder.add_node(c)
builder.add_node(d)

# 定义边
def route(state: State) -> Literal["b", END]:
    if len(state["aggregate"]) < 7:
        return "b"
    else:
        return END

builder.add_edge(START, "a")
builder.add_conditional_edges("a", route)
builder.add_edge("b", "c")
builder.add_edge("b", "d")
builder.add_edge(["c", "d"], "a")
graph = builder.compile()

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

这个图看起来很复杂，但可以把它理解为由超级步组成的循环：

节点 A
节点 B
节点 C 和 D
节点 A
…

我们有一个包含四个超级步的循环，其中节点 C 和 D 会并发执行。像之前一样调用该图，我们会看到在达到终止条件之前，它完成了两整“圈”：

result = graph.invoke({"aggregate": []})

节点 A 看到 []
节点 B 看到 ['A']
节点 D 看到 ['A', 'B']
节点 C 看到 ['A', 'B']
节点 A 看到 ['A', 'B', 'C', 'D']
节点 B 看到 ['A', 'B', 'C', 'D', 'A']
节点 D 看到 ['A', 'B', 'C', 'D', 'A', 'B']
节点 C 看到 ['A', 'B', 'C', 'D', 'A', 'B']
节点 A 看到 ['A', 'B', 'C', 'D', 'A', 'B', 'C', 'D']

不过，如果我们把递归限制设置为四，因为每一圈都是四个超级步，所以只会完成一圈：

from langgraph.errors import GraphRecursionError

try:
    result = graph.invoke({"aggregate": []}, {"recursion_limit": 4})
except GraphRecursionError:
    print("递归错误")

节点 A 看到 []
节点 B 看到 ['A']
节点 C 看到 ['A', 'B']
节点 D 看到 ['A', 'B']
节点 A 看到 ['A', 'B', 'C', 'D']
递归错误

异步

在并发运行 IO 密集型代码时，使用异步编程范式可以带来显著的性能提升（例如，向聊天模型提供商并发发起 API 请求）。要将图的 sync 实现转换为 async 实现，你需要：

将 nodes 从使用 def 更新为使用 async def。
将内部代码更新为适当地使用 await。
根据需要使用 .ainvoke 或 .astream 调用图。

由于许多 LangChain 对象都实现了 Runnable Protocol，而该协议为所有 sync 方法都提供了 async 变体，因此通常可以很快地把一个 sync 图升级为 async 图。请看下面的示例。为了演示底层 LLM 的异步调用，我们会包含一个聊天模型：

👉 Read the OpenAI chat model integration docs

pip install -U "langchain[openai]"

import os
from langchain.chat_models import init_chat_model

os.environ["OPENAI_API_KEY"] = "sk-..."

model = init_chat_model("gpt-5.4")

👉 Read the Anthropic chat model integration docs

pip install -U "langchain[anthropic]"

import os
from langchain.chat_models import init_chat_model

os.environ["ANTHROPIC_API_KEY"] = "sk-..."

model = init_chat_model("claude-sonnet-4-6")

👉 Read the Azure chat model integration docs

pip install -U "langchain[openai]"

import os
from langchain.chat_models import init_chat_model

os.environ["AZURE_OPENAI_API_KEY"] = "..."
os.environ["AZURE_OPENAI_ENDPOINT"] = "..."
os.environ["OPENAI_API_VERSION"] = "2025-03-01-preview"

model = init_chat_model(
    "azure_openai:gpt-5.4",
    azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"],
)

👉 Read the Google GenAI chat model integration docs

pip install -U "langchain[google-genai]"

import os
from langchain.chat_models import init_chat_model

os.environ["GOOGLE_API_KEY"] = "..."

model = init_chat_model("google_genai:gemini-2.5-flash-lite")

👉 Read the AWS Bedrock chat model integration docs

pip install -U "langchain[aws]"

from langchain.chat_models import init_chat_model

# Follow the steps here to configure your credentials:
# https://docs.aws.amazon.com/bedrock/latest/userguide/getting-started.html

model = init_chat_model(
    "anthropic.claude-3-5-sonnet-20240620-v1:0",
    model_provider="bedrock_converse",
)

👉 Read the HuggingFace chat model integration docs

pip install -U "langchain[huggingface]"

import os
from langchain.chat_models import init_chat_model

os.environ["HUGGINGFACEHUB_API_TOKEN"] = "hf_..."

model = init_chat_model(
    "microsoft/Phi-3-mini-4k-instruct",
    model_provider="huggingface",
    temperature=0.7,
    max_tokens=1024,
)

👉 Read the OpenRouter chat model integration docs

pip install -U "langchain-openrouter"

import os
from langchain.chat_models import init_chat_model

os.environ["OPENROUTER_API_KEY"] = "sk-..."

model = init_chat_model(
    "auto",
    model_provider="openrouter",
)

from langchain.chat_models import init_chat_model
from langgraph.graph import MessagesState, StateGraph

async def node(state: MessagesState):
    new_message = await llm.ainvoke(state["messages"])
    return {"messages": [new_message]}

builder = StateGraph(MessagesState).add_node(node).set_entry_point("node")
graph = builder.compile()

input_message = {"role": "user", "content": "你好"}
result = await graph.ainvoke({"messages": [input_message]})

异步流式传输 有关使用异步进行流式传输的示例，请参阅流式传输指南。

使用 `Command` 组合控制流和状态更新

将控制流（边）和状态更新（节点）组合起来会很有用。例如，你可能希望在同一个节点中既执行状态更新，又决定接下来要前往哪个节点。LangGraph 提供了一种方式：从节点函数返回一个 Command 对象：

def my_node(state: State) -> Command[Literal["my_other_node"]]:
    return Command(
        # 状态更新
        update={"foo": "bar"},
        # 控制流
        goto="my_other_node"
    )

下面我们展示一个端到端示例。让我们创建一个包含 3 个节点的简单图：A、B 和 C。我们会先执行节点 A，然后根据节点 A 的输出决定接下来前往节点 B 还是节点 C。

import random
from typing_extensions import TypedDict, Literal
from langgraph.graph import StateGraph, START
from langgraph.types import Command

# 定义图状态
class State(TypedDict):
    foo: str

# 定义节点

def node_a(state: State) -> Command[Literal["node_b", "node_c"]]:
    print("调用了 A")
    value = random.choice(["b", "c"])
    # 这是条件边函数的替代方案
    if value == "b":
        goto = "node_b"
    else:
        goto = "node_c"

    # 注意 Command 如何允许你同时更新图状态并路由到下一个节点
    return Command(
        # 这是状态更新
        update={"foo": value},
        # 这是边的替代方案
        goto=goto,
    )

def node_b(state: State):
    print("调用了 B")
    return {"foo": state["foo"] + "b"}

def node_c(state: State):
    print("调用了 C")
    return {"foo": state["foo"] + "c"}

现在我们可以使用上面的节点创建 StateGraph。请注意，这个图没有用于路由的条件边！这是因为控制流是在 node_a 内部通过 Command 定义的。

builder = StateGraph(State)
builder.add_edge(START, "node_a")
builder.add_node(node_a)
builder.add_node(node_b)
builder.add_node(node_c)
# 注意：节点 A、B 和 C 之间没有边！

graph = builder.compile()

你可能已经注意到，我们使用 Command 作为返回类型注解，例如 Command[Literal["node_b", "node_c"]]。这对于图渲染是必需的，它告诉 LangGraph node_a 可以导航到 node_b 和 node_c。

from IPython.display import display, Image

display(Image(graph.get_graph().draw_mermaid_png()))

如果多次运行该图，我们会看到它会根据节点 A 中的随机选择走不同路径（A -> B 或 A -> C）。

graph.invoke({"foo": ""})

调用了 A
调用了 C

导航到父图中的节点

如果你正在使用子图，你可能希望从子图中的某个节点导航到另一个子图（也就是父图中的另一个节点）。为此，你可以在 Command 中指定 graph=Command.PARENT：

def my_node(state: State) -> Command[Literal["my_other_node"]]:
    return Command(
        update={"foo": "bar"},
        goto="other_subgraph",  # 其中 `other_subgraph` 是父图中的一个节点
        graph=Command.PARENT
    )

让我们用上面的示例来演示这一点。我们会把上面示例中的 nodeA 改成一个单节点图，然后把它作为子图添加到父图中。

使用 Command.PARENT 进行状态更新 当你从子图节点向父图节点发送更新，并且更新的键同时存在于父图和子图的状态模式中时，你必须为父图状态中要更新的键定义一个 reducer。请看下面的示例。

import operator
from typing_extensions import Annotated

class State(TypedDict):
    # 注意：我们在这里定义了一个 reducer
    foo: Annotated[str, operator.add]

def node_a(state: State):
    print("调用了 A")
    value = random.choice(["a", "b"])
    # 这是条件边函数的替代方案
    if value == "a":
        goto = "node_b"
    else:
        goto = "node_c"

    # 注意 Command 如何允许你同时更新图状态并路由到下一个节点
    return Command(
        update={"foo": value},
        goto=goto,
        # 这会告诉 LangGraph 导航到父图中的 node_b 或 node_c
        # 注意：这会导航到相对于子图最近的父图
        graph=Command.PARENT,
    )

subgraph = StateGraph(State).add_node(node_a).add_edge(START, "node_a").compile()

def node_b(state: State):
    print("调用了 B")
    # 注意：因为我们已经定义了 reducer，所以不需要手动追加
    # 新字符到现有的 'foo' 值中。相反，reducer 会自动追加这些字符
    # （通过 operator.add）
    return {"foo": "b"}

def node_c(state: State):
    print("调用了 C")
    return {"foo": "c"}

builder = StateGraph(State)
builder.add_edge(START, "subgraph")
builder.add_node("subgraph", subgraph)
builder.add_node(node_b)
builder.add_node(node_c)

graph = builder.compile()

graph.invoke({"foo": ""})

调用了 A
调用了 C

在工具内部使用

一个常见用例是在工具内部更新图状态。例如，在客户支持应用中，你可能希望在对话开始时根据客户的账号或 ID 查询客户信息。要从工具更新图状态，你可以从工具返回 Command(update={"my_custom_key": "foo", "messages": [...]})：

from langchain.tools import ToolRuntime

@tool
def lookup_user_info(runtime: ToolRuntime):
    """使用此工具查询用户信息，以便更好地帮助他们解决问题。"""
    user_info = get_user_info(runtime.server_info.user.identity)
    return Command(
        update={
            # 更新状态键
            "user_info": user_info,
            # 更新消息历史
            "messages": [ToolMessage("已成功查询用户信息", tool_call_id=runtime.tool_call_id)]
        }
    )

当从工具返回 Command 时，你必须在 Command.update 中包含 messages（或任何用于消息历史的状态键），并且 messages 中的消息列表必须包含一个 ToolMessage。这是生成有效消息历史所必需的（LLM 提供商要求带有工具调用的 AI 消息后面必须跟随工具结果消息）。

如果你使用的是通过 Command 更新状态的工具，我们建议使用预构建的 ToolNode，它会自动处理返回 Command 对象的工具，并将其传播到图状态。如果你正在编写一个调用工具的自定义节点，则需要手动将工具返回的 Command 对象作为节点更新进行传播。

可视化你的图

这里我们演示如何可视化你创建的图。你可以可视化任意 Graph，包括 StateGraph。让我们通过绘制分形图来找点乐子 :)。

import random
from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]

class MyNode:
    def __init__(self, name: str):
        self.name = name
    def __call__(self, state: State):
        return {"messages": [("assistant", f"调用了节点 {self.name}")]}

def route(state) -> Literal["entry_node", END]:
    if len(state["messages"]) > 10:
        return END
    return "entry_node"

def add_fractal_nodes(builder, current_node, level, max_level):
    if level > max_level:
        return
    # 本层要创建的节点数量
    num_nodes = random.randint(1, 3)  # 可根据需要调整随机性
    for i in range(num_nodes):
        nm = ["A", "B", "C"][i]
        node_name = f"node_{current_node}_{nm}"
        builder.add_node(node_name, MyNode(node_name))
        builder.add_edge(current_node, node_name)
        # 递归添加更多节点
        r = random.random()
        if r > 0.2 and level + 1 < max_level:
            add_fractal_nodes(builder, node_name, level + 1, max_level)
        elif r > 0.05:
            builder.add_conditional_edges(node_name, route, node_name)
        else:
            # 结束
            builder.add_edge(node_name, END)

def build_fractal_graph(max_level: int):
    builder = StateGraph(State)
    entry_point = "entry_node"
    builder.add_node(entry_point, MyNode(entry_point))
    builder.add_edge(START, entry_point)
    add_fractal_nodes(builder, entry_point, 1, max_level)
    # 可选：如果需要，可以设置一个结束点
    builder.add_edge(entry_point, END)  # 或任何特定节点
    return builder.compile()

app = build_fractal_graph(3)

Mermaid

我们也可以将图类转换为 Mermaid 语法。

print(app.get_graph().draw_mermaid())

%%{init: {'flowchart': {'curve': 'linear'}}}%%
graph TD;
    tart__([<p>__start__</p>]):::first
    ry_node(entry_node)
    e_entry_node_A(node_entry_node_A)
    e_entry_node_B(node_entry_node_B)
    e_node_entry_node_B_A(node_node_entry_node_B_A)
    e_node_entry_node_B_B(node_node_entry_node_B_B)
    e_node_entry_node_B_C(node_node_entry_node_B_C)
    nd__([<p>__end__</p>]):::last
    tart__ --> entry_node;
    ry_node --> __end__;
    ry_node --> node_entry_node_A;
    ry_node --> node_entry_node_B;
    e_entry_node_B --> node_node_entry_node_B_A;
    e_entry_node_B --> node_node_entry_node_B_B;
    e_entry_node_B --> node_node_entry_node_B_C;
    e_entry_node_A -.-> entry_node;
    e_entry_node_A -.-> __end__;
    e_node_entry_node_B_A -.-> entry_node;
    e_node_entry_node_B_A -.-> __end__;
    e_node_entry_node_B_B -.-> entry_node;
    e_node_entry_node_B_B -.-> __end__;
    e_node_entry_node_B_C -.-> entry_node;
    e_node_entry_node_B_C -.-> __end__;
    ssDef default fill:#f2f0ff,line-height:1.2
    ssDef first fill-opacity:0
    ssDef last fill:#bfb6fc

PNG

如果愿意，我们也可以把 Graph 渲染成 .png。这里可以使用三种方式：

使用 Mermaid.ink API（不需要额外包）
使用 Mermaid + Pyppeteer（需要 pip install pyppeteer）
使用 graphviz（需要 pip install graphviz）

使用 Mermaid.Ink 默认情况下，draw_mermaid_png() 会使用 Mermaid.Ink 的 API 来生成图表。

from IPython.display import Image, display
from langchain_core.runnables.graph import CurveStyle, MermaidDrawMethod, NodeStyles

display(Image(app.get_graph().draw_mermaid_png()))

使用 Mermaid + Pyppeteer

import nest_asyncio

nest_asyncio.apply()  # Jupyter Notebook 需要它来运行异步函数

display(
    Image(
        app.get_graph().draw_mermaid_png(
            curve_style=CurveStyle.LINEAR,
            node_colors=NodeStyles(first="#ffdfba", last="#baffc9", default="#fad7de"),
            wrap_label_n_words=9,
            output_file_path=None,
            draw_method=MermaidDrawMethod.PYPPETEER,
            background_color="white",
            padding=10,
        )
    )
)

使用 Graphviz

try:
    display(Image(app.get_graph().draw_png()))
except ImportError:
    print(
        "你可能需要为 pygraphviz 安装依赖，更多信息请参见 https://github.com/pygraphviz/pygraphviz/blob/main/INSTALL.txt"
    )

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.

Edit this page on GitHub or file an issue.

​设置

​定义和更新状态

​定义状态

​更新状态

​使用 reducer 处理状态更新

​MessagesState

​使用 Overwrite 绕过 reducer

​定义输入和输出 schema

​在节点之间传递私有状态

​使用 pydantic 模型作为图状态

​添加运行时配置

​添加重试策略

​设置节点超时

​配置节点超时

​处理节点错误

​在节点内部访问执行信息

​访问线程 ID 和运行 ID

​根据重试状态调整行为

​在节点内部访问服务器信息

​在节点内部访问排空状态

​添加节点缓存

​创建一系列步骤

​创建分支

​并行运行图节点

​延迟节点执行

​条件分支

​Map-Reduce 和 send API

​创建和控制循环

​施加递归限制

​异步

​使用 Command 组合控制流和状态更新

​导航到父图中的节点

​在工具内部使用

​可视化你的图

​Mermaid

​PNG

设置

定义和更新状态

定义状态

更新状态

使用 reducer 处理状态更新

MessagesState

使用 `Overwrite` 绕过 reducer

定义输入和输出 schema

在节点之间传递私有状态

使用 pydantic 模型作为图状态

添加运行时配置

添加重试策略

设置节点超时

配置节点超时

处理节点错误

在节点内部访问执行信息

访问线程 ID 和运行 ID

根据重试状态调整行为

在节点内部访问服务器信息

在节点内部访问排空状态

添加节点缓存

创建一系列步骤

创建分支

并行运行图节点

延迟节点执行

条件分支

Map-Reduce 和 send API

创建和控制循环

施加递归限制

异步

使用 `Command` 组合控制流和状态更新

导航到父图中的节点

在工具内部使用

可视化你的图

Mermaid

PNG