LangGraph 架构拆解

繁體中文

简体中文
English
日本語
Español
Français
Deutsch

繁體中文

简体中文
English
日本語
Español
Français
Deutsch

Appearance

BaseChannel 抽象:通道即狀態原語 ​

源码版本1.2.9

職責 ​

在 LangGraph 的執行時裡,「通道 (channel)」是狀態 (state) 的最小單元。你寫一個 StateGraph 編譯出來的 Pregel 實例,內部並不直接持有使用者的 dict,而是持有一組 BaseChannel 實例——每個通道一個 key,通道自己負責存值、合併寫、對外暴露讀、還能把當前狀態序列化成檢查點 (checkpoint) 形式。BaseChannel 就是所有通道類別共用的抽象基類 (BaseChannel:19)。

它本身不存資料(__slots__ 裡只有 key 和 typ,見 __slots__:22-26)。它只規定四個抽象方法子類必須實作:ValueType / UpdateType 兩個型別屬性,以及 update / get / from_checkpoint 三個動作 (abstract methods:60-99)。另外有 checkpoint / is_available / consume / finish / copy 五個帶預設實作的方法,預設實作要嘛走 get() 兜底,要嘛回傳 False 當 no-op。

換句話說,BaseChannel 把「狀態怎麼存、怎麼變、怎麼留痕」這三件事一次性釘成介面,Pregel 引擎只需要呼叫 update / get / checkpoint 這些方法,不需要知道具體是 LastValue 還是 Topic 還是 BinaryOperatorAggregate 在背後怎麼做合併。這就是為什麼引擎和狀態形狀可以徹底解耦。

設計動機 ​

為什麼把狀態做成「通道」這種物件,而不是直接拿一個 dict 在節點間傳?

  • 統一寫合併語意:同一個超步 (superstep) 內,多個節點可能並行往同一個 key 寫。dict 沒法表達「list 要追加,純量要覆蓋,計數器要相加」這幾套不同的合併規則;通道用一個 update(values: Sequence[Update]) -> bool 把策略收口,子類自己決定怎麼 fold。
  • 不可變步內視圖:節點讀到的是步開始時的快照,本步別的節點寫的值要等下一步才可見。get 是唯讀當前值,update 只在步邊界由 apply_writes 統一呼叫 (apply_writes:317-323),從機制上杜絕了競爭。
  • 可序列化:每個通道都能 checkpoint() 出一個可序列化值(checkpoint:49-58)和 from_checkpoint 重建自己 (from_checkpoint:60-65)。BaseCheckpointSaver 族就是靠這倆介面把整個執行緒狀態寫進 SQLite/Postgres 的。
  • 生命週期鉤子:consume 和 finish 兩個方法(consume / finish:101-121)給特殊通道(比如 Topic(accumulate=False) 或 LastValueAfterFinish)一個在步邊界或執行結尾改自己狀態的機會,預設 no-op 表示大多數通道不需要這層語意。
  • 型別自描述:ValueType / UpdateType 兩個屬性(ValueType / UpdateType:28-36)讓編譯期和執行期都能反查「這個通道存什麼、接受什麼寫」,StateGraph 在 _is_field_channel:1862-1887 裡就用它來識別 Annotated[..., SomeChannel] 這種宣告。

關鍵檔案 ​

  • BaseChannel class:19-26 — 抽象基類定義,泛型參數 Value / Update / Checkpoint,__slots__ 只宣告 key 和 typ。
  • ValueType / UpdateType:28-36 — 兩個抽象 property,子類用來宣告存值型別和寫值型別。
  • checkpoint:49-58 — 預設實作:直接回傳 self.get(),空通道回傳 MISSING。
  • from_checkpoint:60-65 — 抽象方法:從一個 checkpoint 值構造一個等價的新通道,子類必須實作。
  • get:69-73 — 抽象讀介面,空通道拋 EmptyChannelError。
  • is_available:75-85 — 預設實作走 get() + 捕獲 EmptyChannelError,子類一般覆寫成更快的判斷。
  • update:89-99 — 抽象寫介面,Pregel 在每個超步結尾呼叫一次,順序任意;回傳 True 表示通道被改動。
  • consume / finish:101-121 — 生命週期鉤子,預設 no-op 回傳 False。
  • apply_writes:315-345 — Pregel 把本步所有任務寫回合併到通道的唯一入口,update / consume / finish 都在這段裡被統一排程。
  • _is_field_channel:1862-1887 — StateGraph 編譯時用 isinstance(item, BaseChannel) 把 Annotated[..., channel] 翻譯成 channel 實例。

資料流 ​

通道的寫發生在 apply_writes 裡,Pregel 把本步所有任務的 writes 按通道聚合,然後呼叫 update(apply_writes:317-323):

python
# Apply writes to channels
updated_channels: set[str] = set()
for chan, vals in pending_writes_by_channel.items():
    if chan in channels:
        if channels[chan].update(vals) and next_version is not None:
            checkpoint["channel_versions"][chan] = next_version
            # unavailable channels can't trigger tasks, so don't add them
            if channels[chan].is_available():
                updated_channels.add(chan)

# Channels that weren't updated in this step are notified of a new step
if bump_step:
    for chan in channels:
        if channels[chan].is_available() and chan not in updated_channels:
            if channels[chan].update(EMPTY_SEQ) and next_version is not None:
                checkpoint["channel_versions"][chan] = next_version
                if channels[chan].is_available():
                    updated_channels.add(chan)

注意第二段:本步沒被任何任務寫的通道也會收到一次 update(EMPTY_SEQ) 呼叫。這是 BaseChannel 介面的隱含契約——子類的 update 必須能接受空序列,通常回傳 False 表示「沒改動」。Topic(accumulate=False) 這種「每步清空」語意就是靠這個空呼叫觸發的:空 update 讓它把上一步的值清掉,從而在下一步不可見。

整個通道生命週期在 Pregel 一輪 tick 裡的位置如下:

邊界與失敗 ​

  • EmptyChannelError:get() 在通道從未被寫過時拋 (get:70-73)。is_available 預設實作就靠 catch 這個例外判斷可用性,子類若能更便宜地判斷應覆寫。
  • MISSING 哨兵:checkpoint() 在 get() 拋 EmptyChannelError 時回傳 langgraph._internal._typing.MISSING 而不是 None (checkpoint fallback:55-58),這樣 None 可以作為合法存值出現,不與「空」混淆。
  • update 回傳 False:回傳 False 表示通道沒改動,Pregel 就不會更新 channel_versions,後續 prepare_next_tasks 也就不會因為這條通道觸發任何節點 (update call:319);這是 Pregel 收斂判斷的基礎。
  • 空序列呼叫:apply_writes 對未變更通道也會呼叫 update(EMPTY_SEQ)(EMPTY_SEQ update:329)。BaseChannel.update 的契約是空輸入回傳 False,但 Topic(accumulate=False) 會在此時清空自己並可能回傳 True,所以子類不能假設「空輸入 = 啥也不做」。
  • consume 與 finish 預設 no-op(consume / finish:101-121):回傳 False 意味著大多數通道不參與生命週期管理。apply_writes 在 bump_step 結尾會呼叫 finish(finish call:338),只有 LastValueAfterFinish 這類特殊通道會用它做「等執行結束才可見」的語意。
  • copy 預設走 checkpoint:BaseChannel.copy 預設 self.from_checkpoint(self.checkpoint())(copy:40-47)。子類如果 checkpoint 代價高(比如要深拷貝大 list),應該覆寫 copy 提供更便宜的實作,Topic.copy 就是這麼做的。

小結 ​

BaseChannel 把「狀態怎麼存、怎麼合併、怎麼序列化」一次性釘成介面,Pregel 引擎只需要按 update → is_available → get → checkpoint → finish 這套固定順序呼叫,就能跑在任意通道實作上。要理解具體實作,看 /channels/last-value 和 /channels/topic-binop;通道被 Pregel 讀寫的過程見 /pregel/algo;檢查點如何把 checkpoint() 的回傳值持久化見 /checkpoint/base-saver。

對照官方資料:LangGraph 文件 · README