个人主页 for_ever_love__ 欢迎各位大佬莅临其他栏目: 大模型开发从0到1 其他栏目: iOS项目总结大全 其他栏目: 我想学python了 其他栏目: iOS UI 文章目录LangGraph 子图流程太长时别再往一张图里塞节点了一、子图到底解决什么问题二、最小可用例子把子图当节点挂上去三、状态对不上怎么办四、子图里能不能中断五、子图内部的流式输出六、什么时候不该拆子图七、常见报错排查小结LangGraph 子图流程太长时别再往一张图里塞节点了一个 Agent 画到 30 个节点的时候你会发现两件事一是图看不懂了二是其中查资料→整理→校验这一小段明明在另外两个流程里也用得上却只能复制一遍。子图Subgraph就是为了解决这两件事。一、子图到底解决什么问题先说清楚子图不是什么它不是新语法也不是某种高级特性。LangGraph 里所谓子图本质就是——一个编译好的图当作另一个图的节点来用。它带来三个实际好处好处说明拆分一张 30 节点的图拆成 5 节点的父图 3 张 8 节点的子图每张图单独看得懂复用同一段检索 重排 过滤逻辑能在多个父图里直接当节点挂独立演进子图内部怎么改换模型、加分支、加重试父图不受影响反过来如果你的图只有五六个节点不要上子图。多一层嵌套就多一层状态转译的麻烦得不偿失。二、最小可用例子把子图当节点挂上去先建一个子图负责取数 清洗fromtypingimportTypedDictfromlanggraph.graphimportStateGraph,START,ENDclassCleanState(TypedDict):raw:strcleaned:strdeffetch(state:CleanState):# 假装这里真的去查了接口return{raw: Hello World }defclean(state:CleanState):return{cleaned: .join(state[raw].split())}subStateGraph(CleanState)sub.add_node(fetch,fetch)sub.add_node(clean,clean)sub.add_edge(START,fetch)sub.add_edge(fetch,clean)sub.add_edge(clean,END)clean_graphsub.compile()# ← 编译产物然后在父图里把它当成一个普通节点classMainState(TypedDict):question:strcleaned:stranswer:strdefask_llm(state:MainState):# 这里只是示意真实场景用 init_chat_modelreturn{answer:f处理结果是{state[cleaned]}}mainStateGraph(MainState)main.add_node(prepare,clean_graph)# ← 直接传编译好的图main.add_node(ask,ask_llm)main.add_edge(START,prepare)main.add_edge(prepare,ask)main.add_edge(ask,END)appmain.compile()就这么简单。编译后的图实现了和普通节点一样的调用协议接收 state返回 state 更新所以父图根本不需要知道它接到的是一个图还是一个函数。三、状态对不上怎么办这是子图最容易踩的坑。规则很直接父子 state 里同名的 key 会自动透传不同名的 key 不会。CleanState.cleaned在MainState里也有 → 子图写的值会流回父图 ✅CleanState.raw在MainState里没有 → 只在子图内部流转父图看不到 ✅这也是我们想要的父图的question想传给子图 →不会自动传因为子图 state 里没这个 key ❌最后一条最常出问题。解决办法有两个办法一给子图加一个转换节点推荐defto_sub(state:MainState):父图 → 子图显式做一次字段映射return{raw:state[question]}main.add_node(to_sub,to_sub)main.add_node(prepare,clean_graph)main.add_edge(START,to_sub)main.add_edge(to_sub,prepare)转换节点看起来啰嗦但它把两个模块之间的接口写成了显式代码。三个月后你回头看一眼就知道父图给了子图什么——这比隐式透传好维护得多。办法二父子共用同一套 State如果子图本来就是为这个父图服务的直接让两边用同一个TypedDict。代价是子图没法单独复用了好处是零转换。我的选择习惯准备复用的子图用办法一纯粹为了拆分而拆的子图用办法二。四、子图里能不能中断能而且这是子图一个很有价值的用法——把需要人工确认的环节封装成一个可复用模块。fromlanggraph.checkpoint.memoryimportInMemorySaverfromlanggraph.typesimportinterrupt,Commanddefhuman_review(state:ReviewState):decisioninterrupt({question:这段内容可以发布吗,content:state[draft],})return{approved:decisionok}# 子图编译时挂 checkpointerreview_subStateGraph(ReviewState)review_sub.add_node(human_review,human_review)review_sub.add_edge(START,human_review)review_sub.add_edge(human_review,END)review_graphreview_sub.compile(checkpointerInMemorySaver())父图调用时正常 invoke遇到中断会抛出恢复时app.invoke(Command(resumeok),config{configurable:{thread_id:t-1}})这里有个必须注意的点中断的恢复是靠thread_id定位检查点的。如果父图和子图用的是同一个 checkpointer父子共用一条 thread 记录如果子图自己单独挂了一个 checkpointer恢复时要保证 config 能传到子图那一层。实践建议除非你很清楚自己在做什么否则只在最外层挂一个 checkpointer子图单独挂会让状态恢复的链路变得非常难调。五、子图内部的流式输出父图stream默认看不到子图内部的中间步骤只会看到子图整体结束时的输出。想看子图里面发生了什么加subgraphsTruefornamespace,chunkinapp.stream({question:...},stream_modeupdates,subgraphsTrue,):node_namenamespace[-1]ifnamespaceelserootprint(node_name,-,chunk)namespace是一个元组表示当前 chunk 来自哪一层根图()或者只带父图信息子图末尾会多出子图的节点名例如(prepare:xxx, clean:yyy)前端展示时可以用 namespace 的层级做缩进父图一级、子图缩进一级用户就能看到这一步里面还做了两件小事。六、什么时候不该拆子图列三条我自己的判断标准供参考节点数少于 10不要拆。拆了维护成本反而上升。子图只有一个节点不要拆。那直接写个函数当节点就行。父子之间需要频繁互相读写状态不要拆。这说明它们本来就是一个耦合体强行拆开只会让你写一堆转换代码。另外补充一个经验按职责拆不要按步骤拆。把1→2→3→4切成1→2和3→4毫无意义把检索 / 生成 / 审核拆成三块才有意义因为这三块各自的失败处理、重试策略、模型选择都不一样。七、常见报错排查现象原因处理子图节点执行了父图读不到结果父图 state 没有同名 key加转换节点或父子共用 stateInvalidUpdateError子图返回了父图 state 里声明之外的 key检查两个 TypedDict 是否对齐中断后恢复报找不到检查点子图单独挂了 checkpointer把 checkpointer 提到最外层子图反复执行父图有环且子图在环里确认子图是幂等的或把子图移出循环最后一点值得展开子图一旦被放进循环里每次循环都会完整跑一遍。如果子图里有 LLM 调用成本和耗时会翻倍增长。能提到循环外面就提到外面这是最省钱的优化。小结子图 把编译后的图当节点用本质没有魔法状态靠同名 key 透传不同名就写转换节点checkpointer 尽量只在最外层挂一次看子图内部过程用subgraphsTrue图小于 10 个节点就别拆拆了是给自己找麻烦。
