​
图

LangGraph 的核心是将 Agent 工作流建模为图。您使用三个关键组件来定义 Agent 的行为：
  1. State：表示应用程序当前快照的共享数据结构。它可以是任何数据类型，但通常使用共享状态模式定义。
  2. Nodes：编码 Agent 逻辑的函数。它们接收当前状态作为输入，执行一些计算或副作用，并返回更新后的状态。
  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 由图的模式以及指定如何将更新应用于状态的reducer函数组成。State 的模式将是图中所有 Nodes 和 Edges 的输入模式，可以是 TypedDict 或 Pydantic 模型。所有 Nodes 将发出对 State 的更新，然后使用指定的 reducer 函数应用。

​
模式

定义图模式的主要记录方式是使用TypedDict。如果您想提供默认值，请使用dataclass。如果需要递归数据验证，我们也支持使用 Pydantic BaseModel作为您的图状态（请注意，Pydantic 的性能不如 TypedDict 或 dataclass）。 默认情况下，图具有相同的输入和输出模式。如果您想更改此设置，也可以直接指定显式的输入和输出模式。当您有很多键，其中一些明确用于输入而另一些用于输出时，这很有用。请参阅指南获取更多信息。
高级create_agent工厂在 langchain 中不支持 Pydantic 状态模式。

​
多重模式

通常，所有图形节点使用单一模式通信。这意味着它们将读取和写入相同的状态通道。但是，在某些情况下，我们需要更多的控制：
  • 内部节点可以传递图中输入/输出不需要的信息。
  • 我们可能还想为图使用不同的输入/输出模式。例如，输出可能只包含一个相关的输出键。
可以让节点写入图内部的私有状态通道以进行内部节点通信。我们可以简单地定义一个私有模式 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_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 作为输入模式传递给 node_1。但是，我们写入了 foo，这是 OverallState 中的一个通道。我们怎么可能写入不包含在输入模式中的状态通道？这是因为节点_可以写入图状态中的任何状态通道_。图状态是在初始化时定义的状态通道的并集，其中包括 OverallState 和过滤器 InputState 和 OutputState。
  2. 我们使用以下方式初始化图： 
    StateGraph(
    OverallState,
    input_schema=InputState,
    output_schema=OutputState
)
    我们如何在 node_2 中写入 PrivateState？如果它没有在 StateGraph 初始化中传递，图如何访问这个模式？ 我们可以这样做，因为 _nodes 也可以声明额外的状态 channels_，只要状态模式定义存在。在这种情况下，PrivateState 模式已定义，因此我们可以将 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}。这被视为对状态的更新。请注意，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）指定了 reducer 函数（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 键通过将两个列表相加来更新。

​
覆盖

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

​
在图状态中使用消息

​
为什么使用消息？

大多数现代 LLM 提供商都有接受消息列表作为输入的聊天模型接口。LangChain 的聊天模型接口特别接受消息对象列表作为输入。这些消息有多种形式，如HumanMessage（用户输入）或AIMessage（LLM 响应）。 要了解消息对象的更多信息，请参阅消息概念指南。

​
在图中使用消息

在许多情况下，在图状态中存储先前的对话历史作为消息列表很有帮助。为此，我们可以向图状态添加一个存储 Message 对象列表的键（通道），并使用 reducer 函数对其进行注解（请参阅下面示例中的 messages 键）。reducer 函数对于告诉图如何用每次状态更新（例如，当节点发送更新时）更新状态中的 Message 对象列表至关重要。如果您没有指定 reducer，则每次状态更新都将用最近提供的值覆盖消息列表。如果您想简单地将消息追加到现有列表，可以使用 operator.add 作为 reducer。 但是，您可能还希望手动更新图状态中的消息（例如人工介入）。如果您使用 operator.add，您发送给图的手动状态更新将被追加到现有消息列表，而不是更新现有消息。为了避免这种情况，您需要一个可以跟踪消息 ID 并在更新时覆盖现有消息的 reducer。为此，您可以使用预构建的add_messages函数。对于全新的消息，它将简单地追加到现有列表，但它也会正确处理现有消息的更新。

​
序列化

除了跟踪消息 ID，add_messages函数还会在状态更新收到时尝试将消息反序列化为 LangChain Message 对象。 有关更多信息，请参阅 LangChain 序列化/反序列化。这允许以下列格式发送图输入/状态更新： 
# 这种方式支持
{"messages": [HumanMessage(content="message")]}

# 这种方式也支持
{"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—包含运行时context和其他信息（如 store、stream_writer、execution_info 和 server_info）的 Runtime 对象
与 NetworkX 类似，您使用add_node方法将这些节点添加到图中： 
from dataclasses import dataclass
from typing_extensions import TypedDict

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("在节点中: ", runtime.context.user_id)
    return {"results": f"你好, {state['input']}!"}

def node_with_execution_info(state: State, runtime: Runtime):
    print("在节点中，thread_id: ", runtime.execution_info.thread_id)
    return {"results": f"你好, {state['input']}!"}


builder.add_node("plain_node", plain_node)
builder.add_node("node_with_runtime", node_with_runtime)
builder.add_node("node_with_execution_info", node_with_execution_info)
...
在幕后，函数被转换为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)

​
节点缓存

LangGraph 支持基于节点输入的任务/节点缓存。要使用缓存：
  • 在编译图时指定缓存（或指定入口点）
  • 为节点指定缓存策略。每个缓存策略支持：
    • key_func 用于基于节点输入生成缓存键，默认为输入的 hash（使用 pickle）。
    • 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]:
    # 昂贵的计算
    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. 第二次运行使用缓存并快速返回。

​
边

边定义如何路由逻辑以及图如何决定停止。这是您的 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 接受图的当前 state 并返回一个值。 默认情况下，routing_function 的返回值用作下一节点的名称（或节点列表）来发送状态。所有那些节点将作为下一超步骤的一部分并行运行。 您可以选择提供一个字典，将 routing_function 的输出映射到下一节点的名称。 
graph.add_conditional_edges("node_a", routing_function, {True: "node_b", False: "node_c"})
如果您想在一个函数中组合状态更新和路由，请使用Command而不是条件边。

​
入口点

入口点是图开始时运行的第一个节点。您可以使用从虚拟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"})

​
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

Command 是一个用于控制图执行的多功能原语。它接受四个参数：
  • update：应用状态更新（类似于从节点返回更新）。
  • goto：导航到特定节点（类似于条件边）。
  • graph：从子图导航时定位父图。
  • resume：在中断后提供值以恢复执行。
Command 在三种上下文中使用：

​
从节点返回

​
update 和 goto

从节点函数返回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")
当您需要同时更新状态和路由到不同节点时使用Command。如果您只需要路由而不更新状态，请使用条件边。
在节点函数中返回Command时，必须使用节点名称列表添加返回类型注解，例如 Command[Literal["my_other_node"]]。这对于图形渲染是必要的，并告诉 LangGraph my_node 可以导航到 my_other_node。
Command 仅添加动态边——使用 add_edge / addEdge 定义的静态边仍然执行。例如，如果 node_a 返回 Command(goto="my_other_node") 并且您还有 graph.add_edge("node_a", "node_b")，则 node_b 和 my_other_node 都将运行。
查看此操作指南获取如何使用Command的端到端示例。

​
graph

如果您正在使用子图，您可以通过在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
    )
将 graph 设置为 Command.PARENT 将导航到最近的父图。当您将更新从子图节点发送到父图节点时，对于父图和子图状态模式共享的键，您必须为父图状态中要更新的键定义一个 reducer。请参阅此示例。
这对于实现多 Agent 交接特别有用。请参阅导航到父图中的节点获取详情。

​
输入到 invoke 或 stream

Command(resume=...) 是作为 invoke()/stream() 输入的唯一 Command 模式。不要使用 Command(update=...) 作为输入来继续多轮对话——因为将任何 Command 作为输入会从最新检查点恢复（即最后运行的步骤，而不是 __start__），如果图已经完成，图将看起来卡住。要在现有线程上继续对话，请传递普通输入字典：
# 错误 - 图从最新检查点恢复
#（最后运行的步骤），看起来卡住
graph.invoke(Command(update={
    "messages": [{"role": "user", "content": "follow up"}]
}), config)

# 正确 - 普通字典从 __start__ 重新开始
graph.invoke( {
    "messages": [{"role": "user", "content": "follow up"}]
}, config)

​
resume

使用 Command(resume=...) 在中断后提供值并恢复图执行。传递给 resume 的值成为暂停节点内 interrupt() 调用的返回值： 
from langgraph.types import Command, interrupt

def human_review(state: State):
    # 暂停图并等待值
    answer = interrupt("您批准吗？")
    return {"messages": [{"role": "user", "content": answer}]}

# 第一次调用 - 命中中断并暂停
result = graph.invoke({"messages": [...]}, config)

# 用值恢复 - interrupt() 调用返回 "yes"
result = graph.invoke(Command(resume="yes"), config)
查看中断概念指南获取中断模式的完整详情，包括多个中断和验证循环。

​
从工具返回

您可以从工具返回Command来更新图状态和控制流。使用 update 修改状态（例如，保存对话期间查找的客户信息），使用 goto 在工具完成后路由到特定节点。
在工具内部使用时，goto 添加动态边——在调用工具的节点上定义的任何静态边仍将执行。
请参阅在工具内部使用获取详情。

​
图迁移

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)
    # ...