使用图 API - Docs by LangChain

本指南演示了 LangGraph 图 API 的基础知识。它将引导你了解状态，以及如何组合常见的图结构，如序列、分支和循环。它还涵盖了 LangGraph 的控制功能，包括用于 Map-Reduce 工作流的 [Send API](#map-reduce 和 send api) 以及用于将状态更新与节点间“跳转”相结合的 [Command API](#使用 command 结合控制流和状态更新)。

设置

安装 langgraph：

pip install -U langgraph

uv add langgraph

设置 LangSmith 以获得更好的调试体验注册 LangSmith 以快速发现问题并提升你的 LangGraph 项目的性能。LangSmith 允许你使用跟踪数据来调试、测试和监控使用 LangGraph 构建的 LLM 应用——在文档中了解更多入门信息。

定义和更新状态

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

如何使用状态来定义图的模式
如何使用归约器来控制状态更新的处理方式。

定义状态

LangGraph 中的状态可以是 TypedDict、Pydantic 模型或数据类。下面我们将使用 TypedDict。有关使用 Pydantic 的详细信息，请参阅[使用 Pydantic 模型作为图状态](#使用 pydantic 模型作为图状态)。默认情况下，图将具有相同的输入和输出模式，状态决定了该模式。有关如何定义不同的输入和输出模式，请参阅定义输入和输出模式。让我们考虑一个使用消息的简单示例。这代表了用于许多 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("Hello!")
    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("Hi")]})
result

{'messages': [HumanMessage(content='Hi'), AIMessage(content='Hello!')], 'extra_field': 10}

请注意：

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

为了方便，我们经常通过漂亮打印来检查消息对象的内容：

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

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

Hi
================================== Ai Message ==================================

Hello!

使用归约器处理状态更新

状态中的每个键都可以有自己的独立归约器函数，该函数控制如何应用来自节点的更新。如果没有明确指定归约器函数，则假定对该键的所有更新都应覆盖它。对于 TypedDict 状态模式，我们可以通过用归约器函数注解状态的相应字段来定义归约器。在前面的示例中，我们的节点通过向其追加消息来更新状态中的 "messages" 键。下面，我们为这个键添加一个归约器，以便更新自动追加：

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("Hello!")
    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("Hi")]})

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

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

Hi
================================== Ai Message ==================================

Hello!

MessagesState

实际上，更新消息列表还有额外的考虑因素：

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

LangGraph 包含一个内置的归约器 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("Hello!")
    return {"messages": [new_message], "extra_field": 10}

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

input_message = {"role": "user", "content": "Hi"}

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

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

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

Hi
================================== Ai Message ==================================

Hello!

这是涉及聊天模型的应用程序的一种通用状态表示。为了方便，LangGraph 包含一个预构建的 MessagesState，这样我们可以有：

from langgraph.graph import MessagesState

class State(MessagesState):
    extra_field: int

使用 `Overwrite` 绕过归约器

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

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": ["first message"]}

def replace_messages(state: State):
    # 绕过归约器并替换整个消息列表
    return {"messages": Overwrite(["replacement message"])}

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": ["initial"]})
print(result["messages"])

['replacement message']

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

def replace_messages(state: State):
    return {"messages": {"__overwrite__": ["replacement message"]}}

当节点并行执行时，在给定的超级步骤中，只有一个节点可以使用 Overwrite 来处理同一个状态键。如果多个节点尝试在同一个超级步骤中覆盖同一个键，将引发 InvalidUpdateError。

定义输入和输出模式

默认情况下，StateGraph 使用单一模式操作，所有节点都期望使用该模式进行通信。但是，也可以为图定义不同的输入和输出模式。当指定了不同的模式时，内部模式仍将用于节点之间的通信。输入模式确保提供的输入符合预期的结构，而输出模式根据定义的输出模式过滤内部数据，仅返回相关信息。下面，我们将看到如何定义不同的输入和输出模式。

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

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

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

# 定义整体模式，结合输入和输出
class OverallState(InputState, OutputState):
    pass

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

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

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

{'answer': 'bye'}

请注意，调用的输出仅包含输出模式。

在节点之间传递私有状态

在某些情况下，你可能希望节点交换对中间逻辑至关重要但不需要成为图主要模式一部分的信息。这些私有数据与图的整体输入/输出无关，只应在特定节点之间共享。下面，我们将创建一个由三个节点（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": "set by node_1"}
    print(f"Entered node `node_1`:\n\tInput: {state}.\n\tReturned: {output}")
    return output

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

def node_2(state: Node2Input) -> OverallState:
    output = {"a": "set by node_2"}
    print(f"Entered node `node_2`:\n\tInput: {state}.\n\tReturned: {output}")
    return output

# Node 3 只能访问整体状态（无法访问来自 node_1 的私有数据）
def node_3(state: OverallState) -> OverallState:
    output = {"a": "set by node_3"}
    print(f"Entered node `node_3`:\n\tInput: {state}.\n\tReturned: {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": "set at start",
    }
)

print()
print(f"Output of graph invocation: {response}")

Entered node `node_1`:
    Input: {'a': 'set at start'}.
    Returned: {'private_data': 'set by node_1'}
Entered node `node_2`:
    Input: {'private_data': 'set by node_1'}.
    Returned: {'a': 'set by node_2'}
Entered node `node_3`:
    Input: {'a': 'set by node_2'}.
    Returned: {'a': 'set by node_3'}

Output of graph invocation: {'a': 'set by node_3'}

使用 pydantic 模型作为图状态

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

已知限制 * 目前，图的输出不会是 pydantic 模型的实例。 * 运行时验证仅发生在图第一个节点的输入上，而不是后续节点或输出上。 * pydantic 的验证错误跟踪不会显示错误发生在哪个节点。 * 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": "goodbye"}

# 构建状态图
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": "hello"})

使用无效输入调用图

try:
    graph.invoke({"a": 123})  # 应该是字符串
except Exception as e:
    print("An exception was raised because `a` is an integer rather than a string.")
    print(e)

An exception was raised because `a` is an integer rather than a string.
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 模型作为状态模式时，了解序列化的工作方式非常重要，特别是在以下情况下：

将 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"Input state type: {type(state)}")
    print(f"Nested type: {type(state.nested)}")
    # 返回字典更新
    return {"text": state.text + " processed", "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="hello", count=0, nested=NestedModel(value="test"))
print(f"Input object type: {type(input_state)}")

# 使用 Pydantic 实例调用图
result = graph.invoke(input_state)
print(f"Output type: {type(result)}")
print(f"Output content: {result}")

# 如果需要，转换回 Pydantic 模型
output_model = ComplexState(**result)
print(f"Converted back to 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"\nExpected validation error: {e}")

处理消息模型

在状态模式中使用 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="Hello there!")]}

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="Hi")], context="Customer support chat"
)

result = graph.invoke(initial_state)
print(f"Output: {result}")

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

添加运行时配置

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

为你的配置指定一个模式
将配置添加到节点或条件边的函数签名中
将配置传递到图中。

请参见下面的简单示例：

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

# 1. 指定配置模式
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("Unknown values.")

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": "hi"}
# 没有配置时，使用默认值（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": "hi"}
response = graph.invoke({"messages": [input_message]}, context={"model_provider": "openai", "system_message": "Respond in Italian."})
for message in response["messages"]:
    message.pretty_print()

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

hi
================================== 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_community.utilities import SQLDatabase
from langchain.messages import AIMessage

db = SQLDatabase.from_uri("sqlite:///:memory:")
model = init_chat_model("claude-haiku-4-5-20251001")

def query_database(state: MessagesState):
    query_result = db.run("SELECT * FROM Artist LIMIT 10;")
    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= 参数限制单个异步节点尝试可能运行的时间。传递一个数字（秒）、一个 timedelta 或一个 TimeoutPolicy 以对运行和空闲超时进行更精细的控制。当超过限制时，LangGraph 会引发 NodeTimeoutError，并让重试策略决定是否重试。

每个节点的超时需要 langgraph>=1.2，目前处于 alpha 阶段。

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，目前处于 alpha 阶段。

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。没有检查点器时为 `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

使用 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()

即使没有重试策略，execution_info 也可在 Runtime 对象上使用——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`	经过身份验证的用户（如果配置了自定义身份验证）。

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。

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

在节点内访问排空状态

当请求优雅关闭时，runtime.drain_requested 为 True。在节点内读取此值，以便在下一个超级步骤边界之前跳过昂贵的工作：

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，目前处于 alpha 阶段。有关完整的 RunControl API，请参阅优雅关闭。

添加节点缓存

节点缓存对于希望避免重复操作的情况很有用，例如在执行昂贵操作（无论是时间还是成本方面）时。LangGraph 允许你为图中的节点添加个性化的缓存策略。要配置缓存策略，请将 cache_policy 参数传递给 add_node 函数。在以下示例中，一个 CachePolicy 对象被实例化，其生存时间为 120 秒，并使用默认的 key_func 生成器。然后将其与节点关联：

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())

创建步骤序列

前提条件 本指南假设你熟悉上面关于状态的部分。

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

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

要添加一系列节点，我们使用 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 节点控制：

状态更新如何被检查点化
Human in the Loop工作流中的中断如何恢复
我们如何使用 LangGraph 的时间旅行功能“回退”和分支执行

它们还决定了执行步骤如何被流式传输，以及你的应用程序如何使用 Studio 进行可视化和调试。让我们演示一个端到端的示例。我们将创建三个步骤的序列：

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

让我们首先定义我们的状态。这控制着图的模式，也可以指定如何应用更新。有关更多详细信息，请参阅使用归约器处理状态更新。在我们的例子中，我们将只跟踪两个值：

from typing_extensions import TypedDict

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

我们的节点只是读取图状态并对其进行更新的 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}

请注意，当向状态发出更新时，每个节点只需指定它希望更新的键的值。默认情况下，这将覆盖相应键的值。你也可以使用归约器来控制更新的处理方式——例如，你可以将连续的更新追加到一个键中。有关更多详细信息，请参阅使用归约器处理状态更新。

最后，我们定义图。我们使用 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 节点的边。
当没有更多节点可执行时，图停止。

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

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 提供了对节点并行执行的原生支持，这可以显著提高基于图的工作流的性能。这种并行化是通过扇出和扇入机制实现的，利用标准边和 conditional_edges。以下是一些示例，展示如何添加创建适合你的分支数据流。

并行运行图节点

在这个示例中，我们从 Node A 扇出到 B and C，然后扇入到 D。对于我们的状态，我们指定了归约器 add 操作。这将组合或累积状态中特定键的值，而不是简单地覆盖现有值。对于列表，这意味着将新列表与现有列表连接起来。有关使用归约器更新状态的更多详细信息，请参阅上面的状态归约器部分。

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

class State(TypedDict):
    # operator.add 归约器函数使其仅追加
    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()))

使用归约器，你可以看到在每个节点中添加的值被累积了。

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" 在同一个超级步骤中并发执行。因为它们在同一个步骤中，所以节点 "d" 在 "b" 和 "c" 都完成后执行。重要的是，并行超级步骤的更新可能不会按一致的顺序排列。如果你需要并行超级步骤的更新具有一致的、预定的顺序，你应该将输出与用于排序的值一起写入状态中的单独字段。

异常处理？

LangGraph 在超级步骤内执行节点，这意味着虽然并行分支是并行执行的，但整个超级步骤是事务性的。如果这些分支中的任何一个引发异常，则不会有任何更新应用到状态（整个超级步骤出错）。重要的是，当使用检查点器时，超级步骤内成功节点的结果会被保存，并且在恢复时不会重复。如果你有容易出错的（可能想处理不稳定的 API 调用），LangGraph 提供了两种解决方法：

你可以在节点内编写常规的 Python 代码来捕获和处理异常。
你可以设置一个 retry_policy 来指示图重试引发某些类型异常的节点。只有失败的分支会被重试，因此你不必担心执行冗余工作。

这些结合在一起，让你可以执行并行执行并完全控制异常处理。

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

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

延迟节点执行

当你想要延迟节点的执行直到所有其他待处理任务完成时，延迟节点执行很有用。当分支具有不同的长度时，这在 Map-Reduce 流程等工作流中尤其相关。上面的示例展示了当每个路径只有一步时如何扇出和扇入。但如果一个分支有多个步骤呢？让我们在 "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 归约器函数使其仅追加
    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" 在同一个超级步骤中并发执行。我们在节点 d 上设置了 defer=True，因此它将不会执行，直到所有待处理任务完成。在这种情况下，这意味着 "d" 等待执行，直到整个 "b" 分支完成。

条件分支

如果你的扇出应该在运行时根据状态变化，你可以使用 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'}}

创建和控制循环

在创建带有循环的图时，我们需要一种终止执行的机制。最常见的是通过添加一个条件边来实现，该边在达到某个终止条件时路由到 END 节点。你也可以在调用或流式传输图时设置图的递归限制。递归限制设置了图在引发错误之前允许执行的超级步骤数量。阅读更多关于递归限制概念的信息。让我们考虑一个带有循环的简单图，以更好地理解这些机制的工作原理。

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

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

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()

要控制递归限制，请在配置中指定 "recursion_limit"。这将引发一个 GraphRecursionError，你可以捕获并处理它：

from langgraph.errors import GraphRecursionError

try:
    graph.invoke(inputs, {"recursion_limit": 3})
except GraphRecursionError:
    print("Recursion Error")

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

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

class State(TypedDict):
    # operator.add 归约器函数使其仅追加
    aggregate: Annotated[list, operator.add]

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

def b(state: State):
    print(f'Node B sees {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 代理，其中节点 "a" 是一个工具调用模型，节点 "b" 代表工具。在我们的 route 条件边中，我们指定当状态中的 "aggregate" 列表超过阈值长度后应结束。调用图，我们看到在达到终止条件之前，我们在节点 "a" 和 "b" 之间交替。

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

Node A sees []
Node B sees ['A']
Node A sees ['A', 'B']
Node B sees ['A', 'B', 'A']
Node A sees ['A', 'B', 'A', 'B']
Node B sees ['A', 'B', 'A', 'B', 'A']
Node A sees ['A', 'B', 'A', 'B', 'A', 'B']

施加递归限制

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

from langgraph.errors import GraphRecursionError

try:
    graph.invoke({"aggregate": []}, {"recursion_limit": 4})
except GraphRecursionError:
    print("Recursion Error")

Node A sees []
Node B sees ['A']
Node C sees ['A', 'B']
Node D sees ['A', 'B']
Node A sees ['A', 'B', 'C', 'D']
Recursion Error

扩展示例：达到递归限制时返回状态

与其引发 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'Node A sees {state["aggregate"]}')
    return {"aggregate": ["A"]}

def b(state: State):
    print(f'Node B sees {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)

Node A sees []
Node B sees ['A']
Node A sees ['A', 'B']
{'aggregate': ['A', 'B', 'A']}

扩展示例：带有分支的循环

为了更好地理解递归限制的工作原理，让我们考虑一个更复杂的例子。下面我们实现一个循环，但其中一个步骤扇出到两个节点：

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'Node A sees {state["aggregate"]}')
    return {"aggregate": ["A"]}

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

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

def d(state: State):
    print(f'Node D sees {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": []})

Node A sees []
Node B sees ['A']
Node D sees ['A', 'B']
Node C sees ['A', 'B']
Node A sees ['A', 'B', 'C', 'D']
Node B sees ['A', 'B', 'C', 'D', 'A']
Node D sees ['A', 'B', 'C', 'D', 'A', 'B']
Node C sees ['A', 'B', 'C', 'D', 'A', 'B']
Node A sees ['A', 'B', 'C', 'D', 'A', 'B', 'C', 'D']

然而，如果我们设置递归限制为四，我们只完成一圈，因为每圈是四个超级步骤：

from langgraph.errors import GraphRecursionError

try:
    result = graph.invoke({"aggregate": []}, {"recursion_limit": 4})
except GraphRecursionError:
    print("Recursion Error")

Node A sees []
Node B sees ['A']
Node C sees ['A', 'B']
Node D sees ['A', 'B']
Node A sees ['A', 'B', 'C', 'D']
Recursion Error

异步

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

更新 nodes 使用 async def 而不是 def。
更新内部代码以适当地使用 await。
根据需要使用 .ainvoke 或 .astream 调用图。

因为许多 LangChain 对象实现了 Runnable Protocol，该协议具有所有 sync 方法的 async 变体，所以通常将 sync 图升级为 async 图相当快。请参见下面的示例。为了演示底层 LLM 的异步调用，我们将包含一个聊天模型：

👉 阅读 OpenAI 聊天模型集成文档

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")

import os
from langchain_openai import ChatOpenAI

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

model = ChatOpenAI(model="gpt-5.4")

👉 阅读 Anthropic 聊天模型集成文档

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")

import os
from langchain_anthropic import ChatAnthropic

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

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

👉 阅读 Azure 聊天模型集成文档

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"],
)

import os
from langchain_openai import AzureChatOpenAI

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

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

👉 阅读 Google GenAI 聊天模型集成文档

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")

import os
from langchain_google_genai import ChatGoogleGenerativeAI

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

model = ChatGoogleGenerativeAI(model="gemini-2.5-flash-lite")

👉 阅读 AWS Bedrock 聊天模型集成文档

pip install -U "langchain[aws]"

from langchain.chat_models import init_chat_model

# 按照此处步骤配置您的凭证：
# 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",
)

from langchain_aws import ChatBedrock

model = ChatBedrock(model="anthropic.claude-3-5-sonnet-20240620-v1:0")

👉 阅读 HuggingFace 聊天模型集成文档

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,
)

import os
from langchain_huggingface import ChatHuggingFace, HuggingFaceEndpoint

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

llm = HuggingFaceEndpoint(
    repo_id="microsoft/Phi-3-mini-4k-instruct",
    temperature=0.7,
    max_length=1024,
)
model = ChatHuggingFace(llm=llm)

👉 阅读 OpenRouter 聊天模型集成文档

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",
)

import os
from langchain_openrouter import ChatOpenRouter

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

model = ChatOpenRouter(model="auto")

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": "Hello"}
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("Called 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("Called B")
    return {"foo": state["foo"] + "b"}

def node_c(state: State):
    print("Called 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": ""})

Called A
Called 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 进行状态更新 当你从子图节点向父图节点发送更新时，对于父图和子图状态模式共享的键，你必须在父图状态中为要更新的键定义一个归约器。请参见下面的示例。

import operator
from typing_extensions import Annotated

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

def node_a(state: State):
    print("Called 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("Called B")
    # 注意：由于我们已经定义了归约器，我们不需要手动追加
    # 新字符到现有的 'foo' 值。相反，归约器将自动追加这些
    # （通过 operator.add）
    return {"foo": "b"}

def node_c(state: State):
    print("Called 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": ""})

Called A
Called C

在工具内使用

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

from langchain.tools import ToolRuntime

@tool
def lookup_user_info(runtime: ToolRuntime):
    """Use this to look up user information to better assist them with their questions."""
    user_info = get_user_info(runtime.server_info.user.identity)
    return Command(
        update={
            # 更新状态键
            "user_info": user_info,
            # 更新消息历史
            "messages": [ToolMessage("Successfully looked up user information", 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"Called node {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

如果需要，我们可以将图渲染为 .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(
        "You likely need to install dependencies for pygraphviz, see more here https://github.com/pygraphviz/pygraphviz/blob/main/INSTALL.txt"
    )

将这些文档连接到 Claude、VSCode 等，通过 MCP 获得实时答案。

在 GitHub 上编辑此页面或提交问题。

​设置

​定义和更新状态

​定义状态

​更新状态

​使用归约器处理状态更新

​MessagesState

​使用 Overwrite 绕过归约器

​定义输入和输出模式

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

​使用 pydantic 模型作为图状态

​添加运行时配置

​添加重试策略

​配置节点超时

​处理节点错误

​在节点内访问执行信息

​访问线程和运行 ID

​根据重试状态调整行为

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

​在节点内访问排空状态

​添加节点缓存

​创建步骤序列

​创建分支

​并行运行图节点

​延迟节点执行

​条件分支

​Map-Reduce 和 send API

​创建和控制循环

​施加递归限制

​异步

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

​导航到父图中的节点

​在工具内使用

​可视化你的图

​Mermaid

​PNG

设置

定义和更新状态

定义状态

更新状态

使用归约器处理状态更新

MessagesState

使用 `Overwrite` 绕过归约器

定义输入和输出模式

在节点之间传递私有状态

使用 pydantic 模型作为图状态

添加运行时配置

添加重试策略

配置节点超时

处理节点错误

在节点内访问执行信息

访问线程和运行 ID

根据重试状态调整行为

在节点内访问服务器信息

在节点内访问排空状态

添加节点缓存

创建步骤序列

创建分支

并行运行图节点

延迟节点执行

条件分支

Map-Reduce 和 send API

创建和控制循环

施加递归限制

异步

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

导航到父图中的节点

在工具内使用

可视化你的图

Mermaid

PNG