LangGraph 术语表¶
图形¶
在核心层面，LangGraph 将代理工作流建模为图形。您使用三个关键组件定义代理的行为：
-
状态：一个共享的数据结构，表示应用程序当前的状态快照。它可以是任何 Python 类型，但通常是一个TypedDict或 Pydantic 的BaseModel。 -
节点：编码代理逻辑的 Python 函数。它们接收当前的状态作为输入，执行某些计算或副作用，并返回更新后的状态。 -
边：根据当前状态确定下一个要执行的节点的 Python 函数。这些函数可以是条件分支或固定转换。
通过组合 节点 和 边，您可以创建复杂的循环工作流，随着时间推移不断演化 状态。真正的力量在于 LangGraph 如何管理这个 状态。强调一点：节点 和 边 不过是 Python 函数——它们可以包含大型语言模型（LLM）或仅仅是普通的 Python 代码。
简而言之：节点负责执行任务，边决定下一步做什么。
LangGraph 的底层图算法使用 消息传递 来定义一个通用程序。当一个节点完成其操作时，它会沿一条或多条边向其他节点发送消息。这些接收节点随后执行其函数，将生成的消息传递给下一组节点，过程如此继续。受 Google 的 Pregel 系统启发，该程序以离散的“超级步骤”形式推进。
一个超级步骤可以视为对图节点的一次迭代。并行运行的节点属于同一个超级步骤，而顺序运行的节点则属于不同的超级步骤。在图执行开始时，所有节点都处于 非活动 状态。当一个节点在其任意入边（或“通道”）上收到新消息（状态）时，它就会变为 活动 状态。活动节点然后运行其函数并作出响应更新。每个超级步骤结束时，没有入边消息的节点投票将其自身标记为 非活动 状态。当所有节点都处于 非活动 状态且没有消息在传输时，图执行终止。
状态图¶
StateGraph 类是主要使用的图类。此图由用户定义的 状态 对象参数化。
编译您的图¶
为了构建您的图，您首先定义 状态，然后添加 节点 和 边，最后编译它。那么，究竟什么是编译您的图以及为什么需要它？
编译其实是一个相当简单的步骤。它提供了一些基本检查来确保您的图结构正确（无孤立节点等）。这也是指定运行时参数的地方，如 检查点器 和 断点。通过调用 .compile 方法即可编译您的图：
您 必须 在使用图之前先对其进行编译。
状态¶
当你定义一个图时，首先要定义该图的 State。State 包含了图的模式以及用于指定如何应用状态更新的reducer函数。State 的模式将是图中所有 Nodes 和 Edges 的输入模式，并且可以是 TypedDict 或 Pydantic 模型。所有的 Nodes 都会发出对 State 的更新，然后使用指定的 reducer 函数来应用这些更新。
模式¶
指定图模式的主要方法是使用 TypedDict。然而，我们还支持使用 Pydantic BaseModel 作为图的状态以添加**默认值**和额外的数据验证。
默认情况下，图具有相同的输入和输出模式。如果你想要改变这一点，也可以直接指定显式的输入和输出模式。这在你有许多键，并且其中一些明确为输入而另一些为输出时很有用。参见此处的笔记本了解如何使用。
多个模式¶
通常，图的所有节点都与单个模式通信。这意味着它们将读写相同的态通道。但是，在某些情况下，我们希望对此有更多的控制：
- 内部节点可以传递不需要在图的输入/输出中显示的信息。
- 我们可能也希望使用不同的输入/输出模式来处理图。例如，输出可能只包含一个相关的输出键。
节点可以在图内部写入私有态通道以供内部节点通信。我们只需定义一个私有模式 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=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？如果它没有在StateGraph初始化时传递，图是如何获得对该模式的访问权限的？我们可以这样做是因为**节点也可以声明附加的状态通道**，只要状态模式定义存在即可。在这种情况下，PrivateState模式已定义，因此我们可以在图中添加新的状态通道bar并写入它。
减少器¶
减少器是理解节点更新如何应用于 State 的关键。State 中的每个键都有自己的独立减少器函数。如果没有显式指定减少器函数，则假定所有对该键的更新都应该覆盖它。有几种不同类型的减少器，从默认类型的减少器开始：
默认减少器¶
这两个示例展示了如何使用默认减少器：
示例A：
在这个示例中，没有为任何键指定减少器函数。假设图的输入是 {"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 键通过将两个列表相加来进行更新。
在图状态中使用消息¶
为什么使用消息？¶
大多数现代语言模型提供商都提供了一个聊天模型接口，该接口接受消息列表作为输入。特别是 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 作为其减少器函数的图示例。
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 减少器。通常，要跟踪的状态比仅仅消息更多，所以我们看到人们子类化此状态并添加更多的字段，如下所示：
节点¶
在LangGraph中，节点通常是Python函数（同步或异步），其中第一个位置参数是状态，第二个位置参数（可选）是一个“配置”，包含可选的配置参数（如thread_id）。
类似于NetworkX，你可以使用add_node方法将这些节点添加到图中：
API Reference: RunnableConfig | StateGraph
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)
...
在幕后，函数会被转换为RunnableLambda，这会为你的函数添加批处理和异步支持，以及本机跟踪和调试功能。
如果你向图中添加一个节点而不指定名称，则该节点将被赋予默认名称，等同于函数名称。
START 节点¶
START 节点是一个特殊的节点，表示发送用户输入到图中的节点。主要目的是确定哪些节点应该首先被调用。
API Reference: START
END 节点¶
END 节点是一个特殊的终端节点。当您希望表示哪些边在其完成后没有进一步操作时，可以引用这个节点。
边缘¶
边缘定义了逻辑如何被路由以及图何时停止运行。这是代理工作方式的重要部分，也是不同节点之间如何通信的关键。有几种关键类型的边：
- 普通边：直接从一个节点到下一个节点。
- 条件边：调用一个函数来确定下一步去哪个（些）节点。
- 入口点：当用户输入到达时首先调用的节点。
- 条件入口点：调用一个函数来确定当用户输入到达时首先调用哪些节点。
一个节点可以有多条出边。如果一个节点有多条出边，则所有这些目标节点将在下一个超级步骤中并行执行。
普通边¶
如果你总是希望从节点A到节点B，你可以直接使用add_edge方法。
条件边¶
如果你想**可选地**路由到一个或多个边（或可选地终止），你可以使用add_conditional_edges方法。此方法接受一个节点名称和一个“路由函数”，在该节点执行后调用：
类似于节点，routing_function接受当前图的状态，并返回一个值。
默认情况下，routing_function的返回值将作为要发送状态的下一个节点（或节点列表）的名称。所有这些节点将在下一个超级步骤中并行执行。
你可以选择提供一个字典，将routing_function的输出映射到下一个节点的名称。
Tip
如果你想在一个函数中结合状态更新和路由，请使用Command而不是条件边。
入口点¶
入口点是图开始时首先运行的节点。你可以使用虚拟的START节点的add_edge方法指定进入图的位置。
API Reference: START
条件入口点¶
条件入口点允许你根据自定义逻辑从不同的节点开始。你可以使用虚拟的START节点的add_conditional_edges方法来实现这一点。
API Reference: START
你可以选择提供一个字典，将routing_function的输出映射到下一个节点的名称。
发送 (Send)¶
默认情况下，节点 和 边 是预先定义好的，并且它们操作的是同一个共享状态。然而，在某些情况下，具体的边可能无法提前确定，或者您希望同时存在不同版本的 状态。一个常见的例子是 Map-Reduce 设计模式。在这种设计模式中，第一个节点可能会生成一个对象列表，而您可能希望将另一个节点应用于这些所有对象。对象的数量可能无法提前确定（这意味着边的数量也可能未知），并且输入到下游节点的 状态 应该是不同的（每个生成的对象对应一个状态）。
为了支持这种设计模式，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中指定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将导航到最近的父图。
这在实现多代理交接时特别有用。
在工具内部使用¶
常见的用例是从工具内部更新图状态。例如，在客户支持应用程序中，您可能希望在对话开始时根据客户的账户号码或ID查找客户信息。为了从工具中更新图状态，您可以返回Command(update={"my_custom_key": "foo", "messages": [...]})：
@tool
def lookup_user_info(tool_call_id: Annotated[str, InjectedToolCallId], config: RunnableConfig):
"""使用此功能查找用户信息，以便更好地回答他们的问题。"""
user_info = get_user_info(config.get("configurable", {}).get("user_id"))
return Command(
update={
# 更新状态键
"user_info": user_info,
# 更新消息历史记录
"messages": [ToolMessage("成功查找用户信息", tool_call_id=tool_call_id)]
}
)
Important
当您从工具返回Command时，您必须在Command.update中包含messages（或任何用于消息历史记录的状态键），并且messages中的消息列表必须包含一个ToolMessage。这是为了确保结果的消息历史记录有效（LLM提供商要求AI消息与工具调用相结合，然后跟随工具结果消息）。
如果您正在使用的工具通过Command更新状态，我们建议使用预构建的ToolNode，它自动处理返回Command对象的工具，并将其传播到图状态。如果您编写了一个自定义节点，该节点调用了工具，则需要手动传播工具返回的Command对象作为节点的更新。
人机交互¶
Command是人机交互工作流程的重要组成部分：当使用interrupt()收集用户输入时，Command随后用于提供输入并通过Command(resume="User input")恢复执行。查看此概念性指南以获取更多信息。
持久化¶
LangGraph 提供了内置的持久性功能，用于保存代理的状态，使用的是检查点器。检查点器在每个超级步骤后保存图状态的快照，允许在任何时候恢复执行。这使得诸如人机交互、内存管理以及容错等功能成为可能。您甚至可以直接在执行后通过适当的get和update方法来操作图的状态。更多详细信息，请参阅persistence概念指南。
线程¶
在LangGraph中，线程代表了图与用户之间单独会话或对话。当使用检查点功能时，单个对话中的轮次（甚至是单次图执行中的步骤）都会通过一个唯一的线程ID来组织。
存储¶
LangGraph 通过 BaseStore 接口提供内置的文档存储功能。与按线程ID保存状态的检查点不同，存储使用自定义命名空间来组织数据。这使得跨线程持久化成为可能，允许代理维持长期记忆，从过去的交互中学习，并随着时间的推移积累知识。常见的应用场景包括存储用户资料、构建知识库以及管理所有线程之间的全局偏好设置。
图迁移¶
LangGraph 可以轻松处理图定义（节点、边和状态）的迁移，即使在使用检查点程序来跟踪状态的情况下也是如此。
- 对于位于图末端的线程（即未中断的线程），您可以更改整个图的拓扑结构（即所有节点和边，删除、添加、重命名等）
- 对于当前被中断的线程，我们支持除重命名/删除节点之外的所有拓扑更改（因为该线程可能会进入一个不再存在的节点）——如果这是一个阻碍因素，请联系我们，我们可以优先解决这个问题。
- 对于修改状态，我们对添加和删除键具有完全的前后兼容性
- 被重命名的状态键将失去其在现有线程中的保存状态
- 如果状态键的类型以不兼容的方式发生变化，则这些变化可能会在具有变更前状态的线程中引起问题——如果这是一个阻碍因素，请联系我们，我们可以优先解决这个问题。
配置¶
在创建图表时，您还可以标记图表中某些部分是可配置的。这通常是为了方便在不同模型或系统提示之间切换。这样，您可以创建一个单一的“认知架构”（即图表），但可以拥有多个不同的实例。
您可以选择性地指定一个 config_schema 在创建图表时。
然后可以通过 configurable 配置字段将此配置传递给图表。
您可以在节点内部访问并使用此配置：
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 键中。请参见以下示例：
阅读本教程以了解递归限制的工作原理。
中断¶
使用interrupt函数在特定点暂停图以收集用户输入。interrupt函数将中断信息传递给客户端，允许开发人员收集用户输入、验证图的状态或在继续执行之前做出决策。
API Reference: interrupt
from langgraph.types import interrupt
def human_approval_node(state: State):
...
answer = interrupt(
# 此值将发送到客户端。
# 它可以是任何可序列化的JSON值。
{"问题": "是否可以继续?"},
)
...
通过将一个包含resume键设置为interrupt函数返回值的Command对象传递给图来恢复图的执行。
更多关于如何使用interrupt函数实现**人机交互**工作流的内容，请参阅人机交互概念指南。
断点¶
断点可以在特定位置暂停图执行，并允许逐步骤地执行。断点由LangGraph的持久层提供支持，该层在每个图步骤后保存状态。断点也可以用于启用人机交互的工作流，尽管我们建议使用interrupt函数来实现这一目的。
更多关于断点的信息，请参阅断点概念指南。
子图¶
子图是一种用于作为另一个图中的节点的图。这不过是封装概念在LangGraph中的应用。使用子图的一些原因包括：
-
构建多代理系统
-
当您希望在多个图中重用一组节点时，这些节点可能共享某些状态，您可以将它们定义为一个子图，并在多个父图中使用它们。
-
当您希望不同的团队可以独立地工作在图的不同部分时，您可以将每个部分定义为一个子图，只要子图接口（输入和输出模式）得到尊重，父图就可以在不知道子图细节的情况下构建。
有两种方法可以向父图添加子图：
- 添加带有编译后子图的节点：当父图和子图共享状态键且不需要转换状态时，此方法非常有用。
- 添加带有调用子图函数的节点：当父图和子图具有不同的状态模式且需要在调用子图之前或之后转换状态时，此方法非常有用。
subgraph = subgraph_builder.compile()
def call_subgraph(state: State):
return subgraph.invoke({"subgraph_key": state["parent_key"]})
builder.add_node("subgraph", call_subgraph)
让我们来看一下每种情况的例子。
作为编译后的图¶
创建子图节点最简单的方法是直接使用一个编译后的子图。这样做时，**重要**的是父图和子图的状态模式至少共享一个键，以便它们能够通信。如果您的图和子图没有共享任何键，则应编写一个调用子图的函数作为函数。
注意事项
如果您传递额外的键给子图节点（即除了共享键之外），它们将被子图节点忽略。同样，如果您从子图返回额外的键，它们将被父图忽略。
API Reference: StateGraph
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 在设计时充分考虑了流式处理的支持，包括在执行过程中从图节点获取流式更新、从大语言模型调用中获取令牌流等。更多详细信息，请参阅此概念性指南。