图 API 概念¶

图¶

LangGraph 的核心是将智能代理工作流建模为图。你使用三个关键组件来定义智能代理的行为：

State：表示应用程序当前快照的共享数据结构。它可以是任何数据类型，但通常使用共享状态模式定义。
Nodes：编码智能代理逻辑的函数。它们接收当前状态作为输入，执行一些计算或副作用，并返回更新后的状态。
Edges：根据当前状态确定下一个要执行的 Node 的函数。它们可以是条件分支或固定转换。

通过组合 Nodes 和 Edges，你可以创建复杂的、循环的工作流，随着时间的推移演变状态。然而，真正的力量来自于 LangGraph 如何管理该状态。需要强调的是：Nodes 和 Edges 只不过是函数——它们可以包含 LLM 或只是普通的代码。

简而言之：节点执行工作，边告诉下一步做什么。

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

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

StateGraph¶

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

编译你的图¶

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

编译是一个相当简单的步骤。它对图的结构进行一些基本检查（没有孤立节点等）。它也是你可以指定运行时参数（如检查点器和断点）的地方。你只需调用 .compile 方法来编译你的图：

Python

graph = graph_builder.compile(...)

JavaScript

const graph = new StateGraph(StateAnnotation)
  .addNode("nodeA", nodeA)
  .addEdge(START, "nodeA")
  .addEdge("nodeA", END)
  .compile();

你**必须**在使用图之前编译它。

状态¶

Python

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

JavaScript

定义图时，你首先要做的是定义图的 State。State 由图的模式以及指定如何将更新应用于状态的 reducer 函数组成。State 的模式将是图中所有 Nodes 和 Edges 的输入模式，可以是 Zod 模式或使用 Annotation.Root 构建的模式。所有 Nodes 将发出对 State 的更新，然后使用指定的 reducer 函数应用这些更新。

模式¶

Python

指定图模式的主要文档化方式是使用 TypedDict。如果你想在状态中提供默认值，请使用 dataclass。如果你想要递归数据验证，我们也支持使用 Pydantic BaseModel 作为图状态（但请注意，pydantic 的性能不如 TypedDict 或 dataclass）。

默认情况下，图将具有相同的输入和输出模式。如果你想更改此设置，你也可以直接指定显式的输入和输出模式。当你有很多键，其中一些明确用于输入，另一些用于输出时，这很有用。请参阅此处的指南了解如何使用。

JavaScript

指定图模式的主要文档化方式是使用 Zod 模式。但是，我们也支持使用 Annotation API 来定义图的模式。

默认情况下，图将具有相同的输入和输出模式。如果你想更改此设置，你也可以直接指定显式的输入和输出模式。当你有很多键，其中一些明确用于输入，另一些用于输出时，这很有用。

多个模式¶

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

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

可以让节点写入图内部的私有状态通道以进行内部节点通信。我们可以简单地定义一个私有模式 PrivateState。

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

让我们看一个例子：

Python

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

JavaScript

const InputState = z.object({
  userInput: z.string(),
});

const OutputState = z.object({
  graphOutput: z.string(),
});

const OverallState = z.object({
  foo: z.string(),
  userInput: z.string(),
  graphOutput: z.string(),
});

const PrivateState = z.object({
  bar: z.string(),
});

const graph = new StateGraph({
  state: OverallState,
  input: InputState,
  output: OutputState,
})
  .addNode("node1", (state) => {
    // Write to OverallState
    return { foo: state.userInput + " name" };
  })
  .addNode("node2", (state) => {
    // Read from OverallState, write to PrivateState
    return { bar: state.foo + " is" };
  })
  .addNode(
    "node3",
    (state) => {
      // Read from PrivateState, write to OutputState
      return { graphOutput: state.bar + " Lance" };
    },
    { input: PrivateState }
  )
  .addEdge(START, "node1")
  .addEdge("node1", "node2")
  .addEdge("node2", "node3")
  .addEdge("node3", END)
  .compile();

await graph.invoke({ userInput: "My" });
// { graphOutput: 'My name is Lance' }

这里有两个微妙而重要的点需要注意：

Python

我们将 state: InputState 作为输入模式传递给 node_1。但是，我们写出到 foo，这是 OverallState 中的一个通道。我们如何能写出到一个不包含在输入模式中的状态通道？这是因为节点_可以写入图状态中的任何状态通道_。图状态是在初始化时定义的状态通道的并集，包括 OverallState 和过滤器 InputState 和 OutputState。
我们用 StateGraph(OverallState,input_schema=InputState,output_schema=OutputState) 初始化图。那么，我们如何能在 node_2 中写入 PrivateState？如果它没有在 StateGraph 初始化中传递，图如何获得对此模式的访问？我们可以这样做是因为_节点也可以声明额外的状态通道_，只要状态模式定义存在。在这种情况下，PrivateState 模式已定义，所以我们可以在图中添加 bar 作为新的状态通道并写入它。

JavaScript

我们将 state 作为输入模式传递给 node1。但是，我们写出到 foo，这是 OverallState 中的一个通道。我们如何能写出到一个不包含在输入模式中的状态通道？这是因为节点_可以写入图状态中的任何状态通道_。图状态是在初始化时定义的状态通道的并集，包括 OverallState 和过滤器 InputState 和 OutputState。
我们用 StateGraph({ state: OverallState, input: InputState, output: OutputState }) 初始化图。那么，我们如何能在 node2 中写入 PrivateState？如果它没有在 StateGraph 初始化中传递，图如何获得对此模式的访问？我们可以这样做是因为_节点也可以声明额外的状态通道_，只要状态模式定义存在。在这种情况下，PrivateState 模式已定义，所以我们可以在图中添加 bar 作为新的状态通道并写入它。

归约器¶

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

默认归约器¶

这两个例子展示了如何使用默认归约器：

示例 A：

Python

from typing_extensions import TypedDict

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

JavaScript

const State = z.object({
  foo: z.number(),
  bar: z.array(z.string()),
});

在这个例子中，没有为任何键指定归约器函数。假设图的输入是：

Python

{"foo": 1, "bar": ["hi"]}。然后假设第一个 Node 返回 {"foo": 2}。这被视为对状态的更新。注意 Node 不需要返回整个 State 模式——只需要一个更新。应用此更新后，State 将变为 {"foo": 2, "bar": ["hi"]}。如果第二个节点返回 {"bar": ["bye"]}，则 State 将变为 {"foo": 2, "bar": ["bye"]}

JavaScript

{ foo: 1, bar: ["hi"] }。然后假设第一个 Node 返回 { foo: 2 }。这被视为对状态的更新。注意 Node 不需要返回整个 State 模式——只需要一个更新。应用此更新后，State 将变为 { foo: 2, bar: ["hi"] }。如果第二个节点返回 { bar: ["bye"] }，则 State 将变为 { foo: 2, bar: ["bye"] }

示例 B：

Python

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 键是通过将两个列表相加来更新的。

JavaScript

import { z } from "zod";
import { withLangGraph } from "@langchain/langgraph/zod";

const State = z.object({
  foo: z.number(),
  bar: withLangGraph(z.array(z.string()), {
    reducer: {
      fn: (x, y) => x.concat(y),
    },
  }),
});

在这个例子中，我们使用 withLangGraph 函数为第二个键（bar）指定了一个归约器函数。注意第一个键保持不变。假设图的输入是 { foo: 1, bar: ["hi"] }。然后假设第一个 Node 返回 { foo: 2 }。这被视为对状态的更新。注意 Node 不需要返回整个 State 模式——只需要一个更新。应用此更新后，State 将变为 { foo: 2, bar: ["hi"] }。如果第二个节点返回 { bar: ["bye"] }，则 State 将变为 { foo: 2, bar: ["hi", "bye"] }。注意这里 bar 键是通过将两个数组相加来更新的。

在图状态中使用消息¶

为什么使用消息？¶

Python

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

JavaScript

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

在图中使用消息¶

Python

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

但是，你可能还想手动更新图状态中的消息（例如，人机协同）。如果你使用 operator.add，你发送到图的手动状态更新将被附加到现有的消息列表，而不是更新现有消息。为避免这种情况，你需要一个可以跟踪消息 ID 并覆盖现有消息（如果更新）的归约器。为此，你可以使用预构建的 add_messages 函数。对于全新的消息，它将简单地附加到现有列表，但它也会正确处理现有消息的更新。

JavaScript

在许多情况下，将先前的对话历史记录作为消息列表存储在图状态中是有帮助的。为此，我们可以向图状态添加一个键（通道），用于存储 Message 对象列表，并使用归约器函数对其进行注释（参见下面示例中的 messages 键）。归约器函数对于告诉图如何在每次状态更新时更新 State 中的 Message 对象列表至关重要（例如，当节点发送更新时）。如果你不指定归约器，每次状态更新都会用最近提供的值覆盖消息列表。如果你想简单地将消息附加到现有列表，你可以使用一个连接数组的函数作为归约器。

但是，你可能还想手动更新图状态中的消息（例如，人机协同）。如果你使用简单的连接函数，你发送到图的手动状态更新将被附加到现有的消息列表，而不是更新现有消息。为避免这种情况，你需要一个可以跟踪消息 ID 并覆盖现有消息（如果更新）的归约器。为此，你可以使用预构建的 MessagesZodState 模式。对于全新的消息，它将简单地附加到现有列表，但它也会正确处理现有消息的更新。

序列化¶

Python

除了跟踪消息 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 作为归约器函数的图的示例。

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]

JavaScript

除了跟踪消息 ID 之外，MessagesZodState 还会尝试在 messages 通道上收到状态更新时将消息反序列化为 LangChain Message 对象。这允许以以下格式发送图输入/状态更新：

// this is supported
{
  messages: [new HumanMessage("message")];
}

// and this is also supported
{
  messages: [{ role: "human", content: "message" }];
}

由于使用 MessagesZodState 时状态更新始终反序列化为 LangChain Messages，因此你应该使用点表示法访问消息属性，如 state.messages[state.messages.length - 1].content。下面是使用 MessagesZodState 的图的示例：

import { StateGraph, MessagesZodState } from "@langchain/langgraph";

const graph = new StateGraph(MessagesZodState)
  ...

MessagesZodState 定义了一个单独的 messages 键，它是 BaseMessage 对象的列表并使用适当的归约器。通常，需要跟踪的状态不仅仅是消息，所以我们看到人们扩展此状态并添加更多字段，如：

const State = z.object({
  messages: MessagesZodState.shape.messages,
  documents: z.array(z.string()),
});

Python

MessagesState¶

由于在状态中拥有消息列表非常常见，因此存在一个名为 MessagesState 的预构建状态，使其易于使用消息。MessagesState 定义了一个单独的 messages 键，它是 AnyMessage 对象的列表并使用 add_messages 归约器。通常，需要跟踪的状态不仅仅是消息，所以我们看到人们子类化此状态并添加更多字段，如：

from langgraph.graph import MessagesState

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

节点¶

Python

在 LangGraph 中，节点是接受以下参数的 Python 函数（同步或异步）：

state：图的状态
config：包含配置信息（如 thread_id）和跟踪信息（如 tags）的 RunnableConfig 对象
runtime：包含运行时 context和其他信息（如 store 和 stream_writer）的 Runtime 对象

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

JavaScript

在 LangGraph 中，节点通常是接受以下参数的函数（同步或异步）：

state：图的状态
config：包含配置信息（如 thread_id）和跟踪信息（如 tags）的 RunnableConfig 对象

你可以使用 addNode 方法将节点添加到图中。

import { StateGraph } from "@langchain/langgraph";
import { RunnableConfig } from "@langchain/core/runnables";
import { z } from "zod";

const State = z.object({
  input: z.string(),
  results: z.string(),
});

const builder = new StateGraph(State);
  .addNode("myNode", (state, config) => {
    console.log("In node: ", config?.configurable?.user_id);
    return { results: `Hello, ${state.input}!` };
  })
  addNode("otherNode", (state) => {
    return state;
  })
  ...

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

如果你在不指定名称的情况下将节点添加到图中，它将被赋予与函数名称等效的默认名称。

Python

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

JavaScript

builder.addNode(myNode);
// You can then create edges to/from this node by referencing it as `"myNode"`

`START` 节点¶

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

Python

from langgraph.graph import START

graph.add_edge(START, "node_a")

JavaScript

import { START } from "@langchain/langgraph";

graph.addEdge(START, "nodeA");

`END` 节点¶

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

Python

from langgraph.graph import END

graph.add_edge("node_a", END)

JavaScript

import { END } from "@langchain/langgraph";

graph.addEdge("nodeA", END);

节点缓存¶

Python

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'))  # (1)!
[{'expensive_node': {'result': 10}}]
print(graph.invoke({"x": 5}, stream_mode='updates'))  # (2)!
[{'expensive_node': {'result': 10}, '__metadata__': {'cached': True}}]

第一次运行需要两秒钟（由于模拟的昂贵计算）。
第二次运行利用缓存并快速返回。

JavaScript

LangGraph 支持基于节点输入的任务/节点缓存。要使用缓存：

在编译图（或指定入口点）时指定缓存
为节点指定缓存策略。每个缓存策略支持：
keyFunc，用于根据节点输入生成缓存键。
ttl，缓存的生存时间（以秒为单位）。如果未指定，缓存将永不过期。

import { StateGraph, MessagesZodState } from "@langchain/langgraph";
import { InMemoryCache } from "@langchain/langgraph-checkpoint";

const graph = new StateGraph(MessagesZodState)
  .addNode(
    "expensive_node",
    async () => {
      // Simulate an expensive operation
      await new Promise((resolve) => setTimeout(resolve, 3000));
      return { result: 10 };
    },
    { cachePolicy: { ttl: 3 } }
  )
  .addEdge(START, "expensive_node")
  .compile({ cache: new InMemoryCache() });

await graph.invoke({ x: 5 }, { streamMode: "updates" }); // (1)!
// [{"expensive_node": {"result": 10}}]
await graph.invoke({ x: 5 }, { streamMode: "updates" }); // (2)!
// [{"expensive_node": {"result": 10}, "__metadata__": {"cached": true}}]

边¶

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

普通边：直接从一个节点到下一个节点。
条件边：调用函数以确定下一个要去的节点。
入口点：当用户输入到达时首先调用哪个节点。
条件入口点：调用函数以确定当用户输入到达时首先调用哪个节点。

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

普通边¶

Python

如果你**总是**想从节点 A 到节点 B，你可以直接使用 @[add_edge][] 方法。

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

JavaScript

如果你**总是**想从节点 A 到节点 B，你可以直接使用 @[addEdge][add_edge] 方法。

graph.addEdge("nodeA", "nodeB");

条件边¶

Python

如果你想**可选地**路由到 1 条或多条边（或可选地终止），你可以使用 @[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"})

JavaScript

如果你想**可选地**路由到 1 条或多条边（或可选地终止），你可以使用 @[addConditionalEdges][add_conditional_edges] 方法。此方法接受节点的名称和在该节点执行后要调用的"路由函数"：

graph.addConditionalEdges("nodeA", routingFunction);

与节点类似，routingFunction 接受图的当前 state 并返回一个值。

默认情况下，routingFunction 的返回值用作下一个要发送状态的节点（或节点列表）的名称。所有这些节点将作为下一个超步的一部分并行运行。

你可以选择提供一个对象，将 routingFunction 的输出映射到下一个节点的名称。

graph.addConditionalEdges("nodeA", routingFunction, {
  true: "nodeB",
  false: "nodeC",
});

Tip

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

入口点¶

Python

入口点是图启动时运行的第一个节点。你可以使用 @[add_edge][] 方法从虚拟 @[START][] 节点到第一个要执行的节点来指定从哪里进入图。

from langgraph.graph import START

graph.add_edge(START, "node_a")

JavaScript

入口点是图启动时运行的第一个节点。你可以使用 @[addEdge][add_edge] 方法从虚拟 @[START][] 节点到第一个要执行的节点来指定从哪里进入图。

import { START } from "@langchain/langgraph";

graph.addEdge(START, "nodeA");

条件入口点¶

Python

条件入口点允许你根据自定义逻辑从不同的节点开始。你可以使用从虚拟 @[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"})

JavaScript

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

import { START } from "@langchain/langgraph";

graph.addConditionalEdges(START, routingFunction);

你可以选择提供一个对象，将 routingFunction 的输出映射到下一个节点的名称。

graph.addConditionalEdges(START, routingFunction, {
  true: "nodeB",
  false: "nodeC",
});

`Send`¶

Python

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

JavaScript

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

为了支持这种设计模式，LangGraph 支持从条件边返回 @[Send][] 对象。Send 接受两个参数：首先是节点的名称，其次是要传递给该节点的状态。

import { Send } from "@langchain/langgraph";

graph.addConditionalEdges("nodeA", (state) => {
  return state.subjects.map((subject) => new Send("generateJoke", { subject }));
});

`Command`¶

Python

将控制流（边）和状态更新（节点）结合起来可能很有用。例如，你可能希望在同一个节点中同时执行状态更新并决定下一个要去的节点。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")

JavaScript

将控制流（边）和状态更新（节点）结合起来可能很有用。例如，你可能希望在同一个节点中同时执行状态更新并决定下一个要去的节点。LangGraph 通过从节点函数返回 Command 对象来提供这样做的方法：

import { Command } from "@langchain/langgraph";

graph.addNode("myNode", (state) => {
  return new Command({
    update: { foo: "bar" },
    goto: "myOtherNode",
  });
});

使用 Command，你还可以实现动态控制流行为（与条件边相同）：

import { Command } from "@langchain/langgraph";

graph.addNode("myNode", (state) => {
  if (state.foo === "bar") {
    return new Command({
      update: { foo: "baz" },
      goto: "myOtherNode",
    });
  }
});

在节点函数中使用 Command 时，你必须在添加节点时添加 ends 参数来指定它可以路由到哪些节点：

builder.addNode("myNode", myNode, {
  ends: ["myOtherNode", END],
});

Important

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

查看这个操作指南以获取如何使用 Command 的端到端示例。

什么时候应该使用 Command 而不是条件边？¶

当你需要**同时**更新图状态**和**路由到不同节点时，使用 Command。例如，在实现多智能代理切换时，路由到不同的智能代理并将一些信息传递给该智能代理很重要。
使用条件边在节点之间有条件地路由而不更新状态。

导航到父图中的节点¶

Python

如果你使用子图，你可能希望从子图内的节点导航到不同的子图（即父图中的不同节点）。为此，你可以在 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
    )

Note

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

使用 Command.PARENT 的状态更新

当你从子图节点向父图节点发送共享父图和子图状态模式的键的更新时，你**必须**为要更新的键在父图状态中定义一个归约器。参见此示例。

JavaScript

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

import { Command } from "@langchain/langgraph";

graph.addNode("myNode", (state) => {
  return new Command({
    update: { foo: "bar" },
    goto: "otherSubgraph", // where `otherSubgraph` is a node in the parent graph
    graph: Command.PARENT,
  });
});

Note

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

使用 Command.PARENT 的状态更新

当你从子图节点向父图节点发送共享父图和子图状态模式的键的更新时，你**必须**为要更新的键在父图状态中定义一个归约器。

JavaScript

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

import { Command } from "@langchain/langgraph";

graph.addNode("myNode", (state) => {
  return new Command({
    update: { foo: "bar" },
    goto: "otherSubgraph", // where `otherSubgraph` is a node in the parent graph
    graph: Command.PARENT,
  });
});

Note

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

使用 Command.PARENT 的状态更新

当你从子图节点向父图节点发送共享父图和子图状态模式的键的更新时，你**必须**为要更新的键在父图状态中定义一个归约器。

这在实现多智能代理切换时特别有用。

查看此指南了解详情。

在工具内部使用¶

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

有关详细信息，请参阅此指南。

人机协同¶

Python

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

JavaScript

Command 是人机协同工作流的重要组成部分：当使用 interrupt() 收集用户输入时，然后使用 Command 通过 new Command({ resume: "User input" }) 提供输入并恢复执行。查看人机协同概念指南了解更多信息。

图迁移¶

LangGraph 可以轻松处理图定义（节点、边和状态）的迁移，即使使用检查点器来跟踪状态。

对于图末尾的线程（即未中断），你可以更改图的整个拓扑结构（即所有节点和边，删除、添加、重命名等）
对于当前中断的线程，我们支持除重命名/删除节点之外的所有拓扑更改（因为该线程现在可能即将进入一个不再存在的节点）——如果这是一个阻碍，请联系我们，我们可以优先考虑解决方案。
对于修改状态，我们对添加和删除键具有完全的向后和向前兼容性
重命名的状态键在现有线程中会丢失其保存的状态
以不兼容方式更改类型的状态键可能会在更改之前具有状态的线程中导致问题——如果这是一个阻碍，请联系我们，我们可以优先考虑解决方案。

Python

运行时上下文¶

创建图时，你可以为传递给节点的运行时上下文指定 context_schema。这对于传递不属于图状态的信息给节点很有用。例如，你可能希望传递依赖项，如模型名称或数据库连接。

@dataclass
class ContextSchema:
    llm_provider: str = "openai"

graph = StateGraph(State, context_schema=ContextSchema)

JavaScript

创建图时，你还可以标记图的某些部分是可配置的。这通常是为了能够轻松切换模型或系统提示。这允许你创建单个"认知架构"（图），但拥有它的多个不同实例。

创建图时，你可以选择指定配置模式。

import { z } from "zod";

const ConfigSchema = z.object({
  llm: z.string(),
});

const graph = new StateGraph(State, ConfigSchema);

Python

然后，你可以使用 invoke 方法的 context 参数将此上下文传递到图中。

graph.invoke(inputs, context={"llm_provider": "anthropic"})

JavaScript

然后，你可以使用 configurable 配置字段将此配置传递到图中。

const config = { configurable: { llm: "anthropic" } };

await graph.invoke(inputs, config);

然后，你可以在节点或条件边内访问和使用此上下文：

from langgraph.runtime import Runtime

def node_a(state: State, runtime: Runtime[ContextSchema]):
    llm = get_llm(runtime.context.llm_provider)
    ...

有关配置的完整详细信息，请参阅此指南。

JavaScript

graph.addNode("myNode", (state, config) => {
  const llmType = config?.configurable?.llm || "openai";
  const llm = getLlm(llmType);
  return { results: `Hello, ${state.input}!` };
});

递归限制¶

Python

递归限制设置图在单次执行期间可以执行的最大超步数。一旦达到限制，LangGraph 将引发 GraphRecursionError。默认情况下，此值设置为 25 步。可以在运行时为任何图设置递归限制，并通过配置字典传递给 .invoke/.stream。重要的是，recursion_limit 是一个独立的 config 键，不应该像所有其他用户定义的配置一样传递到 configurable 键内。参见下面的示例：

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

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

JavaScript

递归限制设置图在单次执行期间可以执行的最大超步数。一旦达到限制，LangGraph 将引发 GraphRecursionError。默认情况下，此值设置为 25 步。可以在运行时为任何图设置递归限制，并通过配置对象传递给 .invoke/.stream。重要的是，recursionLimit 是一个独立的 config 键，不应该像所有其他用户定义的配置一样传递到 configurable 键内。参见下面的示例：

await graph.invoke(inputs, {
  recursionLimit: 5,
  configurable: { llm: "anthropic" },
});

可视化¶

能够可视化图通常很好，特别是当它们变得更复杂时。LangGraph 带有几种内置的可视化图的方法。有关更多信息，请参阅此操作指南。

图 API 概念¶

图¶

StateGraph¶

编译你的图¶

状态¶

模式¶

多个模式¶

归约器¶

默认归约器¶

在图状态中使用消息¶

为什么使用消息？¶

在图中使用消息¶

序列化¶

MessagesState¶

节点¶

START 节点¶

END 节点¶

节点缓存¶

边¶

普通边¶

条件边¶

入口点¶

条件入口点¶

Send¶

Command¶

什么时候应该使用 Command 而不是条件边？¶

导航到父图中的节点¶

在工具内部使用¶

人机协同¶

图迁移¶

运行时上下文¶

递归限制¶

可视化¶

`START` 节点¶

`END` 节点¶

`Send`¶

`Command`¶