Skip to content

语言图(LangGraph)术语表

从核心上来说,LangGraph 将代理工作流建模为图。你可以使用三个关键组件来定义代理的行为:

  1. 状态(State:一种共享的数据结构,代表应用程序的当前快照。它可以是任何 Python 类型,但通常是 TypedDict 或 Pydantic 的 BaseModel

  2. 节点(Nodes:编码代理逻辑的 Python 函数。它们接收当前的 State 作为输入,执行一些计算或副作用操作,并返回更新后的 State

  3. 边(Edges:根据当前的 State 确定接下来要执行哪个 Node 的 Python 函数。它们可以是条件分支或固定的转换。

通过组合 NodesEdges,你可以创建随着时间推移演变 State 的复杂循环工作流。不过,真正的强大之处在于 LangGraph 对 State 的管理方式。需要强调的是:NodesEdges 只不过是 Python 函数 —— 它们可以包含大语言模型(LLM),也可以只是普通的 Python 代码。

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

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

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

状态图(StateGraph)

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

编译你的图

要构建你的图,你首先要定义状态(state,然后添加节点(nodes边(edges,最后进行编译。编译图到底是什么意思,为什么需要编译呢?

编译是一个相当简单的步骤。它会对图的结构进行一些基本检查(例如没有孤立节点等)。你还可以在此处指定运行时参数,如检查点器(checkpointers)断点(breakpoints)。你只需调用 .compile 方法来编译你的图:

graph = graph_builder.compile(...)

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

状态

定义图时要做的第一件事是定义图的 State(状态)。State图的模式以及reducer 函数组成,这些函数指定了如何将更新应用到状态上。State 的模式将作为图中所有 Node(节点)和 Edge(边)的输入模式,可以是 TypedDictPydantic 模型。所有 Node 都会向 State 发出更新,然后使用指定的 reducer 函数应用这些更新。

模式

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

默认情况下,图的输入和输出模式相同。如果想改变这一点,也可以直接指定明确的输入和输出模式。当有很多键,且有些键明确用于输入,有些用于输出时,这很有用。有关如何使用,请参阅此处的笔记本

多种模式

通常,所有图节点都使用单一模式进行通信。这意味着它们会读写相同的状态通道。但在某些情况下,我们希望对此有更多的控制: - 内部节点可以传递图的输入/输出中不需要的信息。 - 我们可能还希望为图使用不同的输入/输出模式。例如,输出可能只包含一个相关的输出键。

节点可以在图内部的私有状态通道中写入信息,用于内部节点通信。我们可以简单地定义一个私有模式 PrivateState。更多细节请参阅这个笔记本

也可以为图定义明确的输入和输出模式。在这些情况下,我们定义一个“内部”模式,其中包含与图操作相关的*所有*键。此外,我们还定义 inputoutput 模式,它们是“内部”模式的子集,用于限制图的输入和输出。更多细节请参阅这个笔记本

让我们看一个例子:

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

这里有两个微妙但重要的点需要注意: 1. 我们将 state: InputState 作为输入模式传递给 node_1。但是,我们向 fooOverallState 中的一个通道)写入数据。我们如何能向输入模式中未包含的状态通道写入数据呢?这是因为节点*可以向图状态中的任何状态通道写入数据*。图状态是初始化时定义的状态通道的并集,其中包括 OverallState 以及过滤器 InputStateOutputState。 2. 我们使用 StateGraph(OverallState,input=InputState,output=OutputState) 初始化图。那么,我们如何能在 node_2 中向 PrivateState 写入数据呢?如果在 StateGraph 初始化时没有传递这个模式,图是如何访问它的呢?我们可以这样做是因为*只要状态模式定义存在,节点也可以声明额外的状态通道*。在这种情况下,定义了 PrivateState 模式,所以我们可以在图中添加 bar 作为新的状态通道并向其写入数据。

归约器

归约器是理解如何将节点的更新应用到 State 的关键。State 中的每个键都有自己独立的归约器函数。如果没有明确指定归约器函数,则假定对该键的所有更新都应覆盖它。有几种不同类型的归约器,从默认类型的归约器开始:

默认归约器

以下两个示例展示了如何使用默认归约器:

示例 A:

from typing_extensions import TypedDict

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

在这个示例中,没有为任何键指定归约器函数。假设图的输入是 {"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)指定了一个归约器函数(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(大语言模型响应)。要了解更多关于消息对象的信息,请参阅概念指南。

在图中使用消息

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

但是,你可能还想手动更新图状态中的消息(例如,人工干预)。如果使用 operator.add,你发送给图的手动状态更新将被追加到现有消息列表中,而不是更新现有消息。为避免这种情况,你需要一个能够跟踪消息 ID 并在更新时覆盖现有消息的归约器。为此,可以使用预构建的 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 作为归约器函数的图的示例:

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 Reference: AnyMessage

MessagesState

由于在状态中包含消息列表非常常见,因此存在一个名为 MessagesState 的预构建状态,它使消息的使用变得简单。MessagesState 定义了一个单一的 messages 键,该键是一个 AnyMessage 对象列表,并使用 add_messages 归约器。通常,需要跟踪的状态不仅仅是消息,因此我们看到人们会继承这个状态并添加更多字段,例如:

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


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


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

API Reference: RunnableConfig

在底层,函数会被转换为 RunnableLambda,它为你的函数添加了批量处理和异步支持,以及原生的跟踪和调试功能。

如果你在不指定名称的情况下将节点添加到图中,它将被赋予一个与函数名相同的默认名称。

builder.add_node(my_node)
# 然后你可以通过将其引用为 `"my_node"` 来创建指向/来自此节点的边

START 节点

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

from langgraph.graph import START

graph.add_edge(START, "node_a")

END 节点

END 节点是一个特殊节点,代表终端节点。当你想表示哪些边在完成后没有后续操作时,可以引用此节点。

from langgraph.graph import END

graph.add_edge("node_a", END)

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

  • 普通边:直接从一个节点到下一个节点。
  • 条件边:调用一个函数来确定接下来要前往的节点。
  • 入口点:当用户输入到达时首先调用的节点。
  • 条件入口点:当用户输入到达时,调用一个函数来确定首先要调用的节点。

一个节点可以有**多条**出边。如果一个节点有多个出边,那么**所有**这些目标节点将作为下一个超级步的一部分并行执行。

普通边

如果你**总是**想从节点 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 方法来指定图的入口位置。

from langgraph.graph import START

graph.add_edge(START, "node_a")

条件入口点

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

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 进行状态更新

当你从子图节点向父图节点发送更新,且更新的键在父图和子图的状态模式中都存在时,你**必须**为你要更新的父图状态中的键定义一个归约器。请参阅这个示例

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

在工具中使用

一个常见的用例是从工具内部更新图形状态。例如,在客户支持应用程序中,你可能希望在对话开始时根据客户的账户号码或 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={
            # 更新状态键
            "user_info": user_info,
            # 更新消息历史记录
            "messages": [ToolMessage("Successfully looked up user information", tool_call_id=tool_call_id)]
        }
    )

Important

当从工具中返回 Command 时,你**必须**在 Command.update 中包含 messages(或用于消息历史记录的任何状态键),并且 messages 中的消息列表**必须**包含一个 ToolMessage。这对于使生成的消息历史记录有效是必要的(大语言模型提供商要求带有工具调用的 AI 消息后面必须跟随工具结果消息)。

如果你正在使用通过 Command 更新状态的工具,我们建议使用预构建的 ToolNode,它会自动处理工具返回的 Command 对象并将其传播到图形状态。如果你正在编写一个调用工具的自定义节点,则需要手动将工具返回的 Command 对象作为节点的更新进行传播。

人工介入

Command 是人工介入工作流的重要组成部分:当使用 interrupt() 收集用户输入时,然后使用 Command 来提供输入并通过 Command(resume="User input") 恢复执行。查看这个概念指南以获取更多信息。

持久化

LangGraph 使用 检查点保存器 为你的代理状态提供内置的持久化功能。检查点保存器会在每个超级步保存图状态的快照,从而允许在任何时候恢复。这支持了诸如人工介入交互、内存管理和容错等功能。你甚至可以在图执行后使用相应的 getupdate 方法直接操作图的状态。有关更多详细信息,请参阅 持久化概念指南

线程

LangGraph 中的线程代表你的图与用户之间的各个会话或对话。使用检查点功能时,单个对话中的轮次(甚至单个图执行中的步骤)会通过唯一的线程 ID 进行组织。

存储

LangGraph 通过 BaseStore 接口提供了内置的文档存储功能。与通过线程 ID 保存状态的检查点器不同,存储使用自定义命名空间来组织数据。这实现了跨线程持久化,使智能体能够保留长期记忆、从过去的交互中学习,并随着时间积累知识。常见用例包括存储用户配置文件、构建知识库以及管理所有线程的全局偏好设置。

图迁移

即使使用检查点器来跟踪状态,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"}})

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

interrupt

使用 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?"},
    )
    ...

要恢复图的执行,可将一个 Command 对象传递给图,并将 resume 键设置为 interrupt 函数返回的值。

人在环概念指南中了解更多关于如何将 interrupt 用于**人在环**工作流的信息。

断点

断点可在特定点暂停图执行,并允许逐步执行。断点由 LangGraph 的持久化层提供支持,该层会在每个图步骤后保存状态。断点也可用于启用人工介入工作流,不过我们建议为此目的使用interrupt 函数

断点概念指南中了解更多关于断点的信息。

子图

子图是一种,它被用作另一个图中的节点。这不过是将古老的封装概念应用到了 LangGraph 中。使用子图的一些原因如下:

  • 构建多智能体系统

  • 当你想在多个图中复用一组节点,且这些节点可能共享某些状态时,你可以在一个子图中定义它们一次,然后在多个父图中使用它们

  • 当你希望不同的团队独立地处理图的不同部分时,你可以将每个部分定义为一个子图,只要遵循子图接口(输入和输出模式),父图就可以在不了解子图任何细节的情况下构建

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

  • 添加一个包含已编译子图的节点:当父图和子图共享状态键,且你不需要在输入或输出时转换状态时,这种方法很有用
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  # 注意这个键与父图状态共享
    bar: str

# 定义子图
def subgraph_node(state: SubgraphState):
    # 注意这个子图节点可以通过共享的 "foo" 键与父图通信
    return {"foo": state["foo"] + "bar"}

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

# 定义父图
builder = StateGraph(State)
builder.add_node("subgraph", subgraph)
...
graph = builder.compile()

作为一个函数

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

class State(TypedDict):
    foo: str

class SubgraphState(TypedDict):
    # 注意这些键都不与父图状态共享
    bar: str
    baz: str

# 定义子图
def subgraph_node(state: SubgraphState):
    return {"bar": state["bar"] + "baz"}

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

# 定义父图
def node(state: State):
    # 将状态转换为子图状态
    response = subgraph.invoke({"bar": state["foo"]})
    # 将响应转换回父状态
    return {"foo": response["bar"]}

builder = StateGraph(State)
# 注意我们使用的是 `node` 函数而不是已编译的子图
builder.add_node(node)
...
graph = builder.compile()

可视化

能够对图进行可视化通常是很有用的,尤其是当图变得更加复杂时。LangGraph 提供了几种内置的图可视化方法。有关更多信息,请参阅此操作指南

流式处理

LangGraph 从设计之初就对流式处理提供了一流的支持,包括在执行过程中从图节点进行流式更新、从大语言模型(LLM)调用中流式传输令牌等。有关更多信息,请参阅此概念指南

Comments