AI 教學 17 min read

LangGraph 核心:State / Node / Edge 完整教學(第一個會跑的 graph)

LangGraph 的核心只有三個概念:State(跨步驟共用的狀態)、Node(收 state 回傳要改的欄位的函式)、Edge(把 node 串起來的線)。這篇用血脂決策 pipeline 的真實場景,帶你從 StateGraph 定義、add_node、add_edge 一路到 compile 與 invoke,給一個複製貼上就能跑的最小範例,並講三個新手最常踩的坑:回傳整份 state、忘記接 START/END、在 node 裡 mutate state。「用真實專案學 LangGraph」系列 EP2。

#LangGraph #LangGraph 教學 #StateGraph +4 tags
LangGraph 核心:State / Node / Edge 完整教學(第一個會跑的 graph)
本文目錄 · 10
TL;DR：LangGraph 的核心只有三個概念——State（一份跨步驟共用的狀態）、Node（一個收 state、回傳「我要改哪些欄位」的函式）、Edge（把 node 串起來的線）。這篇用一個血脂決策 pipeline 的真實場景，帶你寫出第一個會跑的 graph：StateGraph 定義 → add_node → add_edge → compile → invoke。看完你就抓得住 LangGraph 的心智模型，後面分流跟重試都是在這之上加東西。這是「用真實專案學 LangGraph」系列 EP2。

上一篇（EP1）講完選型，結論是：複雜流程下沉到 LangGraph，而學 LangGraph 先學三件事——state、node、edge。這篇就把這三件事一次教完，而且直接給你一個跑得起來的最小範例。

📌 目錄

先建立心智模型：graph 就是一條流水線

別被「graph」這個字嚇到。對 9 成的應用來說，LangGraph 的圖就是一條流水線：資料從頭進來，一站一站處理，最後出去。

舉個真實場景：一個血脂用藥決策的 pipeline。一筆病人資料進來，要經過這幾站：

開始

補齊衍生指標
BMI / eGFR / LDL

跑公式
判斷風險

載入決策規則

輸出建議

每一站就是一個 node。站跟站之間的箭頭就是 edge。而那筆「一路被加工的病人資料」，就是 state——它從頭流到尾，每一站往上面寫一點東西。

抓住這個畫面，剩下的都是把它翻成 code。

🧠 State：跨步驟共用的那份狀態

State 是整張圖的 single source of truth。所有 node 讀它、寫它。在 LangGraph 裡，最常見的定義方式是 Python 的 TypedDict：

from typing import TypedDict, Any

class AnalyzeState(TypedDict, total=False):
    """跨 node 傳遞的狀態。total=False = 每個欄位都可選填。"""


patient: dict[str, Any]      # 病人資料（會被第一站 enrich）
    computed: dict[str, Any]     # 公式算出來的事實
    active_rule: dict[str, Any]  # 載入的決策規則
    advice: str                  # 最終建議

兩個關鍵設計：

  • total=False：代表每個 key 都是可選的。第一站跑完只有 patient，第二站才填 computed，不會因為「現在還沒有 advice」就報錯。
  • 欄位就是合約：哪個 node 負責寫哪個欄位，要講清楚。這是團隊協作時最重要的紀律——patient 由第一站寫，computed 由第二站寫，其他站只讀不改。
進階：state 的欄位可以用更嚴格的型別（像 Pydantic model 配 extra="forbid"），讓「寫錯欄位」當場炸出來，而不是靜悄悄帶著髒資料往下走。新手先用 TypedDict 就好。

⚡ Node：收 state、回傳要改的欄位

這是 LangGraph 最容易誤解的地方，先把規則記死：

一個 node 就是一個函式，收整份 state，回傳一個 dict——dict 裡只放「我這站要改的欄位」。

def patient_loader(state: AnalyzeState) -> dict:
    """第一站：補齊衍生指標。"""
    patient = state["patient"]
    # 假裝這裡算了 BMI / eGFR / LDL
    enriched = {**patient, "bmi": 24.5, "ldl_c": 71}
    return {"patient": enriched}   # 只回傳我改的那個 key

注意 return {"patient": enriched}——它不是回傳整份 state，只回傳要更新的部分。LangGraph 會自動把這個 dict 合併進 state。你不用自己去 mutate state，也不該去 mutate（不可變原則，避免隱性 side effect）。

再寫第二、第三站，長得一模一樣：

def precompute(state: AnalyzeState) -> dict:
    """第二站：跑公式判斷風險。"""
    ldl = state["patient"]["ldl_c"]
    computed = {"ldl_on_target": ldl <= 55}  # 達標門檻
    return {"computed": computed}

def load_rule(state: AnalyzeState) -> dict:
    """第三站：根據風險載入規則 + 出建議。"""
    on_target = state["computed"]["ldl_on_target"]
    advice = "維持現用藥" if on_target else "建議加上 ezetimibe 強化降脂"
    return {"advice": advice}

每個 node 都是純函式：輸入 state、輸出「要改什麼」，不碰外面、不改傳進來的東西。這個紀律一守住，你的 graph 就好 debug、好測試、好擴充。

🔗 Edge：把 node 串成流程

有了 node，用 edge 把它們串起來。線性流程超簡單：

from langgraph.graph import StateGraph, START, END

g = StateGraph(AnalyzeState)


# 1. 註冊 node（給名字 + 對應函式）
g.add_node("patient_loader", patient_loader)
g.add_node("precompute", precompute)
g.add_node("load_rule", load_rule)


# 2. 用 edge 串起來：START → 第一站 → 第二站 → 第三站 → END
g.add_edge(START, "patient_loader")
g.add_edge("patient_loader", "precompute")
g.add_edge("precompute", "load_rule")
g.add_edge("load_rule", END)

START 跟 END 是 LangGraph 內建的兩個特殊節點，標記圖的入口跟出口。add_edge(A, B) 的意思是「A 跑完就跑 B」。

就這樣，一條線性流水線就接好了。下一篇 EP3 會講，當你需要「跑完第三站，視結果決定去哪」的時候，把 add_edge 換成 add_conditional_edges 就能分流。

🎯 compile 跟 invoke：讓 graph 跑起來

圖畫完要 compile() 才能跑：

app = g.compile()

compile() 回傳一個可執行的物件。然後 invoke() 丟初始 state 進去，它會跑完整條流水線，回傳最終 state：

initial = {"patient": {"name": "test", "ldl_c": 71}}
final = app.invoke(initial)

print(final["advice"])
# → 建議加上 ezetimibe 強化降脂

如果你的 node 裡有 async 操作（查 DB、打 API），用 await app.ainvoke(initial) 的 async 版本。production 服務幾乎都用 ainvoke。

一個實務小技巧：compile() 只需要做一次。在 web 服務裡，把它放在 module 載入時跑一次、存成全域變數，每個 request 重複用同一個 compiled graph，不要每次 request 都重 compile：

# module 載入時 compile 一次
_app = g.compile()

async def handle_request(patient: dict):
    return await _app.ainvoke({"patient": patient})

📝 完整可跑範例

把上面全部拼起來，這是一個你複製貼上就能跑的最小 LangGraph：

from typing import TypedDict, Any
from langgraph.graph import StateGraph, START, END

class AnalyzeState(TypedDict, total=False):
    patient: dict[str, Any]
    computed: dict[str, Any]
    advice: str


def patient_loader(state: AnalyzeState) -> dict:
    enriched = {**state["patient"], "bmi": 24.5}
    return {"patient": enriched}


def precompute(state: AnalyzeState) -> dict:
    ldl = state["patient"]["ldl_c"]
    return {"computed": {"ldl_on_target": ldl <= 55}}


def load_rule(state: AnalyzeState) -> dict:
    on_target = state["computed"]["ldl_on_target"]
    advice = "維持現用藥" if on_target else "建議加上 ezetimibe"
    return {"advice": advice}


g = StateGraph(AnalyzeState)
g.add_node("patient_loader", patient_loader)
g.add_node("precompute", precompute)
g.add_node("load_rule", load_rule)
g.add_edge(START, "patient_loader")
g.add_edge("patient_loader", "precompute")
g.add_edge("precompute", "load_rule")
g.add_edge("load_rule", END)


app = g.compile()


result = app.invoke({"patient": {"ldl_c": 71}})
print(result["advice"])   # 建議加上 ezetimibe

裝好 LangGraph 就能跑：

pip install langgraph
python first_graph.py

驗證成功的標準：終端機印出 建議加上 ezetimibe。把 ldl_c 改成 50 再跑一次，應該印出 維持現用藥——這證明你的 state 真的一路流過三站、每站都正確讀寫。

🔥 我踩過的坑

坑一：node 回傳整份 state，結果欄位被洗掉

新手最常見錯誤——以為要回傳完整 state：

# ❌ 錯：手動組整份 state，很容易漏欄位
def precompute(state):
    state["computed"] = {...}
    return state   # 看起來沒事，但養成壞習慣

正確做法是只回傳你要改的 key，讓 LangGraph 自己合併：

# ✓ 對：只回傳這站負責的欄位
def precompute(state):
    return {"computed": {...}}

回傳整份 state 在簡單情況下不會出錯，但一旦多個 node 並行、或用到 reducer（下一篇講），「回傳整份」就會把別人寫的東西洗掉。從第一天就養成「只回傳 diff」的習慣。

坑二：忘記接 START 或 END

add_node 只是「註冊」這個 node 存在，不代表它會被跑。你一定要用 edge 把它接進流程，而且整條流程的頭要接 START、尾要接 END。漏掉 START 那條 edge，compile 不會報錯，但 invoke 時 graph 不知道從哪開始，直接給你怪結果。

坑三：在 node 裡 mutate 傳進來的 state

def bad_node(state):
    state["patient"]["bmi"] = 24.5   # ❌ 直接改傳進來的物件
    return {}

這違反不可變原則。改傳進來的 dict 會造成難追的 side effect，尤其多 node 共讀同一份資料時。永遠建新物件：{**state["patient"], "bmi": 24.5}，回傳新的，不動原本的。

❓ 常見問題 FAQ

Q：State 一定要用 TypedDict 嗎？
A：不一定，但建議。LangGraph 支援 TypedDict、Pydantic model、dataclass。新手用 TypedDict 最直觀；要嚴格驗證欄位（防打錯字）就上 Pydantic。重點是 state 的欄位要當成合約寫清楚，用哪種型別是次要的。

Q：node 一定要是函式嗎？可以是 class 嗎？
A：node 本質是「可呼叫物件」（callable），函式最常見，但任何 __call__ 接收 state、回傳 dict 的物件都行。複雜 node 要帶設定或依賴時，用 class 包起來會比較乾淨。

Q：一個 node 可以同時改多個欄位嗎？
A：可以。回傳的 dict 想放幾個 key 都行，例如 return {"computed": ..., "advice": ...}。只要那些 key 是這個 node 的職責就好。

Q：graph 跑到一半出錯會怎樣？
A：預設 node raise exception 就整條 graph 中斷、把 exception 往外丟。你可以在 node 內自己 try/except 處理，或用 LangGraph 的 durable state（1.0 強化的功能）讓流程能從中斷點接回。新手先讓它 raise，把錯誤暴露出來比較好 debug。

Q：怎麼看 graph 實際跑了哪些 node？
A：用 app.stream(initial) 取代 invoke，它會 yield 每一步的中間結果，你能看到 state 一站一站怎麼變。debug 流程超好用。

🔗 延伸資源

author
陳
陳彥彤

AI 工程師 · AI 顧問。Java 後端 8 年、AI 工程師 2 年。AI 內訓 · AI 導入顧問 · 前後端與雲端培訓。

support

覺得文章有用可以到 GitHub 給個 star，或是透過信箱聊聊 AI 內訓、AI 導入顧問或前後端 / 雲端培訓。

email github
related

相關文章

[AI 教學] · 16min
LangGraph Conditional Edge + Reducer:讓 graph 會分流、會重試
線性 graph 不夠用時你需要兩個東西:Conditional Edge(跑完一個 node 看結果決定下一步去哪——過了往前、沒過退回重試、重試太多次直接退件)跟 Reducer(讓某個 state 欄位累加而不是覆蓋,最典型就是 retry 計數器)。這篇用 add_conditional_edges + router 函式 + Annotated[int, operator.add] 把這兩件事講透,並解釋一個關鍵設計決策:MAX_RETRY 該設多少、為什麼很多嚴肅場景答案是 0。「用真實專案學 LangGraph」系列 EP3。
[AI 教學] · 13min
LangChain vs LangGraph 是什麼?2026 該學哪個(完整比較)
LangChain 跟 LangGraph 不是二選一,是上下層:LangGraph 是底層 runtime(管 state、流程、重試),LangChain 是建在它上面的高階 API。2025-10-22 兩者都上 1.0 正式版。這篇用比較表、三個情境判斷、三個真實踩坑,講清楚標準 agent loop 該用 LangChain、要精細控制流程分流與重試該用 LangGraph,以及完全新手該從哪個開始學。「用真實專案學 LangGraph」系列第一篇。
[AI 工具] · 12min
我原本用 Hermes Agent CLI,後來為什麼換成 LangGraph(遷移覆盤)
系列番外。我這套醫療決策 pipeline 最早不是用 LangGraph,是用 Hermes Agent CLI(NousResearch 2026 年初開源的自主 agent 框架)。Hermes 很強,但強在通用對話助理、會自我學習、接 20+ 平台;而我要的是確定性步驟為主、流程可畫成圖、每步可審計、出錯能精準退件的決策管線。方向不對再強也是錯的工具。這篇覆盤這個選型決策:Hermes 哪裡好、為什麼對我的場景不對、換到 LangGraph 解決了什麼。「用真實專案學 LangGraph」系列 EP6(完結)。
← all posts