跳到主要内容

​
图

LangGraph 的核心是将代理工作流建模为图。你使用三个关键组件来定义代理的行为
  1. State：一个共享数据结构，表示应用程序的当前快照。它可以是任何数据类型，但通常使用共享状态 schema 来定义。
  2. Nodes：编码代理逻辑的函数。它们接收当前状态作为输入，执行一些计算或副作用，并返回更新后的状态。
  3. Edges：根据当前状态确定接下来执行哪个 Node 的函数。它们可以是条件分支或固定转换。
通过组合 Nodes 和 Edges，你可以创建复杂的、循环的工作流，这些工作流会随着时间的推移演变状态。然而，真正的力量来自于 LangGraph 管理状态的方式。需要强调的是：Nodes 和 Edges 仅仅是函数——它们可以包含 LLM 或只是普通的旧代码。 简而言之：节点执行工作，边告诉下一步做什么。 LangGraph 的底层图算法使用消息传递来定义一个通用程序。当一个节点完成其操作时，它会通过一条或多条边向其他节点发送消息。这些接收节点随后执行其函数，将结果消息传递给下一组节点，并继续这个过程。受 Google 的 Pregel 系统的启发，该程序以离散的“超级步骤”进行。 一个超级步骤可以被认为是图节点的一次迭代。并行运行的节点属于同一个超级步骤，而顺序运行的节点属于不同的超级步骤。在图执行开始时，所有节点都处于 inactive 状态。当一个节点在其任何传入边（或“通道”）上接收到新消息（状态）时，它会变为 active。然后，活跃节点运行其函数并响应更新。在每个超级步骤结束时，没有传入消息的节点通过将自身标记为 inactive 来投票 halt。当所有节点都 inactive 并且没有消息正在传输时，图执行终止。

​
StateGraph

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

​
编译你的图

要构建你的图，你首先定义状态，然后添加节点和边，最后编译它。究竟什么是编译你的图以及为什么需要它？ 编译是一个非常简单的步骤。它对你的图的结构进行一些基本检查（例如没有孤立节点）。你还可以在其中指定运行时参数，例如检查点和断点。你只需调用 .compile 方法即可编译你的图：
graph = graph_builder.compile(...)
你 必须 在使用你的图之前编译它。

​
状态

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

​
Schema

指定图 schema 的主要文档方式是使用 TypedDict。如果你想在状态中提供默认值，请使用 dataclass。我们还支持使用 Pydantic BaseModel 作为你的图状态，如果你需要递归数据验证（但请注意 Pydantic 的性能低于 TypedDict 或 dataclass）。 默认情况下，图将具有相同的输入和输出 schema。如果你想更改这一点，你还可以直接指定显式的输入和输出 schema。当你有很多键，其中一些明确用于输入而另一些用于输出时，这会很有用。有关如何使用的信息，请参阅此处的指南。

​
多个 schema

通常，所有图节点都使用一个单一的 schema 进行通信。这意味着它们将读取和写入相同的状态通道。但是，在某些情况下，我们希望对此有更多的控制。
  • 内部节点可以传递图中不需要的输入/输出信息。
  • 我们可能还希望为图使用不同的输入/输出 schema。例如，输出可能只包含一个相关的输出键。
内部节点可以在图中写入私有状态通道以进行内部节点通信。我们可以简单地定义一个私有 schema，PrivateState。 也可以为图定义显式的输入和输出 schema。在这些情况下，我们定义一个包含与图操作相关的所有键的“内部” schema。但是，我们还定义了作为“内部” schema 子集的 input 和 output 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:
    # 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_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'}
这里有两点微妙而重要的注意事项
  1. 我们将 state: InputState 作为输入 schema 传递给 node_1。但是，我们写入到 foo，这是 OverallState 中的一个通道。我们如何写入不包含在输入 schema 中的状态通道？这是因为节点可以写入图状态中的任何状态通道。图状态是初始化时定义的状态通道的并集，其中包括 OverallState 以及过滤器 InputState 和 OutputState。
  2. 我们用 StateGraph(OverallState,input_schema=InputState,output_schema=OutputState) 初始化图。那么，我们如何在 node_2 中写入 PrivateState 呢？如果 PrivateState 没有在 StateGraph 初始化时传入，图如何访问此 schema？我们可以这样做，因为节点也可以声明额外的状态通道，只要状态 schema 定义存在即可。在这种情况下，PrivateState schema 已定义，因此我们可以将 bar 添加为图中的新状态通道并写入其中。

​
Reducer

Reducer 是理解节点更新如何应用于 State 的关键。State 中的每个键都有自己的独立 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"]}。然后我们假设第一个 Node 返回 {"foo": 2}。这被视为对状态的更新。请注意，Node 不需要返回整个 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}。这被视为对状态的更新。请注意，Node 不需要返回整个 State schema - 只需返回一个更新。应用此更新后，State 将是 {"foo": 2, "bar": ["hi"]}。如果第二个节点返回 {"bar": ["bye"]}，则 State 将是 {"foo": 2, "bar": ["hi", "bye"]}。请注意，这里的 bar 键是通过将两个列表相加来更新的。

​
覆盖

在某些情况下，你可能希望绕过 reducer 并直接覆盖状态值。LangGraph 为此提供了 Overwrite 类型。在此处了解如何使用 Overwrite。

​
在图状态中使用消息

​
为什么要使用消息？

大多数现代 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.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]

​
节点

在 LangGraph 中，节点是接受以下参数的 Python 函数（同步或异步）
  1. state：图的状态
  2. config：一个 RunnableConfig 对象，包含配置信息，如 thread_id，以及跟踪信息，如 tags
  3. runtime：一个 Runtime 对象，包含运行时 context 和其他信息，如 store 和 stream_writer
与 NetworkX 类似，你使用 add_node 方法将这些节点添加到图中。 
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)
...
在幕后，函数被转换为 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")

​
END 节点

END 节点是一个特殊节点，表示一个终止节点。当你希望表示哪些边在完成操作后没有后续操作时，会引用此节点。 
from langgraph.graph import END

graph.add_edge("node_a", END)

​
节点缓存

LangGraph 支持根据节点的输入缓存任务/节点。要使用缓存：
  • 在编译图（或指定入口点）时指定缓存。
  • 为节点指定缓存策略。每个缓存策略都支持：
    • key_func 用于根据节点的输入生成缓存键，默认为使用 pickle 对输入进行 hash。
    • ttl，缓存的存活时间（以秒为单位）。如果未指定，则缓存永不失效。
例如： 
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]:
    # expensive computation
    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'))    
# [{'expensive_node': {'result': 10}}]
print(graph.invoke({"x": 5}, stream_mode='updates'))    
# [{'expensive_node': {'result': 10}, '__metadata__': {'cached': True}}]
  1. 第一次运行需要两秒钟（由于模拟的昂贵计算）。
  2. 第二次运行利用缓存并快速返回。

​
边

边定义了逻辑如何路由以及图如何决定停止。这是代理如何工作以及不同节点如何相互通信的重要组成部分。有几种关键类型的边：
  • 普通边：直接从一个节点到下一个节点。
  • 条件边：调用一个函数来决定接下来要到哪个（或哪些）节点。
  • 入口点：当用户输入到达时，首先调用哪个节点。
  • 条件入口点：当用户输入到达时，调用一个函数来决定首先调用哪个（或哪些）节点。
一个节点可以有多个出边。如果一个节点有多个出边，所有这些目标节点都将在下一个超级步骤中并行执行。

​
普通边

如果你想 总是 从节点 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"})
如果你想在一个函数中结合状态更新和路由，请使用 Command 而不是条件边。

​
入口点

入口点是图启动时运行的第一个节点。你可以使用 add_edge 方法从虚拟 START 节点到要执行的第一个节点，以指定图的入口。 
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"})

​
发送

默认情况下，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)

​
命令

将控制流（边）和状态更新（节点）结合起来会很有用。例如，你可能希望在同一个节点中既执行状态更新又决定下一步要到哪个节点。LangGraph 提供了一种通过从节点函数返回 Command 对象来实现此目的的方法。 
def my_node(state: State) -> Command[Literal["my_other_node"]]:
    return Command(
        # state update
        update={"foo": "bar"},
        # control flow
        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")
当你在节点函数中返回 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",  # where `other_subgraph` is a node in the parent graph
        graph=Command.PARENT
    )
将 graph 设置为 Command.PARENT 将导航到最近的父图。当你从子图节点向父图节点发送更新时，如果该键同时由父图和子图状态模式共享，你必须在父图状态中为要更新的键定义一个reducer。请参阅此示例。
这在实现多代理移交时特别有用。 请查阅此指南了解详情。

​
在工具中使用

一个常见的用例是从工具内部更新图状态。例如，在客户支持应用程序中，你可能希望在对话开始时根据客户的帐号或 ID 查找客户信息。 请参阅此指南了解详情。

​
人工干预

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

​
图迁移

即使在使用检查点来跟踪状态时，LangGraph 也可以轻松处理图定义（节点、边和状态）的迁移。
  • 对于图末尾的线程（即未中断的线程），你可以更改整个图的拓扑结构（即所有节点和边，删除、添加、重命名等）。
  • 对于当前中断的线程，我们支持除重命名/删除节点之外的所有拓扑更改（因为该线程现在可能即将进入一个不再存在的节点）——如果这是一个障碍，请联系我们，我们可以优先解决此问题。
  • 对于修改状态，我们完全支持添加和删除键的向后和向前兼容性。
  • 重命名的状态键会在现有线程中丢失其保存的状态。
  • 类型以不兼容方式更改的状态键目前可能会导致在更改前具有状态的线程中出现问题——如果这是一个障碍，请联系我们，我们可以优先解决此问题。

​
运行时上下文

创建图时，可以为传递给节点的运行时上下文指定 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)
    # ...
有关配置的完整说明，请参阅此指南。

​
递归限制

递归限制设置了图在单次执行期间可以执行的超级步骤的最大数量。一旦达到限制，LangGraph 将引发 GraphRecursionError。默认情况下，此值设置为 25 步。递归限制可以在运行时在任何图上设置，并通过配置字典传递给 invoke/stream。重要的是，recursion_limit 是一个独立的 config 键，不应像所有其他用户定义配置一样在 configurable 键中传递。请参见以下示例： 
graph.invoke(inputs, config={"recursion_limit": 5}, context={"llm": "anthropic"})
阅读此操作指南，了解有关递归限制如何工作的更多信息。

​
可视化

可视化图通常很有用，尤其是当它们变得更复杂时。LangGraph 提供了几种内置方法来可视化图。请参阅此操作指南了解更多信息。
在 GitHub 上编辑此页面源文件。
以编程方式连接这些文档到 Claude、VSCode 等，通过 MCP 获取实时答案。
© . This site is unofficial and not affiliated with LangChain, Inc.