AI 教學 13 min read

LangGraph 完整 Pipeline + Human-in-the-loop:8 步決策管線全貌

前四篇的零件——state、node、edge、conditional edge、reducer、LLM 結構化輸出——這篇全部組起來,串成一條完整的醫療 AI 決策 pipeline:載入資料 → 補衍生指標 → 載入規則 → 組 few-shot → LLM 出建議 → 業務驗證 → 適配輸出 → 持久化,中間帶分流重試。然後加上 LangGraph 的殺手級功能 human-in-the-loop:用 interrupt() 在高風險決策點暫停整張 graph,等人點頭才繼續,靠 durable state 讓這件事變簡單。「用真實專案學 LangGraph」系列 EP5。

本文目錄 · 9
TL;DR:前四篇的零件——state、node、edge、conditional edge、reducer、LLM 結構化輸出——這篇全部組起來,串成一條完整的醫療 AI 決策 pipeline:載入資料 → 補衍生指標 → 載入規則 → 組 few-shot → LLM 出建議 → 業務驗證 → 適配輸出 → 持久化,中間帶分流重試。然後加上 LangGraph 的殺手級功能 human-in-the-loop(HITL):在高風險決策點用 interrupt() 暫停整張 graph,等人點頭才繼續。這篇講「真實 production pipeline 長什麼樣」跟「為什麼 LangGraph 的 durable state 讓 HITL 變得簡單」。系列 EP5。

EP4 把 LLM 接進單一 node。這篇是收網——把整套組起來,並加上「跑到一半停下來等人審核」這個只有 LangGraph 做得漂亮的能力。

📌 目錄

完整 pipeline 長什麼樣

一筆病人血脂資料進來,到一份可用的處置建議出去,中間我把它拆成八個 node:

ok

retry

reject

START

1 載入病人資料

2 補衍生指標 / 跑公式

3 載入決策規則

4 組 few-shot 範例

5 LLM 出結構化建議

6 業務驗證

7 適配輸出格式

END 退件

8 持久化

END

看這張圖,你會發現它大部分是直線,只有一個分流點(步驟 6 的業務驗證)。這很正常——真實 pipeline 不需要到處都是 if-else,把分流留給真正需要決策的那一兩個點,其他用 add_edge 直連,流程才讀得懂。

🧠 八個步驟各做什麼

#步驟類型做什麼
1載入病人資料確定性把進來的原始資料補齊缺漏欄位,標準化格式
2補衍生指標 / 跑公式確定性算 BMI、跑風險評分公式——這些是數學,不該丟給 LLM
3載入決策規則確定性根據病人類型載入對應的臨床指引規則
4組 few-shot 範例確定性挑 2 正 1 反的範例塞進待會的 prompt(EP4 講過)
5LLM 出結構化建議AI把前面備好的料丟給 LLM,拿 Pydantic 結構化輸出(EP4)
6業務驗證確定性用規則檢查 LLM 輸出合不合理,分流 ok/retry/reject(EP3)
7適配輸出格式確定性把內部結構轉成上游 API 要的形狀
8持久化確定性存決策紀錄,供稽核
注意八步裡只有一步(步驟 5)是 AI,其他七步全是確定性的。這正是 LangGraph 的甜蜜點——當你的流程是「一堆確定性步驟 + 幾個關鍵 AI 步驟」的混合,LangGraph 讓你用同一張 graph、同一份 state 把它們串起來,AI node 跟純算 node 平起平坐。

🎯 為什麼確定性步驟不該丟給 LLM

這是設計上最重要的判斷,新手最容易做錯:能用程式算的,就不要叫 LLM 算。

步驟 2 的 BMI、風險評分,是固定公式。叫 LLM 算它:

  • 可能算錯——LLM 數學不可靠,尤其多步驟計算
  • 不可審計——它怎麼算出來的你不知道,醫療場景要能交代每個數字的來源
  • 慢又貴——一次 LLM 呼叫的延遲跟成本,去算一個 weight / height² 完全不划算
  • 不穩定——同樣輸入可能給不同結果
把 LLM 用在它真正擅長的地方:需要綜合判斷、難以用規則窮舉的步驟(步驟 5 的處置建議)。其他能寫成公式、能寫成規則的,老老實實用 Python 算。LLM 是 pipeline 裡的一個 node,不是整個 pipeline。 這個觀念建立起來,你設計出來的系統才會可靠又省錢。

⚙️ Human-in-the-loop:interrupt() 讓 graph 停下來等人

醫療、金融這種高風險場景,AI 出的建議常常不能直接生效,要醫師/審核人員點頭。LangGraph 1.0 把這個能力做成第一級 API:interrupt()

在需要人工審核的 node 裡呼叫 interrupt(),整張 graph 會就地凍結,把當前狀態回傳給你,等外部把人的決定餵回來再從原地繼續:

from langgraph.types import interrupt, Command

def human_review(state: AnalyzeState) -> dict:
# 高風險建議:暫停,把建議丟出去給人看
decision = state["decision"]
human_input = interrupt({
"question": "請審核此處置建議是否採用",
"ai_suggestion": decision,
"reason": state["reason"],
})
# 人點頭後,graph 從這裡繼續
return {"approved": human_input["approved"], "reviewer": human_input["by"]}

呼叫端拿到中斷後,把人的決定用 Command(resume=...) 餵回去:

config = {"configurable": {"thread_id": "case-001"}}

# 第一次跑,跑到 interrupt 就停
result = app.invoke({"patient": patient}, config=config)
print(result["__interrupt__"]) # → 顯示要審核的建議

# ... 過了一小時,醫師在前端按了「採用」...

# 餵回決定,graph 從 interrupt 原地繼續,不是重跑
final = app.invoke(
Command(resume={"approved": True, "by": "Dr. Lin"}),
config=config,
)

把這個 human_review node 插在步驟 6(驗證過)跟步驟 7(適配輸出)之間,就得到一條「AI 出建議 → 規則驗證 → 人工審核 → 才正式輸出」的 pipeline。

🔥 為什麼 HITL 在 LangGraph 特別簡單

關鍵在 durable state(持久化狀態)

你想想,「暫停一個跑到一半的流程、等一小時後人來按確認、再從原地繼續」這件事,自己手刻有多痛:你要把當前所有變數狀態序列化存起來、要記住「卡在哪一步」、人回來時要把狀態反序列化、要跳回正確的位置繼續。這是一整套 state machine + 持久化機制。

LangGraph 把這整套包掉了。你只要:

  • compile 時掛一個 checkpointer(記憶體 / Postgres / Redis 都行)

  • 在要停的地方呼叫 interrupt()

  • 人回來時用 Command(resume=...) 餵回
  • 中間「狀態存哪、卡在哪、怎麼接回」全是 LangGraph 用它的 durable state 機制處理掉。這就是 EP1 比較表裡「durable state、server 重啟也接得回」那一格的實戰意義——它讓「跑很久、中途要等人、可能被打斷」的長流程變得可行。這也是為什麼 EP1 我說「要 human-in-the-loop 就下沉到 LangGraph」。

    🔥 我踩過的坑

    坑一:HITL 沒掛 checkpointer

    interrupt() 要能「停下來再接回」,靠的是 checkpointer 把狀態存起來。compile 時忘了傳 checkpointerinterrupt() 會行為異常或直接報錯——因為沒有地方存「卡在哪」。HITL 跟 checkpointer 是綁在一起的,要用前者必掛後者。

    坑二:把整條 pipeline 都做成 conditional edge

    新手學會 conditional edge 後容易上癮,每個 node 後面都接一個 router。結果 graph 變成一團義大利麵,沒人看得懂流程。大部分的線該用 add_edge 直連,conditional 只留給真正的決策點。我這條八步 pipeline 只有一個 conditional 點。

    坑三:thread_id 沒管好,HITL 接到別人的狀態

    durable state 是用 thread_id 區分不同的執行實例。如果你兩個 case 共用同一個 thread_id,resume 時會接到錯的狀態,建議跨 case 串線。每個獨立的執行(每個病人 case)要有自己唯一的 thread_id。

    ❓ 常見問題 FAQ

    Q:interrupt() 之後 graph 真的「停住」嗎?會占著資源嗎?
    A:不會占著 process。interrupt() 是把狀態存進 checkpointer 然後結束本次 invoke,process 就釋放了。等你之後用 Command(resume=...) 再次 invoke,它才從 checkpointer 把狀態載回來繼續。所以人等一小時,這一小時你的 server 沒有任何東西卡著。

    Q:八個 node 一定要拆這麼細嗎?
    A:不一定,看你的流程。拆細的好處是每個 node 職責單一、好測試、好除錯(哪一步出錯一目了然),而且 state 在 node 間的流動很清楚。但太細會增加樣板。原則:一個 node 做一件可以獨立描述的事,能獨立講清楚就值得拆。

    Q:checkpointer 用哪種?記憶體 vs Postgres?
    A:開發測試用記憶體版(MemorySaver)最快。Production 一定要用持久化的(Postgres / Redis),因為 server 重啟記憶體就沒了,HITL 等人那段時間若重啟就接不回。LangGraph 官方有提供 Postgres checkpointer。

    Q:HITL 一定要在前端做嗎?
    A:不一定。interrupt() 只是把「要審核的內容」吐出來,怎麼呈現給人、人怎麼回,是你應用層的事——可以是前端按鈕、可以是 Slack 訊息按 approve、可以是 email 回覆。LangGraph 只管「停 / 接回」,UI 你自己決定。

    Q:這條 pipeline 可以直接抄去用嗎?
    A:結構可以參考,但每一步的具體邏輯(規則、公式、prompt、驗證條件)一定是你自己領域的。這系列教的是 LangGraph 的「骨架怎麼搭」,骨架上的「肉」是你的 domain knowledge。把骨架學起來,肉自己填。

    🔗 延伸資源

    author
    陳彥彤

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

    support

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

    related

    相關文章

    [AI 工具] · 12min
    我原本用 Hermes Agent CLI,後來為什麼換成 LangGraph(遷移覆盤)
    系列番外。我這套醫療決策 pipeline 最早不是用 LangGraph,是用 Hermes Agent CLI(NousResearch 2026 年初開源的自主 agent 框架)。Hermes 很強,但強在通用對話助理、會自我學習、接 20+ 平台;而我要的是確定性步驟為主、流程可畫成圖、每步可審計、出錯能精準退件的決策管線。方向不對再強也是錯的工具。這篇覆盤這個選型決策:Hermes 哪裡好、為什麼對我的場景不對、換到 LangGraph 解決了什麼。「用真實專案學 LangGraph」系列 EP6(完結)。
    [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 教學] · 17min
    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。