子图 - 以编程方式使用文档

本指南解释了使用子图的机制。子图是一个图，用作节点在另一个图中。

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

设置

npm install @langchain/langgraph

定义子图通信

添加子图时，您需要定义父图与子图之间的通信方式：

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

<a id="invoke-a-graph-from-a-node"></a> ### 在节点内调用子图

当父图和子图具有 **不同的状态模式** （没有共享的键）时，在节点函数中调用子图。当您想要为每个代理保留私有消息历史时，这很常见 multi-agent system.

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

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

  // 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 });   // [!code highlight]
      return { foo: response.bar };   // [!code highlight]
    })
    .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);
    }
  }

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

  [[], { 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)

这是一个具有两层子图的示例：父图 -> 子图 -> 孙图。

  // 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 };   // [!code highlight]
      const grandchildGraphOutput = await grandchildGraph.invoke(grandchildGraphInput);
      return { myChildKey: grandchildGraphOutput.myGrandchildKey + " today?" };   // [!code highlight]
    })   // [!code highlight]
    .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 };   // [!code highlight]
      const childGraphOutput = await childGraph.invoke(childGraphInput);
      return { myKey: childGraphOutput.myChildKey };   // [!code highlight]
    })   // [!code highlight]
    .addNode("parent2", (state) => {
      return { myKey: state.myKey + " bye!" };
    })
    .addEdge(START, "parent1")
    .addEdge("parent1", "child")
    .addEdge("child", "parent2")
    .addEdge("parent2", END);

  const parentGraph = parent.compile();

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

我们正在从子图状态通道 (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!' } }]

<a id="add-a-graph-as-a-node"></a> ### 将子图添加为节点

当父图和子图 **共享状态键**，您可以直接将编译好的子图传递给 add_node。无需包装函数——子图会自动读写父级的状态通道。例如，在 multi-agent 系统中，代理通常通过共享的消息 key.

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

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

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

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

  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();

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

此键与父图状态共享
此密钥是私有的，归属 SubgraphState 对父图不可见

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

子图持久化

使用子图时，需要决定其内部数据在调用之间的处理方式。考虑一个将请求委托给专业子代理的客户支持机器人："账单专家"子代理应该记住客户之前的问题，还是每次调用时都重新开始？

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

模式	`checkpointer=`	行为
Per-invocation	`None` （默认）	每次调用都会重新开始，并继承父图的检查点以支持中断和持久执行在单次调用中。
Per-thread	`True`	状态在同一线程的调用之间累积。每次调用都会从上一次的结束位置继续。
无状态	`False`	完全不进行保存点检查——像普通函数调用一样运行。不支持中断或持久执行。

对于大多数应用程序来说，每次调用是正确选择，包括 multi-agent 子代理处理独立请求的系统。当子代理需要多轮对话记忆时使用按线程（例如，一个在多次交互中积累上下文的研究助手）。

有状态

有状态子图继承父图的检查点器，这使得中断, 持久化和状态检查成为可能。两种有状态模式的区别在于状态保留的时长。

每次调用（默认）

当每次调用子图都是独立的，且子代理不需要记住之前调用的任何信息时，请使用每次调用持久化。这是最常见的模式，特别是对于 multi-agent 子代理处理一次性请求的系统，如“查询此客户的订单”或“总结此文档”。

省略 checkpointer 或将其设置为 None。每次调用都是全新的，但在单次调用内，子图继承父图的检查点器，并可以使用 interrupt() 来暂停和恢复。

以下示例使用两个子代理（水果专家、蔬菜专家）作为工具包装，供外部代理调用：

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

每次调用都可以使用 interrupt() 来暂停和恢复。添加 interrupt() 到工具函数中，以要求用户批准后才能继续：

  const fruitInfo = tool(
    (input) => {
      interrupt("continue?");  // [!code highlight]
      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);  // [!code highlight]
  // Subagent message count: 4

Multi-turn

每次调用都以全新的子代理状态开始。子代理不记得之前的调用：

  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!)

Multiple subgraph calls

对同一子图的多次调用不会产生冲突，因为每次调用都有各自独立的检查点命名空间：

  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 来启用此行为。

以下示例使用水果专家子代理进行编译 checkpointer=True:

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,  // [!code highlight]
});

// 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: [  // [!code highlight]
    toolCallLimitMiddleware({ toolName: "ask_fruit_expert", runLimit: 1 }),  // [!code highlight]
  ],  // [!code highlight]
  checkpointer: new MemorySaver(),
});

Interrupts

每线程子代理支持 interrupt() 与每次调用相同。添加 interrupt() 到工具函数以要求用户批准：

  const fruitInfo = tool(
    (input) => {
      interrupt("continue?");  // [!code highlight]
      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);  // [!code highlight]
  // Subagent message count: 4

Multi-turn

状态在多次调用中累积——子代理会记住过去的对话：

  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!)

Multiple subgraph calls

当你有多个 **不同的** 每线程子图（例如，水果专家和蔬菜专家），每个子图都需要自己的存储空间，这样它们的检查点就不会相互覆盖。这被称为 **命名空间隔离**.

如果你在节点内调用子图，LangGraph会根据调用顺序分配命名空间（第一次调用、第二次调用等）。这意味着重新排序调用可能会混淆哪个子图加载哪个状态。为了避免这种情况，请将每个子代理包装在其自己的 StateGraph 中并使用唯一的节点名称——这为每个子图提供了稳定、唯一的命名空间：

  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  // [!code highlight]
      .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!)

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

无状态

Use this when you want to run a subagent like a plain function call with no checkpointing overhead. The subgraph cannot pause/resume and does not benefit from 持久执行。使用以下方式编译 checkpointer=False.

const subgraphBuilder = new StateGraph(...);
const subgraph = subgraphBuilder.compile({ checkpointer: false });  // [!code highlight]

检查点引用

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

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

功能	每次调用（默认）	每线程	无状态
`checkpointer=`	`None`	`True`	`False`
中断（HITL）	✅	✅	❌
多轮记忆	❌	✅	❌
多次调用（不同子图）	✅	⚠️	✅
多次调用（相同子图）	✅	❌	✅
状态检查	⚠️	✅	❌

中断 (HITL)：子图可以使用 interrupt() 来暂停执行并等待用户输入，然后从中断处继续。
多轮记忆：子图在同一个线程内的多次调用中保留其状态。每次调用都从上一个的断点继续，而不是重新开始。
多次调用（不同子图）：多个不同的子图实例可以在单个节点内被调用，而不会产生检查点命名空间冲突。
多次调用（相同子图）：同一个子图实例可以在单个节点内被多次调用。使用有状态持久化时，这些调用会写入相同的检查点命名空间并产生冲突——应改用每次调用的持久化。
状态检查：子图的状态可通过以下方式获取 get_state(config, subgraphs=True) 用于调试和监控。

查看子图状态

当您启用持久化时，您可以使用 subgraphs 选项检查子图状态。使用无状态检查点（checkpointer=False）时，不会保存子图检查点，因此子图状态不可用。

Per-invocation

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

  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;  // [!code highlight]

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

Per-thread

返回 **累积的** 此线程上所有调用的子图状态。

  // 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 });  // [!code highlight]

  // 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;  // [!code highlight]

流式输出子图

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

const stream = await graph.streamEvents(
  { foo: "foo" },
  {
    subgraphs: true,   // [!code highlight]
    version: "v3",
  }
);
for await (const snapshot of stream.values) {
  console.log(snapshot);
}

设置 subgraphs: true 来从子图流式输出。

Stream from subgraphs

  // 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();

  const stream = await graph.streamEvents(
    { foo: "foo" },
    {
      subgraphs: true,   // [!code highlight]
      version: "v3",
    }
  );
  for await (const snapshot of stream.values) {
    console.log(snapshot);
  }

设置 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' } }]