iT邦幫忙

2026 iThome 鐵人賽

DAY 3
0
Vibe Coding

agent工作流系列 第 14

【Day 14】LangGraph Human-in-the-Loop:斷點中斷、人工審批與狀態修改實戰

  • 分享至 

  • xImage
  •  

在前面的章節中,我們實現了單一 ReAct Agent 與 Supervisor 多智能體協同系統。這些系統雖然具備高度自主性,但在生產級企業應用中,「完全放手讓 Agent 自主執行」往往伴隨著巨大的業務風險:

  • 如果 Agent 誤判意圖,自動向 10,000 名客戶發送了錯誤的折扣通知怎麼辦?
  • 如果 Agent 即將執行一筆 50 萬元的資料庫退款操作,誰來做最後把關?
  • 如果 Agent 在長程任務中理解偏差,人類能否在「中途」修正它的狀態,而不是讓它重頭來過?

解決這些問題的核心機制,就是 Human-in-the-Loop(人機協同,簡稱 HITL)

今天我們將深入探討:如何利用 LangGraph 內建的持久化檢查點(Checkpointer)與中斷機制(Interrupts),在關鍵節點安全暫停、交由人類審批,並動態修改 State 後繼續執行!


一、Human-in-the-Loop 的三大核心模式

在 Agent 系統中,人類介入主要有三種形態:

1. 審批確認 (Approval / Gatekeeper)
   [Agent 規劃操作] ──> [ 暫停中斷 ⏸️ ] ──> 人類點擊 [核准/拒絕] ──> [執行動作]

2. 狀態干預與編輯 (State Editing)
   [Agent 產生草稿] ──> [ 暫停中斷 ⏸️ ] ──> 人類手動修改草稿 ──> [Agent 基於修改後內容繼續]

3. 追問與輸入補充 (Input Clarification)
   [Agent 資訊不足] ──> [ 暫停中斷 ⏸️ ] ──> 人類補充必要資訊 ──> [Agent 恢復執行]

LangGraph 實現 HITL 的底層技術依賴於 Checkpointer(狀態快照持久化)。當設定了中斷點時,圖會在執行到該節點前後將整個 State 序列化存入資料庫,並將執行緒(Thread)掛起。


二、實戰場景:電商退款審批狀態圖設計

我們要建立一個退款審核工作流:

  1. triage_node:LLM 根據使用者請求評估退款金額。
  2. human_review_node(中斷閘口):若退款金額超過 1,000 元,觸發中斷掛起,等待人工審核。
  3. execute_refund_node:審核通過後,執行實際退款與資料庫寫入。
       ┌───────────────┐
       │   __start__   │
       └───────┬───────┘
               │
               ▼
       ┌───────────────┐
       │  triage_node  │ (評估退款金額)
       └───────┬───────┘
               │ (條件邊: 金額 > 1000 元?)
        ├── 超過 ──> ⏸️ [ 中斷掛起: 等待人工審查 ]
        │                 │ (人工確認 / 修改狀態)
        │                 ▼
        │            ┌─────────────────────┐
        │            │  human_review_node  │
        │            └──────────┬──────────┘
        │                       │
        └── 未超過 ─────────────┴──> ┌─────────────────────────┐
                                    │   execute_refund_node   │
                                    └────────────┬────────────┘
                                                 │
                                                 ▼
                                          ┌─────────────┐
                                          │   __end__   │
                                          └─────────────┘

三、步驟 1:定義狀態與審核資料結構

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]

四、步驟 2:定義節點邏輯與條件路由

# 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}

五、步驟 3:組裝狀態圖並設定中斷點(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"]  # 關鍵:在進入此節點前中斷
)

六、步驟 4:執行工作流、觸發中斷與手動恢復

我們透過一個金額為 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 元

七、步驟 5:人工干預——修改 State 並恢復執行(State Update)

主管登入後台,審查發現商品瑕疵屬實,但決定與用戶協議給予「折讓退款 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 元。

八、生產環境 Human-in-the-Loop 工程實踐矩陣

設計要點 潛在問題 最佳解決方案
中斷持久化 記憶體儲存重啟後遺失掛起任務 生產環境使用 PostgresSaverRedisSaver 取代 MemorySaver
超時處置 人工遲遲未審批導致流程永久卡死 設置定時任務(Cron / Celery),超過 24 小時未審核自動發送催辦提醒或執行回退
權限隔離 任意使用者可能呼叫 API 偽造核准 在調用 update_state 前,外層 API 必須通過嚴格的 RBAC 角色權限校驗
稽核日誌 無法追溯是誰在何時修改了狀態 每次 update_state 同步寫入 Audit Log 資料庫,記錄操作者 ID 與修改前後 Diff

小結與下集預告

透過 LangGraph 的 interrupt_beforeupdate_state,我們成功建立了人機協同的安全護欄。Agent 不再是不受控制的野馬,而是能隨時在中斷點優雅掛起、接受人類審核與修正的可靠助理。

現在,我們已經完整涵蓋了 Prompt、Context、Structured Output、Tool Calling、LangGraph 狀態機、Multi-Agent 以及 Human-in-the-Loop。

明天 【Day 15】前半程總結與工程實務心法:構建可靠 Agent 的防禦性設計指南,我們將盤點前半程的核心技術拼圖,並梳理一份生產級 Agent 系統必備的架構架構與防禦性工程清單!


上一篇
【Day 13】實作 LangGraph Supervisor Multi-Agent 系統:構建研究與寫作的自動化團隊
下一篇
【Day 15】前半程總結與工程實務心法:構建可靠 Agent 的防禦性設計指南
系列文
agent工作流15
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言