LangGraph 词汇表¶

图¶

LangGraph 的核心是将 Agent 工作流程建模为图。您可以使用三个关键组件定义 Agent 的行为

状态：一个共享的数据结构，表示应用程序的当前快照。它可以是任何 Python 类型，但通常是 TypedDict 或 Pydantic BaseModel。
节点：Python 函数，编码 Agent 的逻辑。它们接收当前的 状态 作为输入，执行一些计算或副作用，并返回更新后的 状态。
边：Python 函数，根据当前的 状态 确定接下来要执行哪个 节点。它们可以是条件分支或固定转换。

通过组合 节点 和 边，您可以创建复杂的、循环的工作流程，随着时间的推移演化 状态。然而，真正的力量来自于 LangGraph 如何管理该 状态。要强调的是：节点 和 边 不过是 Python 函数 - 它们可以包含 LLM 或仅仅是良好的 Python 代码。

简而言之：节点执行工作。边指示下一步做什么。

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

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

StateGraph¶

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

编译您的图¶

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

编译是一个非常简单的步骤。它对图的结构提供了一些基本检查（没有孤立节点等）。您还可以在此处指定运行时参数，例如检查点和断点。您只需调用 .compile 方法即可编译您的图

graph = graph_builder.compile(...)

您必须在可以使用图之前编译它。

状态¶

定义图时，您要做的第一件事是定义图的 状态。状态 由图的模式以及 reducer 函数组成，这些函数指定如何将更新应用于状态。状态 的模式将是图中所有 节点 和 边 的输入模式，并且可以是 TypedDict 或 Pydantic 模型。所有 节点 都将发出对 状态 的更新，然后使用指定的 reducer 函数应用这些更新。

模式¶

指定图模式的主要文档化方法是使用 TypedDict。但是，我们也支持使用 Pydantic BaseModel 作为您的图状态以添加默认值和额外的数据验证。

默认情况下，图将具有相同的输入和输出模式。如果您想更改此设置，您还可以直接指定显式的输入和输出模式。当您有很多键，并且有些键专门用于输入，而另一些键专门用于输出时，这很有用。有关如何使用，请参阅此处的笔记本。

多种模式¶

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

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

可以在图中拥有节点写入私有状态通道以进行内部节点通信。我们可以简单地定义一个私有模式 PrivateState。有关更多详细信息，请参阅此笔记本。

也可以为图定义显式的输入和输出模式。在这些情况下，我们定义一个“内部”模式，其中包含所有与图操作相关的键。但是，我们还定义了 输入 和 输出 模式，这些模式是“内部”模式的子集，用于约束图的输入和输出。有关更多详细信息，请参阅此笔记本。

让我们看一个例子

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=InputState,output=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=InputState,output=OutputState) 初始化图。那么，我们如何在 node_2 中写入 PrivateState？如果 PrivateState 未在 StateGraph 初始化中传递，图如何访问此模式？我们可以这样做，因为节点也可以声明额外的状态通道，只要状态模式定义存在即可。在这种情况下，定义了 PrivateState 模式，因此我们可以添加 bar 作为图中的新状态通道并写入它。

Reducer¶

Reducer 是理解如何将节点更新应用于 状态 的关键。状态 中的每个键都有其自己的独立 reducer 函数。如果未显式指定 reducer 函数，则假定对该键的所有更新都应覆盖它。有几种不同类型的 reducer，从默认类型的 reducer 开始

默认 Reducer¶

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

示例 A

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

示例 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"]}。然后让我们假设第一个 节点 返回 {"foo": 2}。这被视为对状态的更新。请注意，节点 不需要返回整个 状态 模式 - 仅需返回更新。应用此更新后，状态 将变为 {"foo": 2, "bar": ["hi"]}。如果第二个节点返回 {"bar": ["bye"]}，则 状态 将变为 {"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 序列化/反序列化的更多信息，请参阅此处。这允许以以下格式发送图输入/状态更新

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

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

由于在使用 add_messages 时，状态更新始终反序列化为 LangChain Messages，因此您应使用点表示法访问消息属性，例如 state["messages"][-1].content。以下是使用 add_messages 作为其 reducer 函数的图的示例。

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]

API 参考：AnyMessage | add_messages

MessagesState¶

由于在您的状态中拥有消息列表非常常见，因此存在一个预构建状态，称为 MessagesState，它可以轻松使用消息。MessagesState 定义了一个 messages 键，该键是一个 AnyMessage 对象列表，并使用 add_messages reducer。通常，要跟踪的状态不仅仅是消息，因此我们看到人们对该状态进行子类化并添加更多字段，例如

from langgraph.graph import MessagesState

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

节点¶

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

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

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

builder = StateGraph(dict)


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


# The second argument is optional
def my_other_node(state: dict):
    return state


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

API 参考：RunnableConfig | StateGraph

在幕后，函数被转换为 RunnableLambda，它为您的函数添加了批处理和异步支持，以及本机跟踪和调试。

如果您在未指定名称的情况下向图添加节点，则将为其指定一个默认名称，该名称等效于函数名称。

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

`START` 节点¶

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

from langgraph.graph import START

graph.add_edge(START, "node_a")

API 参考：START

`END` 节点¶

END 节点是一个特殊节点，表示终端节点。当您要表示哪些边在完成后没有操作时，将引用此节点。

from langgraph.graph import END

graph.add_edge("node_a", END)

边¶

边定义了逻辑的路由方式以及图如何决定停止。这是您的 Agent 工作方式以及不同节点之间如何相互通信的重要组成部分。有几种关键类型的边

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

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

普通边¶

如果您始终想要从节点 A 转到节点 B，则可以直接使用 add_edge 方法。

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

条件边¶

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

graph.add_conditional_edges("node_a", routing_function)

与节点类似，routing_function 接受图的当前 状态 并返回值。

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

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

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

提示

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

入口点¶

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

from langgraph.graph import START

graph.add_edge(START, "node_a")

API 参考：START

条件入口点¶

条件入口点允许您根据自定义逻辑从不同的节点开始。您可以使用虚拟 START 节点的 add_conditional_edges 来完成此操作。

from langgraph.graph import START

graph.add_conditional_edges(START, routing_function)

API 参考：START

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

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

`发送`¶

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

为了支持这种设计模式，LangGraph 支持从条件边返回 发送 对象。发送 接受两个参数：第一个是节点的名称，第二个是要传递给该节点的状态。

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)

`命令`¶

将控制流（边）和状态更新（节点）组合起来可能很有用。例如，您可能希望在同一节点中同时执行状态更新和决定接下来要转到哪个节点。LangGraph 提供了一种通过从节点函数返回 命令 对象来实现此目的的方法

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

使用 命令，您还可以实现动态控制流行为（与条件边相同）

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

重要提示

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

查看此操作指南，以获取有关如何使用 命令 的端到端示例。

我应该何时使用命令而不是条件边？¶

当您需要同时更新图状态并路由到不同的节点时，请使用 命令。例如，当实现多 Agent 交接时，路由到不同的 Agent 并将一些信息传递给该 Agent 非常重要。

使用条件边在节点之间有条件地路由，而无需更新状态。

导航到父图中的节点¶

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

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

注意

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

使用 Command.PARENT 进行状态更新

当您从子图节点向父图节点发送更新，对于父图和子图状态模式共享的键，您必须为要在父图状态中更新的键定义 reducer。请参阅此示例。

这在实现多 Agent 交接时特别有用。

在工具内部使用¶

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

@tool
def lookup_user_info(tool_call_id: Annotated[str, InjectedToolCallId], config: RunnableConfig):
    """Use this to look up user information to better assist them with their questions."""
    user_info = get_user_info(config.get("configurable", {}).get("user_id"))
    return Command(
        update={
            # update the state keys
            "user_info": user_info,
            # update the message history
            "messages": [ToolMessage("Successfully looked up user information", tool_call_id=tool_call_id)]
        }
    )

重要提示

当从工具返回 命令 并且 messages 中的消息列表必须包含 ToolMessage 时，您必须在 Command.update 中包含 messages（或用于消息历史记录的任何状态键）。这是使生成的消息历史记录有效所必需的（LLM 提供商要求 AI 消息与工具调用之后是工具结果消息）。

如果您正在使用通过 命令 更新状态的工具，我们建议使用预构建的 ToolNode，它可以自动处理工具返回的 命令 对象，并将它们传播到图状态。如果您正在编写调用工具的自定义节点，则需要手动传播工具返回的 命令 对象作为节点的更新。

人机环路¶

命令 是人机环路工作流程的重要组成部分：当使用 interrupt() 收集用户输入时，然后使用 命令 提供输入并通过 Command(resume="User input") 恢复执行。有关更多信息，请查看此概念指南。

持久性¶

LangGraph 使用检查点为您的 Agent 状态提供内置持久性。检查点在每个超级步骤保存图状态的快照，从而允许随时恢复。这实现了诸如人机环路交互、内存管理和容错等功能。您甚至可以使用适当的 get 和 update 方法在图执行后直接操作图的状态。有关更多详细信息，请参阅持久性概念指南。

线程¶

LangGraph 中的线程表示图和用户之间的各个会话或对话。当使用检查点时，单个对话中的轮次（甚至单个图执行中的步骤）都由唯一的线程 ID 组织。

存储¶

LangGraph 通过 BaseStore 接口提供内置文档存储。与通过线程 ID 保存状态的检查点不同，存储使用自定义命名空间来组织数据。这实现了跨线程持久性，允许 Agent 维护长期记忆、从过去的交互中学习并随着时间的推移积累知识。常见的用例包括存储用户配置文件、构建知识库以及管理所有线程的全局首选项。

图迁移¶

即使在使用检查点来跟踪状态时，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 步。递归限制可以在运行时在任何图上设置，并通过 config 字典传递给 .invoke/.stream。重要的是，recursion_limit 是一个独立的 config 键，不应像所有其他用户定义的配置一样在 configurable 键内传递。请参阅以下示例

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

阅读此操作指南，以了解有关递归限制如何工作的更多信息。

`中断`¶

使用 interrupt 函数在特定点暂停图，以收集用户输入。interrupt 函数将中断信息呈现给客户端，允许开发人员在恢复执行之前收集用户输入、验证图状态或做出决策。

from langgraph.types import interrupt

def human_approval_node(state: State):
    ...
    answer = interrupt(
        # This value will be sent to the client.
        # It can be any JSON serializable value.
        {"question": "is it ok to continue?"},
    )
    ...

API 参考：interrupt

通过将 命令 对象传递给图，并将 resume 键设置为 interrupt 函数返回的值，可以恢复图。

在人机环路概念指南中，详细了解如何在 人机环路 工作流程中使用 interrupt。

断点¶

断点在特定点暂停图执行，并允许逐步执行。断点由 LangGraph 的 持久性层 提供支持，该层在每个图步骤后保存状态。断点还可以用于启用 人机环路 工作流程，尽管我们建议为此目的使用 interrupt 函数。

在断点概念指南中，详细了解断点。

子图¶

子图是一个图，在另一个图中用作节点。这不过是应用于 LangGraph 的古老的封装概念。使用子图的一些原因是

构建多 Agent 系统
当您想在多个图中重用一组节点时，这些图可能共享某些状态，您可以在子图中定义它们一次，然后在多个父图中使用它们
当您希望不同的团队独立处理图的不同部分时，您可以将每个部分定义为子图，并且只要子图接口（输入和输出模式）得到遵守，就可以构建父图，而无需了解子图的任何详细信息

有两种方法可以将子图添加到父图

添加带有编译子图的节点：当父图和子图共享状态键，并且您无需在进入或退出时转换状态时，这很有用

builder.add_node("subgraph", subgraph_builder.compile())

添加带有调用子图的函数的节点：当父图和子图具有不同的状态模式，并且您需要在调用子图之前或之后转换状态时，这很有用

subgraph = subgraph_builder.compile()

def call_subgraph(state: State):
    return subgraph.invoke({"subgraph_key": state["parent_key"]})

builder.add_node("subgraph", call_subgraph)

让我们看一下每个示例。

作为编译图¶

创建子图节点最简单的方法是直接使用已编译的子图。这样做时，重要的是父图和子图的状态模式至少共享一个它们可以用来通信的键。如果您的图和子图不共享任何键，您应该编写一个函数调用子图来代替。

注意

如果您将额外的键传递给子图节点（即，除了共享键之外），子图节点将忽略它们。同样地，如果您从子图返回额外的键，父图将忽略它们。

from langgraph.graph import StateGraph
from typing import TypedDict

class State(TypedDict):
    foo: str

class SubgraphState(TypedDict):
    foo: str  # note that this key is shared with the parent graph state
    bar: str

# Define subgraph
def subgraph_node(state: SubgraphState):
    # note that this subgraph node can communicate with the parent graph via the shared "foo" key
    return {"foo": state["foo"] + "bar"}

subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node(subgraph_node)
...
subgraph = subgraph_builder.compile()

# Define parent graph
builder = StateGraph(State)
builder.add_node("subgraph", subgraph)
...
graph = builder.compile()

API 参考：StateGraph

作为函数¶

您可能想要定义一个具有完全不同模式的子图。在这种情况下，您可以创建一个节点函数来调用子图。此函数将需要转换输入（父）状态为子图状态，然后再调用子图，并在从节点返回状态更新之前将结果转换回父状态。

class State(TypedDict):
    foo: str

class SubgraphState(TypedDict):
    # note that none of these keys are shared with the parent graph state
    bar: str
    baz: str

# Define subgraph
def subgraph_node(state: SubgraphState):
    return {"bar": state["bar"] + "baz"}

subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node(subgraph_node)
...
subgraph = subgraph_builder.compile()

# Define parent graph
def node(state: State):
    # transform the state to the subgraph state
    response = subgraph.invoke({"bar": state["foo"]})
    # transform response back to the parent state
    return {"foo": response["bar"]}

builder = StateGraph(State)
# note that we are using `node` function instead of a compiled subgraph
builder.add_node(node)
...
graph = builder.compile()

可视化¶

通常，能够可视化图是很不错的，尤其是在它们变得更复杂时。 LangGraph 附带了几种内置的可视化图的方法。有关更多信息，请参阅此操作指南。

流式处理¶

LangGraph 内置了对流式处理的一流支持，包括在执行期间从图节点流式传输更新、从 LLM 调用流式传输令牌等等。有关更多信息，请参阅此概念指南。

LangGraph 词汇表¶

图¶

StateGraph¶

编译您的图¶

状态¶

模式¶

多种模式¶

Reducer¶

默认 Reducer¶

在图状态中使用消息¶

为什么使用消息？¶

在您的图中使用消息¶

序列化¶

MessagesState¶

节点¶

START 节点¶

END 节点¶

边¶

普通边¶

条件边¶

入口点¶

条件入口点¶

发送¶

命令¶

我应该何时使用命令而不是条件边？¶

导航到父图中的节点¶

在工具内部使用¶

人机环路¶

持久性¶

线程¶

存储¶

图迁移¶

配置¶

递归限制¶

中断¶

断点¶

子图¶

作为编译图¶

作为函数¶

可视化¶

流式处理¶

评论

`START` 节点¶

`END` 节点¶

`发送`¶

`命令`¶

`中断`¶