子图（Subgraphs）

本指南介绍使用子图的机制。子图是一个在另一个图中作为节点使用的图。

子图适用于：

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

安装配置

npm install @langchain/langgraph

TIP

注册 LangSmith 以快速发现问题并提高 LangGraph 项目的性能。LangSmith 允许你使用跟踪数据来调试、测试和监控使用 LangGraph 构建的 LLM 应用程序。

定义子图通信

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

模式	适用场景	状态架构
在节点内调用子图	父图和子图具有不同的状态架构（无共享键），或者你需要在它们之间转换状态	你编写一个包装函数，将父状态映射到子图输入，并将子图输出映射回父状态
将子图添加为节点	父图和子图共享状态键——子图从与父级相同的通道读取并写入	你将编译后的子图直接传递给 addNode——无需包装函数

在节点内调用子图

当父图和子图具有不同的状态架构（无共享键）时，在节点函数内调用子图。这在你想要为多智能体系统中的每个智能体保留私有消息历史记录时很常见。

节点函数在调用子图之前将父状态转换为子图状态，并在返回之前将结果转换回父状态。

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

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

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

const subgraph = subgraphBuilder.compile();

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

// 将状态转换为子图状态并转换回来
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();

完整示例：不同的状态架构

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

// 定义子图
const SubgraphState = new StateSchema({
  // 请注意，这些键均未与父图状态共享
  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();

// 定义父图
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();

const stream = await graph.streamEvents(
  { foo: "foo" },
  { subgraphs: true, version: "v3" }
);
for await (const message of stream.messages) {
  for await (const token of message.text) {
    process.stdout.write(token);
  }
}

完整示例：不同状态架构的两级嵌套子图（Parent -> Child -> Grandchild）

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

// 孙子图
const GrandChildState = new StateSchema({
  myGrandchildKey: z.string(),
});

const grandchild = new StateGraph(GrandChildState)
  .addNode("grandchild1", (state) => {
    return { myGrandchildKey: state.myGrandchildKey + ", how are you" };
  })
  .addEdge(START, "grandchild1")
  .addEdge("grandchild1", END);

const grandchildGraph = grandchild.compile();

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

const child = new StateGraph(ChildState)
  .addNode("child1", async (state) => {
    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();

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

const parent = new StateGraph(ParentState)
  .addNode("parent1", (state) => {
    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();

将子图添加为节点

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

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

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

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

const subgraph = subgraphBuilder.compile();

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

const graph = builder.compile();

子图持久化（Subgraph persistence）

当你使用子图时，你需要决定在调用之间其内部数据会发生什么。考虑一个委托给专家子代理的客户支持机器人：“计费专家”子代理应该记住客户较早的问题，还是每次被调用时都重新开始？

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

模式	checkpointer=	行为
每次调用独立（默认）	默认（未设置）	每次调用都重新开始，并继承父级的检查点保存器，以在单次调用内支持中断和持久化执行。
每线程持久	true	状态在同一线程的多次调用中累积。每次调用都会从上次停止的地方继续。
无状态	false	完全不进行检查点记录——像普通的函数调用一样运行。没有中断或持久化执行。

有状态模式（Stateful）

有状态子图继承父图的检查点保存器，从而启用中断、持久化和状态检查。

每次调用独立（默认）

TIP

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

省略 checkpointer。每次调用都重新开始，但在单次调用内，子图会继承父级的检查点保存器，并可以使用 interrupt() 暂停和恢复。

每线程持久（Per-thread）

在子图上设置 checkpointer: true，使其具有自己的跨调用持久化状态。每次调用子图时，它都会保留之前调用的累积记忆。

无状态模式（Stateless）

设置 checkpointer: false 会禁用子图中的所有检查点功能。它纯粹作为内存中的无状态函数运行，无法调用 interrupt()。

查看子图状态（View subgraph state）

使用带有 subgraphs: true 的 getState 检查子图状态：

const state = await graph.getState(config, { subgraphs: true });
// 检查任务和子图快照
console.log(state.tasks[0]?.state);

流式传输子图输出（Stream subgraph outputs）

在调用 streamEvents 时传递 { subgraphs: true, version: "v3" }，即可同时监听来自父图和所有嵌套子图的事件：

const stream = await graph.streamEvents(input, {
  subgraphs: true,
  version: "v3",
});

for await (const subgraph of stream.subgraphs) {
  console.log(`子图 [${subgraph.name}] 路径: ${subgraph.path}`);
  for await (const message of subgraph.messages) {
    console.log(await message.text);
  }
}