跳到主要內容

AG Day 17 Human-in-the-loop：interrupt 與人工核准

AG Day 17 Human-in-the-loop：interrupt 與人工核准

執行需求：CPU 可跑。在昨天 AG Day 16（原文連結）中，我們替 research-agent 接上了 SQLite 版的 checkpointer，讓代理能透過 thread_id 記住每一個研究任務的完整對話狀態，即使程式重啟也能從上次停下的地方繼續。今天要利用這個「能被中斷、能被恢復」的能力，實作一個對正式系統很重要的設計：讓代理在準備執行高風險動作之前，主動停下來，等待人類明確核准之後才繼續，而不是自顧自地把所有決定都執行完。今天全部範例都能在沒有 API 金鑰的情況下跑完，唯一需要提前準備的是昨天建立的 checkpointer，因為人工核准機制完全建立在「能被中斷、能被恢復」這個能力之上。

引言

到目前為止，research-agent 的工具都相對溫和：查資料、抓網頁，出錯了大不了重試或回報失敗，不會對外部世界造成不可逆的影響。但一個真正要拿去用的研究助理，遲早會需要做一些「有代價」的動作：呼叫按用量計費的付費 API、把整理好的報告寄出去、或是把資料寫進一個別人也在用的正式資料庫。如果代理完全自主地決定什麼時候該做這些事，一旦提示詞設計有漏洞，或模型偶爾判斷失準，後果可能不是「重試一次就好」，而是「已經花了錢」或「已經寄出去了，收不回來」。

Human-in-the-loop（人機協作、簡稱 HITL）的核心想法很直接：在流程裡插入一個「暫停點」，讓真人看過目前的狀態、決定要不要放行，代理才能繼續往下走。這聽起來像是要另外寫一套暫停與恢復的機制，但昨天我們已經把地基打好了：只要圖有 checkpointer，LangGraph 就能在任何節點裡呼叫一個叫 interrupt() 的函式，暫停目前的執行、把控制權交還給呼叫端；等真人做出決定後，呼叫端再用 Command(resume=...) 把決定送回去，圖會從暫停的那一點，帶著這個決定繼續往下跑。今天我們就是要把這個機制，安裝在 research-agent 呼叫外部搜尋服務之前。

原理與觀念

interrupt() 怎麼暫停一個正在執行的節點

interrupt() 是 LangGraph 提供的一個函式，你可以在圖的任何節點函式內部呼叫它，並傳入一個你想讓人類看到的資料（例如「即將呼叫搜尋 API，關鍵字是什麼、預估會用掉多少額度」）。呼叫這個函式時，LangGraph 會利用先前設定好的 checkpointer，把當下的狀態完整存下來，然後拋出一個特殊的訊號讓整個 invoke 或 stream 呼叫提前結束，並把你傳入 interrupt() 的資料回傳給呼叫端。從呼叫端的角度看，這次呼叫並沒有跑完整張圖，而是在某個節點「卡住」了，等待進一步指示。

用 Command(resume=...) 把決定送回去

真人看到中斷資料、做出決定之後（例如核准或拒絕），呼叫端要再呼叫一次 app.invoke(Command(resume=決定內容), config=同一個thread_id的config)。這裡的關鍵是要帶上同一個 thread_id，這樣 LangGraph 才知道要從哪一個被中斷的快照繼續。圖恢復執行時，原本呼叫 interrupt() 的那一行程式碼，會像函式呼叫直接回傳一樣，把 resume 帶的內容當成回傳值，讓節點函式可以根據這個值決定接下來要做什麼，例如核准就真的呼叫外部服務、拒絕就回傳一則說明訊息給使用者。

為什麼一定要有 checkpointer 才能用 interrupt

這也是為什麼我們特意把記憶功能排在人工核准之前講解：interrupt() 能運作的前提，是圖已經設定了 checkpointer。沒有 checkpointer，圖沒有地方可以存放「中斷當下的完整狀態」，resume 時也就無從恢復起，程式只會直接報錯或行為不如預期。這個相依關係在官方文件裡有明確說明，但初學者常常想跳過記憶章節、直接用 interrupt，結果卡在莫名其妙的錯誤訊息上。

中斷不是例外，是一種正常的流程分支

剛接觸這個機制的人，常常把「中斷」跟程式語言裡的例外（exception）搞混，覺得 interrupt() 像是丟出一個錯誤、要用 try/except 接住。實際上中斷是圖執行過程中一種正常、預期內的流程分支，跟工具呼叫失敗完全是兩回事：工具失敗代表「事情沒有照計畫走」，需要昨天談的錯誤處理機制；中斷則代表「事情正照計畫走，只是這個計畫裡有一步需要先問過人」。把這兩種情境的心智模型分開，之後在設計更複雜的核准規則時，才不會把「失敗」與「需要核准」這兩種完全不同的狀態混為一談，錯誤地用同一套處理邏輯去應付。

完整實作

我們把「需要核准」的判斷邏輯，加在 search_web 這個工具前面，模擬一個「單次搜尋若標記為高成本查詢，需要人工核准才能真的送出」的情境。先在 src/research_agent/tools.py 補一個判斷式：

# research-agent/src/research_agent/tools.py（新增）
EXPENSIVE_KEYWORDS = ("深度分析", "大量", "全面")


def is_expensive_query(query: str) -> bool:
    """簡化示範：包含特定關鍵字的查詢視為高成本，正式規則應依實際計費規範調整。"""
    return any(keyword in query for keyword in EXPENSIVE_KEYWORDS)

接著在 src/research_agent/graph.py 新增一個獨立節點 approval_gate，放在 agent 節點與 tools 節點之間，專門負責判斷要不要中斷、以及中斷後怎麼處理人類的決定：

# research-agent/src/research_agent/graph.py（新增節點）
from langgraph.types import interrupt, Command
from research_agent.tools import is_expensive_query


def approval_gate(state):
    """檢查最後一次工具呼叫是否為高成本查詢，是的話中斷等待人工核准。"""
    last_message = state["messages"][-1]
    tool_calls = getattr(last_message, "tool_calls", None) or []

    for call in tool_calls:
        if call["name"] == "search_web" and is_expensive_query(call["args"].get("query", "")):
            decision = interrupt({
                "type": "approval_required",
                "tool": call["name"],
                "args": call["args"],
                "reason": "偵測到高成本查詢關鍵字，需要人工核准後才會實際送出。",
            })
            if decision != "approve":
                # 拒絕時，直接把這個呼叫改標記為已拒絕，讓後續改回覆說明文字
                call["args"]["_rejected"] = True
    return state

把 approval_gate 接進圖裡，注意它要放在 agent 決定呼叫工具之後、真正執行 tools 節點之前：

# research-agent/src/research_agent/graph.py（調整邊的連法）
def build_graph(checkpointer=None):
    graph = StateGraph(MessagesState)
    graph.add_node("agent", call_model)
    graph.add_node("approval_gate", approval_gate)
    graph.add_node("tools", tool_node)
    graph.add_edge(START, "agent")
    graph.add_conditional_edges("agent", tools_condition, {"tools": "approval_gate", END: END})
    graph.add_edge("approval_gate", "tools")
    graph.add_edge("tools", "agent")
    return graph.compile(checkpointer=checkpointer)

然後寫一段驗證腳本，模擬「代理想做高成本查詢 → 中斷 → 人工核准 → 繼續執行」的完整流程：

# research-agent/verify_interrupt.py
from langchain_core.messages import HumanMessage
from langgraph.types import Command
from research_agent.graph import build_graph
from research_agent.memory import get_checkpointer

app = build_graph(checkpointer=get_checkpointer())
config = {"configurable": {"thread_id": "hitl-demo-001"}}

result = app.invoke(
    {"messages": [HumanMessage(content="幫我對這個主題做一次深度分析搜尋")]},
    config=config,
)

if "__interrupt__" in result:
    payload = result["__interrupt__"][0].value
    print("流程已中斷，等待人工核准：")
    print(" - 工具：", payload["tool"])
    print(" - 參數：", payload["args"])
    print(" - 原因：", payload["reason"])

    # 模擬真人核准
    final = app.invoke(Command(resume="approve"), config=config)
    print("\n核准後繼續執行，最終回覆：", final["messages"][-1].content[:80])
else:
    print("此次呼叫未觸發中斷，最終回覆：", result["messages"][-1].content[:80])

離線模式下的示範輸出：

流程已中斷，等待人工核准：
 - 工具： search_web
 - 參數： {'query': '這個主題做一次深度分析搜尋'}
 - 原因： 偵測到高成本查詢關鍵字，需要人工核准後才會實際送出。

核准後繼續執行，最終回覆： （模型根據離線模擬搜尋結果整理出的示範回應）

最後，示範「拒絕」的路徑，讓 CLI 使用者也能明確選擇不核准：

# research-agent/verify_interrupt_reject.py
from langchain_core.messages import HumanMessage
from langgraph.types import Command
from research_agent.graph import build_graph
from research_agent.memory import get_checkpointer

app = build_graph(checkpointer=get_checkpointer())
config = {"configurable": {"thread_id": "hitl-demo-002"}}

result = app.invoke(
    {"messages": [HumanMessage(content="幫我對這個主題做一次深度分析搜尋")]},
    config=config,
)

if "__interrupt__" in result:
    final = app.invoke(Command(resume="reject"), config=config)
    print("已拒絕，最終回覆：", final["messages"][-1].content[:80])

正式系統裡，核准者往往不是寫程式的人，不會直接呼叫 Python 腳本。我們在 runs 資料表補一個 status 欄位的更新函式，讓中斷發生時同步把任務狀態標記為「等待核准」，之後可以接上任何形式的通知或審核介面，而不必假設核准者一定守在終端機前：

# research-agent/src/research_agent/storage.py（新增）
import sqlite3
from pathlib import Path


def mark_run_status(db_path: Path, run_id: str, status: str) -> None:
    """更新任務狀態，供人工核准流程與監控儀表板查詢使用。"""
    with sqlite3.connect(db_path) as conn:
        conn.execute(
            "UPDATE runs SET status = ? WHERE run_id = ?",
            (status, run_id),
        )
        conn.commit()


def list_pending_approvals(db_path: Path) -> list[tuple[str, str]]:
    """列出所有等待人工核准的任務，供審核介面或排程通知使用。"""
    with sqlite3.connect(db_path) as conn:
        rows = conn.execute(
            "SELECT run_id, user_query FROM runs WHERE status = 'PENDING_APPROVAL'"
        ).fetchall()
    return rows

常見錯誤與踩雷

第一個常見錯誤，就是前面提過的忘記設定 checkpointer，導致呼叫 interrupt() 時得到令人困惑的錯誤訊息。看到跟 interrupt 相關的例外，第一件事永遠是先檢查 build_graph() 有沒有真的傳入 checkpointer、invoke 有沒有帶正確的 thread_id。

第二個是把太多邏輯塞進 interrupt() 傳入的資料裡，卻忘記這份資料最終需要被序列化、存進 checkpoint 資料庫。如果傳入的物件裡有不能被序列化的內容（例如某些自訂類別的實例、開啟中的檔案物件），儲存快照時可能失敗。建議 interrupt() 傳入的內容盡量維持成單純的字典、字串、數字這類基本型別，需要顯示更複雜資訊時，先自己轉換成純文字描述。

第三個雷是誤以為 resume 之後，圖會從頭重新執行一次。實際上恢復執行只會從中斷的那個節點繼續，之前已經執行過的節點不會重跑。這代表如果你在 approval_gate 之前的節點裡有副作用（例如寫了一筆日誌），核准或拒絕都只會觸發一次，不會因為呼叫了兩次 invoke 就重複寫入兩次。但如果誤把有副作用的程式碼寫在 approval_gate 節點「呼叫 interrupt() 之後」的部分，這段程式碼確實只會在恢復執行時才跑到，設計時要清楚每一段程式碼相對於中斷點的執行時機。

第四個常見誤解是把人工核准當成萬用的安全機制，對每個節點都無差別地插入中斷。這樣做會讓一個原本能自動跑完的研究任務，變成每一步都要人盯著，完全失去自動化的意義，也會讓使用者覺得系統很難用。人工核准應該只用在真正有代價、不可逆、或牽涉外部資源消耗的少數節點上，其餘節點維持全自動，這也呼應了我們在系列一開始就強調的原則：好的代理系統不是「什麼都問過人」，而是清楚知道哪裡真正需要人。

效能與實務提醒

從系統設計的角度看，加入人工核准之後，一個研究任務的生命週期可能會拉得很長：使用者可能提交查詢後就離開，幾個小時後才回來核准。這代表我們不能假設呼叫 invoke 的行程會一直存活著等待 resume；正式環境應該把「等待核准中」的任務狀態記錄在 runs 資料表（例如新增一個 status = "PENDING_APPROVAL"），並透過通知機制（Email、聊天機器人訊息等）提醒該核准的人，而不是讓一個 HTTP 請求或 CLI 行程一直卡著不釋放資源。這也代表 interrupt() 回傳的中斷資料，最好在 API 層額外整理成一份獨立的「待核准清單」回應格式，讓前端或審核介面不必理解 LangGraph 內部的資料結構，只需要知道「哪個任務、要核准什麼、核准或拒絕的按鈕分別對應哪支 API」。這種介面設計上的解耦，會在 AG Day 38 實作 FastAPI 部署時看到具體效果。

核准的時效性也是實務上容易忽略的一環。如果一個任務等待核准超過合理時間（例如二十四小時），系統應該有機制自動把它標記為逾時、取消，或是重新提醒核准者，而不是讓任務無限期停留在「等待核准」狀態，佔用著一份永遠不會被回收的檢查點快照。

另外要注意 interrupt() 傳入的內容，應該包含足夠的脈絡讓核准者能做出判斷，但不要洩漏不必要的敏感資訊。我們範例裡回傳了完整的工具參數，正式系統中如果參數涉及使用者個資，建議先做遮罩或摘要處理，再呈現給核准介面。

最後，建議把「哪些工具、哪些條件需要核准」抽成設定檔或資料表，而不是像今天範例一樣寫死在 EXPENSIVE_KEYWORDS 這種簡單關鍵字判斷裡。正式系統的高成本判斷邏輯可能牽涉即時的用量與預算查詢，把判斷邏輯與圖的結構解耦，未來調整規則時才不需要改動核心流程程式碼，也比較容易替不同的核准規則各自寫單元測試。

小結

今天我們替 research-agent 加上了人工核准機制：在 agent 決定呼叫工具、真正執行工具之前，插入一個 approval_gate 節點，用 interrupt() 暫停流程並回報「需要核准的原因與細節」；真人做出決定後，用 Command(resume=...) 把決定送回去，讓圖從中斷點繼續執行。這一切之所以可行，是因為昨天已經替圖裝上了 checkpointer——沒有記憶能力，就沒有可以恢復的「中斷點」。

今天新增的關鍵詞：人機協作（Human-in-the-loop）——在自動化流程中安排真人決策點的設計模式；中斷（interrupt）——暫停圖的執行並把控制權交還呼叫端；恢復（resume）——帶著人類的決定，讓圖從中斷點繼續往下跑。這套機制的價值不在於「讓代理變慢」，而是在少數真正有代價的節點上，把最終決定權留給人類，讓自動化與風險控管可以並存，而不是二選一。

結語

目前為止 research-agent 的圖結構還算單純：一個 agent 節點、一個 tools 節點、加上今天新增的 approval_gate。但接下來我們會陸續加入檢索、報告產生等更多功能，如果全部塞在同一張圖裡，圖會變得又肥又難測試。

明天，我們會進入「AG Day 18 子圖：把研究流程模組化」，把研究流程拆成幾個獨立、可以個別測試的子圖模組，並示範子圖之間如何共享或隔離狀態，讓 research-agent 的架構在功能持續增加時依然保持清晰、容易維護。

延伸資源

  • LangGraph 官方文件：Human-in-the-loop 概念與 interrupt()、Command 的使用說明。
  • LangGraph 官方文件：Persistence 章節中關於中斷與恢復對 checkpointer 的相依說明。
  • NIST AI 風險管理框架（AI RMF）：關於自動化系統中人工監督（human oversight）設計原則的參考資料。
標籤：

留言

張貼留言

這個網誌中的熱門文章

Day 2 變數與資料型別

Day 2 變數與資料型別 引言 寫程式的過程中，變數與資料型別是處理資料的基礎。變數是存放資料的容器，資料型別則決定這筆資料有哪些特性、可以進行哪些操作。學會定義變數、認識各種資料型別，是學好 Python 的關鍵一步。 這篇文章會帶你了解 Python 中變數的觀念、如何定義變數，以及常見的資料型別，包括整數、浮點數、字串、布林值，還有串列、元組、字典與集合等容器型別。我們也會介紹變數的命名規則與撰寫風格建議，以及如何用 type() 檢查資料型別。 什麼是變數？如何在 Python 中定義變數 變數是在程式執行時用來存放資料的名稱。透過定義變數，我們可以給一筆資料一個名字，並在程式的其他地方用這個名字取用該筆資料。在 Python 中，變數不需要事先宣告型別，因為 Python 是動態型別語言，變數的型別由指定給它的值決定。 定義變數的基本語法 在 Python 中定義變數非常簡單，只要用賦值符號 = 把值指定給變數即可。例如： x = 5 # 定義變數 x，並把整數 5 賦值給它 name = "Alice" # 定義變數 name，並把字串 "Alice" 賦值給它 在這裡，x 是一個變數，被賦予整數 5；name 是另一個變數，被賦予字串 "Alice"。 變數的更新與覆寫 變數的值可以修改，也就是說，我們可以在程式的不同地方給同一個變數新的值。例如： x = 10 # x 最初被賦予 10 x = 15 # x 的值現在被更新為 15 這樣就能依照需求，在程式執行過程中靈活調整變數的值。 Python 的動態型別系統 Python 和某些靜態型別語言不同，定義變數時不需要宣告型別。賦值時，Python 會根據值自動判斷變數的型別。例如： x = 5 # x 是整數 x = 3.14 # x 變成浮點數 x = "Hi" # x 變成字串 同一個變數在程式執行過程中可以存放不同型別的值，這是 Python 的彈性之一。 常見資料型別 在 Python 中，資料型別決定我們可以對變數進行哪些操作...
張貼留言
閱讀完整內容

Python 從入門到 PyTorch 深度學習：開啟 AI 世界的大門

Python 從入門到 PyTorch 深度學習：開啟 AI 世界的大門 隨著人工智慧（AI）與深度學習（Deep Learning）快速發展，越來越多人對這些技術產生興趣。不論你是想踏入 AI 領域的初學者，還是已經有程式基礎的開發者，學好 Python 與深度學習框架（例如 PyTorch），都能為你打開更多可能。 為什麼選擇 Python？ Python 已經是資料科學與人工智慧領域的首選語言。它的語法簡潔、容易上手，而且擁有龐大的生態系與大量開源函式庫。無論是資料處理、資料視覺化，還是建立機器學習與深度學習模型，Python 都能勝任。對想進入 AI 或資料科學領域的人來說，它幾乎是必備工具。 PyTorch 是什麼？ PyTorch 是由 Meta（原 Facebook）AI 研究團隊開發的開源深度學習框架，以易用、靈活和動態計算圖著稱，是許多 AI 研究人員與開發者的首選。相較於其他框架，PyTorch 的寫法更貼近原生 Python，對初學者相對友善。無論是簡單的實驗，還是複雜的深度學習模型，PyTorch 都能提供強大的支援。 這個系列能帶給你什麼？ 這個系列會從 Python 的基礎開始，帶你一步一步學習，最後能自己用 PyTorch 建立深度學習模型。即使你完全沒有寫過程式，也能跟著文章的節奏累積技能，理解 AI 與深度學習的核心觀念。 本系列涵蓋的主題 Python 基礎：從變數、條件判斷到函式與模組。 資料處理工具：用 NumPy 與 Pandas 有效率地操作資料。 資料視覺化：用 Matplotlib 與 Seaborn 把資料畫成圖表。 深度學習的數學基礎：線性代數、微積分與機率。 PyTorch 入門：理解張量、模型建構與 GPU 加速。 基礎深度學習模型：CNN 與 RNN 的實作應用。 深度學習專案實戰：從資料前處理到模型部署的端到端流程。 誰適合這個系列？ 程式初學者 ：如果你對 AI 充滿好奇，卻還沒寫過程式，系列的第一部分會帶你快速上手 Python，並幫助你理解深度學習的基本觀念。 資料科學愛好者 ：如果你已經熟悉一些資料處理方法，進階部分會教你如何用 PyTorch 建構深度學習模型。 開發者與研究人員 ：想更深入了...
張貼留言
閱讀完整內容

Day 1 Python 簡介與環境設定

Day 1 Python 簡介與環境設定 引言 在現在的科技環境裡，程式設計已經是一項重要技能。無論你是對資料科學有興趣、想成為開發者，或是想踏入人工智慧（AI）領域，學會寫程式都能明顯提升你的競爭力。在眾多程式語言中，Python 因為語法簡單、功能強大、應用範圍廣泛，成為許多人進入程式世界的第一選擇。這篇文章會帶你認識 Python 的背景與優勢，並一步步教你在不同系統上安裝與設定 Python 開發環境，最後寫出第一支 Python 程式。 為什麼選擇 Python？ Python 是一種高階程式語言，由 Guido van Rossum 在 1991 年發布。Python 的設計哲學強調程式碼的可讀性，並用縮排來定義程式區塊，這點和許多使用大括號的語言不同。簡潔的語法讓它成為初學者的理想選擇；就算是經驗豐富的開發者，也能用它完成複雜的專案。 Python 的優勢如下： 簡單易學 ：Python 的語法清楚、結構簡潔，初學者很快就能上手。和其他語言相比，學習曲線相對平緩，不需要先弄懂一堆複雜觀念，就能開始寫程式。 應用範圍廣泛 ：從資料科學、網頁開發、人工智慧、機器學習、自動化測試到網路爬蟲，Python 都有大量開源函式庫與工具支援，而且在這些領域都扮演關鍵角色。 豐富的函式庫與框架 ：Python 的函式庫生態系非常龐大。做資料分析有 NumPy、Pandas；開發網站有 Django、Flask；做深度學習有 TensorFlow、PyTorch。各種需求幾乎都能找到對應的套件，讓開發更有效率。 跨平台支援 ：Python 支援 Windows、macOS、Linux 等作業系統，程式通常不需要太多修改就能跨平台執行，讓開發與部署更有彈性。 活躍的社群 ：Python 擁有龐大的開發者社群。學習或開發上遇到問題，幾乎都能在社群與論壇（例如 Stack Overflow）找到答案，對初學者來說是很強的後盾，也能減少卡關時的挫折感。 Python 的應用領域 Python 的流行與強大功能，讓許多領域都開始大量使用它。以下是幾個常見的應用方向： 資料科學 ：隨著大數據與人工智慧興起，資料科學大量使用 Python。NumPy、Pandas 與 Matplotlib 等工具能處理和分析龐...
張貼留言
閱讀完整內容