图 API 总览 (Graph API overview)

状态图核心模型 (Graphs)

从底层架构来看，LangGraph 将智能体工作流抽象为有向图。你通过三个核心原语来定义智能体的全部行为：

全局状态 (State)：表示应用程序当前数据快照的共享数据结构。它可以是任意数据结构，通常使用共享的 StateSchema 进行强类型规范。
计算节点 (Nodes)：封装智能体具体业务逻辑的函数。它们接收当前状态作为入参，执行业务运算或产生副作用（如大模型推理或工具调用），并返回状态的增量更新。
流转边 (Edges)：决定当前节点执行完毕后下一步流转至哪个节点的决策函数。可以是确定性的固定流转，也可以是根据状态进行分支选择的条件边。

通过组合节点与边，你可以轻松构建具有循环重试、自愈修正能力的复杂有状态工作流。

一言以蔽之：节点负责具体做事，边决定下一步去哪。

LangGraph 的底层图调度算法采用消息传递（Message passing）机制，灵感源自 Google 著名的大规模图处理系统 Pregel。工作流以离散的超步（Super-steps）为周期推进：并行运行的节点属于同一个超步，串行节点则分属先后相继的超步。当所有节点均处于空闲状态且通道中没有未消费的消息时，图执行平稳终结。

A super-step can be considered a single iteration over the graph nodes. Nodes that run in parallel are part of the same super-step, while nodes that run sequentially belong to separate super-steps. At the start of graph execution, all nodes begin in an inactive state. A node becomes active when it receives a new message (state) on any of its incoming edges (or "channels"). The active node then runs its function and responds with updates. At the end of each super-step, nodes with no incoming messages vote to halt by marking themselves as inactive. The graph execution terminates when all nodes are inactive and no messages are in transit.

全局状态 (State)Graph 状态图类

StateGraph 是构建图工作流的核心类，由用户自定义的 State 状态对象进行泛型约束。

编译工作流图 (Compiling your graph)

构建图的标准流程是：定义状态 -> 添加节点与边 -> 执行编译。编译（Compile）是对图拓扑结构的一致性校验（检查是否存在孤立节点、死循环悬空引用等），同时也是挂载持久化检查点（Checkpointer）与人工审批断点的必经步骤：

typescript

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

必须编译

在调用 invoke 或 stream 执行图之前，必须显式调用 .compile()。编译后生成的实例才是可直接运行的 Runnable 实体。

State

The first thing you do when you define a graph is define the State of the graph. The State consists of the schema of the graph as well as reducer functions which specify how to apply updates to the state. The schema of the State will be the input schema to all Nodes and Edges in the graph. You define state using the StateSchema class, which accepts any standard schemas (like Zod) for individual fields along with special value types like ReducedValue and MessagesValue. All Nodes will emit updates to the State which are then applied using the specified reducer function.

Schema

The main way to specify the schema of a graph is by using the StateSchema class. Each field in the schema can be:

A Standard schema for simple fields (becomes a "last value" channel that overwrites on update)
A ReducedValue for fields that need a custom reducer function (when nodes are run in parallel)
A MessagesValue for chat message lists (prebuilt with message-aware reducer)
An UntrackedValue for transient state that should not be checkpointed

typescript

import {
  StateSchema,
  ReducedValue,
  MessagesValue,
  UntrackedValue
} from "@langchain/langgraph";
import { z } from "zod/v4";

const AgentState = new StateSchema({
  // Prebuilt messages value with built-in reducer
  messages: MessagesValue,

  // Simple fields use Zod schemas directly
  currentStep: z.string(),

  // Fields with defaults
  retryCount: z.number().default(0),

  // Custom reducer for accumulating values
  allSteps: new ReducedValue(
    z.array(z.string()).default(() => []),
    {
      inputSchema: z.string(),
      reducer: (current, newStep) => [...current, newStep],
    }
  ),

  // Transient state not saved to checkpoints
  tempCache: new UntrackedValue(z.record(z.string(), z.unknown())),
});

// Type extraction
type State = typeof AgentState.State;   // Full state type
type Update = typeof AgentState.Update; // Partial update type

// Use in graph
const graph = new StateGraph(AgentState)
  .addNode("myNode", ...)
  .compile();

By default, the graph will have the same input and output schemas. If you want to change this, you can also specify explicit input and output schemas directly. This is useful when you have a lot of keys, and some are explicitly for input and others for output.

Multiple schemas

Typically, all graph nodes communicate with a single schema. This means that they will read and write to the same state channels. But, there are cases where we want more control over this:

Internal nodes can pass information that is not required in the graph's input / output.
We may also want to use different input / output schemas for the graph. The output might, for example, only contain a single relevant output key.

It is possible to have nodes write to private state channels inside the graph for internal node communication. We can simply define a private schema, PrivateState.

It is also possible to define explicit input and output schemas for a graph. In these cases, we define an "internal" schema that contains all keys relevant to graph operations. But, we also define input and output schemas that are sub-sets of the "internal" schema to constrain the input and output of the graph. See Define input and output schemas for more detail.

Let's look at an example:

import { END, START, StateGraph, StateSchema } from "@langchain/langgraph";
import * as z from "zod";

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

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

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

const PrivateState = new StateSchema({
  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 the private bar channel
    return { bar: state.foo + " is" };
  })
  .addNode(
    "node3",
    (state) => {
      // Read from the private bar channel, write to OutputState
      return { graphOutput: state.bar + " Lance" };
    },
    // This input schema declares what node3 reads and registers bar as a graph channel.
    { 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' }

There are two subtle and important points to note here:

We pass state as the input schema to node1. But, we write out to foo, a channel in OverallState. How can we write out to a state channel that is not included in the input schema? This is because a node can write to any state channel in the graph state. The graph state is the union of the state channels defined at initialization, which includes OverallState and the filters InputState and OutputState.
We initialize the graph with StateGraph({ state: OverallState, input: InputState, output: OutputState }). How can we write to PrivateState in node2? How does the graph gain access to this schema if it was not passed in the StateGraph initialization? We can do this because nodes can also declare additional state channels as long as the state schema definition exists. In this case, the PrivateState schema is defined, so we can add bar as a new state channel in the graph and write to it.

WARNING

Private channels are not redacted when streaming.

Input, output, and private schemas constrain what each node reads (its input schema) and what invoke returns (the output schema). They do not hide channels from stream.

When you stream with streamMode: "values", the graph emits all of its state channels by default — including private ones — because values streaming defaults to the full set of state channels rather than the output schema. This is why a private channel like bar is hidden by invoke but visible while streaming:

import { END, START, StateGraph, StateSchema } from "@langchain/langgraph";
import * as z from "zod";

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

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

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

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

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

const stream = await graph.streamEvents({ userInput: "My" }, { version: "v3" });
for await (const snapshot of stream.values) {
  console.log(snapshot);
}
// { userInput: 'My' }
// { foo: 'My name', userInput: 'My' }
// { foo: 'My name', userInput: 'My', bar: 'My name is' }            // <-- private channel
// { foo: 'My name', userInput: 'My', graphOutput: 'My name is Lance', bar: 'My name is' }

To restrict the streamed values to a specific set of channels (e.g. only the output schema), pass outputKeys:

typescript

const stream = await graph.streamEvents(
  { userInput: "My" },
  { version: "v3", outputKeys: ["graphOutput"] }  
);
for await (const snapshot of stream.values) {
  console.log(snapshot);
}
// { graphOutput: 'My name is Lance' }

If you only need the channels a node actually produced each step (rather than the full accumulated state), use streamMode: "updates" instead.

Reducers

Reducers are key to understanding how updates from nodes are applied to the State. Each key in the State has its own independent reducer function. If no reducer function is explicitly specified then it is assumed that all updates to that key should override it. There are a few different types of reducers, starting with the default type of reducer:

Reducer arguments

Every reducer is a binary function with two positional arguments:

Left argument: The current value already stored in state for that key.
Right argument: The update for that key returned by a node.

When a node returns a partial update, LangGraph calls the reducer for each updated key and saves the return value as the new state value:

typescript

const newValue = reducer(currentState[key], nodeUpdate[key]); // left, right

The left argument always comes from accumulated state. The right argument always comes from the latest node update. The following example names both arguments explicitly:

import { ReducedValue, StateSchema } from "@langchain/langgraph";
import * as z from "zod";

const State = new StateSchema({
  tags: new ReducedValue(
    z.array(z.string()).default(() => []),
    {
      reducer: (left: string[], right: string[]) => {
        // left: existing state; right: update from a node
        return left.concat(right);
      },
    }
  ),
});

Suppose the state is { tags: ["draft"] } and a node returns { tags: ["review"] }. LangGraph calls:

const reducer = (left: string[], right: string[]) => left.concat(right);

reducer(["draft"], ["review"]); // left, right → ["draft", "review"]

The new state value for tags is ["draft", "review"].

Custom reducers combine the left and right arguments. The default reducer discards the left argument and keeps only the right.

Default reducer

The default reducer ignores the left argument and replaces the state value with the right argument. This example shows how to use the default reducer:

import { StateSchema } from "@langchain/langgraph";
import * as z from "zod";

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

In this example, no reducer functions are specified for any key. Let's assume the input to the graph is:

{ foo: 1, bar: ["hi"] }. Let's then assume the first Node returns { foo: 2 }. This is treated as an update to the state. Notice that the Node does not need to return the whole State schema - just an update. After applying this update, the State would then be { foo: 2, bar: ["hi"] }. If the second node returns { bar: ["bye"] } then the State would then be { foo: 2, bar: ["bye"] }

Custom reducers

A custom reducer combines the left and right arguments instead of replacing the state value, which is useful for accumulating values, such as appending updates to a list. This example shows how to specify a custom reducer:

import { ReducedValue, StateSchema } from "@langchain/langgraph";
import { z } from "zod/v4";

const State = new StateSchema({
  foo: z.number(),
  bar: new ReducedValue(
    z.array(z.string()).default(() => []),
    { reducer: (x, y) => x.concat(y) }
  ),
});

In this example, we've used ReducedValue to specify a reducer function for the second key (bar). Note that the first key remains unchanged. Let's assume the input to the graph is { foo: 1, bar: ["hi"] }. Let's then assume the first Node returns { foo: 2 }. This is treated as an update to the state. Notice that the Node does not need to return the whole State schema - just an update. After applying this update, the State would then be { foo: 2, bar: ["hi"] }. If the second node returns { bar: ["bye"] } then the State would then be { foo: 2, bar: ["hi", "bye"] }. Notice here that the bar key is updated by concatenating the two arrays together.

Resetting a reducer field

A common source of confusion with reducers: with a merging reducer, returning an empty value does not clear the field. Because the reducer merges the right argument into the left one, an empty update is merged in and previously accumulated values are kept.

This pattern matters for error buffers or retry counters that must be cleared between retry attempts:

import { ReducedValue, StateSchema } from "@langchain/langgraph";
import { z } from "zod/v4";

const State = new StateSchema({
  errors: new ReducedValue(
    z.array(z.string()).default(() => []),
    { reducer: (state: string[], update: string[]) => state.concat(update) },
  ),
});

// node A returns { errors: ["bad sql"] }
// node B returns { errors: [] }
// state.errors is still ["bad sql"]; the empty array is merged in, not cleared

To let a node reset (clear) the field, define a custom reducer that replaces the accumulated value instead of merging it:

import { ReducedValue, StateSchema } from "@langchain/langgraph";
import { z } from "zod/v4";

const State = new StateSchema({
  errors: new ReducedValue(
    z.array(z.string()).default(() => []),
    { reducer: (_state: string[], update: string[]) => update }
  ),
});

// node can now clear the field with { errors: [] }

Alternatively, wrap the update with Overwrite to bypass the reducer for a single update, while keeping the field's normal merging behavior for every other update:

import { Overwrite, ReducedValue, StateSchema } from "@langchain/langgraph";
import { z } from "zod/v4";

const State = new StateSchema({
  errors: new ReducedValue(
    z.array(z.string()).default(() => []),
    { reducer: (state: string[], update: string[]) => state.concat(update) },
  ),
});

const clearErrors = (_state: typeof State.State) => {
  return { errors: new Overwrite([]) };
};

// node can clear the field with { errors: new Overwrite([]) }
// while a normal node can still append with { errors: ["new error"] }

For more information, see Bypass reducers with Overwrite.

Untracked values

UntrackedValue is used for state fields that should exist during graph execution but should never be checkpointed. When a graph resumes from a checkpoint, untracked values will be reset to their initial state (or be unavailable).

This is useful for:

Database connections that can't be serialized
Temporary caches that should be rebuilt on resume
Large objects you don't want to persist
Runtime-only configuration that should be passed fresh each time

typescript

import { StateSchema, UntrackedValue, MessagesValue } from "@langchain/langgraph";
import { z } from "zod/v4";

const State = new StateSchema({
  messages: MessagesValue,

  // Untracked: throws if multiple nodes write in same step (guard: true is default)
  dbConnection: new UntrackedValue<DatabaseConnection>(),

  // Untracked with guard: false allows multiple writes, keeps last value
  tempCache: new UntrackedValue(
    z.record(z.string(), z.unknown()),
    { guard: false }
  ),

  // Untracked without a schema (for maximum flexibility)
  runtimeConfig: new UntrackedValue(),
});

Behavior:

During execution: Values are stored and accessible like normal state
On checkpoint: Untracked values are excluded from the checkpoint data
On resume: Untracked values start fresh (empty or with their default value)
With guard: true (default): Throws error if multiple nodes write in the same step
With guard: false: Multiple writes allowed, last value wins

WARNING

Don't use UntrackedValue for data you need to persist across interrupts or time travel. Use regular state fields or ReducedValue for persistent data.

Type utilities

LangGraph provides several type utilities for better TypeScript type safety when defining nodes and conditional edges.

`GraphNode`

Use GraphNode to type node functions defined outside the graph builder:

typescript

import { GraphNode, StateSchema, Command } from "@langchain/langgraph";
import { z } from "zod/v4";

const State = new StateSchema({
  count: z.number().default(0),
  result: z.string(),
});

// Basic node - receives state, returns partial update
const incrementNode: GraphNode<typeof State> = (state) => {
  return { count: state.count + 1 };
};

// Async node
const fetchNode: GraphNode<typeof State> = async (state, config) => {
  const response = await fetch(`/api/data/${state.count}`);
  return { result: await response.text() };
};

// Node with Command routing - specify valid destinations
const routerNode: GraphNode<{ InputSchema: typeof State; Nodes: "process" | "done" }> = (state) => {
  if (state.count >= 10) {
    return new Command({ goto: "done" });
  }
  return new Command({
    update: { count: state.count + 1 },
    goto: "process"
  });
};

`State.Node` shorthand

Each StateSchema instance has a Node property that provides a shorthand for typing nodes:

typescript

const State = new StateSchema({
  messages: MessagesValue,
  step: z.string(),
});

// These are equivalent:
const myNode1: GraphNode<typeof State> = (state) => ({ step: "done" });
const myNode2: typeof State.Node = (state) => ({ step: "done" });

`ConditionalEdgeRouter`

Use ConditionalEdgeRouter for routing functions in conditional edges (no state updates, just routing):

typescript

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

const State = new StateSchema({
  shouldContinue: z.boolean(),
  step: z.string(),
});

// Router returns node name(s) or END
const router: ConditionalEdgeRouter<{ InputSchema: typeof State; Nodes: "process" | "summarize" }> = (state) => {
  if (!state.shouldContinue) {
    return END;
  }
  return state.step === "initial" ? "process" : "summarize";
};

// Use in graph
graph.addConditionalEdges("check", router);

`StateSchema.State` and `StateSchema.Update`

Extract the state and update types from a schema for use in custom type definitions:

typescript

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

const MyStateSchema = new StateSchema({
  messages: MessagesValue,
  count: z.number().default(0),
});

// Extract the full state type
type MyState = typeof MyStateSchema.State;
// { messages: BaseMessage[], count: number }

// Extract the update type (partial, with reducer input types)
type MyUpdate = typeof MyStateSchema.Update;
// { messages?: Messages, count?: number }

:::

Working with messages in graph state

Why use messages?

Most modern LLM providers have a chat model interface that accepts a list of messages as input. LangChain's chat model interface in particular accepts a list of message objects as inputs. These messages come in a variety of forms such as HumanMessage (user input) or AIMessage (LLM response).

To read more about what message objects are, please refer to the Messages conceptual guide.

Using messages in your graph

In many cases, it is helpful to store prior conversation history as a list of messages in your graph state. To do so, you can use the prebuilt MessagesValue which provides a message-aware reducer that handles message IDs, updates, and deletions automatically.

The MessagesValue reducer is vital to telling the graph how to update the list of Message objects in the state with each state update. If you don't specify a reducer, every state update will overwrite the list of messages with the most recently provided value. MessagesValue handles this correctly: for brand new messages, it appends to the existing list, and for existing messages (matched by ID), it updates them in place.

TIP

MessagesValue is actually a special case of ReducedValue, preconfigured with an internal messagesStateReducer that handles message lists and updates. This provides convenient, message-aware state management for chat message history in LangGraph graphs.

Serialization

In addition to keeping track of message IDs, MessagesValue will also try to deserialize messages into LangChain Message objects whenever a state update is received on the messages channel. This allows sending graph inputs / state updates in the following format:

typescript

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

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

Since the state updates are always deserialized into LangChain Messages when using MessagesValue, you should use dot notation to access message attributes, like state.messages.at(-1).content. Below is an example of a graph that uses MessagesValue:

typescript

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

const State = new StateSchema({
  messages: MessagesValue,
});

const graph = new StateGraph(State)
  ...

The messages field is defined as a MessagesValue which is a list of BaseMessage objects with a built-in reducer. Typically, there is more state to track than just messages, so we see people extend this state and add more fields, like:

typescript

import { StateSchema, MessagesValue } from "@langchain/langgraph";
import * as z from "zod";

const State = new StateSchema({
  messages: MessagesValue,
  documents: z.array(z.string()),
});

计算节点 (Nodes)

In LangGraph, nodes are typically functions (sync or async) that accept the following arguments:

state—The state of the graph
config—A RunnableConfig object that contains configuration information like thread_id and tracing information like tags

You can add nodes to a graph using the addNode method. For better type safety, use the GraphNode type utility or State.Node to type your node functions:

typescript

import { StateGraph, StateSchema, GraphNode } from "@langchain/langgraph";
import * as z from "zod";

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

// Option 1: Use GraphNode type utility
const myNode: GraphNode<typeof State> = (state, config) => {
  console.log("In node: ", config?.configurable?.user_id);
  return { results: `Hello, ${state.input}!` };
};

// Option 2: Use State.Node shorthand
const otherNode: typeof State.Node = (state) => {
  return state;
};

const builder = new StateGraph(State)
  .addNode("myNode", myNode)
  .addNode("otherNode", otherNode)
  ...

Behind the scenes, functions are converted to RunnableLambda, which add batch and async support to your function, along with native tracing and debugging.

If you add a node to a graph without specifying a name, it will be given a default name equivalent to the function name.

typescript

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

Re-execution and idempotency

When you compile with a checkpointer, LangGraph saves checkpoints at super-step boundaries, not mid-function inside a node. If execution stops and later resumes (for example after an interrupt or a retry), the affected node runs again from the start of its function. Code and side effects before the pause run again.

Idempotency. Design node logic so re-execution does not corrupt state. If a node inserts a database row, running it twice should not create duplicate rows unless that is intentional. Use idempotency keys, upserts, or read-before-write checks. For effects around interrupt(), see Side effects called before interrupt must be idempotent.

Graph changes. Determinism rules about code changes do not apply to graph structure. You can add or remove nodes and edges without breaking resume for existing threads. Resumed runs use saved state and execute whatever graph you compile now.

Tasks and interrupts inside a node. If a node calls tasks or interrupt, stricter determinism rules apply on resume. LangGraph restores completed task results from the checkpointer, but changing task or interrupt order in code before the resume point can mismatch cached values. A Functional API entrypoint compiles to a single node that runs the whole entrypoint method this way. See Determinism, Idempotency, and Using tasks in nodes.

Using tasks in nodes

If a node contains multiple operations, you may find it easier to implement each operation as a task instead of splitting the logic across multiple nodes. Task results are checkpointed when the graph uses a checkpointer, so resuming a thread can skip completed task work inside the node.

Original

import * as z from "zod";
import { uuid7 } from "langsmith";

import {
  END,
  MemorySaver,
  START,
  StateGraph,
  StateSchema,
} from "@langchain/langgraph";
import type { GraphNode } from "@langchain/langgraph";

const State = new StateSchema({
  url: z.string(),
  result: z.string().optional(),
});

const callApi: GraphNode<typeof State> = async (state) => {
  const response = await fetch(state.url); 
  const text = await response.text();
  const result = text.slice(0, 100);
  return { result };
};

const builder = new StateGraph(State)
  .addNode("callApi", callApi)
  .addEdge(START, "callApi")
  .addEdge("callApi", END);

const checkpointer = new MemorySaver();
const graph = builder.compile({ checkpointer });

const threadId = uuid7();
const config = { configurable: { thread_id: threadId } };

await graph.invoke({ url: "https://www.example.com" }, config);

With task

import * as z from "zod";
import { uuid7 } from "langsmith";

import {
  END,
  MemorySaver,
  START,
  StateGraph,
  StateSchema,
  task,
} from "@langchain/langgraph";
import type { GraphNode } from "@langchain/langgraph";

const State = new StateSchema({
  urls: z.array(z.string()),
  results: z.array(z.string()).optional(),
});

const makeRequest = task("makeRequest", async (url: string) => {
  const response = await fetch(url); 
  const text = await response.text();
  return text.slice(0, 100);
});

const callApi: GraphNode<typeof State> = async (state) => {
  const pending = state.urls.map((url) => makeRequest(url)); 
  const results = await Promise.all(pending);
  return { results };
};

const builder = new StateGraph(State)
  .addNode("callApi", callApi)
  .addEdge(START, "callApi")
  .addEdge("callApi", END);

const checkpointer = new MemorySaver();
const graph = builder.compile({ checkpointer });

const threadId = uuid7();
const config = { configurable: { thread_id: threadId } };

await graph.invoke({ urls: ["https://www.example.com"] }, config);

`START` node

The START Node is a special node that represents the node that sends user input to the graph. The main purpose for referencing this node is to determine which nodes should be called first.

typescript

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

graph.addEdge(START, "nodeA");

`END` node

The END Node is a special node that represents a terminal node. This node is referenced when you want to denote which edges have no actions after they are done.

typescript

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

graph.addEdge("nodeA", END);

Node caching

LangGraph supports caching of tasks/nodes based on the input to the node. To use caching:

Specify a cache when compiling a graph (or specifying an entrypoint)
Specify a cache policy for nodes. Each cache policy supports:
- keyFunc, which is used to generate a cache key based on the input to a node.
- ttl, the time to live for the cache in seconds. If not specified, the cache will never expire.

typescript

import { StateGraph, StateSchema, GraphNode, START } from "@langchain/langgraph";
import { InMemoryCache } from "@langchain/langgraph-checkpoint";
import { z } from "zod/v4";

const State = new StateSchema({
  x: z.number(),
  result: z.number(),
});

const expensiveNode: GraphNode<typeof State> = async (state) => {
  // Simulate an expensive operation
  await new Promise((resolve) => setTimeout(resolve, 3000));
  return { result: state.x * 2 };
};

const graph = new StateGraph(State)
  .addNode("expensive_node", expensiveNode, { cachePolicy: { ttl: 3 } })
  .addEdge(START, "expensive_node")
  .compile({ cache: new InMemoryCache() });

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

流转边 (Edges)

Edges define how the logic is routed and how the graph decides to stop. This is a big part of how your agents work and how different nodes communicate with each other. There are a few key types of edges:

Normal Edges: Go directly from one node to the next.
Conditional Edges: Call a function to determine which node(s) to go to next.
Entry Point: Which node to call first when user input arrives.
Conditional Entry Point: Call a function to determine which node(s) to call first when user input arrives.

A node can have multiple outgoing edges. If a node has multiple outgoing edges, all of those destination nodes will be executed in parallel as a part of the next superstep.

WARNING

For each node, choose one routing mechanism: use normal edges for static routing, or use conditional edges / Command for dynamic routing. Do not mix normal edges and dynamic routing from the same node, because both paths can execute and make graph behavior harder to reason about.

Normal edges

If you always want to go from node A to node B, you can use the addEdge method directly.

typescript

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

Conditional edges

If you want to optionally route to one or more edges (or optionally terminate), you can use the addConditionalEdges method. This method accepts the name of a node and a "routing function" to call after that node is executed:

typescript

graph.addConditionalEdges("nodeA", routingFunction);

Similar to nodes, the routingFunction accepts the current state of the graph and returns a value.

By default, the return value routingFunction is used as the name of the node (or list of nodes) to send the state to next. All those nodes will be run in parallel as a part of the next superstep.

You can optionally provide an object that maps the routingFunction's output to the name of the next node.

typescript

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

TIP

Use Command instead of conditional edges if you want to combine state updates and routing in a single function.

Entry point

The entry point is the first node(s) that are run when the graph starts. You can use the addEdge method from the virtual START node to the first node to execute to specify where to enter the graph.

typescript

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

graph.addEdge(START, "nodeA");

Conditional entry point

A conditional entry point lets you start at different nodes depending on custom logic. You can use addConditionalEdges from the virtual START node to accomplish this.

typescript

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

graph.addConditionalEdges(START, routingFunction);

You can optionally provide an object that maps the routingFunction's output to the name of the next node.

typescript

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

动态并发扇出原语 (Send)

By default, Nodes and Edges are defined ahead of time and operate on the same shared state. However, there can be cases where the exact edges are not known ahead of time and/or you may want different versions of State to exist at the same time. A common example of this is with map-reduce design patterns. In this design pattern, a first node may generate a list of objects, and you may want to apply some other node to all those objects. The number of objects may be unknown ahead of time (meaning the number of edges may not be known) and the input State to the downstream Node should be different (one for each generated object).

To support this design pattern, LangGraph supports returning Send objects from conditional edges. Send takes two arguments: first is the name of the node, and second is the state to pass to that node.

typescript

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

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

动态控制流原语 (Command)

Command is a versatile primitive for controlling graph execution. It accepts four parameters:

update: Apply state updates (similar to returning updates from a node).
goto: Navigate to specific nodes (similar to conditional edges).
graph: Target a parent graph when navigating from subgraphs.
resume: Provide a value to resume execution after an interrupt.

Command is used in three contexts:

Return from nodes: Use update, goto, and graph to combine state updates with control flow.
Input to invoke or stream: Use resume to continue execution after an interrupt.
Return from tools: Similar to return from nodes, combine state updates and control flow from inside a tool.

Return from nodes

`update` and `goto`

Return Command from node functions to update state and route to the next node in a single step:

typescript

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

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

With Command you can also achieve dynamic control flow behavior (identical to conditional edges):

typescript

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

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

Use Command when you need to both update state and route to a different node. If you only need to route without updating state, use conditional edges instead.

When using Command in your node functions, you must add the ends parameter when adding the node to specify which nodes it can route to:

typescript

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

WARNING

Command only adds dynamic edges—static edges defined with add_edge / addEdge still execute. For example, if node_a returns Command(goto="my_other_node") and you also have graph.add_edge("node_a", "node_b"), both node_b and my_other_node will run. For each node, use either Command or static edges to route to the next nodes, not both.

Check out this how-to guide for an end-to-end example of how to use Command.

`graph`

If you are using subgraphs, you can navigate from a node within a subgraph to a different node in the parent graph by specifying graph: Command.PARENT in Command:

typescript

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

Setting graph to Command.PARENT will navigate to the closest parent graph.

When you send updates from a subgraph node to a parent graph node for a key that's shared by both parent and subgraph state schemas, you must define a reducer for the key you're updating in the parent graph state.

This is particularly useful when implementing multi-agent handoffs. Check out Navigate to a node in a parent graph for detail.

Input to `invoke` or `stream`

WARNING

new Command({ resume: ... }) is the only Command pattern intended as input to invoke()/stream() (optionally combined with update to also apply a state change while resuming). Do not use new Command({ update: ... }) alone as input to continue multi-turn conversations—because passing any Command as input resumes from the latest checkpoint (i.e. the last step that ran, not __start__), the graph will appear stuck if it already finished. To continue a conversation on an existing thread, pass a plain input object:

typescript

// WRONG - graph resumes from the latest checkpoint
// (last step that ran), appears stuck
await graph.invoke(new Command({ update: { messages: [{ role: "user", content: "follow up" }] } }), config);  

// CORRECT - plain object restarts from __start__
await graph.invoke({ messages: [{ role: "user", content: "follow up" }] }, config);

`resume`

Use new Command({ resume: ... }) to provide a value and resume graph execution after an interrupt. The value passed to resume becomes the return value of the interrupt() call inside the paused node:

typescript

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

const humanReview = async (state: typeof StateAnnotation.State) => {
  // Pauses the graph and waits for a value
  const answer = interrupt("Do you approve?");
  return { messages: [{ role: "user", content: answer }] };
};

// First invocation - hits the interrupt and pauses
const result = await graph.invoke({ messages: [...] }, config);

// Resume with a value - the interrupt() call returns "yes"
const resumed = await graph.invoke(new Command({ resume: "yes" }), config);

Check out the interrupts conceptual guide for full details on interrupt patterns, including multiple interrupts and validation loops.

Return from tools

You can return Command from tools to update graph state and control flow. Use update to modify state (e.g., saving customer information looked up during a conversation) and goto to route to a specific node after the tool completes.

WARNING

When used inside tools, goto adds a dynamic edge—any static edges already defined on the node that called the tool will still execute. For each node, use either tool-driven dynamic routing or static edges to route to the next nodes, not both.

Refer to Use inside tools for detail.

图状态迁移 (Graph migrations)

LangGraph can easily handle migrations of graph definitions (nodes, edges, and state) even when using a checkpointer to track state.

For threads at the end of the graph (i.e. not interrupted) you can change the entire topology of the graph (i.e. all nodes and edges, remove, add, rename, etc)
For threads currently interrupted, we support all topology changes other than renaming / removing nodes (as that thread could now be about to enter a node that no longer exists) -- if this is a blocker please reach out and we can prioritize a solution.
For modifying state, we have full backwards and forwards compatibility for adding and removing keys
State keys that are renamed lose their saved state in existing threads
State keys whose types change in incompatible ways could currently cause issues in threads with state from before the change -- if this is a blocker please reach out and we can prioritize a solution.

TIP

For changes that are technically compatible but alter business logic, such as rewriting the tool set or restructuring conversation flow, see Business compatibility. That page covers pinning a behavioral version in state so existing threads keep the old path while new threads pick up the latest version.

运行时上下文 (Runtime context)

When creating a graph, you can specify a contextSchema for runtime context passed to nodes. This is useful for passing information to nodes that is not part of the graph state. For example, you might want to pass dependencies such as model name or a database connection.

typescript

import { StateGraph, StateSchema } from "@langchain/langgraph";
import * as z from "zod";

const State = new StateSchema({
  input: z.string(),
  output: z.string(),
});

const ContextSchema = z.object({
  llm: z.union([z.literal("openai"), z.literal("anthropic")]),
});

const graph = new StateGraph(State, ContextSchema);

You can then pass this configuration into the graph using the context property.

typescript

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

await graph.invoke(inputs, config);

You can then access and use this context inside a node or conditional edge:

typescript

import { Runtime, GraphNode } from "@langchain/langgraph";
import * as z from "zod";

const nodeA: GraphNode<typeof State> = (state, config) => {
  const llm = getLLM(runtime.context?.llm);
  // ...
  return {};
};

See Add runtime configuration for a full breakdown on configuration.

typescript

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

Recursion limit

The recursion limit sets the maximum number of super-steps the graph can execute during a single execution. Once the limit is reached, LangGraph will raise GraphRecursionError. By default this value is set to 25 steps. The recursion limit can be set on any graph at runtime, and is passed to invoke/stream via the config object. Importantly, recursionLimit is a standalone config key and should not be passed inside the configurable key as all other user-defined configuration. See the example below:

typescript

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

Accessing and handling the recursion counter

The current step counter is accessible in config.metadata.langgraph_step within any node, allowing for proactive recursion handling before hitting the recursion limit. This enables you to implement graceful degradation strategies within your graph logic.

How it works

The step counter is stored in config.metadata.langgraph_step. LangGraph increments this counter as the graph executes and raises a GraphRecursionError once the configured recursionLimit is exceeded.

Accessing the current step counter

You can access the current step counter within any node to monitor execution progress.

typescript

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

const myNode: GraphNode<typeof State> = async (state, config) => {
  const currentStep = config.metadata?.langgraph_step;
  console.log(`Currently on step: ${currentStep}`);
  return state;
}

Design your graph with explicit termination conditions, and catch GraphRecursionError as a safety net:

typescript

import {
  StateGraph,
  StateSchema,
  ReducedValue,
  GraphNode,
  ConditionalEdgeRouter,
  END,
  GraphRecursionError
} from "@langchain/langgraph";
import { z } from "zod/v4";

const State = new StateSchema({
  messages: new ReducedValue(
    z.array(z.string()).default(() => []),
    { reducer: (x, y) => x.concat(y) }
  ),
});

// Build graph with explicit termination logic
const graph = new StateGraph(State)
  .addNode("reasoning", async (state) => {
    // Normal processing - design your graph with explicit termination conditions
    return {
      messages: ["thinking..."]
    };
  })
  .addConditionalEdges("reasoning", (state) => {
    // Add your termination condition here
    if (state.messages.length >= 5) {
      return END;
    }
    return "reasoning";
  });

const app = graph.compile();

// Catch GraphRecursionError as a safety net
try {
  const result = await app.invoke(
    { messages: [] },
    { recursionLimit: 10 }
  );
} catch (error) {
  if (error instanceof GraphRecursionError) {
    console.log("Recursion limit reached, handling gracefully");
    // Handle the error - return partial results, notify user, etc.
  }
}

Proactive vs reactive approaches

There are two main approaches to handling recursion limits: proactive (monitoring within the graph) and reactive (catching errors externally).

typescript

import {
  StateGraph,
  StateSchema,
  ReducedValue,
  GraphNode,
  ConditionalEdgeRouter,
  END,
  GraphRecursionError
} from "@langchain/langgraph";
import { z } from "zod/v4";

const State = new StateSchema({
  messages: new ReducedValue(
    z.array(z.string()).default(() => []),
    { reducer: (x, y) => x.concat(y) }
  ),
});


// Build graph with explicit termination logic
const builder = new StateGraph(State)
  .addNode("agent", async (state) => {
    return {
      messages: ["Processing..."]
    };
  })
  .addConditionalEdges("agent", (state) => {
    // Design termination conditions into your graph
    if (state.messages.length >= 5) {
      return END;
    }
    return "agent";
  });

const graph = builder.compile();

// Reactive Approach - catch GraphRecursionError as safety net
try {
  const result = await graph.invoke(
    { messages: [] },
    { recursionLimit: 10 }
  );
} catch (error) {
  if (error instanceof GraphRecursionError) {
    // Handle externally after graph execution fails
    console.log("Recursion limit exceeded, handling gracefully");
  }
}

The reactive approach catches GraphRecursionError after the limit is exceeded. Design your graph with explicit termination conditions to avoid hitting the limit in the first place.

Approach	Detection	Handling	Control Flow
Reactive (catching `GraphRecursionError`)	After limit exceeded	Outside graph in try/catch	Graph execution terminated

Reactive advantages:

Simple implementation
No need to modify graph logic
Centralized error handling

Other available metadata

Along with langgraph_step, the following metadata is also available in config.metadata:

typescript

const inspectMetadata: GraphNode<typeof State> = async (state, config) => {
  const metadata = config.metadata;

  console.log(`Step: ${metadata?.langgraph_step}`);
  console.log(`Node: ${metadata?.langgraph_node}`);
  console.log(`Triggers: ${metadata?.langgraph_triggers}`);
  console.log(`Path: ${metadata?.langgraph_path}`);
  console.log(`Checkpoint NS: ${metadata?.langgraph_checkpoint_ns}`);

  return state;
}

图可视化 (Visualization)

It's often nice to be able to visualize graphs, especially as they get more complex. LangGraph comes with several built-in ways to visualize graphs. See Visualize your graph for more info.

可观测性与追踪 (Observability and Tracing)

To trace, debug and evaluate your agents, use LangSmith.

图 API 总览 (Graph API overview) ​

状态图核心模型 (Graphs) ​

全局状态 (State)Graph 状态图类 ​

编译工作流图 (Compiling your graph) ​

State ​

Schema ​

Multiple schemas ​

Reducers ​

Reducer arguments ​

Default reducer ​

Custom reducers ​

Resetting a reducer field ​

Untracked values ​

Type utilities ​

GraphNode ​

State.Node shorthand ​

ConditionalEdgeRouter ​

StateSchema.State and StateSchema.Update ​

Working with messages in graph state ​

Why use messages? ​

Using messages in your graph ​

Serialization ​

计算节点 (Nodes) ​

Re-execution and idempotency ​

Using tasks in nodes ​

START node ​

END node ​

Node caching ​

流转边 (Edges) ​

Normal edges ​

Conditional edges ​

Entry point ​

Conditional entry point ​

动态并发扇出原语 (Send) ​

动态控制流原语 (Command) ​

Return from nodes ​

update and goto ​

graph ​

Input to invoke or stream ​

resume ​

Return from tools ​

图状态迁移 (Graph migrations) ​

运行时上下文 (Runtime context) ​

Recursion limit ​

Accessing and handling the recursion counter ​

How it works ​

Accessing the current step counter ​

Proactive vs reactive approaches ​

Other available metadata ​

图可视化 (Visualization) ​

可观测性与追踪 (Observability and Tracing) ​

延伸阅读 (Learn more) ​

图 API 总览 (Graph API overview)

状态图核心模型 (Graphs)

全局状态 (State)Graph 状态图类

编译工作流图 (Compiling your graph)

State

Schema

Multiple schemas

Reducers

Reducer arguments

Default reducer

Custom reducers

Resetting a reducer field

Untracked values

Type utilities

`GraphNode`

`State.Node` shorthand

`ConditionalEdgeRouter`

`StateSchema.State` and `StateSchema.Update`

Working with messages in graph state

Why use messages?

Using messages in your graph

Serialization

计算节点 (Nodes)

Re-execution and idempotency

Using tasks in nodes

`START` node

`END` node

Node caching

流转边 (Edges)

Normal edges

Conditional edges

Entry point

Conditional entry point

动态并发扇出原语 (Send)

动态控制流原语 (Command)

Return from nodes

`update` and `goto`

`graph`

Input to `invoke` or `stream`

`resume`

Return from tools

图状态迁移 (Graph migrations)

运行时上下文 (Runtime context)

Recursion limit

Accessing and handling the recursion counter

How it works

Accessing the current step counter

Proactive vs reactive approaches

Other available metadata

图可视化 (Visualization)

可观测性与追踪 (Observability and Tracing)

延伸阅读 (Learn more)