LangGraph 的子图 - 共享模式键和不共享模式键的学习

子图允许构建由多个组件组成的复杂系统这些组件本身就是图。使用子图的一个常见用例是构建多智能体系统。添加子图时的主要问题是父图和子图如何通信即它们在图执行期间如何相互传递状态。有两种情况父图和子图共享模式键。在这种情况下可以添加一个带有已编译子图的节点。父图和子图有不同的模式。在这种情况下必须添加一个调用子图的节点函数当父图和子图具有不同的状态模式并且需要在调用子图之前或之后转换状态时这很有用父子图共享模式键介绍状态字段分两类同名 key双向共享子图修改会同步回父图子图独有的 key仅子图生命周期内存在出了子图就消失子图内部可以利用私有临时状态做中间计算不需要污染父图全局状态foo共享状态来自父图子图可以读、可以改修改会传回父图bar子图私有状态只在子图内部生效子图跑完之后 bar 直接丢弃不会回到父图代码import{StateGraph,Annotation}fromlangchain/langgraph;constSubgraphStateAnnotationAnnotation.Root({foo:Annotationstring,// note that this key is shared with the parent graph statebar:Annotationstring,});constsubgraphNode1async(state:typeofSubgraphStateAnnotation.State){return{bar:bar};};constsubgraphNode2async(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.foostate.bar};};constsubgraphBuildernewStateGraph(SubgraphStateAnnotation).addNode(subgraphNode1,subgraphNode1).addNode(subgraphNode2,subgraphNode2).addEdge(__start__,subgraphNode1).addEdge(subgraphNode1,subgraphNode2)constsubgraphsubgraphBuilder.compile();// Define parent graphconstParentStateAnnotationAnnotation.Root({foo:Annotationstring,});constnode1async(state:typeofParentStateAnnotation.State){return{foo:hi! state.foo,};}constbuildernewStateGraph(ParentStateAnnotation).addNode(node1,node1)// note that were adding the compiled subgraph as a node to the parent graph.addNode(node2,subgraph).addEdge(__start__,node1).addEdge(node1,node2)constgraphbuilder.compile();conststreamawaitgraph.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。一步步推演父图 node1foo hi! test→foo hi! test进入子图子图初始 state{ foo:hi! test }bar 为空subgraphNode1设置barbar子图 state 现在{foo:hi! test, bar:bar}subgraphNode2foo hi! testbar→foo hi! testbar子图执行完毕只把 foo 回传给父图bar 丢掉父图最终 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子图子图内部先走 subgraphNode1const subgraphNode1 async (state) { return { bar: bar }; };子图此时 state{foo:hi! foo}执行完增加 bar“bar”chunk 输出{ node2: { subgraphNode1: { bar: bar } } }因为 subgraph 挂载在父图node2子图内部节点的 chunk 会嵌套在node2下面第 3 个 chunk子图跑完 subgraphNode2const subgraphNode2 async (state) { return { foo: state.foo state.bar }; };state.foo“hi! foo”state.bar“bar”更新 foo hi! foobarchunk 输出{ node2: { subgraphNode2: { foo: hi! foobar } } }子图全部执行结束子图退出状态合并回父图子图只把共享 keyfoo回写给父图子图私有bar丢弃不会回到父图状态。不会单独再产生一个 chunk子图最后节点的更新已经包含在上面 subgraphNode2 的 chunk 里面执行结束循环退出for await 遍历完所有 chunk循环结束父子图不共享模式键对于更复杂的系统可能希望定义与父图具有完全不同模式没有共享键的子图。例如在多智能体 RAG 系统中搜索智能体可能只需要跟踪查询和检索到的文档。如果应用程序属于这种情况需要定义一个调用子图的节点函数。此函数需要在调用子图之前将输入父状态转换为子图状态并在从节点返回状态更新之前将结果转换回父状态。下面展示如何修改原始示例从节点内部调用子图import{StateGraph,Annotation}fromlangchain/langgraph;constSubgraphAnnotationAnnotation.Root({bar:Annotationstring,// note that this key is shared with the parent graph statebaz:Annotationstring,});constsubgraphNodeOneasync(state:typeofSubgraphAnnotation.State){return{baz:baz};};constsubgraphNodeTwoasync(state:typeofSubgraphAnnotation.State){return{bar:state.barstate.baz}};constsubgraphCalledInFunctionnewStateGraph(SubgraphAnnotation).addNode(subgraphNode1,subgraphNodeOne).addNode(subgraphNode2,subgraphNodeTwo).addEdge(__start__,subgraphNode1).addEdge(subgraphNode1,subgraphNode2).compile();// Define parent graphconstParentAnnotationAnnotation.Root({foo:Annotationstring,});constnodeOneasync(state:typeofParentAnnotation.State){return{foo:hi! state.foo,};}constnodeTwoasync(state:typeofParentAnnotation.State){constresponseawaitsubgraphCalledInFunction.invoke({bar:state.foo,});return{foo:response.bar}}constgraphWithFunctionnewStateGraph(ParentStateAnnotation).addNode(node1,nodeOne)// note that were adding the compiled subgraph as a node to the parent graph.addNode(node2,nodeTwo).addEdge(__start__,node1).addEdge(node1,node2).compile();constgraphWithFunctionStreamawaitgraphWithFunction.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}}]如果为子图启用了检查点则不能在同一节点内调用多个子图