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 長什麼樣
- 八個步驟各做什麼
- 為什麼確定性步驟不該丟給 LLM
- Human-in-the-loop:讓 graph 停下來等人
- 為什麼 HITL 在 LangGraph 特別簡單
- 我踩過的坑
- 常見問題 FAQ
- 延伸資源
完整 pipeline 長什麼樣
一筆病人血脂資料進來,到一份可用的處置建議出去,中間我把它拆成八個 node:
看這張圖,你會發現它大部分是直線,只有一個分流點(步驟 6 的業務驗證)。這很正常——真實 pipeline 不需要到處都是 if-else,把分流留給真正需要決策的那一兩個點,其他用 add_edge 直連,流程才讀得懂。
🧠 八個步驟各做什麼
| # | 步驟 | 類型 | 做什麼 |
|---|---|---|---|
| 1 | 載入病人資料 | 確定性 | 把進來的原始資料補齊缺漏欄位,標準化格式 |
| 2 | 補衍生指標 / 跑公式 | 確定性 | 算 BMI、跑風險評分公式——這些是數學,不該丟給 LLM |
| 3 | 載入決策規則 | 確定性 | 根據病人類型載入對應的臨床指引規則 |
| 4 | 組 few-shot 範例 | 確定性 | 挑 2 正 1 反的範例塞進待會的 prompt(EP4 講過) |
| 5 | LLM 出結構化建議 | AI | 把前面備好的料丟給 LLM,拿 Pydantic 結構化輸出(EP4) |
| 6 | 業務驗證 | 確定性 | 用規則檢查 LLM 輸出合不合理,分流 ok/retry/reject(EP3) |
| 7 | 適配輸出格式 | 確定性 | 把內部結構轉成上游 API 要的形狀 |
| 8 | 持久化 | 確定性 | 存決策紀錄,供稽核 |
🎯 為什麼確定性步驟不該丟給 LLM
這是設計上最重要的判斷,新手最容易做錯:能用程式算的,就不要叫 LLM 算。
步驟 2 的 BMI、風險評分,是固定公式。叫 LLM 算它:
- 可能算錯——LLM 數學不可靠,尤其多步驟計算
- 不可審計——它怎麼算出來的你不知道,醫療場景要能交代每個數字的來源
- 慢又貴——一次 LLM 呼叫的延遲跟成本,去算一個
weight / height²完全不划算 - 不穩定——同樣輸入可能給不同結果
⚙️ 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 把這整套包掉了。你只要:
interrupt()Command(resume=...) 餵回中間「狀態存哪、卡在哪、怎麼接回」全是 LangGraph 用它的 durable state 機制處理掉。這就是 EP1 比較表裡「durable state、server 重啟也接得回」那一格的實戰意義——它讓「跑很久、中途要等人、可能被打斷」的長流程變得可行。這也是為什麼 EP1 我說「要 human-in-the-loop 就下沉到 LangGraph」。
🔥 我踩過的坑
坑一:HITL 沒掛 checkpointer
interrupt() 要能「停下來再接回」,靠的是 checkpointer 把狀態存起來。compile 時忘了傳 checkpointer,interrupt() 會行為異常或直接報錯——因為沒有地方存「卡在哪」。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。把骨架學起來,肉自己填。
🔗 延伸資源
- LangGraph 官方 repo(langchain-ai/langgraph) —
interrupt/ checkpointer / durable state - LangGraph human-in-the-loop 官方文件 — HITL 完整模式
- LangGraph 1.0 GA 公告 — durable execution 的官方說明
- 上一篇:EP4 — LLM node:結構化輸出 + few-shot + 業務驗證
- 下一篇:EP6 — 番外:我原本用 Hermes Agent CLI,後來為什麼換成 LangGraph