企业官网建设流程全解析

LangGraph 子图：流程太长时，别再往一张图里塞节点了

一个 Agent 画到 30 个节点的时候，你会发现两件事：
一是图看不懂了，二是其中"查资料→整理→校验"这一小段，
明明在另外两个流程里也用得上，却只能复制一遍。

子图（Subgraph）就是为了解决这两件事。

一、子图到底解决什么问题

先说清楚子图不是什么：它不是新语法，也不是某种高级特性。
LangGraph 里所谓子图，本质就是——一个编译好的图，当作另一个图的节点来用。

它带来三个实际好处：

反过来，如果你的图只有五六个节点，不要上子图。
多一层嵌套就多一层状态转译的麻烦，得不偿失。

二、最小可用例子：把子图当节点挂上去

先建一个子图，负责"取数 + 清洗"：

fromtypingimportTypedDictfromlanggraph.graphimportStateGraph,START,ENDclassCleanState(TypedDict):raw:strcleaned:strdeffetch(state:CleanState):# 假装这里真的去查了接口return{"raw":" Hello World "}defclean(state:CleanState):return{"cleaned":" ".join(state["raw"].split())}sub=StateGraph(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_graph=sub.compile()# ← 编译产物

然后在父图里，把它当成一个普通节点：

classMainState(TypedDict):question:strcleaned:stranswer:strdefask_llm(state:MainState):# 这里只是示意，真实场景用 init_chat_modelreturn{"answer":f"处理结果是：{state['cleaned']}"}main=StateGraph(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)app=main.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):decision=interrupt({"question":"这段内容可以发布吗？","content":state["draft"],})return{"approved":decision=="ok"}# 子图编译时挂 checkpointerreview_sub=StateGraph(ReviewState)review_sub.add_node("human_review",human_review)review_sub.add_edge(START,"human_review")review_sub.add_edge("human_review",END)review_graph=review_sub.compile(checkpointer=InMemorySaver())

父图调用时正常 invoke，遇到中断会抛出；恢复时：

app.invoke(Command(resume="ok"),config={"configurable":{"thread_id":"t-1"}})

这里有个必须注意的点：中断的恢复是靠thread_id定位检查点的。
如果父图和子图用的是同一个 checkpointer，父子共用一条 thread 记录；
如果子图自己单独挂了一个 checkpointer，恢复时要保证 config 能传到子图那一层。

实践建议：除非你很清楚自己在做什么，否则只在最外层挂一个 checkpointer，
子图单独挂会让状态恢复的链路变得非常难调。

五、子图内部的流式输出

父图stream默认看不到子图内部的中间步骤，只会看到子图整体结束时的输出。
想看子图里面发生了什么，加subgraphs=True：

fornamespace,chunkinapp.stream({"question":"..."},stream_mode="updates",subgraphs=True,):node_name=namespace[-1]ifnamespaceelse"<root>"print(node_name,"->",chunk)

namespace是一个元组，表示当前 chunk 来自哪一层：

• 根图：()或者只带父图信息
• 子图：末尾会多出子图的节点名，例如("prepare:xxx", "clean:yyy")

前端展示时可以用 namespace 的层级做缩进，父图一级、子图缩进一级，
用户就能看到"这一步里面还做了两件小事"。

六、什么时候不该拆子图

列三条我自己的判断标准，供参考：

1. 节点数少于 10，不要拆。拆了维护成本反而上升。
2. 子图只有一个节点，不要拆。那直接写个函数当节点就行。
3. 父子之间需要频繁互相读写状态，不要拆。这说明它们本来就是一个耦合体，
   强行拆开只会让你写一堆转换代码。

另外补充一个经验：按"职责"拆，不要按"步骤"拆。
把"1→2→3→4"切成"1→2"和"3→4"毫无意义；
把"检索 / 生成 / 审核"拆成三块才有意义，因为这三块各自的失败处理、重试策略、
模型选择都不一样。

七、常见报错排查

最后一点值得展开：子图一旦被放进循环里，每次循环都会完整跑一遍。
如果子图里有 LLM 调用，成本和耗时会翻倍增长。
能提到循环外面就提到外面，这是最省钱的优化。

小结

• 子图 = 把编译后的图当节点用，本质没有魔法；
• 状态靠同名 key 透传，不同名就写转换节点；
• checkpointer 尽量只在最外层挂一次；
• 看子图内部过程用subgraphs=True；
• 图小于 10 个节点就别拆，拆了是给自己找麻烦。