这是用户在 2025-7-28 13:04 为 https://langchain-ai.github.io/langgraph/concepts/low_level/ 保存的双语快照页面,由 沉浸式翻译 提供双语支持。了解如何保存?
Skip to content

Graph API concepts
绘制 API 概念图

Graphs  图表

At its core, LangGraph models agent workflows as graphs. You define the behavior of your agents using three key components:
LangGraph 的核心是将代理工作流建模为图形。您可以使用三个关键组件定义代理的行为:

  1. State: A shared data structure that represents the current snapshot of your application. It can be any Python type, but is typically a TypedDict or Pydantic BaseModel.
    状态 :一个共享的数据结构,表示应用程序的当前快照。它可以是任何 Python 类型,但通常是 TypedDict 或 PydanticBaseModel

  2. Nodes: Python functions that encode the logic of your agents. They receive the current State as input, perform some computation or side-effect, and return an updated State.
    Nodes:编码代理逻辑的 Python 函数。它们接收当前 State 作为输入,执行一些计算或副作用,并返回更新的 State

  3. Edges: Python functions that determine which Node to execute next based on the current State. They can be conditional branches or fixed transitions.
    Edges:Python 函数,根据当前状态确定接下来执行哪个节点 。它们可以是条件分支或固定转换。

By composing Nodes and Edges, you can create complex, looping workflows that evolve the State over time. The real power, though, comes from how LangGraph manages that State. To emphasize: Nodes and Edges are nothing more than Python functions - they can contain an LLM or just good ol' Python code.
通过组合节点,您可以创建复杂的循环工作流,这些工作流会随着时间的推移而演变状态 。然而,真实的力量来自 LangGraph 如何管理那个状态 。需要强调的是: 节点边只不过是 Python 函数--它们可以包含一个 LLM 或只是好的 Python 代码。

In short: nodes do the work, edges tell what to do next.
简而言之: 节点完成工作,边告诉下一步该做什么

LangGraph's underlying graph algorithm uses message passing to define a general program. When a Node completes its operation, it sends messages along one or more edges to other node(s). These recipient nodes then execute their functions, pass the resulting messages to the next set of nodes, and the process continues. Inspired by Google's Pregel system, the program proceeds in discrete "super-steps."
LangGraph 的底层图形算法使用消息传递来定义通用程序。当节点完成其操作时,它沿着沿着一个或多个边向其他节点发送消息。然后,这些接收节点执行它们的功能,将结果消息传递给下一组节点,并且该过程继续。受谷歌 Pregel 系统的启发,该程序以离散的“超级步骤”进行。"

A super-step can be considered a single iteration over the graph nodes. Nodes that run in parallel are part of the same super-step, while nodes that run sequentially belong to separate super-steps. At the start of graph execution, all nodes begin in an inactive state. A node becomes active when it receives a new message (state) on any of its incoming edges (or "channels"). The active node then runs its function and responds with updates. At the end of each super-step, nodes with no incoming messages vote to halt by marking themselves as inactive. The graph execution terminates when all nodes are inactive and no messages are in transit.
超级步骤可以被认为是图节点上的单个迭代。并行运行的节点是同一个超级步骤的一部分,而顺序运行的节点属于单独的超级步骤。在图形执行开始时,所有节点开始都处于非活动状态。当一个节点在它的任何一个输入边(或“通道”)上接收到一个新的消息(状态)时,它就变成活动的 。然后,活动节点运行其功能并响应更新。在每个超级步骤结束时,没有传入消息的节点通过将自己标记为不活动来投票停止 。当所有节点都不活动并且没有消息在传输中时,图形执行终止。

StateGraph  状态图

The StateGraph class is the main graph class to use. This is parameterized by a user defined State object.
StateGraph 类是要使用的主要图形类。这由用户定义的 State 对象参数化。

Compiling your graph
编译您的图表

To build your graph, you first define the state, you then add nodes and edges, and then you compile it. What exactly is compiling your graph and why is it needed?
要构建图,首先要定义状态 ,然后添加节点,然后编译它。究竟是什么在编译图,为什么需要它?

Compiling is a pretty simple step. It provides a few basic checks on the structure of your graph (no orphaned nodes, etc). It is also where you can specify runtime args like checkpointers and breakpoints. You compile your graph by just calling the .compile method:
编译是一个非常简单的步骤。它提供了一些关于图结构的基本检查(没有孤立节点等)。也可以在这里指定运行时参数,如检查指针和断点。你可以通过调用 .compile 方法来编译你的图:

graph = graph_builder.compile(...)

You MUST compile your graph before you can use it.
必须在使用它之前编译你的图表。

State  状态

The first thing you do when you define a graph is define the State of the graph. The State consists of the schema of the graph as well as reducer functions which specify how to apply updates to the state. The schema of the State will be the input schema to all Nodes and Edges in the graph, and can be either a TypedDict or a Pydantic model. All Nodes will emit updates to the State which are then applied using the specified reducer function.
定义图时要做的第一件事是定义图的 StateState图的模式以及 reducer 函数组成,reducer 函数指定如何将更新应用于状态。State 的模式将成为图中所有节点的输入模式,并且可以是 TypedDictPydantic 模型。所有节点将发出状态更新,然后使用指定的 reducer 函数应用。

Schema  模式

The main documented way to specify the schema of a graph is by using a TypedDict. If you want to provide default values in your state, use a dataclass. We also support using a Pydantic BaseModel as your graph state if you want recursive data validation (though note that pydantic is less performant than a TypedDict or dataclass).
指定图模式的主要方法是使用 TypedDict。如果你想在你的状态下提供默认值,使用一个类 。如果你想要递归的数据验证,我们也支持使用 PydanticBaseModel 作为你的图状态(尽管注意 pydantic 的性能不如 TypedDict)。

By default, the graph will have the same input and output schemas. If you want to change this, you can also specify explicit input and output schemas directly. This is useful when you have a lot of keys, and some are explicitly for input and others for output. See the guide here for how to use.
默认情况下,图形将具有相同的输入和输出模式。如果您想改变这一点,也可以直接指定显式的输入和输出模式。当您有很多键时,这很有用,其中一些键显式用于输入,另一些用于输出。请参阅此处的指南了解如何使用。

Multiple schemas
多个模式

Typically, all graph nodes communicate with a single schema. This means that they will read and write to the same state channels. But, there are cases where we want more control over this:
通常,所有图节点都与单个模式通信。这意味着它们将读取和写入相同的状态通道。但是,在某些情况下,我们希望对此进行更多的控制:

  • Internal nodes can pass information that is not required in the graph's input / output.
    内部节点可以传递图的输入/输出中不需要的信息。
  • We may also want to use different input / output schemas for the graph. The output might, for example, only contain a single relevant output key.
    我们可能还希望为图使用不同的输入/输出模式。例如,输出可能仅包含单个相关输出键。

It is possible to have nodes write to private state channels inside the graph for internal node communication. We can simply define a private schema, PrivateState. See this guide for more detail.
可以让节点写入图内部的私有状态通道,以进行内部节点通信。我们可以简单地定义一个私有模式 PrivateState。请参阅本指南了解更多详细信息。

It is also possible to define explicit input and output schemas for a graph. In these cases, we define an "internal" schema that contains all keys relevant to graph operations. But, we also define input and output schemas that are sub-sets of the "internal" schema to constrain the input and output of the graph. See this guide for more detail.
还可以为图定义显式的输入和输出模式。在这些情况下,我们定义了一个“内部”模式,其中包含与图操作相关的所有键。但是,我们还定义了输入输出模式,它们是“内部”模式的子集,以约束图的输入和输出。请参阅本指南了解更多详细信息。

Let's look at an example:
让我们看一个例子:

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:
    # Write to OverallState
    return {"foo": state["user_input"] + " name"}

def node_2(state: OverallState) -> PrivateState:
    # Read from OverallState, write to PrivateState
    return {"bar": state["foo"] + " is"}

def node_3(state: PrivateState) -> OutputState:
    # Read from PrivateState, write to 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'}

There are two subtle and important points to note here:
这里有两个微妙而重要的要点需要注意:

  1. We pass state: InputState as the input schema to node_1. But, we write out to foo, a channel in OverallState. How can we write out to a state channel that is not included in the input schema? This is because a node can write to any state channel in the graph state. The graph state is the union of the state channels defined at initialization, which includes OverallState and the filters InputState and OutputState.
    我们将 state:InputState 作为输入模式传递给 node_1。但是,我们写信给 fooOverallState 中的一个频道。我们如何写出一个不包含在输入模式中的状态通道?这是因为节点可以写入图状态中的任何状态通道。 图状态是在初始化时定义的状态通道的联合,其包括 OverallState 以及过滤器 InputStateOutputState

  2. We initialize the graph with StateGraph(OverallState,input_schema=InputState,output_schema=OutputState). So, how can we write to PrivateState in node_2? How does the graph gain access to this schema if it was not passed in the StateGraph initialization? We can do this because nodes can also declare additional state channels as long as the state schema definition exists. In this case, the PrivateState schema is defined, so we can add bar as a new state channel in the graph and write to it.
    我们用 StateGraph(OverallState,input_schema=InputState,output_schema=OutputState) 初始化图。那么,我们如何在 node_2 中写入 PrivateState?如果在 StateGraph 初始化中没有传递这个模式,那么图如何访问这个模式?我们可以这样做,因为只要状态模式定义存在, 节点也可以声明其他状态通道 。在本例中,定义了 PrivateState 模式,因此我们可以在图中添加 bar 作为新的状态通道,并对其进行写入。

Reducers  还原器

Reducers are key to understanding how updates from nodes are applied to the State. Each key in the State has its own independent reducer function. If no reducer function is explicitly specified then it is assumed that all updates to that key should override it. There are a few different types of reducers, starting with the default type of reducer:
Reducer 是了解如何将节点的更新应用于 State 的关键。State 中的每个键都有自己独立的 reducer 函数。如果没有明确指定 reducer 函数,那么假设对该键的所有更新都应该覆盖它。有几种不同类型的 reducer,从默认的 reducer 类型开始:

Default Reducer
默认 Reducer

These two examples show how to use the default reducer:
这两个例子展示了如何使用默认的 reducer:

Example A:  示例 A:

from typing_extensions import TypedDict

class State(TypedDict):
    foo: int
    bar: list[str]

In this example, no reducer functions are specified for any key. Let's assume the input to the graph is {"foo": 1, "bar": ["hi"]}. Let's then assume the first Node returns {"foo": 2}. This is treated as an update to the state. Notice that the Node does not need to return the whole State schema - just an update. After applying this update, the State would then be {"foo": 2, "bar": ["hi"]}. If the second node returns {"bar": ["bye"]} then the State would then be {"foo": 2, "bar": ["bye"]}
在本例中,没有为任何键指定 reducer 函数。让我们假设图的输入是 {“foo”:1,“bar”:[“hi”]}。假设第一个 Node 返回 {“foo”:2}。这被视为状态的更新。注意,Node 不需要返回整个 State 模式--只需要返回一个更新。应用此更新后,State 将为 {“foo”:2,“bar”:[“hi”]}。如果第二个节点返回 {“bar”:[“bye”]},则 State 将为 {“foo”:2,“bar”:[“bye”]}

Example B:  示例 B:

from typing import Annotated
from typing_extensions import TypedDict
from operator import add

class State(TypedDict):
    foo: int
    bar: Annotated[list[str], add]

In this example, we've used the Annotated type to specify a reducer function (operator.add) for the second key (bar). Note that the first key remains unchanged. Let's assume the input to the graph is {"foo": 1, "bar": ["hi"]}. Let's then assume the first Node returns {"foo": 2}. This is treated as an update to the state. Notice that the Node does not need to return the whole State schema - just an update. After applying this update, the State would then be {"foo": 2, "bar": ["hi"]}. If the second node returns {"bar": ["bye"]} then the State would then be {"foo": 2, "bar": ["hi", "bye"]}. Notice here that the bar key is updated by adding the two lists together.
在这个例子中,我们使用了 Annotated 类型为第二个键(bar)指定了一个 reducer 函数(operator.add)。请注意,第一个键保持不变。让我们假设图的输入是 {“foo”:1,“bar”:[“hi”]}。假设第一个 Node 返回 {“foo”:2}。这被视为状态的更新。注意,Node 不需要返回整个 State 模式--只需要返回一个更新。应用此更新后,State 将为 {“foo”:2,“bar”:[“hi”]}。如果第二个节点返回 {“bar”:[“bye”]},则 State 将是 {"foo": 2, "bar": ["hi", "bye"]} 。请注意,这里的 bar 键是通过将两个列表添加到一起来更新的。

Working with Messages in Graph State
在 Graph State 中使用消息

Why use messages?
为什么使用消息?

Most modern LLM providers have a chat model interface that accepts a list of messages as input. LangChain's ChatModel in particular accepts a list of Message objects as inputs. These messages come in a variety of forms such as HumanMessage (user input) or AIMessage (LLM response). To read more about what message objects are, please refer to this conceptual guide.
大多数现代 LLM 提供者都有一个聊天模型接口,可以接受一系列消息作为输入。LangChain 的 ChatModel 特别接受 Message 对象列表作为输入。这些消息有多种形式,例如 HumanMessage(用户输入)或 AIMessage(LLM 响应)。要了解更多关于什么是消息对象的信息,请参阅概念指南。

Using Messages in your Graph
在图中使用消息

In many cases, it is helpful to store prior conversation history as a list of messages in your graph state. To do so, we can add a key (channel) to the graph state that stores a list of Message objects and annotate it with a reducer function (see messages key in the example below). The reducer function is vital to telling the graph how to update the list of Message objects in the state with each state update (for example, when a node sends an update). If you don't specify a reducer, every state update will overwrite the list of messages with the most recently provided value.
在许多情况下,将先前的会话历史记录存储为图表状态中的消息列表是很有帮助的。为此,我们可以向存储 Message 对象列表的图状态添加一个键(通道),并使用 reducer 函数对其进行注释(参见下面示例中的 messages 键)。reducer 函数对于告诉图如何在每次状态更新时更新状态中的 Message 对象列表至关重要(例如,当节点发送更新时)。 如果没有指定 reducer,每次状态更新都会用最近提供的值覆盖消息列表。

If you wanted to simply append messages to the existing list, you could use operator.add as a reducer.
如果你想简单地将消息追加到现有列表中,你可以使用 operator.add 作为一个 reducer。

However, you might also want to manually update messages in your graph state (e.g. human-in-the-loop). If you were to use operator.add, the manual state updates you send to the graph would be appended to the existing list of messages, instead of updating existing messages. To avoid that, you need a reducer that can keep track of message IDs and overwrite existing messages, if updated.
但是,您可能还希望手动更新图形状态中的消息(例如,人在回路中)。如果使用 operator.add,则发送到图的手动状态更新将附加到现有消息列表中,而不是更新现有消息。为了避免这种情况,您需要一个 reducer,它可以跟踪消息 ID 并在更新时覆盖现有消息。

To achieve this, you can use the prebuilt add_messages function. For brand new messages, it will simply append to existing list, but it will also handle the updates for existing messages correctly.
要实现这一点,您可以使用预构建的 add_messages 函数。对于全新的消息,它将简单地添加到现有列表中,但它也将正确处理现有消息的更新。

Serialization
序列化

In addition to keeping track of message IDs, the add_messages function will also try to deserialize messages into LangChain Message objects whenever a state update is received on the messages channel. See more information on LangChain serialization/deserialization here. This allows sending graph inputs / state updates in the following format:
除了跟踪消息 ID,add_messages 函数还将尝试在消息通道上接收到状态更新时将消息转换为 LangChainMessage 对象。 在这里查看更多关于 LangChain 序列化/并行化的信息。这允许以以下格式发送图形输入/状态更新:

# this is supported
{"messages": [HumanMessage(content="message")]}

# and this is also supported
{"messages": [{"type": "human", "content": "message"}]}

Since the state updates are always deserialized into LangChain Messages when using add_messages, you should use dot notation to access message attributes, like state["messages"][-1].content. Below is an example of a graph that uses add_messages as its reducer function.
由于使用 add_messages 时状态更新总是被转换为 LangChainMessages,因此您应该使用点表示法来访问消息属性,如 state[“messages”][-1].content。下面是一个使用 add_messages 作为其 reducer 函数的图的示例。

API Reference: AnyMessage | add_messages
API 参考:AnyMessage|添加_消息

from langchain_core.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 状态

Since having a list of messages in your state is so common, there exists a prebuilt state called MessagesState which makes it easy to use messages. MessagesState is defined with a single messages key which is a list of AnyMessage objects and uses the add_messages reducer. Typically, there is more state to track than just messages, so we see people subclass this state and add more fields, like:
由于在状态中有一个消息列表是很常见的,因此存在一个名为 MessagesState 的预构建状态,它可以很容易地使用消息。MessagesState 使用单个 messages 键定义,该键是 AnyMessage 对象的列表,并使用 add_messagesreducer。通常情况下,有更多的状态要跟踪,而不仅仅是消息,所以我们看到人们子类化这个状态并添加更多的字段,比如:

from langgraph.graph import MessagesState

class State(MessagesState):
    documents: list[str]

Nodes

In LangGraph, nodes are Python functions (either synchronous or asynchronous) that accept the following arguments:
在 LangGraph 中,节点是接受以下参数的 Python 函数(同步或异步):

  1. state: The state of the graph
    state:图的状态
  2. config: A RunnableConfig object that contains configuration information like thread_id and tracing information like tags
    configRunnableConfig 对象,包含配置信息(如 thread_id) 和跟踪信息(如 tags)
  3. runtime: A Runtime object that contains runtime context and other information like store and stream_writer
    runtime:一包含运行时上下文和其他信息(如 storestream_writer) 的对象

Similar to NetworkX, you add these nodes to a graph using the add_node method:
NetworkX 类似,您可以使用 add_node 方法将这些节点添加到图中:

API Reference: RunnableConfig | StateGraph
API 参考:RunnableConfig| StateGraph

from dataclasses import dataclass
from typing_extensions import TypedDict

from langchain_core.runnables import RunnableConfig
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_config(state: State, config: RunnableConfig):
    print("In node with thread_id: ", config["configurable"]["thread_id"])
    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_config", node_with_config)
...

Behind the scenes, functions are converted to RunnableLambdas, which add batch and async support to your function, along with native tracing and debugging.
在后台,函数被转换为 RunnableLambda,这为您的函数添加了批处理和调试支持,以及本机跟踪和调试沿着。

If you add a node to a graph without specifying a name, it will be given a default name equivalent to the function name.
如果您在没有指定名称的情况下将节点添加到图中,则会为该节点提供与函数名称等效的默认名称。

builder.add_node(my_node)
# You can then create edges to/from this node by referencing it as `"my_node"`

START Node
start node

The START Node is a special node that represents the node that sends user input to the graph. The main purpose for referencing this node is to determine which nodes should be called first.
START 节点是一个特殊的节点,它表示将用户输入发送到图形的节点。引用此节点的主要目的是确定应首先调用哪些节点。

API Reference: START  API 参考:START

from langgraph.graph import START

graph.add_edge(START, "node_a")

END Node
EndNode

The END Node is a special node that represents a terminal node. This node is referenced when you want to denote which edges have no actions after they are done.
END 节点是表示终端节点的特殊节点。当您要表示哪些边在完成后没有动作时,将引用此节点。

from langgraph.graph import END

graph.add_edge("node_a", END)

Node Caching
节点缓存

LangGraph supports caching of tasks/nodes based on the input to the node. To use caching:
LangGraph 支持基于节点输入的任务/节点缓存。要使用缓存,请执行以下操作:

  • Specify a cache when compiling a graph (or specifying an entrypoint)
    编译图形时指定缓存(或指定入口点)
  • Specify a cache policy for nodes. Each cache policy supports:
    为节点指定缓存策略。每个缓存策略支持:
    • key_func used to generate a cache key based on the input to a node, which defaults to a hash of the input with pickle.
      key_func 用于根据节点的输入生成缓存键,默认为 pickle 输入的哈希值
    • ttl, the time to live for the cache in seconds. If not specified, the cache will never expire.
      ttl,该高速缓存的生存时间(秒)。如果未指定,则该高速缓存永远不会过期。

For example:  举例来说:

API Reference: StateGraph
API 参考:StateGraph

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]:
    # expensive computation
    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'))  
[{'expensive_node': {'result': 10}}]
print(graph.invoke({"x": 5}, stream_mode='updates'))  
[{'expensive_node': {'result': 10}, '__metadata__': {'cached': True}}]

Edges  边缘

Edges define how the logic is routed and how the graph decides to stop. This is a big part of how your agents work and how different nodes communicate with each other. There are a few key types of edges:
边定义逻辑如何路由以及图如何决定停止。这是代理如何工作以及不同节点如何相互通信的重要组成部分。有几种关键类型的边:

  • Normal Edges: Go directly from one node to the next.
    法线边:直接从一个节点转到下一个节点。
  • Conditional Edges: Call a function to determine which node(s) to go to next.
    Conditional Edges:调用一个函数来确定下一个节点。
  • Entry Point: Which node to call first when user input arrives.
    入口点:当用户输入到达时,首先调用哪个节点。
  • Conditional Entry Point: Call a function to determine which node(s) to call first when user input arrives.
    条件入口点(Conditional Entry Point):调用一个函数来确定当用户输入到达时首先调用哪个节点。

A node can have MULTIPLE outgoing edges. If a node has multiple out-going edges, all of those destination nodes will be executed in parallel as a part of the next superstep.
一个节点可以有多个传出边。如果一个节点有多个向外的边, 所有这些目标节点将作为下一个超步的一部分并行执行。

Normal Edges
法线边

If you always want to go from node A to node B, you can use the add_edge method directly.
如果您总是想从节点 A 转到节点 B,可以直接使用 add_edge 方法。

graph.add_edge("node_a", "node_b")

Conditional Edges
条件边

If you want to optionally route to 1 or more edges (or optionally terminate), you can use the add_conditional_edges method. This method accepts the name of a node and a "routing function" to call after that node is executed:
如果您希望选择性地布线到 1 条或多条边(或选择性地终止),则可以使用 add_conditional_edges 方法。这个方法接受一个节点的名字和一个“路由函数”,在这个节点被执行后调用:

graph.add_conditional_edges("node_a", routing_function)

Similar to nodes, the routing_function accepts the current state of the graph and returns a value.
与节点类似,routing_function 接受图的当前状态并返回一个值。

By default, the return value routing_function is used as the name of the node (or list of nodes) to send the state to next. All those nodes will be run in parallel as a part of the next superstep.
默认情况下,返回值 routing_function 用作要将状态发送到 next 的节点(或节点列表)的名称。所有这些节点将作为下一个超级步骤的一部分并行运行。

You can optionally provide a dictionary that maps the routing_function's output to the name of the next node.
您可以有选地提供一个字典,将 routing_function 的输出映射到下一个节点的名称。

graph.add_conditional_edges("node_a", routing_function, {True: "node_b", False: "node_c"})

Tip  尖端

Use Command instead of conditional edges if you want to combine state updates and routing in a single function.
如果要将联合收割机状态更新和路由合并到一个函数中,请使用命令而不是条件边。

Entry Point  入口点

The entry point is the first node(s) that are run when the graph starts. You can use the add_edge method from the virtual START node to the first node to execute to specify where to enter the graph.
入口点是图形开始时运行的第一个节点。您可以使用 add_edge 方法从虚拟 START 节点到第一个要执行的节点来指定进入图形的位置。

API Reference: START  API 参考:START

from langgraph.graph import START

graph.add_edge(START, "node_a")

Conditional Entry Point
条件入口点

A conditional entry point lets you start at different nodes depending on custom logic. You can use add_conditional_edges from the virtual START node to accomplish this.
条件入口点允许您根据自定义逻辑从不同的节点开始。您可以使用虚拟 START 节点中的 add_conditional_edges 来完成此操作。

API Reference: START  API 参考:START

from langgraph.graph import START

graph.add_conditional_edges(START, routing_function)

You can optionally provide a dictionary that maps the routing_function's output to the name of the next node.
您可以有选地提供一个字典,将 routing_function 的输出映射到下一个节点的名称。

graph.add_conditional_edges(START, routing_function, {True: "node_b", False: "node_c"})

Send
发送

By default, Nodes and Edges are defined ahead of time and operate on the same shared state. However, there can be cases where the exact edges are not known ahead of time and/or you may want different versions of State to exist at the same time. A common example of this is with map-reduce design patterns. In this design pattern, a first node may generate a list of objects, and you may want to apply some other node to all those objects. The number of objects may be unknown ahead of time (meaning the number of edges may not be known) and the input State to the downstream Node should be different (one for each generated object).
默认情况下, 节点是提前定义的,并在相同的共享状态下操作。但是,在某些情况下,可能无法提前知道确切的边,并且/或者您可能希望同时存在不同版本的 State。一个常见的例子是 map-reduce 设计模式。在此设计模式中,第一个节点可能会生成一个对象列表,您可能希望将其他一些节点应用于所有这些对象。对象的数量可能提前未知(意味着边的数量可能未知),并且下游节点的输入状态应该不同(每个生成的对象一个)。

To support this design pattern, LangGraph supports returning Send objects from conditional edges. Send takes two arguments: first is the name of the node, and second is the state to pass to that node.
为了支持这种设计模式,LangGraph 支持从条件边返回 Send 对象。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
命令

It can be useful to combine control flow (edges) and state updates (nodes). For example, you might want to BOTH perform state updates AND decide which node to go to next in the SAME node. LangGraph provides a way to do so by returning a Command object from node functions:
组合联合收割机控制流(边)和状态更新(节点)可能是有用的。例如,您可能既要执行状态更新,又要决定在 MySQL 节点中下一个要转到哪个节点。LangGraph 提供了一种方法,通过从节点函数返回一个 Command 对象来实现:

def my_node(state: State) -> Command[Literal["my_other_node"]]:
    return Command(
        # state update
        update={"foo": "bar"},
        # control flow
        goto="my_other_node"
    )

With Command you can also achieve dynamic control flow behavior (identical to conditional edges):
使用 Command 还可以实现动态控制流行为(与条件边相同):

def my_node(state: State) -> Command[Literal["my_other_node"]]:
    if state["foo"] == "bar":
        return Command(update={"foo": "baz"}, goto="my_other_node")

Important  重要

When returning Command in your node functions, you must add return type annotations with the list of node names the node is routing to, e.g. Command[Literal["my_other_node"]]. This is necessary for the graph rendering and tells LangGraph that my_node can navigate to my_other_node.
当在节点函数中返回 Command 时,必须添加返回类型注释,其中包含节点路由到的节点名称列表,例如 Command[Literal["my_other_node"]] 。这对于图形渲染是必要的,并告诉 LangGraphmy_node 可以导航到 my_other_node

Check out this how-to guide for an end-to-end example of how to use Command.
查看此操作指南, 了解如何使用 Command 的端到端示例。

When should I use Command instead of conditional edges?
什么时候应该使用 Command 而不是 Conditional edges?

Use Command when you need to both update the graph state and route to a different node. For example, when implementing multi-agent handoffs where it's important to route to a different agent and pass some information to that agent.
当您需要更新图形状态路由到不同节点时 请使用 Command。例如,当实现多代理的路由时,重要的是路由到不同的代理并将一些信息传递给该代理。

Use conditional edges to route between nodes conditionally without updating the state.
使用条件边在节点之间有条件地路由,而不更新状态。

If you are using subgraphs, you might want to navigate from a node within a subgraph to a different subgraph (i.e. a different node in the parent graph). To do so, you can specify graph=Command.PARENT in Command:
如果您使用子图 ,您可能希望从子图中的节点导航到不同的子图(即父图中的不同节点)。为此,您可以在 Command 中指定 graph=Command. PANView

def my_node(state: State) -> Command[Literal["other_subgraph"]]:
    return Command(
        update={"foo": "bar"},
        goto="other_subgraph",  # where `other_subgraph` is a node in the parent graph
        graph=Command.PARENT
    )

Note  注意

Setting graph to Command.PARENT will navigate to the closest parent graph.
图形设置为 Command.PANView 将导航到最近的父图形。

State updates with Command.PARENT
使用 Command. PANDO 更新状态

When you send updates from a subgraph node to a parent graph node for a key that's shared by both parent and subgraph state schemas, you must define a reducer for the key you're updating in the parent graph state. See this example.
当您将更新从子图节点发送到父图节点时,对于父图和子图状态模式共享的键,您必须为在父图状态中更新的键定义一个 reducer。看这个例子

This is particularly useful when implementing multi-agent handoffs.
这在实现多智能体切换时特别有用。

Check out this guide for detail.
查看本指南了解详细信息。

Using inside tools
使用内部工具

A common use case is updating graph state from inside a tool. For example, in a customer support application you might want to look up customer information based on their account number or ID in the beginning of the conversation.
一个常见的用例是从工具内部更新图形状态。例如,在客户支持应用程序中,您可能希望在对话开始时根据客户的帐号或 ID 查找客户信息。

Refer to this guide for detail.
有关详细信息,请参阅本指南

Human-in-the-loop
人在回路

Command is an important part of human-in-the-loop workflows: when using interrupt() to collect user input, Command is then used to supply the input and resume execution via Command(resume="User input"). Check out this conceptual guide for more information.
Command 是人在环工作流的重要组成部分:当使用 interrupt() 收集用户输入时,Command 用于提供输入并通过 Command 恢复执行 (resume=“User input”)。查看此概念指南以了解更多信息。

Graph Migrations
图迁移

LangGraph can easily handle migrations of graph definitions (nodes, edges, and state) even when using a checkpointer to track state.
LangGraph 可以轻松地处理图形定义(节点,边缘和状态)的迁移,即使使用检查指针来跟踪状态。

  • For threads at the end of the graph (i.e. not interrupted) you can change the entire topology of the graph (i.e. all nodes and edges, remove, add, rename, etc)
    对于图末尾的线程(即未中断),您可以更改图的整个拓扑(即所有节点和边,删除,添加,重命名等)
  • For threads currently interrupted, we support all topology changes other than renaming / removing nodes (as that thread could now be about to enter a node that no longer exists) -- if this is a blocker please reach out and we can prioritize a solution.
    对于当前中断的线程,我们支持所有拓扑更改,而不是重命名/删除节点(因为该线程现在可能即将进入一个不再存在的节点)-如果这是一个阻塞,请联系我们,我们可以优先考虑解决方案。
  • For modifying state, we have full backwards and forwards compatibility for adding and removing keys
    对于修改状态,我们有完全的向后和向前兼容性来添加和删除键
  • State keys that are renamed lose their saved state in existing threads
    重命名的状态键将丢失其在现有线程中保存的状态
  • State keys whose types change in incompatible ways could currently cause issues in threads with state from before the change -- if this is a blocker please reach out and we can prioritize a solution.
    如果状态键的类型以不兼容的方式发生变化,那么当前可能会导致线程中的状态发生变化-如果这是一个阻塞,请联系我们,我们可以优先考虑解决方案。

Runtime Context
联系我们

When creating a graph, you can specify a context_schema for runtime context passed to nodes. This is useful for passing information to nodes that is not part of the graph state. For example, you might want to pass dependencies such as model name or a database connection.
创建图时,可以为传递给节点的运行时上下文指定 context_schema。这对于将信息传递给不属于图形状态的节点非常有用。例如,您可能希望传递依赖项,如模型名称或数据库连接。

@dataclass
class ContextSchema:
    llm_provider: str = "openai"

graph = StateGraph(State, context_schema=ContextSchema)

You can then pass this context into the graph using the context parameter of the invoke method.
然后可以使用 invoke 方法的 context 参数将此上下文传递到图中。

graph.invoke(inputs, context={"llm_provider": "anthropic"})

You can then access and use this context inside a node or conditional edge:
然后,您可以在节点或条件边中访问和使用此上下文:

from langgraph.runtime import Runtime

def node_a(state: State, runtime: Runtime[ContextSchema]):
    llm = get_llm(runtime.context.llm_provider)
    ...

See this guide for a full breakdown on configuration.
请参阅本指南了解配置的完整细分。

Recursion Limit
递归限制

The recursion limit sets the maximum number of super-steps the graph can execute during a single execution. Once the limit is reached, LangGraph will raise GraphRecursionError. By default this value is set to 25 steps. The recursion limit can be set on any graph at runtime, and is passed to .invoke/.stream via the config dictionary. Importantly, recursion_limit is a standalone config key and should not be passed inside the configurable key as all other user-defined configuration. See the example below:
递归限制设置图形在单个执行期间可以执行的最大超级步骤数。一旦达到限制,LangGraph 将引发 GraphRecursionError。默认情况下,此值设置为 25 步。递归限制可以在运行时在任何图上设置,并通过配置字典传递给 .invoke/.stream。重要的是,recursion_limit 是一个独立的配置键,不应该像所有其他用户定义的配置一样在可配置键中传递。请参见下面的示例:

graph.invoke(inputs, config={"recursion_limit": 5}, context={"llm": "anthropic"})

Read this how-to to learn more about how the recursion limit works.
阅读这篇指南 ,了解更多关于递归限制的工作原理。

Visualization
Visualization

It's often nice to be able to visualize graphs, especially as they get more complex. LangGraph comes with several built-in ways to visualize graphs. See this how-to guide for more info.
能够可视化图形通常是很好的,特别是当它们变得更复杂时。LangGraph 提供了几种内置的可视化图形的方法。请参阅此操作指南了解更多信息。