图形 API 概念¶

图¶

在本质上，LangGraph 将代理工作流建模为图。你使用以下三个关键组件来定义你的代理行为：

State: 一个共享的数据结构，表示应用程序的当前快照。它可以是任何 Python 类型，但通常是一个 TypedDict 或 Pydantic 的 BaseModel。
Nodes: 编码代理逻辑的 Python 函数。它们接收当前的 State 作为输入，执行某些计算或副作用，并返回更新后的 State。
Edges: 确定基于当前 State 应该执行哪个 Node 的 Python 函数。它们可以是条件分支或固定转换。

通过组合 Nodes 和 Edges，你可以创建复杂的、循环的工作流，随着时间推移演化 State。然而，真正的强大之处来自于 LangGraph 如何管理这个 State。强调一下：Nodes 和 Edges 只不过是 Python 函数——它们可以包含 LLM 或仅仅是传统的 Python 代码。

简而言之：节点负责执行任务，边决定下一步要做什么。

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

超级步骤可以被认为是对图节点的一次迭代。并行运行的节点属于同一个超级步骤，而顺序运行的节点则属于不同的超级步骤。在图执行开始时，所有节点都处于 inactive（非活动）状态。当某个节点在其任意一条入边（或“通道”）上接收到新消息（状态）时，该节点变为 active（活动）。然后，该活动节点运行其函数并作出响应。每个超级步骤结束时，没有入消息的节点会投票 halt（停止），通过将自身标记为 inactive。当所有节点都为 inactive 且没有消息正在传输时，图执行终止。

StateGraph¶

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

编译你的图¶

要构建你的图，首先定义 state，然后添加 nodes 和 edges，最后编译它。那么，什么是编译你的图？为什么需要它？

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

graph = graph_builder.compile(...)

你必须在使用图之前对其进行编译。

状态¶

当你定义一个图时，首先要定义图的 State。State 包含图的 schema 以及用于指定如何将更新应用到状态的 reducer 函数。State 的 schema 将作为图中所有 Nodes 和 Edges 的输入 schema，可以是 TypedDict 或 Pydantic 模型。所有 Nodes 都会向 State 发出更新，这些更新随后通过指定的 reducer 函数进行应用。

Schema¶

指定图的 schema 的主要方式是使用 TypedDict。然而，我们还支持使用 Pydantic BaseModel 作为你的图状态，以添加 默认值 和额外的数据验证。

默认情况下，图将具有相同的输入和输出 schema。如果你想更改这一点，也可以直接指定显式的输入和输出 schema。这在你有很多键，并且其中一些明确用于输入，另一些用于输出时非常有用。请参阅此指南了解如何使用。

多个 schema¶

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

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

可以在图内部节点之间传递私有状态通道，让节点写入私有状态通道。我们只需定义一个私有 schema，PrivateState。有关详细信息，请参阅此指南。

还可以为图定义显式的输入和输出 schema。在这种情况下，我们定义一个“内部” schema，它包含所有与图操作相关的键。但我们也会定义 input 和 output schema，它们是“内部” schema 的子集，用于限制图的输入和输出。有关详细信息，请参阅此指南。

让我们看一个例子：

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 的输入 schema。但，我们写入了 foo，这是 OverallState 中的一个状态通道。我们如何能够写入不在输入 schema 中的状态通道？这是因为节点 可以写入图状态中的任何状态通道。图状态是初始化时定义的所有状态通道的并集，其中包括 OverallState 以及过滤器 InputState 和 OutputState。
我们用 StateGraph(OverallState, input_schema=InputState, output_schema=OutputState) 初始化图。那么，我们如何在 node_2 中写入 PrivateState？如果图没有在 StateGraph 初始化时传入这个 schema，它是如何获得访问权限的？我们可以这样做是因为 节点也可以声明额外的状态通道，只要状态 schema 定义存在。在这种情况下，PrivateState schema 已经定义，因此我们可以在图中添加 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}。这被视为对状态的更新。请注意，节点不需要返回整个 State schema，只需要一个更新。在应用此更新后，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}。这被视为对状态的更新。请注意，节点不需要返回整个 State schema，只需要一个更新。在应用此更新后，State 将变为 {"foo": 2, "bar": ["hi"]}。如果第二个节点返回 {"bar": ["bye"]}，则 State 将变为 {"foo": 2, "bar": ["hi", "bye"]}。请注意，这里的 bar 键是通过将两个列表相加来更新的。

在 Graph State 中处理消息¶

为什么使用消息？¶

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

在你的图中使用消息¶

在许多情况下，将之前的对话历史存储为消息列表是有帮助的。为此，我们可以在图状态中添加一个键（通道），用来存储一组 Message 对象，并用 reducer 函数对其进行注释（见下面示例中的 messages 键）。reducer 函数对于告诉图如何在每次状态更新时更新状态中的消息列表至关重要（例如，当一个节点发送更新时）。如果你不指定 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 键的 state，该键是一个 AnyMessage 对象的列表，并使用 add_messages 作为 reducer。通常，除了消息外还有更多的状态需要跟踪，所以人们通常继承这个 state 并添加更多字段，例如：

from langgraph.graph import MessagesState

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

节点¶

在 LangGraph 中，节点通常是 Python 函数（同步或异步），其中 第一个 位置参数是状态，并且（可选地），第二个 位置参数是一个 "config"，包含可选的可配置参数（例如 thread_id）。

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

^{API Reference: RunnableConfig | StateGraph}

from typing_extensions import TypedDict

from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph

class State(TypedDict):
    input: str
    results: str

builder = StateGraph(State)


def my_node(state: State, config: RunnableConfig):
    print("In node: ", config["configurable"]["user_id"])
    return {"results": f"Hello, {state['input']}!"}


# 第二个参数是可选的
def my_other_node(state: State):
    return state


builder.add_node("my_node", my_node)
builder.add_node("other_node", my_other_node)
...

在幕后，函数会被转换为 RunnableLambda，这会为你的函数添加批量和异步支持，并且原生支持追踪和调试。

如果你在添加节点时没有指定名称，它将被赋予一个默认名称，该名称等同于函数名称。

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

首次运行需要完整的一秒钟来执行（由于模拟了昂贵的计算）。
第二次运行利用了缓存，快速返回结果。

边¶

边定义了逻辑的路由方式以及图如何决定停止。这是你的代理工作方式和不同节点之间如何通信的重要部分。有几种关键类型的边：

正常边：直接从一个节点到下一个节点。
条件边：调用一个函数来确定下一步要前往哪个节点（或多个节点）。
入口点：当用户输入到达时，首先调用哪个节点。
条件入口点：调用一个函数来确定当用户输入到达时，首先调用哪个节点（或多个节点）。

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

正常边¶

如果你**始终**希望从节点 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 而不是条件边。

入口点¶

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

^{API Reference: START}

from langgraph.graph import START

graph.add_edge(START, "node_a")

条件入口点¶

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

^{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][langgraph.types.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][langgraph.types.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 查找客户信息。

详情请参阅此指南。

人机交互¶

Command 是人机交互工作流的重要组成部分：当你使用 interrupt() 收集用户输入时，Command 随后用于提供输入并通过 Command(resume="User input") 恢复执行。更多信息请参阅此概念指南。

图结构迁移¶

即使使用 checkpointer 来跟踪状态，LangGraph 也能轻松处理图定义（节点、边和状态）的迁移。

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

配置¶

在创建图时，你也可以标记图的某些部分是可配置的。这通常是为了方便在模型或系统提示之间进行切换。这样你可以创建一个单一的“认知架构”（即图），但可以有多个不同的实例。

在创建图时，你可以选择性地指定 config_schema。

class ConfigSchema(TypedDict):
    llm: str

graph = StateGraph(State, config_schema=ConfigSchema)

然后，你可以通过 configurable 配置字段将此配置传递给图。

config = {"configurable": {"llm": "anthropic"}}

graph.invoke(inputs, config=config)

之后，你可以在节点或条件边中访问并使用此配置：

def node_a(state, config):
    llm_type = config.get("configurable", {}).get("llm", "openai")
    llm = get_llm(llm_type)
    ...

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

递归限制¶

递归限制设置图在单次执行过程中可以执行的超级步骤的最大数量。一旦达到限制，LangGraph 将抛出 GraphRecursionError。默认情况下，该值设置为 25 步。递归限制可以在运行时在任何图上设置，并通过配置字典传递给 .invoke/.stream。重要的是，recursion_limit 是一个独立的 config 键，不应放在 configurable 键内，因为所有其他用户定义的配置都应放在其中。请看下面的例子：

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

如需了解更多关于递归限制如何工作的信息，请阅读此指南。

可视化¶

能够可视化图结构通常很有帮助，尤其是在图变得复杂时。LangGraph 提供了几种内置的图可视化方法。如需更多信息，请参阅此操作指南。

图形 API 概念¶

图¶

StateGraph¶

编译你的图¶

状态¶

Schema¶

多个 schema¶

Reducers¶

默认 Reducer¶

在 Graph State 中处理消息¶

为什么使用消息？¶

在你的图中使用消息¶

序列化¶

MessagesState¶

节点¶

START 节点¶

END 节点¶

节点缓存¶

边¶

正常边¶

条件边¶

入口点¶

条件入口点¶

Send¶

Command¶

在什么情况下应该使用 Command 而不是条件边？¶

导航到父图中的节点¶

在工具内部使用¶

人机交互¶

图结构迁移¶

配置¶

递归限制¶

可视化¶

`START` 节点¶

`END` 节点¶

`Send`¶

`Command`¶

在什么情况下应该使用 `Command` 而不是条件边？¶