在前面的章節中,我們實現了單一 ReAct Agent 與 Supervisor 多智能體協同系統。這些系統雖然具備高度自主性,但在生產級企業應用中,「完全放手讓 Agent 自主執行」往往伴隨著巨大的業務風險:
解決這些問題的核心機制,就是 Human-in-the-Loop(人機協同,簡稱 HITL)。
今天我們將深入探討:如何利用 LangGraph 內建的持久化檢查點(Checkpointer)與中斷機制(Interrupts),在關鍵節點安全暫停、交由人類審批,並動態修改 State 後繼續執行!
在 Agent 系統中,人類介入主要有三種形態:
1. 審批確認 (Approval / Gatekeeper)
[Agent 規劃操作] ──> [ 暫停中斷 ⏸️ ] ──> 人類點擊 [核准/拒絕] ──> [執行動作]
2. 狀態干預與編輯 (State Editing)
[Agent 產生草稿] ──> [ 暫停中斷 ⏸️ ] ──> 人類手動修改草稿 ──> [Agent 基於修改後內容繼續]
3. 追問與輸入補充 (Input Clarification)
[Agent 資訊不足] ──> [ 暫停中斷 ⏸️ ] ──> 人類補充必要資訊 ──> [Agent 恢復執行]
LangGraph 實現 HITL 的底層技術依賴於 Checkpointer(狀態快照持久化)。當設定了中斷點時,圖會在執行到該節點前後將整個 State 序列化存入資料庫,並將執行緒(Thread)掛起。
我們要建立一個退款審核工作流:
triage_node:LLM 根據使用者請求評估退款金額。human_review_node(中斷閘口):若退款金額超過 1,000 元,觸發中斷掛起,等待人工審核。execute_refund_node:審核通過後,執行實際退款與資料庫寫入。 ┌───────────────┐
│ __start__ │
└───────┬───────┘
│
▼
┌───────────────┐
│ triage_node │ (評估退款金額)
└───────┬───────┘
│ (條件邊: 金額 > 1000 元?)
├── 超過 ──> ⏸️ [ 中斷掛起: 等待人工審查 ]
│ │ (人工確認 / 修改狀態)
│ ▼
│ ┌─────────────────────┐
│ │ human_review_node │
│ └──────────┬──────────┘
│ │
└── 未超過 ─────────────┴──> ┌─────────────────────────┐
│ execute_refund_node │
└────────────┬────────────┘
│
▼
┌─────────────┐
│ __end__ │
└─────────────┘
from typing import Optional, Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
# 定義退款工作流全域狀態
class RefundState(TypedDict):
order_id: str
amount: float
reason: str
review_status: Literal["PENDING", "APPROVED", "REJECTED"]
review_comment: Optional[str]
refund_result: Optional[str]
# 1. 意圖與金額分流節點
def triage_node(state: RefundState):
print(f"\n[系統] 收到退款申請:訂單 {state['order_id']},金額: {state['amount']} 元")
return {"review_status": "PENDING"}
# 2. 條件路由判斷:高額退款需人工介入
def route_refund(state: RefundState) -> str:
if state["amount"] > 1000:
return "human_review"
return "execute_refund"
# 3. 人工審查節點(作為中斷錨點)
def human_review_node(state: RefundState):
print(f"[系統] 進入人工審查節點,當前審核狀態: {state['review_status']}")
return {}
# 4. 實際退款執行節點
def execute_refund_node(state: RefundState):
if state["review_status"] == "REJECTED":
result = f"退款申請已拒絕。原因: {state.get('review_comment', '未符合退款標準')}"
else:
result = f"退款成功!已向訂單 {state['order_id']} 撥付 {state['amount']} 元。"
print(f"[系統] 執行結果 ➔ {result}")
return {"refund_result": result}
interrupt_before)我們在編譯狀態圖時,利用 interrupt_before=["human_review"] 明確指定中斷位置:
# 1. 初始化狀態圖
workflow = StateGraph(RefundState)
# 2. 註冊節點
workflow.add_node("triage", triage_node)
workflow.add_node("human_review", human_review_node)
workflow.add_node("execute_refund", execute_refund_node)
# 3. 建立連線
workflow.add_edge(START, "triage")
workflow.add_conditional_edges(
"triage",
route_refund,
{
"human_review": "human_review",
"execute_refund": "execute_refund"
}
)
workflow.add_edge("human_review", "execute_refund")
workflow.add_edge("execute_refund", END)
# 4. 掛載 MemorySaver,並設定在中斷點前暫停掛起
memory = MemorySaver()
app = workflow.compile(
checkpointer=memory,
interrupt_before=["human_review"] # 關鍵:在進入此節點前中斷
)
我們透過一個金額為 3,500 元的訂單,展示「暫停 ➔ 人工介入 ➔ 恢復執行」的完整生命週期:
# 設定會話 Thread ID
config = {"configurable": {"thread_id": "refund_case_999"}}
# 1. 提交初始申請(觸發執行直到中斷)
initial_input = {
"order_id": "ORD-8877",
"amount": 3500.0,
"reason": "商品有瑕疵要求退款",
"review_status": "PENDING"
}
print("=== 步驟 1: 使用者發起退款申請 ===")
for event in app.stream(initial_input, config=config):
print(event)
# 2. 檢查當前狀態:工作流已在中斷點暫停
snapshot = app.get_state(config)
print("\n=== 步驟 2: 檢查系統暫停狀態 ===")
print(f"下一個待執行節點: {snapshot.next}") # ('human_review',)
print(f"當前 State 金額: {snapshot.values['amount']} 元")
執行後輸出:
=== 步驟 1: 使用者發起退款申請 ===
[系統] 收到退款申請:訂單 ORD-8877,金額: 3500.0 元
{'triage': {'review_status': 'PENDING'}}
=== 步驟 2: 檢查系統暫停狀態 ===
下一個待執行節點: ('human_review',)
當前 State 金額: 3500.0 元
主管登入後台,審查發現商品瑕疵屬實,但決定與用戶協議給予「折讓退款 3,000 元」並核准:
print("\n=== 步驟 3: 人工作業——修改狀態並核准 ===")
# 人工動態更新狀態:修改金額為 3000,狀態改為 APPROVED
app.update_state(
config,
{
"amount": 3000.0,
"review_status": "APPROVED",
"review_comment": "經客服電聯確認,客戶同意部分瑕疵折讓退款 3000 元"
},
as_node="human_review" # 指定以 human_review 節點的身份寫入更新
)
# 傳入 None 恢復執行工作流剩餘部分
print("\n=== 步驟 4: 恢復工作流執行 ===")
for event in app.stream(None, config=config):
print(event)
# 查看最終完成狀態
final_state = app.get_state(config)
print("\n最終 State 結果:")
print(f"最終撥付金額: {final_state.values['amount']}")
print(f"最終結果: {final_state.values['refund_result']}")
最終輸出:
=== 步驟 3: 人工作業——修改狀態並核准 ===
=== 步驟 4: 恢復工作流執行 ===
[系統] 進入人工審查節點,當前審核狀態: APPROVED
{'human_review': {}}
[系統] 執行結果 ➔ 退款成功!已向訂單 ORD-8877 撥付 3000.0 元。
{'execute_refund': {'refund_result': '退款成功!已向訂單 ORD-8877 撥付 3000.0 元。'}}
最終 State 結果:
最終撥付金額: 3000.0
最終結果: 退款成功!已向訂單 ORD-8877 撥付 3000.0 元。
| 設計要點 | 潛在問題 | 最佳解決方案 |
|---|---|---|
| 中斷持久化 | 記憶體儲存重啟後遺失掛起任務 | 生產環境使用 PostgresSaver 或 RedisSaver 取代 MemorySaver |
| 超時處置 | 人工遲遲未審批導致流程永久卡死 | 設置定時任務(Cron / Celery),超過 24 小時未審核自動發送催辦提醒或執行回退 |
| 權限隔離 | 任意使用者可能呼叫 API 偽造核准 | 在調用 update_state 前,外層 API 必須通過嚴格的 RBAC 角色權限校驗 |
| 稽核日誌 | 無法追溯是誰在何時修改了狀態 | 每次 update_state 同步寫入 Audit Log 資料庫,記錄操作者 ID 與修改前後 Diff |
透過 LangGraph 的 interrupt_before 與 update_state,我們成功建立了人機協同的安全護欄。Agent 不再是不受控制的野馬,而是能隨時在中斷點優雅掛起、接受人類審核與修正的可靠助理。
現在,我們已經完整涵蓋了 Prompt、Context、Structured Output、Tool Calling、LangGraph 狀態機、Multi-Agent 以及 Human-in-the-Loop。
明天 【Day 15】前半程總結與工程實務心法:構建可靠 Agent 的防禦性設計指南,我們將盤點前半程的核心技術拼圖,並梳理一份生產級 Agent 系統必備的架構架構與防禦性工程清單!