企业官网建设流程全解析

子图允许构建由多个组件组成的复杂系统，这些组件本身就是图。使用子图的一个常见用例是构建多智能体系统。

添加子图时的主要问题是父图和子图如何通信，即它们在图执行期间如何相互传递状态。有两种情况：

• 父图和子图共享模式键。在这种情况下，可以添加一个带有已编译子图的节点。
• 父图和子图有不同的模式。在这种情况下，必须添加一个调用子图的节点函数：当父图和子图具有不同的状态模式并且需要在调用子图之前或之后转换状态时，这很有用

父子图共享模式键

介绍

状态字段分两类

• 同名 key：双向共享，子图修改会同步回父图

• 子图独有的 key：仅子图生命周期内存在，出了子图就消失
  子图内部可以利用私有临时状态做中间计算，不需要污染父图全局状态

• foo：共享状态，来自父图，子图可以读、可以改，修改会传回父图

• bar：子图私有状态，只在子图内部生效，子图跑完之后 bar 直接丢弃，不会回到父图

代码

import{StateGraph,Annotation}from"@langchain/langgraph";constSubgraphStateAnnotation=Annotation.Root({foo:Annotation<string>,// note that this key is shared with the parent graph statebar:Annotation<string>,});constsubgraphNode1=async(state:typeofSubgraphStateAnnotation.State)=>{return{bar:"bar"};};constsubgraphNode2=async(state:typeofSubgraphStateAnnotation.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};};constsubgraphBuilder=newStateGraph(SubgraphStateAnnotation).addNode("subgraphNode1",subgraphNode1).addNode("subgraphNode2",subgraphNode2).addEdge("__start__","subgraphNode1").addEdge("subgraphNode1","subgraphNode2")constsubgraph=subgraphBuilder.compile();// Define parent graphconstParentStateAnnotation=Annotation.Root({foo:Annotation<string>,});constnode1=async(state:typeofParentStateAnnotation.State)=>{return{foo:"hi! "+state.foo,};}constbuilder=newStateGraph(ParentStateAnnotation).addNode("node1",node1)// note that we're adding the compiled subgraph as a node to the parent graph.addNode("node2",subgraph).addEdge("__start__","node1").addEdge("node1","node2")constgraph=builder.compile();

conststream=awaitgraph.stream({foo:"foo"});forawait(constchunkofstream){console.log(chunk);}

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

解释

可以把子图当成一个封装好的独立智能体，但它不是一个完全隔离独立的 Agent 进程
它是「在 LangGraph 内部打包成节点的一套独立状态图工作流」，为什么说它像一个单独智能体呢？

• 子图有自己独立的状态定义：例子里的bar，只有它自己能看见，父图无权访问
• 子图内部有自己的节点、边、流转逻辑，内部可以循环、分支、调用工具、甚至再嵌套更深的子图
• 从父图视角：子图就只是node2这单个节点。父图不需要关心子图里面跑了多少步骤、内部怎么流转，只关心输入输出（共享的foo）

父图：“我调用 node2（子图 Agent），给它 foo，等它跑完拿回更新后的 foo，完事。内部细节我不管”
这就是主管 - 工作者（Orchestrator-Worker，编排式多 Agent）最经典的实现方式：父图 = 编排主管 Agent，各个子图 = 各个子工作 Agent。

一步步推演：

1. 父图 node1：foo = "hi! " + "test"→foo = "hi! test"
2. 进入子图，子图初始 state：{ foo:"hi! test" }，bar 为空
3. subgraphNode1：设置bar="bar"，子图 state 现在：{foo:"hi! test", bar:"bar"}
4. subgraphNode2：foo = "hi! test"+"bar"→foo = "hi! testbar"
5. 子图执行完毕，只把 foo 回传给父图，bar 丢掉
6. 父图最终 state：{ foo: "hi! testbar" }

graph.stream()

graph.stream()是流式迭代器，会在每一个节点执行完成之后，吐出一次当前 graph 状态快照（chunk）
不是等整张图跑完一次性返回最终结果，而是每走完一个节点就推一块数据，适合前端实时打印 / 流式输出。

图链路：
__start__→node1→node2(子图)
子图内部：__start__→subgraphNode1→subgraphNode2

执行顺序 & 每一轮 chunk
初始输入：{foo:"foo"}

第 1 个 chunk：父图 node1 执行完毕
node1 代码：

const node1 = async (state) => { return { foo: "hi! " + state.foo }; }

输入 foo=“foo” → 返回{ foo: "hi! foo" }
输出 chunk：

{ node1: { foo: "hi! foo" } }

stream 每一个 chunk 的 key =刚跑完的节点名称，value = 这个节点返回的状态更新

第 2 个 chunk：进入子图，跑完 subgraphNode1
现在执行进入 node2（子图），子图内部先走 subgraphNode1

const subgraphNode1 = async (state) => { return { bar: "bar" }; };

子图此时 state：{foo:"hi! foo"}，执行完增加 bar=“bar”
chunk 输出：

{ node2: { subgraphNode1: { bar: "bar" } } }

因为 subgraph 挂载在父图node2，子图内部节点的 chunk 会嵌套在node2下面

第 3 个 chunk：子图跑完 subgraphNode2

const subgraphNode2 = async (state) => { return { foo: state.foo + state.bar }; };

state.foo=“hi! foo”，state.bar=“bar”
更新 foo ="hi! foobar"
chunk 输出：

{ node2: { subgraphNode2: { foo: "hi! foobar" } } }

子图全部执行结束，子图退出，状态合并回父图
子图只把共享 keyfoo回写给父图；子图私有bar丢弃，不会回到父图状态。
不会单独再产生一个 chunk，子图最后节点的更新已经包含在上面 subgraphNode2 的 chunk 里面

执行结束，循环退出
for await 遍历完所有 chunk，循环结束

父子图不共享模式键

对于更复杂的系统，可能希望定义与父图具有完全不同模式（没有共享键）的子图。例如，在多智能体 RAG 系统中，搜索智能体可能只需要跟踪查询和检索到的文档。

如果应用程序属于这种情况，需要定义一个调用子图的节点函数。此函数需要在调用子图之前将输入（父）状态转换为子图状态，并在从节点返回状态更新之前将结果转换回父状态。

下面展示如何修改原始示例，从节点内部调用子图

import{StateGraph,Annotation}from"@langchain/langgraph";constSubgraphAnnotation=Annotation.Root({bar:Annotation<string>,// note that this key is shared with the parent graph statebaz:Annotation<string>,});constsubgraphNodeOne=async(state:typeofSubgraphAnnotation.State)=>{return{baz:"baz"};};constsubgraphNodeTwo=async(state:typeofSubgraphAnnotation.State)=>{return{bar:state.bar+state.baz}};constsubgraphCalledInFunction=newStateGraph(SubgraphAnnotation).addNode("subgraphNode1",subgraphNodeOne).addNode("subgraphNode2",subgraphNodeTwo).addEdge("__start__","subgraphNode1").addEdge("subgraphNode1","subgraphNode2").compile();// Define parent graphconstParentAnnotation=Annotation.Root({foo:Annotation<string>,});constnodeOne=async(state:typeofParentAnnotation.State)=>{return{foo:"hi! "+state.foo,};}constnodeTwo=async(state:typeofParentAnnotation.State)=>{constresponse=awaitsubgraphCalledInFunction.invoke({bar:state.foo,});return{foo:response.bar}}constgraphWithFunction=newStateGraph(ParentStateAnnotation).addNode("node1",nodeOne)// note that we're adding the compiled subgraph as a node to the parent graph.addNode("node2",nodeTwo).addEdge("__start__","node1").addEdge("node1","node2").compile();

constgraphWithFunctionStream=awaitgraphWithFunction.stream({foo:"foo"},{subgraphs:true});forawait(constchunkofgraphWithFunctionStream){console.log(chunk);}

[[],{node1:{foo:'hi! foo'}}][['node2:1d2bb11a-3ed1-5c58-9b6f-c7af36a1eeb7'],{subgraphNode1:{baz:'baz'}}][['node2:1d2bb11a-3ed1-5c58-9b6f-c7af36a1eeb7'],{subgraphNode2:{bar:'hi! foobaz'}}][[],{node2:{foo:'hi! foobaz'}}]

如果为子图启用了检查点，则不能在同一节点内调用多个子图