LangGraph 狀態流動 — Reducer 與覆寫的差別
一個新手向的實驗,用來建立「狀態如何在 LangGraph StateGraph 中流動」的直覺。
1這個實驗在演示什麼問題
在 LangGraph 中,state schema 裡的每一個欄位,在 graph 編譯時都會變成一個 channel。節點(node)不會直接設定 state,而是回傳一份「部分更新」, 真正決定這份更新該怎麼處理的是 channel 本身。預設情況下,channel 只會用新值取代舊值 (最後寫入者獲勝)。如果你在欄位上標註了一個 reducer 函式,channel 就會改成把新值「合併」進既有值,而不是直接覆蓋。
「有沒有 reducer」這個單一設計決定,是初學 LangGraph 時最常見的「我的 state 怎麼不見了?」 困惑來源。這個實驗讓這件事變得可以直接觀察:讓兩個 channel 走過完全相同的三個節點, 並排比較它們的行為差異。
2架構與流程
一個單純的線性 graph,三個節點,沒有分支,也沒有任何 LLM 呼叫:
每個節點回傳的更新形狀完全相同:
{"last_note": "node_X ran", "results": ["x"]}
但這兩個欄位在 state schema 裡的宣告方式不同:
| Channel | 宣告方式 | 更新規則 |
|---|---|---|
last_note |
str(一般欄位) |
覆寫 — 新值取代舊值 |
results |
Annotated[list[str], accumulate_results] |
合併 — reducer 把新舊值合併起來 |
這個 graph 用 stream_mode="values" 串流執行,每個節點跑完都會印出完整的
state 字典——所以兩個 channel 的行為差異,在每一行輸出中都清楚可見。
3學習者要自己完成的部分
除了一個函式以外,其餘全部都已經搭建好:accumulate_results(existing, new)。
這個函式目前標記為 TODO: YOUR CORE,在填寫完成之前會拋出
NotImplementedError——在你實作它之前直接執行腳本本來就會失敗,
這個失敗是起點,不是程式錯誤。
任務目標:讓 results 依序累積每個節點的貢獻,同時不要動到
last_note、節點本身,或是 graph 的連線方式。
4操作步驟與可觀察的訊號
- 先完整讀過腳手架程式碼(
graph_lab.py),再開始動手。 - 在執行 graph 之前,先在每個 checkpoint 寫下你的預測。
- 在標記
TODO: YOUR CORE的地方實作 reducer。 - 執行
pip install -r requirements.txt && python graph_lab.py, 觀察每個節點執行後印出的 state。 - 確認:
last_note是不是只顯示最新一個節點的內容,而results是不是每個節點都多累積一筆?
5失敗/對照案例
這是整個實驗的核心,兩者在同一次執行中並排出現:
- 被蓋掉 —
last_note沒有 reducer, 所以每次印出的 state 只會看到最新那個節點留下的訊息,前面節點寫的內容一跑完下個節點 就消失了。 - 持續累積 —
results有 reducer, 所以每次印出的 state 都會看到清單在成長:node_a 後有一筆、node_b 後有兩筆、 node_c 後有三筆。
同一個 graph、同樣的三個節點、回傳值的形狀也一樣。「被蓋掉」跟「持續累積」唯一的差別,
就是 state schema 裡 results 欄位上那一個
Annotated[...] type hint。
6前置需求、成本、mock 替代方案
- 前置需求:Python 3.10+,以及
pip install langgraph。 不需要任何 API key、帳號,執行時也不會發出任何網路請求。 - 成本:無 — 這個 graph 完全不會呼叫模型或外部服務。
- mock 替代方案:不適用;這個實驗本來就設計成完全本地、
自我封閉(
local-mock執行模式)。
7選讀的加深方向
非必要,而且為了讓這個實驗保持精簡,故意沒有預先搭建好:
- Checkpoint/時間回溯 — 加上一個 checkpointer,
並回放
node_b當下的 state。 - Human-in-the-loop 中斷 — 在
node_c前加一個中斷點, 觀察執行到一半時results累積到什麼程度。