Graph API 概念¶

图（Graphs）¶

LangGraph 的核心是将智能体工作流建模为图。你可以使用三个关键组件定义智能体的行为：

状态（State）：一个共享的数据结构，表示应用程序当前的快照。它可以是任何数据类型，但通常通过共享的状态模式来定义。
节点（Nodes）：编码智能体逻辑的函数。它们接收当前状态作为输入，执行某些计算或副作用，并返回更新后的状态。
边（Edges）：根据当前状态决定下一步执行哪个 Node 的函数。它们可以是条件分支或固定转换。

通过组合 Nodes 和 Edges，你可以创建随着时间推移不断演化状态的复杂循环工作流。然而，真正的强大之处在于 LangGraph 如何管理该状态。再次强调：Nodes 和 Edges 只不过是函数——它们可以包含大语言模型（LLM），也可以只是普通的代码。

简而言之：节点负责工作，边决定下一步做什么。

LangGraph 底层的图算法使用消息传递来定义通用程序。当一个节点完成其操作后，它会沿着一条或多条边向其他节点发送消息。这些接收节点随后执行其函数，将结果消息传递给下一组节点，此过程持续进行。受 Google Pregel 系统启发，程序以离散的“超步”推进。

一个超步可视为对图中所有节点的一次迭代。并行运行的节点属于同一个超步，而顺序运行的节点则属于不同的超步。在图执行开始时，所有节点都处于 inactive 状态。当某个节点在其任意入边（或“通道”）上接收到新消息（状态）时，它变为 active。活跃节点执行其函数并返回更新。每个超步结束时，没有入站消息的节点通过将自身标记为 inactive 来投票 halt。当所有节点均为 inactive 且无消息正在传输时，图执行终止。

StateGraph¶

StateGraph 类是主要使用的图类，它由用户自定义的 State 对象参数化。

编译你的图¶

构建图时，首先定义状态，然后添加节点和边，最后编译它。编译图到底是什么意思？为什么需要它？

编译是一个相当简单的步骤。它会对图的结构进行一些基本检查（如无孤立节点等）。同时，你还可以在此指定运行时参数，例如检查点器（checkpointers）和断点。只需调用 .compile 方法即可编译你的图：

graph = graph_builder.compile(...)

你**必须**在使用图之前先编译它。

状态（State）¶

定义图时首先要做的就是定义图的 State。State 包含图的模式（schema）以及指定如何应用状态更新的reducer 函数。State 的模式将是图中所有 Nodes 和 Edges 的输入模式，可以是 TypedDict 或 Pydantic 模型。所有 Nodes 都会发出对 State 的更新，然后使用指定的 reducer 函数应用这些更新。

模式（Schema）¶

指定图模式的主要文档方法是使用 TypedDict。如果你想在状态中提供默认值，请使用 dataclass。如果你希望递归数据验证，我们也支持使用 Pydantic BaseModel 作为图状态（但请注意，pydantic 的性能低于 TypedDict 或 dataclass）。

默认情况下，图的输入和输出模式相同。如果你想更改这一点，也可以直接指定显式的输入和输出模式。当你拥有大量键，其中一些专门用于输入，另一些专门用于输出时，这非常有用。请参阅此处指南了解如何使用。

多种模式¶

通常，所有图节点都使用单一模式通信。这意味着它们将读取和写入相同的状态通道。但在某些情况下，我们希望对此有更多控制：

内部节点可以传递图的输入/输出中不需要的信息。
我们可能还想为图使用不同的输入/输出模式。例如，输出可能只包含单个相关输出键。

可以让节点在图内部写入私有状态通道以供内部节点通信。我们可以简单地定义一个私有模式 PrivateState。

还可以为图定义显式的输入和输出模式。在这种情况下，我们定义一个包含图操作所有相关键的“内部”模式。但同时，我们还定义了 input 和 output 模式，它们是“内部”模式的子集，用于约束图的输入和输出。有关更多详细信息，请参阅本指南。

让我们看一个例子：

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'}

这里有两个微妙而重要的要点需要注意：

我们将 state: InputState 作为输入模式传递给 node_1。但是，我们写入了 foo，这是 OverallState 中的一个通道。我们怎么能写入一个未包含在输入模式中的状态通道呢？这是因为节点_可以写入图状态中的任何状态通道_。图状态是在初始化时定义的状态通道的并集，包括 OverallState 以及过滤器 InputState 和 OutputState。
我们使用 StateGraph(OverallState,input_schema=InputState,output_schema=OutputState) 初始化图。那么，我们如何在 node_2 中写入 PrivateState 呢？如果未在 StateGraph 初始化时传入，图是如何获得对该模式的访问权限的？我们可以这样做是因为_节点也可以声明额外的状态通道_，只要状态模式定义存在即可。在这种情况下，PrivateState 模式已定义，因此我们可以在图中添加 bar 作为新的状态通道并对其进行写入。

Reducers¶

Reducers 是理解节点更新如何应用于 State 的关键。State 中的每个键都有其独立的 reducer 函数。如果没有显式指定 reducer 函数，则默认所有对该键的更新都将覆盖它。以下是几种不同类型的 reducers，从默认类型的 reducer 开始：

默认 Reducer¶

以下两个示例展示了如何使用默认 reducer：

示例 A：

from typing_extensions import TypedDict

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

在此示例中，未为任何键指定 reducer 函数。假设图的输入为：

{"foo": 1, "bar": ["hi"]}。然后假设第一个 Node 返回 {"foo": 2}。这被视为对状态的更新。注意，Node 不需要返回整个 State 模式——只需返回更新部分。应用此更新后，State 将变为 {"foo": 2, "bar": ["hi"]}。如果第二个节点返回 {"bar": ["bye"]}，则 State 将变为 {"foo": 2, "bar": ["bye"]}。

示例 B：

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"]}。然后假设第一个 Node 返回 {"foo": 2}。这被视为对状态的更新。注意，Node 不需要返回整个 State 模式——只需返回更新部分。应用此更新后，State 将变为 {"foo": 2, "bar": ["hi"]}。如果第二个节点返回 {"bar": ["bye"]}，则 State 将变为 {"foo": 2, "bar": ["hi", "bye"]}。请注意，这里的 bar 键是通过将两个列表相加来更新的。

在图状态中处理消息¶

为什么要使用消息？¶

大多数现代 LLM 提供商都有一个接受消息列表作为输入的聊天模型接口。特别是 LangChain 的 ChatModel 接受 Message 对象列表作为输入。这些消息有多种形式，例如 HumanMessage（用户输入）或 AIMessage（LLM 响应）。要了解更多关于消息对象的内容，请参阅此概念指南。

在图中使用消息¶

在许多情况下，将先前的对话历史记录作为消息列表存储在图状态中非常有用。为此，我们可以在图状态中添加一个键（通道）来存储 Message 对象列表，并为其标注一个 reducer 函数（见下面示例中的 messages 键）。reducer 函数对于告诉图如何在每次状态更新时（例如，当节点发送更新时）更新状态中的 Message 对象列表至关重要。如果不指定 reducer，每次状态更新都会用最新提供的值覆盖消息列表。如果只想将消息追加到现有列表，可以使用 operator.add 作为 reducer。

然而，你也可能希望手动更新图状态中的消息（例如，人在环中）。如果你使用 operator.add，你发送给图的手动状态更新将被追加到现有消息列表中，而不是更新现有消息。为避免这种情况，你需要一个能跟踪消息 ID 并在更新时覆盖现有消息的 reducer。为此，你可以使用预构建的 add_messages 函数。对于全新的消息，它会简单地追加到现有列表，但也能正确处理现有消息的更新。

序列化¶

除了跟踪消息 ID，add_messages 函数还会在 messages 通道上接收到状态更新时尝试将消息反序列化为 LangChain Message 对象。有关 LangChain 序列化/反序列化的更多信息，请参阅此处。这允许以以下格式发送图输入/状态更新：

# 支持此格式
{"messages": [HumanMessage(content="message")]}

# 此格式也支持
{"messages": [{"type": "human", "content": "message"}]}

由于使用 add_messages 时状态更新始终被反序列化为 LangChain Messages，你应该使用点符号访问消息属性，例如 state["messages"][-1].content。以下是一个使用 add_messages 作为其 reducer 函数的图示例。

^{API Reference: AnyMessage | add_messages}

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 的预构建状态，便于使用消息。MessagesState 定义了一个单一的 messages 键，它是一个 AnyMessage 对象列表，并使用 add_messages reducer。通常，需要跟踪的状态不止消息，因此我们看到人们继承此状态并添加更多字段，例如：

from langgraph.graph import MessagesState

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

节点（Nodes）¶

在 LangGraph 中，节点是 Python 函数（同步或异步），接受以下参数：

state：图的状态
config：一个 RunnableConfig 对象，包含诸如 thread_id 的配置信息和诸如 tags 的追踪信息
runtime：一个 Runtime 对象，包含运行时 context 以及其他信息，如 store 和 stream_writer

与 NetworkX 类似，你可以使用 add_node 方法将这些节点添加到图中：

^{API Reference: 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)
...

在后台，函数会被转换为 RunnableLambdas，为你的函数添加批处理和异步支持，以及原生追踪和调试功能。

如果你在添加节点时未指定名称，它将被赋予一个默认名称，即函数名。

builder.add_node(my_node)
# 然后你可以通过引用 `"my_node"` 创建到/从此节点的边

`START` 节点¶

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

^{API Reference: START}

from langgraph.graph import START

graph.add_edge(START, "node_a")

`END` 节点¶

END 节点是一个特殊节点，代表终端节点。当你想表示某条边完成后没有后续操作时，会引用此节点。

^{API Reference: END}

from langgraph.graph import END

graph.add_edge("node_a", END)

节点缓存¶

LangGraph 支持基于节点输入的任务/节点缓存。要使用缓存：

在编译图时（或指定入口点时）指定缓存
为节点指定缓存策略。每个缓存策略支持：
key_func 用于根据节点输入生成缓存键，默认使用 pickle 对输入进行哈希。
ttl，缓存的生存时间（秒）。如果未指定，缓存将永不过期。

例如：

^{API Reference: 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]:
    # 昂贵的计算
    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'))  # (1)!
[{'expensive_node': {'result': 10}}]
print(graph.invoke({"x": 5}, stream_mode='updates'))  # (2)!
[{'expensive_node': {'result': 10}, '__metadata__': {'cached': True}}]

第一次运行需要两秒（由于模拟的昂贵计算）。
第二次运行利用缓存并快速返回。

边（Edges）¶

边定义了逻辑路由方式以及图决定何时停止的方式。这是你的智能体如何工作以及不同节点如何相互通信的重要部分。有几种关键类型的边：

普通边：直接从一个节点到下一个节点。
条件边：调用函数以确定接下来去哪个节点（或多个节点）。
入口点：当用户输入到达时首先调用的节点。
条件入口点：调用函数以确定当用户输入到达时首先调用哪个节点（或多个节点）。

一个节点可以有**多个**出边。如果一个节点有多个出边，**所有**目标节点将在下一个超步中并行执行。

普通边¶

如果你**总是**想从节点 A 到节点 B，可以直接使用 add_edge 方法。

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

条件边¶

如果你想要**选择性**路由到一个或多个边（或选择性终止），可以使用 add_conditional_edges 方法。此方法接受节点名称和在该节点执行后调用的“路由函数”：

graph.add_conditional_edges("node_a", routing_function)

与节点类似，routing_function 接受图的当前 state 并返回一个值。

默认情况下，routing_function 的返回值用作下一个要发送状态的节点（或节点列表）的名称。所有这些节点将在下一个超步中并行运行。

你可以选择提供一个字典，将 routing_function 的输出映射到下一个节点的名称。

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

Tip

如果你想在单个函数中结合状态更新和路由，请使用 Command 而不是条件边。

入口点¶

入口点是图启动时首先运行的节点。你可以使用 add_edge 方法从虚拟 START 节点到要执行的第一个节点，以指定进入图的位置。

^{API Reference: START}

from langgraph.graph import START

graph.add_edge(START, "node_a")

条件入口点¶

条件入口点允许你根据自定义逻辑从不同节点开始。你可以使用 add_conditional_edges 方法从虚拟 START 节点实现这一点。

^{API Reference: START}

from langgraph.graph import START

graph.add_conditional_edges(START, routing_function)

你可以选择提供一个字典，将 routing_function 的输出映射到下一个节点的名称。

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

`Send`¶

默认情况下，Nodes 和 Edges 是预先定义的，并在相同的共享状态上操作。然而，在某些情况下，确切的边可能无法提前知道，或者你可能希望同时存在不同版本的 State。一个常见的例子是 map-reduce 设计模式。在这种设计模式中，第一个节点可能生成一个对象列表，你可能希望将另一个节点应用于所有这些对象。对象的数量可能无法提前知道（意味着边的数量可能未知），并且下游 Node 的输入 State 应该不同（每个生成的对象对应一个）。

为了支持这种设计模式，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`¶

结合控制流（边）和状态更新（节点）可能很有用。例如，你可能希望在**同一个节点中**既执行状态更新又决定下一步去哪个节点。LangGraph 提供了一种方法，通过从节点函数返回 Command 对象来实现：

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

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

在节点函数中返回 Command 时，必须添加返回类型注解，列出节点要路由到的节点名称，例如 Command[Literal["my_other_node"]]。这对于图渲染是必要的，并告诉 LangGraph my_node 可以导航到 my_other_node。

查看此操作指南以获取如何使用 Command 的端到端示例。

何时应使用 Command 而不是条件边？¶

当你需要**同时**更新图状态**并**路由到不同节点时，使用 Command。例如，在实现多智能体交接时，重要的是路由到不同的智能体并向该智能体传递一些信息。
使用条件边在不更新状态的情况下有条件地在节点之间路由。

导航到父图中的节点¶

如果你正在使用子图，你可能希望从子图内的节点导航到不同的子图（即父图中的不同节点）。为此，你可以在 Command 中指定 graph=Command.PARENT：

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

Note

将 graph 设置为 Command.PARENT 将导航到最近的父图。

使用 Command.PARENT 的状态更新

当你从子图节点向父图节点发送更新时，如果更新的键在父图和子图状态模式中共享，你**必须**在父图状态中为该键定义reducer。请参阅此示例。

这在实现多智能体交接时特别有用。

查看此指南以获取详细信息。

在工具内部使用¶

一个常见的用例是从工具内部更新图状态。例如，在客户支持应用中，你可能希望在对话开始时根据客户的账号或 ID 查找客户信息。

请参阅此指南以获取详细信息。

人在环中（Human-in-the-loop）¶

Command 是人在环中工作流的重要组成部分：当使用 interrupt() 收集用户输入时，Command 随后用于提供输入并通过 Command(resume="User input") 恢复执行。查看此概念指南以获取更多信息。

图迁移¶

LangGraph 可以轻松处理图定义（节点、边和状态）的迁移，即使使用检查点器跟踪状态也是如此。

对于处于图末尾的线程（即未中断），你可以更改整个图的拓扑结构（即所有节点和边，删除、添加、重命名等）
对于当前中断的线程，我们支持除重命名/删除节点外的所有拓扑更改（因为该线程可能即将进入一个不再存在的节点）——如果这成为阻碍，请联系我们，我们将优先解决。
对于修改状态，我们完全支持向前和向后兼容地添加和删除键
重命名的状态键在现有线程中会丢失其保存的状态
类型以不兼容方式更改的状态键可能会在具有更改前状态的线程中引发问题——如果这成为阻碍，请联系我们，我们将优先解决。

运行时上下文（Runtime Context）¶

创建图时，你可以为传递给节点的运行时上下文指定 context_schema。这对于向节点传递不属于图状态的信息非常有用。例如，你可能希望传递依赖项，如模型名称或数据库连接。

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

graph = StateGraph(State, context_schema=ContextSchema)

然后你可以使用 invoke 方法的 context 参数将此上下文传递到图中。

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

然后你可以在节点或条件边内部访问和使用此上下文：

from langgraph.runtime import Runtime

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

有关配置的完整说明，请参阅此指南。 :::

递归限制（Recursion Limit）¶

递归限制设置图在单次执行期间可以执行的最大超步数。一旦达到限制，LangGraph 将抛出 GraphRecursionError。默认情况下，此值设置为 25 步。递归限制可以在运行时设置在任何图上，并通过 config 字典传递给 .invoke/.stream。重要的是，recursion_limit 是一个独立的 config 键，不应像所有其他用户定义的配置那样传递到 configurable 键内。请参见以下示例：

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

阅读此操作指南以了解更多关于递归限制的工作原理。

可视化（Visualization）¶

能够可视化图通常很有帮助，尤其是当图变得越来越复杂时。LangGraph 自带了几种内置的图可视化方法。有关更多信息，请参阅此操作指南。

Graph API 概念¶

图（Graphs）¶

StateGraph¶

编译你的图¶

状态（State）¶

模式（Schema）¶

多种模式¶

Reducers¶

默认 Reducer¶

在图状态中处理消息¶

为什么要使用消息？¶

在图中使用消息¶

序列化¶

MessagesState¶

节点（Nodes）¶

START 节点¶

END 节点¶

节点缓存¶

边（Edges）¶

普通边¶

条件边¶

入口点¶

条件入口点¶

Send¶

Command¶

何时应使用 Command 而不是条件边？¶

导航到父图中的节点¶

在工具内部使用¶

人在环中（Human-in-the-loop）¶

图迁移¶

运行时上下文（Runtime Context）¶

递归限制（Recursion Limit）¶

可视化（Visualization）¶

`START` 节点¶

`END` 节点¶

`Send`¶

`Command`¶