> ## Documentation Index
> Fetch the complete documentation index at: https://nvd-54.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.
# 图 API 概述
## 图
LangGraph 的核心是将智能体工作流建模为图。你通过三个关键组件来定义智能体的行为：
1. [`状态`](#state)：一个共享数据结构，表示应用程序的当前快照。它可以是任何数据类型，但通常使用共享的状态模式来定义。
2. [`节点`](#nodes)：编码智能体逻辑的函数。它们接收当前状态作为输入，执行某些计算或副作用，并返回更新后的状态。
3. [`边`](#edges)：根据当前状态决定下一步执行哪个`节点`的函数。它们可以是条件分支或固定转换。
通过组合`节点`和`边`，你可以创建复杂的、可循环的工作流，使状态随时间演化。但真正强大之处在于 LangGraph 如何管理该状态。
需要强调的是：`节点`和`边`不过是函数——它们可以包含大语言模型(LLM)或普通代码。
简而言之：*节点执行工作，边决定接下来做什么*。
LangGraph 的底层图算法使用[消息传递](https://en.wikipedia.org/wiki/Message_passing)来定义通用程序。当一个节点完成操作时，它会沿着一条或多条边向其他节点发送消息。这些接收节点随后执行其函数，将生成的消息传递给下一组节点，整个过程如此循环。受到 Google [Pregel](https://research.google/pubs/pregel-a-system-for-large-scale-graph-processing/) 系统的启发，程序以离散的"超级步"方式运行。
超级步可以被理解为对图节点的一次迭代。并行运行的节点属于同一个超级步，而顺序运行的节点属于不同的超级步。在图执行开始时，所有节点都处于 `inactive` 状态。当节点在其任何传入边（或"通道"）上接收到新消息（状态）时，它变为 `active` 状态。活跃节点随后运行其函数并以更新作为响应。在每个超级步结束时，没有传入消息的节点通过将自己标记为 `inactive` 来投票 `halt`。当所有节点都处于 `inactive` 状态且没有消息在传输中时，图执行终止。
### StateGraph
[`StateGraph`](https://reference.langchain.com/python/langgraph/graph/state/StateGraph) 类是主要使用的图类。它由用户定义的 `State` 对象参数化。
### 编译你的图
要构建图，你首先定义[状态](#state)，然后添加[节点](#nodes)和[边](#edges)，最后编译它。编译图究竟是什么，为什么需要它？
编译是一个相当简单的步骤。它对图的结构进行一些基本检查（如没有孤立节点等）。编译也是你可以指定运行时参数（如[检查点器](/oss/python/langgraph/persistence)和断点）的地方。你只需调用 `.compile` 方法即可编译图：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
graph = graph_builder.compile(...)
```
你**必须**在使用图之前对其进行编译。
## 状态
定义图时首先要做的是定义图的`状态`。`状态`由图的[模式](#schema)以及指定如何将更新应用于状态的[`reducer`函数](#reducers)组成。模式将作为图中所有`节点`和`边`的输入模式，可以是 `TypedDict` 或 `Pydantic` 模型。所有`节点`将发出对`状态`的更新，这些更新随后使用指定的 `reducer` 函数应用。
### 模式
指定图模式的主要文档化方式是使用 [`TypedDict`](https://docs.python.org/3/library/typing.html#typing.TypedDict)。如果你想在状态中提供默认值，请使用 [`dataclass`](https://docs.python.org/3/library/dataclasses.html)。如果你需要递归数据验证，我们也支持使用 Pydantic [`BaseModel`](/oss/python/langgraph/use-graph-api#use-pydantic-models-for-graph-state) 作为图状态（但请注意 Pydantic 的性能不如 `TypedDict` 或 `dataclass`）。
默认情况下，图的输入和输出模式相同。如果你想更改此设置，也可以直接指定显式的输入和输出模式。当你有很多键，其中一些明确用于输入，另一些用于输出时，这很有用。更多信息请参阅[指南](/oss/python/langgraph/use-graph-api#define-input-and-output-schemas)。
`langchain` 中的高级 [`create_agent`](/oss/python/langchain/agents) 工厂不支持 Pydantic 状态模式。
#### 多重模式
通常，所有图节点使用单一模式通信。这意味着它们将读取和写入相同的状态通道。但在某些情况下，我们希望对此进行更多控制：
* 内部节点可以传递图输入/输出中不需要的信息。
* 我们可能还希望为图使用不同的输入/输出模式。例如，输出可能只包含一个相关的输出键。
节点可以写入图内部的私有状态通道进行内部节点通信。我们可以简单地定义一个私有模式 `PrivateState`。
也可以为图定义显式的输入和输出模式。在这种情况下，我们定义一个包含与图操作相关的\_所有\_键的"内部"模式。但我们也定义 `input` 和 `output` 模式作为"内部"模式的子集，以约束图的输入和输出。更多详情请参阅[定义输入和输出模式](/oss/python/langgraph/use-graph-api#define-input-and-output-schemas)。
让我们看一个示例：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
class InputState(TypedDict):
user_input: str
class OutputState(TypedDict):
graph_output: str
class OverallState(TypedDict):
foo: str
user_input: str
graph_output: str
class PrivateState(TypedDict):
bar: str
def node_1(state: InputState) -> OverallState:
# 写入 OverallState
return {"foo": state["user_input"] + " name"}
def node_2(state: OverallState) -> PrivateState:
# 从 OverallState 读取，写入 PrivateState
return {"bar": state["foo"] + " is"}
def node_3(state: PrivateState) -> OutputState:
# 从 PrivateState 读取，写入 OutputState
return {"graph_output": state["bar"] + " Lance"}
builder = StateGraph(OverallState,input_schema=InputState,output_schema=OutputState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_2", "node_3")
builder.add_edge("node_3", END)
graph = builder.compile()
graph.invoke({"user_input":"My"})
# {'graph_output': 'My name is Lance'}
```
这里有两个微妙但重要的要点：
1. 我们将 `state: InputState` 作为输入模式传递给 `node_1`。但我们写入了 `foo`，这是 `OverallState` 中的一个通道。我们如何能写入一个不包含在输入模式中的状态通道？这是因为节点\_可以写入图状态中的任何状态通道\_。图状态是在初始化时定义的状态通道的联合，包括 `OverallState` 以及过滤器 `InputState` 和 `OutputState`。
2. 我们用以下方式初始化图：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
StateGraph(
OverallState,
input_schema=InputState,
output_schema=OutputState
)
```
我们如何能在 `node_2` 中写入 `PrivateState`？如果它没有在 `StateGraph` 初始化中传入，图如何访问这个模式？
我们可以这样做，因为`节点也可以声明额外的状态通道`，只要状态模式定义存在。在这种情况下，`PrivateState` 模式已定义，因此我们可以将 `bar` 添加为图中的新状态通道并写入它。
### Reducer
Reducer 是理解节点更新如何应用于`状态`的关键。`状态`中的每个键都有其独立的 reducer 函数。如果没有显式指定 reducer 函数，则假定对该键的所有更新都应覆盖它。从默认类型的 reducer 开始，有几种不同类型的 reducer：
#### 默认 reducer
以下两个示例展示了如何使用默认 reducer：
```python Example A theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from typing_extensions import TypedDict
class State(TypedDict):
foo: int
bar: list[str]
```
在此示例中，没有为任何键指定 reducer 函数。假设图的输入为：
`{"foo": 1, "bar": ["hi"]}`。然后假设第一个`节点`返回 `{"foo": 2}`。这被视为对状态的更新。请注意，`节点`不需要返回完整的`状态`模式——只需返回更新即可。应用此更新后，`状态`将变为 `{"foo": 2, "bar": ["hi"]}`。如果第二个节点返回 `{"bar": ["bye"]}`，则`状态`将变为 `{"foo": 2, "bar": ["bye"]}`。
```python Example B theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from typing import Annotated
from typing_extensions import TypedDict
from operator import add
class State(TypedDict):
foo: int
bar: Annotated[list[str], add]
```
在此示例中，我们使用 `Annotated` 类型为第二个键（`bar`）指定了 reducer 函数（`operator.add`）。请注意，第一个键保持不变。假设图的输入为 `{"foo": 1, "bar": ["hi"]}`。然后假设第一个`节点`返回 `{"foo": 2}`。这被视为对状态的更新。请注意，`节点`不需要返回完整的`状态`模式——只需返回更新即可。应用此更新后，`状态`将变为 `{"foo": 2, "bar": ["hi"]}`。如果第二个节点返回 `{"bar": ["bye"]}`，则`状态`将变为 `{"foo": 2, "bar": ["hi", "bye"]}`。请注意，这里 `bar` 键是通过将两个列表合并来更新的。
#### Overwrite
在某些情况下，你可能希望绕过 reducer 直接覆盖状态值。LangGraph 为此提供了 [`Overwrite`](https://reference.langchain.com/python/langgraph/types/) 类型。[了解如何使用 `Overwrite`](/oss/python/langgraph/use-graph-api#bypass-reducers-with-overwrite)。
### 在图状态中使用消息
#### 为什么使用消息？
大多数现代 LLM 提供商都有一个接受消息列表作为输入的聊天模型接口。LangChain 的[聊天模型接口](/oss/python/langchain/models)特别接受消息对象列表作为输入。这些消息有多种形式，如 [`HumanMessage`](https://reference.langchain.com/python/langchain-core/messages/human/HumanMessage)（用户输入）或 [`AIMessage`](https://reference.langchain.com/python/langchain-core/messages/ai/AIMessage)（LLM 响应）。
要了解更多关于消息对象的信息，请参阅[消息概念指南](/oss/python/langchain/messages)。
#### 在图中使用消息
在许多情况下，将先前的对话历史作为消息列表存储在图状态中很有帮助。为此，我们可以向图状态添加一个存储 `Message` 对象列表的键（通道），并用 reducer 函数对其进行注解（参见下面示例中的 `messages` 键）。reducer 函数对于告诉图如何在每次状态更新时更新`状态`中的 `Message` 对象列表至关重要（例如，当节点发送更新时）。如果你不指定 reducer，每次状态更新都会用最新提供的值覆盖消息列表。如果你只想将消息追加到现有列表中，可以使用 `operator.add` 作为 reducer。
然而，你可能还想手动更新图状态中的消息（例如人机协作）。如果你使用 `operator.add`，你发送给图的手动状态更新将被追加到现有消息列表中，而不是更新现有消息。为避免这种情况，你需要一个能跟踪消息 ID 并正确覆盖现有消息的 reducer。为此，你可以使用预构建的 [`add_messages`](https://reference.langchain.com/python/langgraph/graph/message/add_messages) 函数。对于全新的消息，它会简单地追加到现有列表中，但也会正确处理对现有消息的更新。
#### 序列化
除了跟踪消息 ID 之外，[`add_messages`](https://reference.langchain.com/python/langgraph/graph/message/add_messages) 函数还会在 `messages` 通道上收到状态更新时，尝试将消息反序列化为 LangChain `Message` 对象。
更多信息请参阅 [LangChain 序列化/反序列化](https://python.langchain.com/docs/how_to/serialization/)。这允许以以下格式发送图输入/状态更新：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
# 支持这种方式
{"messages": [HumanMessage(content="message")]}
# 也支持这种方式
{"messages": [{"type": "human", "content": "message"}]}
```
由于使用 [`add_messages`](https://reference.langchain.com/python/langgraph/graph/message/add_messages) 时状态更新总是被反序列化为 LangChain `Messages`，你应该使用点号语法访问消息属性，如 `state["messages"][-1].content`。
下面是一个使用 [`add_messages`](https://reference.langchain.com/python/langgraph/graph/message/add_messages) 作为 reducer 函数的图示例。
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langchain.messages import AnyMessage
from langgraph.graph.message import add_messages
from typing import Annotated
from typing_extensions import TypedDict
class GraphState(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
```
#### MessagesState
由于在状态中包含消息列表非常常见，因此存在一个预构建的状态 `MessagesState`，使使用消息变得简单。`MessagesState` 定义了一个 `messages` 键，它是 `AnyMessage` 对象的列表，并使用 [`add_messages`](https://reference.langchain.com/python/langgraph/graph/message/add_messages) reducer。通常，除了消息之外还需要跟踪更多状态，因此我们经常看到人们继承此状态并添加更多字段，如：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langgraph.graph import MessagesState
class State(MessagesState):
documents: list[str]
```
## 节点
在 LangGraph 中，节点是 Python 函数（同步或异步），接受以下参数：
1. `state`——图的[状态](#state)
2. `config`——一个 [`RunnableConfig`](https://reference.langchain.com/python/langchain-core/runnables/config/RunnableConfig) 对象，包含配置信息（如 `thread_id`）和追踪信息（如 `tags`）
3. `runtime`——一个 `Runtime` 对象，包含[运行时 `context`](#runtime-context) 以及其他信息，如 `store`、`stream_writer`、`execution_info`、`server_info`、`heartbeat`（用于空闲超时刷新）和 `control`（用于[优雅关闭](/oss/python/langgraph/durable-execution#graceful-shutdown)）
与 NetworkX 类似，你使用 [`add_node`](https://reference.langchain.com/python/langgraph/graph/state/StateGraph/add_node) 方法将这些节点添加到图中：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from dataclasses import dataclass
from typing_extensions import TypedDict
from langgraph.graph import StateGraph
from langgraph.runtime import Runtime
class State(TypedDict):
input: str
results: str
@dataclass
class Context:
user_id: str
builder = StateGraph(State)
def plain_node(state: State):
return state
def node_with_runtime(state: State, runtime: Runtime[Context]):
print("In node: ", runtime.context.user_id)
return {"results": f"Hello, {state['input']}!"}
def node_with_execution_info(state: State, runtime: Runtime):
print("In node with thread_id: ", runtime.execution_info.thread_id) # [!code highlight]
return {"results": f"Hello, {state['input']}!"}
builder.add_node("plain_node", plain_node)
builder.add_node("node_with_runtime", node_with_runtime)
builder.add_node("node_with_execution_info", node_with_execution_info)
...
```
在幕后，函数被转换为 [`RunnableLambda`](https://reference.langchain.com/python/langchain-core/runnables/base/RunnableLambda)，它为你的函数添加了批处理和异步支持，以及[原生追踪和调试](/langsmith/home)。
如果你在添加节点到图时没有指定名称，它将被赋予一个等同于函数名的默认名称。
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
builder.add_node(my_node)
# 然后你可以通过引用 `"my_node"` 来创建到/从此节点的边
```
### `START` 节点
[`START`](https://reference.langchain.com/python/langgraph/constants/START) 节点是一个特殊节点，表示向图发送用户输入的节点。引用此节点的主要目的是确定应首先调用哪些节点。
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langgraph.graph import START
graph.add_edge(START, "node_a")
```
### `END` 节点
`END` 节点是一个特殊节点，表示终端节点。当你想表示哪些边在完成后没有后续操作时，引用此节点。
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langgraph.graph import END
graph.add_edge("node_a", END)
```
### 节点缓存
LangGraph 支持基于节点输入的任务/节点缓存。要使用缓存：
* 在编译图（或指定入口点）时指定缓存
* 为节点指定缓存策略。每个缓存策略支持：
* `key_func`：用于基于节点输入生成缓存键，默认使用 pickle 的输入 `hash`。
* `ttl`：缓存的存活时间（秒）。如果未指定，缓存将永不过期。
例如：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
import time
from typing_extensions import TypedDict
from langgraph.graph import StateGraph
from langgraph.cache.memory import InMemoryCache
from langgraph.types import CachePolicy
class State(TypedDict):
x: int
result: int
builder = StateGraph(State)
def expensive_node(state: State) -> dict[str, int]:
# 耗时计算
time.sleep(2)
return {"result": state["x"] * 2}
builder.add_node("expensive_node", expensive_node, cache_policy=CachePolicy(ttl=3))
builder.set_entry_point("expensive_node")
builder.set_finish_point("expensive_node")
graph = builder.compile(cache=InMemoryCache())
print(graph.invoke({"x": 5}, stream_mode='updates')) # [!code highlight]
# [{'expensive_node': {'result': 10}}]
print(graph.invoke({"x": 5}, stream_mode='updates')) # [!code highlight]
# [{'expensive_node': {'result': 10}, '__metadata__': {'cached': True}}]
```
1. 第一次运行需要两秒（由于模拟的耗时计算）。
2. 第二次运行利用缓存并快速返回。
## 边
边定义了逻辑如何路由以及图如何决定停止。这是智能体工作方式以及不同节点之间通信的重要部分。有几种关键类型的边：
* 普通边：直接从一个节点转到下一个节点。
* 条件边：调用函数来确定接下来转到哪个节点。
* 入口点：当用户输入到达时首先调用哪个节点。
* 条件入口点：调用函数来确定当用户输入到达时首先调用哪个节点。
一个节点可以有多个出边。如果一个节点有多个出边，**所有**目标节点将作为下一个超级步的一部分并行执行。
对于每个节点，选择一种路由机制：使用普通边进行静态路由，或使用条件边/[`Command`](https://reference.langchain.com/python/langgraph/types/Command) 进行动态路由。不要混合使用普通边和来自同一节点的动态路由，因为两条路径都可能执行，使图行为更难理解。
### 普通边
如果你**总是**想从节点 A 转到节点 B，可以直接使用 [`add_edge`](https://reference.langchain.com/python/langgraph/pregel/_draw/add_edge) 方法。
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
graph.add_edge("node_a", "node_b")
```
### 条件边
如果你想**有条件地**路由到一个或多个边（或有条件地终止），可以使用 [`add_conditional_edges`](https://reference.langchain.com/python/langgraph/graph/state/StateGraph/add_conditional_edges) 方法。此方法接受节点名称和在该节点执行后调用的"路由函数"：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
graph.add_conditional_edges("node_a", routing_function)
```
与节点类似，`routing_function` 接受图的当前`状态`并返回一个值。
默认情况下，`routing_function` 的返回值用作要将状态发送到的节点（或节点列表）的名称。所有这些节点将作为下一个超级步的一部分并行运行。
你可以选择性地提供一个字典，将 `routing_function` 的输出映射到下一个节点的名称。
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
graph.add_conditional_edges("node_a", routing_function, {True: "node_b", False: "node_c"})
```
如果你想在单个函数中组合状态更新和路由，请使用 [`Command`](#command) 代替条件边。
### 入口点
入口点是图启动时运行的第一个节点。你可以使用 [`add_edge`](https://reference.langchain.com/python/langgraph/pregel/_draw/add_edge) 方法从虚拟 [`START`](https://reference.langchain.com/python/langgraph/constants/START) 节点到第一个要执行的节点来指定图的入口。
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langgraph.graph import START
graph.add_edge(START, "node_a")
```
### 条件入口点
条件入口点允许你根据自定义逻辑从不同的节点开始。你可以使用从虚拟 [`START`](https://reference.langchain.com/python/langgraph/constants/START) 节点出发的 [`add_conditional_edges`](https://reference.langchain.com/python/langgraph/graph/state/StateGraph/add_conditional_edges) 来实现。
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langgraph.graph import START
graph.add_conditional_edges(START, routing_function)
```
你可以选择性地提供一个字典，将 `routing_function` 的输出映射到下一个节点的名称。
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
graph.add_conditional_edges(START, routing_function, {True: "node_b", False: "node_c"})
```
## `Send`
默认情况下，`节点`和`边`是预先定义的，并在相同的共享状态上运行。然而，在某些情况下，确切的边可能事先未知，和/或你可能希望不同版本的`状态`同时存在。一个常见的例子是 [map-reduce](/oss/python/langgraph/use-graph-api#map-reduce-and-the-send-api) 设计模式。在此设计模式中，第一个节点可能生成一个对象列表，你可能希望将某个其他节点应用于所有这些对象。对象的数量可能事先未知（意味着边的数量可能未知），并且下游`节点`的输入`状态`应该不同（每个生成的对象一个）。
为了支持此设计模式，LangGraph 支持从条件边返回 [`Send`](https://reference.langchain.com/python/langgraph/types/Send) 对象。`Send` 接受两个参数：第一个是节点名称，第二个是要传递给该节点的状态。
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langgraph.types import Send
def continue_to_jokes(state: OverallState):
return [Send("generate_joke", {"subject": s}) for s in state['subjects']]
graph.add_conditional_edges("node_a", continue_to_jokes)
```
## `Command`
[`Command`](https://reference.langchain.com/python/langgraph/types/Command) 是一个用于控制图执行的多功能原语。它接受四个参数：
* `update`：应用状态更新（类似于从节点返回更新）。
* `goto`：导航到特定节点（类似于[条件边](#conditional-edges)）。
* `graph`：从[子图](/oss/python/langgraph/use-subgraphs)导航时指定父图。
* `resume`：提供一个值以在[中断](/oss/python/langgraph/interrupts)后恢复执行。
`Command` 在三种上下文中使用：
* **[从节点返回](#return-from-nodes)**：使用 `update`、`goto` 和 `graph` 将状态更新与控制流结合。
* **[作为 `invoke` 或 `stream` 的输入](#input-to-invoke-or-stream)**：使用 `resume` 在中断后继续执行。
* **[从工具返回](#return-from-tools)**：类似于从节点返回，从工具内部组合状态更新和控制流。
### 从节点返回
#### `update` 和 `goto`
从节点函数返回 [`Command`](https://reference.langchain.com/python/langgraph/types/Command) 以在单步中更新状态并路由到下一个节点：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
def my_node(state: State) -> Command[Literal["my_other_node"]]:
return Command(
# 状态更新
update={"foo": "bar"},
# 控制流
goto="my_other_node"
)
```
使用 [`Command`](https://reference.langchain.com/python/langgraph/types/Command) 你还可以实现动态控制流行为（与[条件边](#conditional-edges)相同）：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
def my_node(state: State) -> Command[Literal["my_other_node"]]:
if state["foo"] == "bar":
return Command(update={"foo": "baz"}, goto="my_other_node")
```
当你需要**同时**更新状态**和**路由到不同节点时，使用 [`Command`](https://reference.langchain.com/python/langgraph/types/Command)。如果你只需要路由而不更新状态，请改用[条件边](#conditional-edges)。
在节点函数中返回 [`Command`](https://reference.langchain.com/python/langgraph/types/Command) 时，你必须添加包含节点可路由到的节点名称列表的返回类型注解，例如 `Command[Literal["my_other_node"]]`。这对于图渲染是必要的，并告诉 LangGraph `my_node` 可以导航到 `my_other_node`。
[`Command`](https://reference.langchain.com/python/langgraph/types/Command) 只添加动态边——使用 `add_edge` / `addEdge` 定义的静态边仍然会执行。例如，如果 `node_a` 返回 `Command(goto="my_other_node")` 而你也有 `graph.add_edge("node_a", "node_b")`，则 `node_b` 和 `my_other_node` 都会运行。对于每个节点，使用 [`Command`](https://reference.langchain.com/python/langgraph/types/Command) 或静态边路由到下一个节点，而不是两者兼用。
查看此[操作指南](/oss/python/langgraph/use-graph-api#combine-control-flow-and-state-updates-with-command)了解如何使用 [`Command`](https://reference.langchain.com/python/langgraph/types/Command) 的端到端示例。
#### `graph`
如果你使用[子图](/oss/python/langgraph/use-subgraphs)，你可以通过在 [`Command`](https://reference.langchain.com/python/langgraph/types/Command) 中指定 `graph=Command.PARENT` 来从子图中的节点导航到父图中的不同节点：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
def my_node(state: State) -> Command[Literal["other_subgraph"]]:
return Command(
update={"foo": "bar"},
goto="other_subgraph", # 其中 `other_subgraph` 是父图中的一个节点
graph=Command.PARENT
)
```
将 `graph` 设置为 `Command.PARENT` 将导航到最近的父图。
当你从子图节点向父图节点发送一个同时被父图和子图[状态模式](#schema)共享的键的更新时，你**必须**在父图状态中为你要更新的键定义一个 [reducer](#reducers)。参见此[示例](/oss/python/langgraph/use-graph-api#navigate-to-a-node-in-a-parent-graph)。
这在实现[多智能体交接](/oss/python/langchain/multi-agent/handoffs)时特别有用。查看[导航到父图中的节点](/oss/python/langgraph/use-graph-api#navigate-to-a-node-in-a-parent-graph)了解详情。
### 作为 `invoke` 或 `stream` 的输入
`Command(resume=...)` 是**唯一**用作 `invoke()`/`stream()` 输入的 `Command` 模式。不要使用 `Command(update=...)` 作为输入来继续多轮对话——因为将任何 `Command` 作为输入传递会从最新检查点（即最后运行的步骤，而不是 `__start__`）恢复，如果图已完成，图将看起来卡住了。要在现有线程上继续对话，请传递一个普通输入字典：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
# 错误 - 图从最新检查点（最后运行的步骤）恢复，看起来卡住了
graph.invoke(Command(update={ # [!code --]
"messages": [{"role": "user", "content": "follow up"}] # [!code --]
}), config) # [!code --]
# 正确 - 普通字典从 __start__ 重新开始
graph.invoke( { # [!code ++]
"messages": [{"role": "user", "content": "follow up"}] # [!code ++]
}, config) # [!code ++]
```
#### `resume`
使用 `Command(resume=...)` 提供一个值并在[中断](/oss/python/langgraph/interrupts)后恢复图执行。传递给 `resume` 的值成为暂停节点内 `interrupt()` 调用的返回值：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langgraph.types import Command, interrupt
def human_review(state: State):
# 暂停图并等待一个值
answer = interrupt("Do you approve?")
return {"messages": [{"role": "user", "content": answer}]}
# 第一次调用 - 命中中断并暂停
result = graph.invoke({"messages": [...]}, config)
# 使用一个值恢复 - interrupt() 调用返回 "yes"
result = graph.invoke(Command(resume="yes"), config)
```
查看[中断概念指南](/oss/python/langgraph/interrupts)了解中断模式的完整详情，包括多重中断和验证循环。
### 从工具返回
你可以从工具返回 [`Command`](https://reference.langchain.com/python/langgraph/types/Command) 以更新图状态和控制流。使用 `update` 修改状态（例如，保存在对话中查找到的客户信息），使用 `goto` 在工具完成后路由到特定节点。
在工具内部使用时，`goto` 添加动态边——调用工具的节点上已定义的任何静态边仍将执行。对于每个节点，使用工具驱动的动态路由或静态边路由到下一个节点，而不是两者兼用。
详情请参阅[在工具内部使用](/oss/python/langgraph/use-graph-api#use-inside-tools)。
## 图迁移
LangGraph 可以轻松处理图定义（节点、边和状态）的迁移，即使在使用检查点器跟踪状态时也是如此。
* 对于处于图末端的线程（即未中断），你可以更改图的整个拓扑结构（即所有节点和边、移除、添加、重命名等）
* 对于当前被中断的线程，我们支持除重命名/移除节点之外的所有拓扑更改（因为该线程可能即将进入一个不再存在的节点）——如果这是一个阻碍因素，请联系我们，我们可以优先解决。
* 对于修改状态，我们对添加和删除键具有完全的向后和向前兼容性
* 被重命名的状态键会丢失现有线程中的保存状态
* 以不兼容方式更改类型的状态键目前可能在包含更改前状态的线程中导致问题——如果这是一个阻碍因素，请联系我们，我们可以优先解决。
## 运行时上下文
创建图时，你可以指定一个 `context_schema` 用于传递给节点的运行时上下文。这对于向节点传递不属于图状态的信息很有用。例如，你可能想传递依赖项，如模型名称或数据库连接。
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
@dataclass
class ContextSchema:
llm_provider: str = "openai"
graph = StateGraph(State, context_schema=ContextSchema)
```
然后你可以使用 `invoke` 方法的 `context` 参数将此上下文传递给图。
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
graph.invoke(inputs, context={"llm_provider": "anthropic"})
```
你可以在节点或条件边内部访问和使用此上下文：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langgraph.runtime import Runtime
def node_a(state: State, runtime: Runtime[ContextSchema]):
llm = get_llm(runtime.context.llm_provider)
# ...
```
参见[添加运行时配置](/oss/python/langgraph/use-graph-api#add-runtime-configuration)了解配置的完整说明。
### 递归限制
递归限制设置了图在单次执行期间可以执行的最大[超级步](#graphs)数。一旦达到限制，LangGraph 将引发 `GraphRecursionError`。从 1.0.6 版本开始，默认递归限制设置为 1000 步。递归限制可以在运行时对任何图设置，并通过 config 字典传递给 `invoke`/`stream`。重要的是，`recursion_limit` 是一个独立的 `config` 键，不应在 `configurable` 键内传递，因为所有其他用户定义的配置都在那里。参见下面的示例：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
graph.invoke(inputs, config={"recursion_limit": 5}, context={"llm": "anthropic"})
```
阅读[递归限制](/oss/python/langgraph/graph-api#recursion-limit)了解更多关于递归限制工作原理的信息。
### 访问和处理递归计数器
当前步骤计数器可在任何节点中通过 `config["metadata"]["langgraph_step"]` 访问，允许在达到递归限制之前进行主动的递归处理。这使你能够在图逻辑中实现优雅降级策略。
#### 工作原理
步骤计数器存储在 `config["metadata"]["langgraph_step"]` 中。递归限制检查遵循以下逻辑：`step > stop`，其中 `stop = step + recursion_limit + 1`。当超过限制时，LangGraph 引发 `GraphRecursionError`。
#### 访问当前步骤计数器
你可以在任何节点中访问当前步骤计数器以监控执行进度。
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph
def my_node(state: dict, config: RunnableConfig) -> dict:
current_step = config["metadata"]["langgraph_step"]
print(f"Currently on step: {current_step}")
return state
```
#### 主动递归处理
LangGraph 提供了一个 `RemainingSteps` 管理值，用于跟踪在达到递归限制之前还剩多少步。这允许在图中进行优雅降级。
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from typing import Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.managed import RemainingSteps
class State(TypedDict):
messages: Annotated[list, lambda x, y: x + y]
remaining_steps: RemainingSteps # 管理值 - 跟踪直到限制的步数
def reasoning_node(state: State) -> dict:
# RemainingSteps 由 LangGraph 自动填充
remaining = state["remaining_steps"]
# 检查是否步数即将用尽
if remaining <= 2:
return {"messages": ["Approaching limit, wrapping up..."]}
# 正常处理
return {"messages": ["thinking..."]}
def route_decision(state: State) -> Literal["reasoning_node", "fallback_node"]:
"""根据剩余步数进行路由"""
if state["remaining_steps"] <= 2:
return "fallback_node"
return "reasoning_node"
def fallback_node(state: State) -> dict:
"""处理递归限制即将到达的情况"""
return {"messages": ["Reached complexity limit, providing best effort answer"]}
# 构建图
builder = StateGraph(State)
builder.add_node("reasoning_node", reasoning_node)
builder.add_node("fallback_node", fallback_node)
builder.add_edge(START, "reasoning_node")
builder.add_conditional_edges("reasoning_node", route_decision)
builder.add_edge("fallback_node", END)
graph = builder.compile()
# RemainingSteps 适用于任何 recursion_limit
result = graph.invoke({"messages": []}, {"recursion_limit": 10})
```
#### 主动与被动方法
处理递归限制有两种主要方法：主动（在图内监控）和被动（在外部捕获错误）。
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from typing import Annotated, Literal, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.managed import RemainingSteps
from langgraph.errors import GraphRecursionError
class State(TypedDict):
messages: Annotated[list, lambda x, y: x + y]
remaining_steps: RemainingSteps
# 主动方法（推荐）- 使用 RemainingSteps
def agent_with_monitoring(state: State) -> dict:
"""在图内主动监控和处理递归"""
remaining = state["remaining_steps"]
# 提前检测 - 路由到内部处理
if remaining <= 2:
return {
"messages": ["Approaching limit, returning partial result"]
}
# 正常处理
return {"messages": [f"Processing... ({remaining} steps remaining)"]}
def route_decision(state: State) -> Literal["agent", END]:
if state["remaining_steps"] <= 2:
return END
return "agent"
# 构建图
builder = StateGraph(State)
builder.add_node("agent", agent_with_monitoring)
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", route_decision)
graph = builder.compile()
# 主动：图优雅完成
result = graph.invoke({"messages": []}, {"recursion_limit": 10})
# 被动方法（后备）- 在外部捕获错误
try:
result = graph.invoke({"messages": []}, {"recursion_limit": 10})
except GraphRecursionError as e:
# 在图执行失败后在外部处理
result = {"messages": ["Fallback: recursion limit exceeded"]}
```
这些方法之间的主要区别是：
| 方法 | 检测 | 处理 | 控制流 |
| ---------------------------- | ------- | ----------------- | -------- |
| 主动（使用 `RemainingSteps`） | 在达到限制之前 | 在图内部通过条件路由 | 图继续到完成节点 |
| 被动（捕获 `GraphRecursionError`） | 在超过限制之后 | 在图外部的 try/catch 中 | 图执行终止 |
**主动方法的优势：**
* 在图内优雅降级
* 可以在检查点中保存中间状态
* 通过部分结果提供更好的用户体验
* 图正常完成（无异常）
**被动方法的优势：**
* 实现更简单
* 无需修改图逻辑
* 集中式错误处理
#### 其他可用元数据
除了 `langgraph_step` 之外，以下元数据也可在 `config["metadata"]` 中获取：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
def inspect_metadata(state: dict, config: RunnableConfig) -> dict:
metadata = config["metadata"]
print(f"Step: {metadata['langgraph_step']}")
print(f"Node: {metadata['langgraph_node']}")
print(f"Triggers: {metadata['langgraph_triggers']}")
print(f"Path: {metadata['langgraph_path']}")
print(f"Checkpoint NS: {metadata['langgraph_checkpoint_ns']}")
return state
```
## 可视化
能够可视化图通常很有帮助，特别是当图变得更复杂时。LangGraph 提供了几种内置的图可视化方式。更多信息请参阅[可视化你的图](/oss/python/langgraph/use-graph-api#visualize-your-graph)。
## 可观测性与追踪
要追踪、调试和评估你的智能体，请使用 [LangSmith](/langsmith/home)。
## 了解更多
* [如何使用图 API](/oss/python/langgraph/use-graph-api)
* [函数式 API 概念概述](/oss/python/langgraph/functional-api)
* [选择图 API 还是函数式 API](/oss/python/langgraph/choosing-apis)
***
通过 MCP [连接这些文档](/use-these-docs)到 Claude、VSCode 等工具，获取实时答案。
[在 GitHub 上编辑此页面](https://github.com/langchain-ai/docs/edit/main/src/oss/langgraph/graph-api.mdx)或[提交问题](https://github.com/langchain-ai/docs/issues/new/choose)。