在前面幾篇的探討中，我們確立了高可靠 Agent 系統的核心觀念：「職責分離（AI 做 vs. 程式做）」——讓 AI 專注於非結構化語意理解，而由確定性的程式（Pydantic、State、Router）負責資料驗證與邊界守門。但當 Agent 的職責從「唯讀診斷」跨入具備副作用的「系統處置」時，就必須引入架構中的第三個關鍵維度 —— 「人做（Human-in-the-Loop，人工介入與審核）」。
直接讓 AI 拿著變更權限去碰正式環境（Production）是絕對的安全紅線。但在測試與 Staging 環境，系統常因測試殘留卡死，最適合用來安全驗證處置流程。維護時若直接重啟伺服器往往過於粗暴，容易中斷測試，實務上更傾向採用副作用較低的處置手段（例如強制清除閒置連線、隔離異常節點流量）。然而，即使在非正式環境，依然不能放任模型「全自動」處置；一旦判斷錯誤誤砍關鍵連線，同樣會打亂團隊步調，因此高風險操作必須由程式中斷並交由工程師裁決。
我們延續上一篇的微服務事故排查情境：Staging 環境的結帳服務 checkout 出現大量 500 錯誤，顧客在付款步驟逾時。當時 Agent 查出資料庫連線池滿載的根因後，輸出分析報告即結案；本篇則進一步賦予它執行處置的能力，並在提議清理連線時觸發人工審核，展示從診斷、安全攔截到授權修復的完整流程。
這套處置流程透過終端機互動腳本（uv run hitl-demo）展示：Agent 收到通報自主排查日誌與指標，當它鎖定問題並提議執行高風險工具 terminate_idle_connections 時，中介軟體立即攔截該動作並暫停，終端機會停下來等待工程師在 CLI 輸入裁決：
⏸️ 【觸發安全中斷】中介軟體攔截到高風險系統處置操作！
處置動作：terminate_idle_connections
提議參數：{'service': 'database', 'idle_seconds': 600, 'max_terminate': 50}
安全政策：高風險處置必須由現場工程師裁決
👤 【請在 CLI 選擇處理方式】
[1] approve - 核准：依照 AI 提議的參數直接執行
[2] reject - 拒絕：禁止執行該動作，並傳送指示給 AI
[3] exit - 放棄並結束
請輸入選項 (1-3) [預設: 1]:
整個審核流程包含 5 個階段：
search_logs 檢索日誌，發現資料庫連線池已滿（100/100），其中有 50 筆閒置逾時的連線未釋放。terminate_idle_connections(service="database", idle_seconds=600, max_terminate=50) 來清理連線。check_service_health 驗證資料庫連線池是否恢復正常，確認環境可用後輸出結案報告。在 src/langgraph_human_in_the_loop/tools.py 中定義排查與處置工具：
from langchain_core.tools import tool
@tool
def search_logs(service: str, query: str) -> str:
"""唯讀工具：檢索指定微服務的日誌記錄。安全操作。"""
if service == "checkout" or "database" in query:
return (
"【日誌檢索結果 - checkout】\n"
"- 10:02:15 ERROR: DB connection pool exhausted (active=100/100, idle=0)\n"
"- 10:02:18 ERROR: checkout request timed out waiting for db connection\n"
"- 診斷發現：有 50 筆來自 batch-job 的閒置連線已超時未釋放。"
)
return f"【日誌檢索結果 - {service}】未發現明顯異常日誌。"
@tool
def check_service_health(service: str) -> str:
"""唯讀工具：檢查指定微服務或資料庫的即時健康度與指標。安全操作。"""
if service == "database":
return "【健康狀態 - database】連線池使用率 98%，延遲 4500ms（嚴重異常）。"
return f"【健康狀態 - {service}】狀態正常（HTTP 200，延遲 15ms）。"
@tool
def terminate_idle_connections(
service: str,
idle_seconds: int,
max_terminate: int,
) -> str:
"""高風險處置工具：強制中斷資料庫閒置連線以釋放連線池。涉及線上連線中斷風險。"""
return (
f"【處置成功】已強制中斷 {service} 中閒置超過 {idle_seconds} 秒的連線，"
f"共釋放 {max_terminate} 筆連線。連線池已恢復可用狀態。"
)
@tool
def isolate_problematic_pod(pod_name: str) -> str:
"""高風險處置工具：將異常 Pod 隔離並摘除流量（Cordon & Drain）。
保留現場供後續排查。涉及節點容量縮減。
"""
return f"【處置成功】已將 Pod {pod_name} 流量摘除並完成節點隔離（Cordon）。"
@tool
def scale_up_replicas(service: str, count: int) -> str:
"""高風險處置工具：水平擴展（Scale out）副本數量以分擔線上負載。
涉及資源配額與成本。
"""
return f"【處置成功】服務 {service} 副本數已增加 {count} 個節點。"
@tool
def toggle_feature_flag(feature: str, enabled: bool) -> str:
"""高風險處置工具：緊急切換功能旗標以進行功能降級或恢復。
涉及業務功能可用性。
"""
state_str = "啟用" if enabled else "關閉（降級）"
return f"【處置成功】功能旗標 {feature} 已切換為 {state_str}。"
@tool
def ask_engineer(question: str) -> str:
"""向現場工程師提問以獲取確認（由人類輸入直接作為工具結果）。"""
return ""
在 agent.py 中，我們用 HumanInTheLoopMiddleware 設定審核規則。
原則很清楚：查日誌、看指標等唯讀操作直接放行，讓 Agent 自主排查；只有會改動系統狀態的處置動作，或是需要向工程師提問時，才停下來等待人工介入。
因為中介軟體預設會放行未列出的工具，所以我們在 interrupt_on 把要攔截的工具加上：
middleware = [
HumanInTheLoopMiddleware(
interrupt_on={
# 處置工具：強制中斷審核，僅允許核准或拒絕
"terminate_idle_connections": {
"allowed_decisions": ["approve", "reject"],
},
# 提問工具：暫停並等待工程師直接代答
"ask_engineer": {
"allowed_decisions": ["respond"],
},
# 唯讀工具（search_logs、check_service_health）沒列出，預設自動放行
},
description_prefix="高風險處置操作待工程師核准",
)
]
定義好中介軟體後，使用 create_agent 將模型、工具清單、中介軟體與 Checkpointer 組裝在一起。
Human-in-the-Loop 機制在觸發 interrupt 時，必須將當前圖的節點狀態、尚未執行的 Tool Call 與上下文完整持久化；工程師審核完成後，才能依據同一個 Session（thread_id）從中斷點安全喚醒並繼續執行。因此，組裝 Agent 時必須掛載 Checkpointer（範例使用記憶體型的 InMemorySaver，實務上可替換為 PostgreSQL 等持久化儲存）：
def build_agent(
model: Any,
checkpointer: BaseCheckpointSaver[Any] | None = None,
) -> CompiledStateGraph[Any]:
if checkpointer is None:
checkpointer = InMemorySaver()
return create_agent(
model=model,
tools=tools,
middleware=middleware,
checkpointer=checkpointer,
)
定義好 Agent 後，應用程式（例如維運 API、Slack 機器人或本範例的 CLI 互動腳本）必須負責偵測中斷點、收集工程師決策，並通知 LangGraph 恢復執行。
這段生命週期在 demo.py 的 run_interactive_demo 函式中實作：
config = {"configurable": {"thread_id": "sre-thread-interactive"}}
# 1. 啟動 Agent 進行自主排查
result = agent.invoke(
{"messages": [{"role": "user", "content": prompt}]},
config=config,
version="v2",
)
# 2. 檢查是否觸發高風險處置中斷
if result.interrupts:
action = result.interrupts[0].value["action_requests"][0]
# 3. 取得工程師在 CLI 輸入的選項（choice = input(...)），組裝對應的 Command
if choice == "1": # 核准
resume_cmd = Command(resume={"decisions": [{"type": "approve"}]})
elif choice == "2": # 拒絕
resume_cmd = Command(resume={"decisions": [{"type": "reject", "message": ...}]})
# 4. 帶入相同的 thread_id 喚醒 Agent 繼續執行
final_result = agent.invoke(resume_cmd, config=config, version="v2")
當應用程式呼叫 agent.invoke(resume_cmd, config=config) 時，LangGraph 會依 thread_id 從 Checkpointer 載入先前凍結的狀態，並依據決策類型執行相應邏輯：
approve（原樣核准）：approve，正式呼叫底層的 terminate_idle_connections 清理連線，並將成功訊息包裝成 ToolMessage 寫入歷史狀態。Agent 甦醒後自主呼叫 check_service_health 驗證資料庫連線池恢復正常，輸出結案報告。reject（拒絕處置）：ToolMessage 傳回給 Agent。Agent 讀取反饋後調整策略，改採其他處置方案。respond（人類直接代答）：ask_engineer 詢問特定批次排程時，工程師在終端機輸入的回覆直接作為工具結果返回給 Agent，Agent 讀取後繼續推論。雖然 Human-in-the-Loop 機制為高風險操作加上了人工審核的防線，但賦予 Agent 系統處置能力時，依然必須保持謹慎。人類的審核放行不是萬靈丹，要避免系統在處置過程中失控，設計上仍需守住幾個核心考量：
iThome鐵人賽
看影片追技術