別再硬寫LangGraph了！學會SubGraph，復雜度直接降維打擊 原創
在折騰 LangGraph 有段時間了，最近項目里碰上個特別棘手的問題，才真正體會到子圖（Subgraph）這個設計的妙處。以前總覺得把節點連來連去就夠了，直到工作流復雜到自己都快看不懂的時候，才明白模塊化不是說說而已。
什么是SubGraph
簡單來說，子圖就是一個圖，但它被用作另一個圖中的一個節點 。聽起來有點繞，但實際用起來就是把一堆相關的邏輯打包成一個“黑盒子”。比如，我最近在搞一個多智能體的玩意兒，里面有個專門負責“數據預處理”的環節，步驟還挺多，又是清洗又是轉換的。以前這些節點都散落在主圖里，改個參數都得小心翼翼，生怕牽一發而動全身。后來干脆把它們全塞進一個子圖里，對外只暴露一個輸入和一個輸出。主圖瞬間清爽了，那個預處理模塊也變成了一個可以到處復用的模塊。
使用SubGraph的原因
子圖（subgraph）是一個在另一個圖中作為節點使用的圖——這是封裝概念在 LangGraph 中的應用。子圖允許您構建包含多個組件的復雜系統，而這些組件本身就是圖。
使用子圖的一些原因包括：
- 構建多智能體系統
- 當您想在多個圖中重用一組節點時
- 當您希望不同的團隊獨立開發圖的不同部分時，您可以將每個部分定義為一個子圖。只要遵守子圖的接口（輸入和輸出模式），父圖就可以在不了解子圖任何細節的情況下進行構建。
如何使用
添加子圖時，您需要定義父圖和子圖如何通信
- 共享狀態模式 — 父圖和子圖在其狀態模式中擁有共享的狀態鍵
- 不同狀態模式 — 父圖和子圖的模式中沒有共享的狀態鍵
共享狀態模式
一種常見情況是父圖和子圖通過模式中的共享狀態鍵（通道）進行通信。例如，在多智能體系統中，智能體通常通過共享的 messages 鍵進行通信。
如果您的子圖與父圖共享狀態鍵，您可以按照以下步驟將其添加到您的圖中
定義子圖工作流（在下面的示例中為 subgraph_builder）并編譯它 在定義父圖工作流時，將編譯后的子圖傳遞給 .add_node 方法
from typing_extensions import TypedDict
from langgraph.graph.state import StateGraph, START
# Define subgraph
class SubgraphState(TypedDict):
foo: str
bar: str
def subgraph_node_1(state: SubgraphState):
print("[子圖] subgraph_node_1 輸入狀態:", state)
result = {"bar": "bar"}
print("[子圖] subgraph_node_1 輸出更新:", result)
return result
def subgraph_node_2(state: SubgraphState):
print("[子圖] subgraph_node_2 輸入狀態:", 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')
updated_foo = state["foo"] + state["bar"]
result = {"foo": updated_foo}
print("[子圖] subgraph_node_2 輸出更新:", result)
return result
subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_node(subgraph_node_2)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
subgraph = subgraph_builder.compile()
print("? 子圖編譯完成: subgraph")
# Define parent graph
class ParentState(TypedDict):
foo: str
def node_1(state: ParentState):
print("[父圖] node_1 輸入狀態:", state)
result = {"foo": "hi! " + state["foo"]}
print("[父圖] node_1 輸出更新:", result)
return result
builder = StateGraph(ParentState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", subgraph)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
graph = builder.compile()
print("? 父圖編譯完成: graph")
print("=== 開始執行圖 ===")
initial_input = {"foo": "foo"}
print("初始輸入:", initial_input)
for chunk in graph.stream(initial_input):
print("[流事件] 節點輸出:", chunk)
print("=== 執行結束 ===")輸出結果
? 子圖編譯完成: subgraph
? 父圖編譯完成: graph
=== 開始執行圖 ===
初始輸入: {'foo': 'foo'}
[父圖] node_1 輸入狀態: {'foo': 'foo'}
[父圖] node_1 輸出更新: {'foo': 'hi! foo'}
[流事件] 節點輸出: {'node_1': {'foo': 'hi! foo'}}
[子圖] subgraph_node_1 輸入狀態: {'foo': 'hi! foo'}
[子圖] subgraph_node_1 輸出更新: {'bar': 'bar'}
[子圖] subgraph_node_2 輸入狀態: {'foo': 'hi! foo', 'bar': 'bar'}
[子圖] subgraph_node_2 輸出更新: {'foo': 'hi! foobar'}
[流事件] 節點輸出: {'node_2': {'foo': 'hi! foobar'}}
=== 執行結束 ===示例講解：共享狀態模式
- 狀態類型：父圖 ParentState 與子圖 SubgraphState 都含有 foo 鍵（共享），子圖內部新增 bar（私有）。
- 子圖節點職責：
a.subgraph_node_1 只負責在子圖內部產出 bar，不觸碰父圖的狀態鍵。
b.subgraph_node_2 使用子圖私有的 bar 與共享鍵 foo 組合，更新共享鍵 foo，從而把子圖計算結果“寫回”父圖可見的通道。
- 父圖如何接入：
a.??builder.add_node("node_2", subgraph)?? 直接把已 compile 的子圖作為一個節點；
b.這意味著父圖與子圖共享的鍵可以直接貫通，省去輸入/輸出映射的樣板代碼。
- 執行軌跡：
a.先運行父圖 node_1 預處理 foo → 然后進入子圖（node_2）依次執行 subgraph_node_1 與 subgraph_node_2 → 回到父圖繼續流。
運行與驗證（共享狀態）
- 運行方式：將示例保存為 Python 文件后直接執行。
- 期望輸出：你將看到“[父圖]… → [子圖]…” 的日志順序，以及流式事件中 node_1 與 node_2 的增量更新。
- 關注點：
a.子圖更新 foo 后，父圖后續節點都能看到該更新（因為 foo 為共享鍵）。
b.子圖內的 bar 不會“泄漏”到父圖（除非顯式設計為共享鍵）。
不同狀態模式
示例講解：不同狀態模式
- 狀態類型：父圖 ParentState 只有 foo；子圖 SubgraphState 擁有 bar、baz，二者完全不共享。
- 為什么需要節點包裝：由于沒有共享鍵，父圖無法直接把自身狀態交給子圖，也無法直接讀取子圖結果；因此在 node_2 中進行“輸入映射 → 子圖調用 → 輸出映射”。
- 數據流：
a.輸入映射：在 node_2 中構造??subgraph_input = {"bar": state["foo"]}??。
b.子圖內部：??subgraph_node_1?? 產出??baz??；??subgraph_node_2?? 基于??bar?? 與??baz?? 計算新的??bar??。
c.輸出映射：node_2 將子圖返回的??response["bar"]?? 映射回父圖的??foo??。
- 可觀測性：
a.??for chunk in graph.stream(initial_input, subgraphs=True)?? 會展開子圖內部的節點事件，便于調試與排錯。
運行與驗證（不同狀態）
- 運行方式：與上一個示例相同。
- 期望輸出：流式事件中會出現帶有子圖節點名的條目（如 subgraph_node_1、subgraph_node_2），且最終父圖 foo 被子圖計算結果覆蓋。
- 關注點：
a.通過 invoke 模式，父圖對子圖的輸入/輸出擁有完全控制權，邊界清晰，利于團隊協作與版本演進。
對于更復雜的系統，您可能希望定義與父圖具有完全不同模式（沒有共享鍵）的子圖。例如，您可能希望為多智能體系統中的每個智能體保留私有的消息歷史記錄。
如果您的應用程序屬于這種情況，您需要定義一個調用子圖的節點函數。此函數需要在調用子圖之前將輸入（父）狀態轉換為子圖狀態，并在從節點返回狀態更新之前將結果轉換回父狀態。
from typing_extensions import TypedDict
from langgraph.graph.state import StateGraph, START
# Define subgraph
class SubgraphState(TypedDict):
# note that none of these keys are shared with the parent graph state
bar: str
baz: str
def subgraph_node_1(state: SubgraphState):
print("[子圖] subgraph_node_1 輸入狀態:", state)
result = {"baz": "baz"}
print("[子圖] subgraph_node_1 輸出更新:", result)
return result
def subgraph_node_2(state: SubgraphState):
print("[子圖] subgraph_node_2 輸入狀態:", state)
result = {"bar": state["bar"] + state["baz"]}
print("[子圖] subgraph_node_2 輸出更新:", result)
return result
subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_node(subgraph_node_2)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
subgraph = subgraph_builder.compile()
print("? 子圖編譯完成: subgraph")
# Define parent graph
class ParentState(TypedDict):
foo: str
def node_1(state: ParentState):
print("[父圖] node_1 輸入狀態:", state)
result = {"foo": "hi! " + state["foo"]}
print("[父圖] node_1 輸出更新:", result)
return result
def node_2(state: ParentState):
print("[父圖] node_2 輸入狀態:", state)
subgraph_input = {"bar": state["foo"]}
print("[父圖] node_2 調用子圖 subgraph.invoke 輸入:", subgraph_input)
response = subgraph.invoke(subgraph_input)
print("[父圖] node_2 收到子圖響應:", response)
result = {"foo": response["bar"]}
print("[父圖] node_2 輸出更新:", result)
return result
builder = StateGraph(ParentState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
graph = builder.compile()
print("? 父圖編譯完成: graph")
print("=== 開始執行圖 ===")
initial_input = {"foo": "foo"}
print("初始輸入:", initial_input)
for chunk in graph.stream(initial_input, subgraphs=True):
print("[流事件] 節點輸出:", chunk)
print("=== 執行結束 ===")輸出結果
? 子圖編譯完成: subgraph
? 父圖編譯完成: graph
=== 開始執行圖 ===
初始輸入: {'foo': 'foo'}
[父圖] node_1 輸入狀態: {'foo': 'foo'}
[父圖] node_1 輸出更新: {'foo': 'hi! foo'}
[流事件] 節點輸出: ((), {'node_1': {'foo': 'hi! foo'}})
[父圖] node_2 輸入狀態: {'foo': 'hi! foo'}
[父圖] node_2 調用子圖 subgraph.invoke 輸入: {'bar': 'hi! foo'}
[子圖] subgraph_node_1 輸入狀態: {'bar': 'hi! foo'}
[子圖] subgraph_node_1 輸出更新: {'baz': 'baz'}
[子圖] subgraph_node_2 輸入狀態: {'bar': 'hi! foo', 'baz': 'baz'}
[子圖] subgraph_node_2 輸出更新: {'bar': 'hi! foobaz'}
[父圖] node_2 收到子圖響應: {'bar': 'hi! foobaz', 'baz': 'baz'}
[父圖] node_2 輸出更新: {'foo': 'hi! foobaz'}
[流事件] 節點輸出: (('node_2:7f700b62-93c9-f5aa-b232-57b5eb87953a',), {'subgraph_node_1': {'baz': 'baz'}})
[流事件] 節點輸出: (('node_2:7f700b62-93c9-f5aa-b232-57b5eb87953a',), {'subgraph_node_2': {'bar': 'hi! foobaz'}})
[流事件] 節點輸出: ((), {'node_2': {'foo': 'hi! foobaz'}})
=== 執行結束 ===SubGraph 還能怎么玩
除了封裝模塊，Subgraph 還能解鎖一些其他操作：
- 并行執行：把三個獨立任務（比如同時查用戶畫像、訂單記錄、活動規則）各自做成子圖，主圖里用?
?START → [A, B, C] → END?? 并行調用。比在主圖里硬塞三個并行節點清爽十倍。 - 遞歸調用：子圖里還能再嵌套子圖。比如“生成回復”子圖內部，又需要調用“敏感詞過濾”子圖。層級分明，像俄羅斯套娃，但邏輯反而更清晰。
- 獨立測試：寫個 pytest，直接喂數據給子圖編譯后的對象，秒出結果。不用跑完整個流程就能驗證模塊正確性 —— 這對 CI/CD 太友好了。
最佳實踐與工程化建議
- 明確狀態契約（Schema First）：在團隊協作時先約定子圖輸入/輸出模式，減少后續重構成本。
- 保持最小共享面：共享鍵越多，耦合越強。只有在確需貫通上下文時才共享；其他均通過映射注入/回傳。
- 統一日志前綴：為父圖與子圖日志分別加上“[父圖]/[子圖]”前綴，定位跨圖問題更高效。
- 可觀測性開關：開發態建議開啟 subgraphs=True 觀察內部事件；生產態根據成本與需求選擇性關閉或降采樣。
- 漸進式抽取：先在父圖內完成串聯，待邏輯穩定后再抽取為子圖，減少過度抽象帶來的返工。
常見問題
- 子圖更新不到父圖？
a.檢查是否為共享狀態鍵；若非共享模式，需要在父圖節點中把返回值顯式映射回父圖鍵。
- 子圖內部狀態“泄漏”？
a.確保子圖私有鍵未被誤設為共享；或在父圖側僅接收需要的輸出字段。
- 流事件沒有展示子圖細節？
a.運行時啟用 subgraphs=True；若仍無事件，檢查子圖是否以節點方式被直接掛載或以 invoke 方式調用。
- 多子圖之間如何通信？
a.通過父圖中轉：要么共享鍵、要么在父圖節點內做顯式映射，避免子圖間彼此耦合。
總結
LangGraph 的 Subgraph，表面上是技術特性，骨子里是工程思維。它逼著你把“能跑就行”的代碼，重構為“能維護、能協作、能擴展”的系統。我見過太多人（包括半年前的我）把 LangGraph 當高級版 if-else 用，結果項目越做越重，最后只能推倒重來。
如果你也在用 LangGraph，還沒用 Subgraph —— 別等了，盡快動手嘗試下。
本文轉載自??????AI 博物院?????? 作者：longyunfeigu
