子图 - Docs by LangChain中文

本指南说明如何使用子图的机制。子图是一个作为另一个图中的节点使用的图。子图适用于：

构建多智能体系统
在多个图中复用一组节点
分布式开发：当你希望不同团队独立开发图的不同部分时，可以将每个部分定义为子图。只要遵守子图接口（输入和输出 schema），父图就可以在不了解子图任何实现细节的情况下构建完成

Setup

npm install @langchain/langgraph

为 LangGraph 开发设置 LangSmith 注册 LangSmith，以便快速发现问题并提升 LangGraph 项目的性能。LangSmith 让你可以使用跟踪数据来调试、测试和监控使用 LangGraph 构建的 LLM 应用——阅读更多关于如何开始使用 LangSmith的信息。

Define subgraph communication

添加子图时，你需要定义父图和子图如何通信：

Pattern	When to use	State schemas
在节点内部调用子图	父图和子图具有不同的状态 schema（没有共享键），或者你需要在它们之间转换状态	你需要编写一个包装函数，将父状态映射为子图输入，并将子图输出映射回父状态
将子图作为节点添加	父图和子图共享状态键——子图从与父图相同的通道读取并写入	你可以将已编译的子图直接传给 `add_node`——不需要包装函数

Call a subgraph inside a node

当父图和子图具有不同的状态 schema（没有共享键）时，可以在节点函数内部调用子图。当你希望在多智能体系统中为每个智能体保留私有消息历史时，这种方式很常见。节点函数会先将父状态转换为子图状态，再调用子图；然后在返回前，将结果转换回父状态。

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

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

// Subgraph
const subgraphBuilder = new StateGraph(SubgraphState)
  .addNode("subgraphNode1", (state) => {
    return { bar: "hi! " + state.bar };
  })
  .addEdge(START, "subgraphNode1");

const subgraph = subgraphBuilder.compile();

// Parent graph
const State = new StateSchema({
  foo: z.string(),
});

// Transform the state to the subgraph state and back
const builder = new StateGraph(State)
  .addNode("node1", async (state) => {
    const subgraphOutput = await subgraph.invoke({ bar: state.foo });
    return { foo: subgraphOutput.bar };
  })
  .addEdge(START, "node1");

const graph = builder.compile();

Full example: different state schemas

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

// Define subgraph
const SubgraphState = new StateSchema({
  // note that none of these keys are shared with the parent graph state
  bar: z.string(),
  baz: z.string(),
});

const subgraphBuilder = new StateGraph(SubgraphState)
  .addNode("subgraphNode1", (state) => {
    return { baz: "baz" };
  })
  .addNode("subgraphNode2", (state) => {
    return { bar: state.bar + state.baz };
  })
  .addEdge(START, "subgraphNode1")
  .addEdge("subgraphNode1", "subgraphNode2");

const subgraph = subgraphBuilder.compile();

// Define parent graph
const ParentState = new StateSchema({
  foo: z.string(),
});

const builder = new StateGraph(ParentState)
  .addNode("node1", (state) => {
    return { foo: "hi! " + state.foo };
  })
  .addNode("node2", async (state) => {
    const response = await subgraph.invoke({ bar: state.foo });
    return { foo: response.bar };
  })
  .addEdge(START, "node1")
  .addEdge("node1", "node2");

const graph = builder.compile();

for await (const chunk of await graph.stream(
  { foo: "foo" },
  { subgraphs: true }
)) {
  console.log(chunk);
}

将状态转换为子图状态
将响应转换回父状态

[[], { node1: { foo: 'hi! foo' } }]
[['node2:9c36dd0f-151a-cb42-cbad-fa2f851f9ab7'], { subgraphNode1: { baz: 'baz' } }]
[['node2:9c36dd0f-151a-cb42-cbad-fa2f851f9ab7'], { subgraphNode2: { bar: 'hi! foobaz' } }]
[[], { node2: { foo: 'hi! foobaz' } }]

Full example: different state schemas (two levels of subgraphs)

这是一个包含两层子图的示例：父级 -> 子级 -> 孙级。

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

// Grandchild graph
const GrandChildState = new StateSchema({
  myGrandchildKey: z.string(),
});

const grandchild = new StateGraph(GrandChildState)
  .addNode("grandchild1", (state) => {
    // NOTE: child or parent keys will not be accessible here
    return { myGrandchildKey: state.myGrandchildKey + ", how are you" };
  })
  .addEdge(START, "grandchild1")
  .addEdge("grandchild1", END);

const grandchildGraph = grandchild.compile();

// Child graph
const ChildState = new StateSchema({
  myChildKey: z.string(),
});

const child = new StateGraph(ChildState)
  .addNode("child1", async (state) => {
    // NOTE: parent or grandchild keys won't be accessible here
    const grandchildGraphInput = { myGrandchildKey: state.myChildKey };
    const grandchildGraphOutput = await grandchildGraph.invoke(grandchildGraphInput);
    return { myChildKey: grandchildGraphOutput.myGrandchildKey + " today?" };
  })   
  .addEdge(START, "child1")
  .addEdge("child1", END);

const childGraph = child.compile();

// Parent graph
const ParentState = new StateSchema({
  myKey: z.string(),
});

const parent = new StateGraph(ParentState)
  .addNode("parent1", (state) => {
    // NOTE: child or grandchild keys won't be accessible here
    return { myKey: "hi " + state.myKey };
  })
  .addNode("child", async (state) => {
    const childGraphInput = { myChildKey: state.myKey };
    const childGraphOutput = await childGraph.invoke(childGraphInput);
    return { myKey: childGraphOutput.myChildKey };
  })   
  .addNode("parent2", (state) => {
    return { myKey: state.myKey + " bye!" };
  })
  .addEdge(START, "parent1")
  .addEdge("parent1", "child")
  .addEdge("child", "parent2")
  .addEdge("parent2", END);

const parentGraph = parent.compile();

for await (const chunk of await parentGraph.stream(
  { myKey: "Bob" },
  { subgraphs: true }
)) {
  console.log(chunk);
}

我们将状态从子级状态通道（myChildKey）转换为孙级状态通道（myGrandchildKey）
我们将状态从孙级状态通道（myGrandchildKey）转换回子级状态通道（myChildKey）
这里传入的是一个函数，而不仅仅是已编译的图（grandchildGraph）
我们将状态从父级状态通道（myKey）转换为子级状态通道（myChildKey）
我们将状态从子级状态通道（myChildKey）转换回父级状态通道（myKey）
这里传入的是一个函数，而不仅仅是已编译的图（childGraph）

[[], { parent1: { myKey: 'hi Bob' } }]
[['child:2e26e9ce-602f-862c-aa66-1ea5a4655e3b', 'child1:781bb3b1-3971-84ce-810b-acf819a03f9c'], { grandchild1: { myGrandchildKey: 'hi Bob, how are you' } }]
[['child:2e26e9ce-602f-862c-aa66-1ea5a4655e3b'], { child1: { myChildKey: 'hi Bob, how are you today?' } }]
[[], { child: { myKey: 'hi Bob, how are you today?' } }]
[[], { parent2: { myKey: 'hi Bob, how are you today? bye!' } }]

Add a subgraph as a node

当父图和子图共享状态键时，你可以将已编译的子图直接传给 add_node。不需要包装函数——子图会自动从父图的状态通道中读取并写入。例如，在多智能体系统中，智能体通常会通过共享的 messages 键进行通信。 SQL agent graph

如果你的子图与父图共享状态键，可以按照以下步骤将其添加到图中：

定义子图工作流（下面示例中的 subgraphBuilder）并编译它
在定义父图工作流时，将已编译的子图传给 .addNode 方法

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

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

// Subgraph
const subgraphBuilder = new StateGraph(State)
  .addNode("subgraphNode1", (state) => {
    return { foo: "hi! " + state.foo };
  })
  .addEdge(START, "subgraphNode1");

const subgraph = subgraphBuilder.compile();

// Parent graph
const builder = new StateGraph(State)
  .addNode("node1", subgraph)
  .addEdge(START, "node1");

const graph = builder.compile();

Full example: shared state schemas

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

// Define subgraph
const SubgraphState = new StateSchema({
  foo: z.string(),
  bar: z.string(),
});

const subgraphBuilder = new StateGraph(SubgraphState)
  .addNode("subgraphNode1", (state) => {
    return { bar: "bar" };
  })
  .addNode("subgraphNode2", (state) => {
    // note that this node is using a state key ('bar') that is only available in the subgraph
    // and is sending update on the shared state key ('foo')
    return { foo: state.foo + state.bar };
  })
  .addEdge(START, "subgraphNode1")
  .addEdge("subgraphNode1", "subgraphNode2");

const subgraph = subgraphBuilder.compile();

// Define parent graph
const ParentState = new StateSchema({
  foo: z.string(),
});

const builder = new StateGraph(ParentState)
  .addNode("node1", (state) => {
    return { foo: "hi! " + state.foo };
  })
  .addNode("node2", subgraph)
  .addEdge(START, "node1")
  .addEdge("node1", "node2");

const graph = builder.compile();

for await (const chunk of await graph.stream({ foo: "foo" })) {
  console.log(chunk);
}

这个键与父图状态共享
这个键是 SubgraphState 私有的，对父图不可见

{ node1: { foo: 'hi! foo' } }
{ node2: { foo: 'hi! foobar' } }

Subgraph persistence

使用子图时，你需要决定在两次调用之间如何处理它的内部数据。想象一个客服机器人会委派任务给专业子智能体：这个“账单专家”子智能体应该记住客户之前的问题，还是每次被调用时都重新开始？ .compile() 上的 checkpointer 参数控制子图持久化：

Mode	`checkpointer=`	Behavior
按调用	`None`（默认）	每次调用都重新开始，并继承父图的 checkpointer，以支持单次调用内的中断和持久执行。
按线程	`True`	状态会在同一线程的多次调用之间累积。每次调用都会从上一次结束的位置继续。
无状态	`False`	完全不进行 checkpoint——像普通函数调用一样运行。不支持中断或持久执行。

对于大多数应用来说，按调用是正确选择，包括子智能体处理独立请求的多智能体系统。当子智能体需要多轮对话记忆时（例如，一个会在多次交流中逐步建立上下文的研究助手），请使用按线程模式。

父图必须使用 checkpointer 编译，子图持久化功能（中断、状态检查、按线程记忆）才能工作。参见持久化。

下面的示例使用 LangChain 的 create_agent，这是构建智能体的常见方式。create_agent 底层会生成一个 LangGraph 图，因此所有子图持久化概念都可以直接应用。如果你使用原始的 LangGraph StateGraph 构建，同样的模式和配置选项也适用——详情参见 Graph API。

Stateful

有状态子图会继承父图的 checkpointer，从而启用中断、持久化和状态检查。这两种有状态模式的区别在于状态保留多久。

Per-invocation (default)

这是大多数应用推荐使用的模式，包括子智能体作为工具被调用的多智能体系统。它支持中断、持久化和并行调用，同时保持每次调用相互隔离。

当每次对子图的调用都是独立的，并且子智能体不需要记住之前调用中的任何内容时，请使用按调用持久化。这是最常见的模式，尤其适用于多智能体系统：子智能体处理一次性请求，比如“查询这个客户的订单”或“总结这份文档”。省略 checkpointer 或将其设置为 None。每次调用都会重新开始，但在单次调用内部，子图会继承父图的 checkpointer，并且可以使用 interrupt() 暂停和恢复。下面的示例使用两个子智能体（水果专家、蔬菜专家），并将它们包装为外层智能体的工具：

import { createAgent, tool } from "langchain";
import { MemorySaver, Command, interrupt } from "@langchain/langgraph";
import * as z from "zod";

const fruitInfo = tool(
  (input) => `Info about ${input.fruitName}`,
  {
    name: "fruit_info",
    description: "Look up fruit info.",
    schema: z.object({ fruitName: z.string() }),
  }
);

const veggieInfo = tool(
  (input) => `Info about ${input.veggieName}`,
  {
    name: "veggie_info",
    description: "Look up veggie info.",
    schema: z.object({ veggieName: z.string() }),
  }
);

// Subagents - no checkpointer setting (inherits parent)
const fruitAgent = createAgent({
  model: "gpt-5.4-mini",
  tools: [fruitInfo],
  prompt: "You are a fruit expert. Use the fruit_info tool. Respond in one sentence.",
});

const veggieAgent = createAgent({
  model: "gpt-5.4-mini",
  tools: [veggieInfo],
  prompt: "You are a veggie expert. Use the veggie_info tool. Respond in one sentence.",
});

// Wrap subagents as tools for the outer agent
const askFruitExpert = tool(
  async (input) => {
    const response = await fruitAgent.invoke({
      messages: [{ role: "user", content: input.question }],
    });
    return response.messages[response.messages.length - 1].content;
  },
  {
    name: "ask_fruit_expert",
    description: "Ask the fruit expert. Use for ALL fruit questions.",
    schema: z.object({ question: z.string() }),
  }
);

const askVeggieExpert = tool(
  async (input) => {
    const response = await veggieAgent.invoke({
      messages: [{ role: "user", content: input.question }],
    });
    return response.messages[response.messages.length - 1].content;
  },
  {
    name: "ask_veggie_expert",
    description: "Ask the veggie expert. Use for ALL veggie questions.",
    schema: z.object({ question: z.string() }),
  }
);

// Outer agent with checkpointer
const agent = createAgent({
  model: "gpt-5.4-mini",
  tools: [askFruitExpert, askVeggieExpert],
  prompt:
    "You have two experts: ask_fruit_expert and ask_veggie_expert. " +
    "ALWAYS delegate questions to the appropriate expert.",
  checkpointer: new MemorySaver(),
});

Interrupts
Multi-turn
Multiple subgraph calls

每次调用都可以使用 interrupt() 暂停和恢复。在工具函数中添加 interrupt()，即可在继续执行前要求用户批准：

const fruitInfo = tool(
  (input) => {
    interrupt("continue?");
    return `Info about ${input.fruitName}`;
  },
  {
    name: "fruit_info",
    description: "Look up fruit info.",
    schema: z.object({ fruitName: z.string() }),
  }
);

const config = { configurable: { thread_id: "1" } };

// Invoke - the subagent's tool calls interrupt()
let response = await agent.invoke(
  { messages: [{ role: "user", content: "Tell me about apples" }] },
  config,
);
// response contains __interrupt__

// Resume - approve the interrupt
response = await agent.invoke(new Command({ resume: true }), config);
// Subagent message count: 4

每次调用都会从全新的子智能体状态开始。子智能体不会记住之前的调用：

const config = { configurable: { thread_id: "1" } };

// First call
let response = await agent.invoke(
  { messages: [{ role: "user", content: "Tell me about apples" }] },
  config,
);
// Subagent message count: 4

// Second call - subagent starts fresh, no memory of apples
response = await agent.invoke(
  { messages: [{ role: "user", content: "Now tell me about bananas" }] },
  config,
);
// Subagent message count: 4 (still fresh!)

对同一个子图的多次调用不会产生冲突，因为每次调用都会获得自己的 checkpoint 命名空间：

const config = { configurable: { thread_id: "1" } };

// LLM calls ask_fruit_expert for both apples and bananas
const response = await agent.invoke(
  { messages: [{ role: "user", content: "Tell me about apples and bananas" }] },
  config,
);
// Subagent message count: 4 (apples - fresh)
// Subagent message count: 4 (bananas - fresh)

Per-thread

当子智能体需要记住之前的交互时，请使用按线程持久化。例如，一个会在多次交流中逐步积累上下文的研究助手，或者一个会跟踪已经编辑过哪些文件的编码助手。子智能体的对话历史和数据会在同一线程的多次调用之间累积。每次调用都会从上一次结束的位置继续。使用 checkpointer=True 编译即可启用这种行为。

按线程子图不支持并行工具调用。当 LLM 可以将按线程子智能体作为工具使用时，它可能会尝试并行多次调用该工具（例如，同时询问水果专家关于苹果和香蕉的问题）。这会导致 checkpoint 冲突，因为两次调用都会写入同一个命名空间。下面的示例使用 LangChain 的 ToolCallLimitMiddleware 来避免这种情况。如果你使用纯 LangGraph StateGraph 构建，则需要自行阻止并行工具调用——例如，通过配置模型禁用并行工具调用，或者添加逻辑确保同一个子图不会被并行调用多次。

下面的示例使用一个通过 checkpointer=True 编译的水果专家子智能体：

import { createAgent, tool, toolCallLimitMiddleware } from "langchain";
import { MemorySaver, Command, interrupt } from "@langchain/langgraph";
import * as z from "zod";

const fruitInfo = tool(
  (input) => `Info about ${input.fruitName}`,
  {
    name: "fruit_info",
    description: "Look up fruit info.",
    schema: z.object({ fruitName: z.string() }),
  }
);

// Subagent with checkpointer=true for persistent state
const fruitAgent = createAgent({
  model: "gpt-5.4-mini",
  tools: [fruitInfo],
  prompt: "You are a fruit expert. Use the fruit_info tool. Respond in one sentence.",
  checkpointer: true,
});

// Wrap subagent as a tool for the outer agent
const askFruitExpert = tool(
  async (input) => {
    const response = await fruitAgent.invoke({
      messages: [{ role: "user", content: input.question }],
    });
    return response.messages[response.messages.length - 1].content;
  },
  {
    name: "ask_fruit_expert",
    description: "Ask the fruit expert. Use for ALL fruit questions.",
    schema: z.object({ question: z.string() }),
  }
);

// Outer agent with checkpointer
// Use toolCallLimitMiddleware to prevent parallel calls to per-thread subagents,
// which would cause checkpoint conflicts.
const agent = createAgent({
  model: "gpt-5.4-mini",
  tools: [askFruitExpert],
  prompt: "You have a fruit expert. ALWAYS delegate fruit questions to ask_fruit_expert.",
  middleware: [  
    toolCallLimitMiddleware({ toolName: "ask_fruit_expert", runLimit: 1 }),
  ],
  checkpointer: new MemorySaver(),
});

Interrupts
Multi-turn
Multiple subgraph calls

按线程子智能体像按调用模式一样支持 interrupt()。在工具函数中添加 interrupt()，即可要求用户批准：

const fruitInfo = tool(
  (input) => {
    interrupt("continue?");
    return `Info about ${input.fruitName}`;
  },
  {
    name: "fruit_info",
    description: "Look up fruit info.",
    schema: z.object({ fruitName: z.string() }),
  }
);

const config = { configurable: { thread_id: "1" } };

// Invoke - the subagent's tool calls interrupt()
let response = await agent.invoke(
  { messages: [{ role: "user", content: "Tell me about apples" }] },
  config,
);
// response contains __interrupt__

// Resume - approve the interrupt
response = await agent.invoke(new Command({ resume: true }), config);
// Subagent message count: 4

状态会在多次调用之间累积——子智能体会记住过去的对话：

const config = { configurable: { thread_id: "1" } };

// First call
let response = await agent.invoke(
  { messages: [{ role: "user", content: "Tell me about apples" }] },
  config,
);
// Subagent message count: 4

// Second call - subagent REMEMBERS apples conversation
response = await agent.invoke(
  { messages: [{ role: "user", content: "Now tell me about bananas" }] },
  config,
);
// Subagent message count: 8 (accumulated!)

当你有多个不同的按线程子图时（例如，一个水果专家和一个蔬菜专家），每个子图都需要自己的存储空间，这样它们的 checkpoint 才不会互相覆盖。这称为命名空间隔离。如果你在节点内部调用子图，LangGraph 会根据调用顺序分配命名空间（第一次调用、第二次调用，等等）。这意味着如果你重新排序调用，可能会混淆哪个子图加载哪个状态。为避免这种情况，请将每个子智能体包装到自己的 StateGraph 中，并使用唯一的节点名称——这会为每个子图提供稳定且唯一的命名空间：

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

function createSubAgent(model: string, { name, ...kwargs }: { name: string; [key: string]: any }) {
  const agent = createAgent({ model, name, ...kwargs });
  return new StateGraph(new StateSchema({ messages: MessagesValue }))
    .addNode(name, agent)  // unique name → stable namespace
    .addEdge(START, name)
    .compile();
}

const fruitAgent = createSubAgent("gpt-5.4-mini", {
  name: "fruit_agent", tools: [fruitInfo], prompt: "...", checkpointer: true,
});
const veggieAgent = createSubAgent("gpt-5.4-mini", {
  name: "veggie_agent", tools: [veggieInfo], prompt: "...", checkpointer: true,
});
const config = { configurable: { thread_id: "1" } };

// First call - LLM calls both fruit and veggie experts
let response = await agent.invoke(
  { messages: [{ role: "user", content: "Tell me about cherries and broccoli" }] },
  config,
);
// Fruit subagent message count: 4
// Veggie subagent message count: 4

// Second call - both agents accumulate independently
response = await agent.invoke(
  { messages: [{ role: "user", content: "Now tell me about oranges and carrots" }] },
  config,
);
// Fruit subagent message count: 8 (remembers cherries!)
// Veggie subagent message count: 8 (remembers broccoli!)

作为节点添加的子图已经会自动获得基于名称的命名空间，因此不需要这个包装器。

Stateless

当你希望像普通函数调用一样运行子智能体、且不想承担 checkpoint 开销时，请使用此模式。子图不能暂停/恢复，也无法受益于持久执行。使用 checkpointer=False 编译。

如果没有 checkpoint，子图就没有持久执行能力。如果进程在运行中崩溃，子图无法恢复，必须从头重新运行。

const subgraphBuilder = new StateGraph(...);
const subgraph = subgraphBuilder.compile({ checkpointer: false });

Checkpointer reference

通过 .compile() 上的 checkpointer 参数控制子图持久化：

const subgraph = builder.compile({ checkpointer: false });  // or true, or null

Feature	Per-invocation (default)	Per-thread	Stateless
`checkpointer=`	`None`	`True`	`False`
Interrupts (HITL)	✅	✅	❌
Multi-turn memory	❌	✅	❌
Multiple calls (different subgraphs)	✅		✅
Multiple calls (same subgraph)	✅	❌	✅
State inspection		✅	❌

Interrupts (HITL)：子图可以使用 interrupt() 暂停执行并等待用户输入，然后从暂停处继续。
Multi-turn memory：子图会在同一线程内的多次调用之间保留其状态。每次调用都会从上一次结束的位置继续，而不是重新开始。
Multiple calls (different subgraphs)：可以在单个节点中调用多个不同的子图实例，而不会产生 checkpoint 命名空间冲突。
Multiple calls (same subgraph)：同一个子图实例可以在单个节点中被调用多次。使用有状态持久化时，这些调用会写入同一个 checkpoint 命名空间并产生冲突——请改用按调用持久化。
State inspection：可通过 get_state(config, subgraphs=True) 获取子图状态，用于调试和监控。

View subgraph state

启用持久化后，你可以使用 subgraphs 选项检查子图状态。使用无状态 checkpoint（checkpointer=False）时，不会保存任何子图 checkpoint，因此无法获取子图状态。

查看子图状态要求 LangGraph 能够静态发现该子图——也就是说，它是作为节点添加的，或是在节点内部调用的。当子图在工具函数内部或其他间接方式中被调用时（例如，subagents 模式），该功能不起作用。无论嵌套层级如何，中断仍会传播到顶层图。

Per-invocation
Per-thread

仅返回当前调用的子图状态。每次调用都会重新开始。

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

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

// Subgraph
const subgraphBuilder = new StateGraph(State)
  .addNode("subgraphNode1", (state) => {
    const value = interrupt("Provide value:");
    return { foo: state.foo + value };
  })
  .addEdge(START, "subgraphNode1");

const subgraph = subgraphBuilder.compile();  // inherits parent checkpointer

// Parent graph
const builder = new StateGraph(State)
  .addNode("node1", subgraph)
  .addEdge(START, "node1");

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

const config = { configurable: { thread_id: "1" } };

await graph.invoke({ foo: "" }, config);

// View subgraph state for the current invocation
const subgraphState = (await graph.getState(config, { subgraphs: true })).tasks[0].state;

// Resume the subgraph
await graph.invoke(new Command({ resume: "bar" }), config);

返回该线程上所有调用累积得到的子图状态。

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

// Subgraph with its own persistent state
const SubgraphState = new StateSchema({
  messages: MessagesValue,
});

const subgraphBuilder = new StateGraph(SubgraphState);
// ... add nodes and edges
const subgraph = subgraphBuilder.compile({ checkpointer: true });

// Parent graph
const builder = new StateGraph(SubgraphState)
  .addNode("agent", subgraph)
  .addEdge(START, "agent");

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

const config = { configurable: { thread_id: "1" } };

await graph.invoke({ messages: [{ role: "user", content: "hi" }] }, config);
await graph.invoke({ messages: [{ role: "user", content: "what did I say?" }] }, config);

// View accumulated subgraph state (includes messages from both invocations)
const subgraphState = (await graph.getState(config, { subgraphs: true })).tasks[0].state;

Stream subgraph outputs

若要观察嵌套图的执行，我们推荐使用事件流：stream.subgraphs 投影会发现每个嵌套运行，并暴露其 path、messages 和 values，无需解析命名空间字符串。

for await (const chunk of await graph.stream(
  { foo: "foo" },
  {
    subgraphs: true,
    streamMode: "updates",
  }
)) {
  console.log(chunk);
}

设置 subgraphs: true 以流式传输子图的输出。

Stream from subgraphs

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

// Define subgraph
const SubgraphState = new StateSchema({
  foo: z.string(),
  bar: z.string(),
});

const subgraphBuilder = new StateGraph(SubgraphState)
  .addNode("subgraphNode1", (state) => {
    return { bar: "bar" };
  })
  .addNode("subgraphNode2", (state) => {
    // note that this node is using a state key ('bar') that is only available in the subgraph
    // and is sending update on the shared state key ('foo')
    return { foo: state.foo + state.bar };
  })
  .addEdge(START, "subgraphNode1")
  .addEdge("subgraphNode1", "subgraphNode2");

const subgraph = subgraphBuilder.compile();

// Define parent graph
const ParentState = new StateSchema({
  foo: z.string(),
});

const builder = new StateGraph(ParentState)
  .addNode("node1", (state) => {
    return { foo: "hi! " + state.foo };
  })
  .addNode("node2", subgraph)
  .addEdge(START, "node1")
  .addEdge("node1", "node2");

const graph = builder.compile();

for await (const chunk of await graph.stream(
  { foo: "foo" },
  {
    streamMode: "updates",
    subgraphs: true,
  }
)) {
  console.log(chunk);
}

设置 subgraphs: true 以流式传输子图的输出。

[[], { node1: { foo: 'hi! foo' } }]
[['node2:e58e5673-a661-ebb0-70d4-e298a7fc28b7'], { subgraphNode1: { bar: 'bar' } }]
[['node2:e58e5673-a661-ebb0-70d4-e298a7fc28b7'], { subgraphNode2: { foo: 'hi! foobar' } }]
[[], { node2: { foo: 'hi! foobar' } }]

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.

Edit this page on GitHub or file an issue.

​Setup

​Define subgraph communication

​Call a subgraph inside a node

​Add a subgraph as a node

​Subgraph persistence

​Stateful

​Per-invocation (default)

​Per-thread

​Stateless

​Checkpointer reference

​View subgraph state

​Stream subgraph outputs

Setup

Define subgraph communication

Call a subgraph inside a node

Add a subgraph as a node

Subgraph persistence

Stateful

Per-invocation (default)

Per-thread

Stateless

Checkpointer reference

View subgraph state

Stream subgraph outputs