在前一篇文章中，我們介紹了如何設計 多代理系統，讓不同的 Agent 各自專精於任務並協同合作。然而，在實際應用中，僅依賴全自動化流程往往難以應付所有情境。當任務涉及專業判斷、決策風險較高，或需要額外的人工審核時，引入人類參與就顯得格外重要。
今天的主題是 人類參與流程（Human-in-the-loop, HITL），也常被稱為 人機協作。我們將探討如何在 LangGraph 中設置 中斷點（Interrupts），讓 Agent 在執行過程中能暫停，等待使用者輸入或確認，再繼續往下執行。透過這樣的設計，我們能建立一個更靈活、可控的互動式決策機制。
雖然 AI Agent 能透過 LLM 自動完成推理與決策，但在許多情境中仍無法完全取代人類的角色，常見原因包括：
因此，在設計應用流程時，我們往往會加入人工確認的環節。AI 負責生成初步建議，而最終是否採納，則交由人類進行檢視、修正或批准，再讓後續流程繼續執行。這樣的設計既能發揮 AI 的效率，又能保留人類的專業判斷，讓整體系統更可靠、更可控。
在人機協作的流程中，AI 與人類會交替處理不同的任務，藉此兼顧 AI 的運算效率與人類的專業判斷。以下是常見的應用場景：
這些設計的核心目標，是讓 AI 自動化處理大部分任務，同時保留人工介入的彈性。如此一來，不僅能提升準確度與降低風險，也能增強使用者對系統的信任感，讓 AI 真正成為可協作的智慧助手。
實務上，這樣的模式已廣泛應用在各種工具中。以 Codex CLI、Claude Code、Gemini CLI 等程式碼輔助工具為例，AI 可以快速產生程式碼片段或建議，但開發者往往需要檢查、測試甚至重構，才能確保結果正確並符合專案需求。這正是一種典型的 Human-in-the-loop（HITL） 協作流程：AI 負責加速產出，而人類則扮演審核與把關的角色。
在 LangGraph 中，如果我們希望流程能在特定位置「暫停」，並等待人類輸入，就可以透過 interrupt() 函式來實現。這是一個專為人機協作模式設計的工具，能讓流程在某個節點中斷，把訊息或狀態交給使用者確認，待使用者回覆後再繼續執行。
interrupt() 如何運作？interrupt() 是 LangGraph 實現 Human-in-the-loop 的核心方法。它的基本運作流程如下：
interrupt() 的節點時，流程會停止，並輸出一個特殊的 __interrupt__ 訊息。interrupt() 可以輸出任何可序列化為 JSON 的資料，例如一段文字說明、問題提示或狀態資訊，讓使用者能清楚理解當前情境。new Command({ resume: value }) 回傳給流程，LangGraph 會將這筆資料寫回狀態，並繼續往下執行後續節點。為了讓 interrupt() 能正確運作，還需要搭配一些設定與規範：
checkpointer（例如 MemorySaver），用來保存中斷時的狀態，否則流程無法在恢復時回到正確位置。thread_id。這個 ID 會對應到一次完整的執行脈絡，確保中斷後能順利追蹤並正確銜接後續步驟。透過這樣的設計，LangGraph 讓我們可以輕鬆地在自動化流程中插入人工確認節點，進而建立靈活可控的人機協作系統。
以下程式展示一個簡單的場景：AI 產生文字後，交由人類檢查與修改，再繼續流程。
import { MemorySaver, Annotation, interrupt, Command, StateGraph, START, END } from '@langchain/langgraph';
// 定義狀態
const StateAnnotation = Annotation.Root({
text: Annotation<string>(),
});
// 定義需要人工介入的節點
function humanNode(state: typeof StateAnnotation.State) {
const value = interrupt({
textToRevise: state.text, // 要給人類檢查的內容
});
return { text: value }; // 將人類的輸入更新回狀態
}
// 建立流程
const workflow = new StateGraph(StateAnnotation)
.addNode('human_node', humanNode)
.addEdge(START, 'human_node')
.addEdge('human_node', END);
// 使用 MemorySaver 作為 checkpointer
const checkpointer = new MemorySaver();
const graph = workflow.compile({ checkpointer });
// 指定 thread ID，用於追蹤與恢復
const threadConfig = { configurable: { thread_id: 'thread-1' } };
// 第一次執行，流程會停在 interrupt
const firstResult = await graph.invoke(
{ text: 'Original text' },
threadConfig,
);
console.log(firstResult); // 輸出 __interrupt__ 訊息，等待人工輸入
// 模擬人類輸入並恢復執行
const resumedResult = await graph.invoke(
new Command({ resume: 'Edited text' }),
threadConfig,
);
console.log(resumedResult); // 狀態更新為人類修改後的內容
第一次執行會輸出中斷訊息，提示使用者需要修改文字。以下是一個輸出範例：
{
text: 'Original text',
__interrupt__: [
{
id: 'c03575ef6254c31aa5a27ce593148d68',
value: {
textToRevise: 'Original text'
}
}
]
}
當使用者輸入 Edited text 並恢復流程後，狀態會更新為新的內容：
{ text: 'Edited text' }
透過 interrupt()，我們可以很自然地在 AI 流程中插入人工審核或確認的步驟。這不僅讓系統更安全、可靠，也讓 AI 與人類能夠真正「協作」，而不是單向輸出結果。
interrupt() 提供了暫停與恢復的能力，但真正的價值在於如何運用它設計人機協作的流程。在實務中，常見的人機協作模式大致可以分為三類：
接下來，我們將透過範例展示如何利用 interrupt() 來實現這些協作模式。
在許多高風險或需要人工把關的場景中（例如 API 呼叫、資料庫寫入、金融交易或敏感操作），我們不希望讓 AI 全自動執行。這時候就可以透過 interrupt() 插入 批准或拒絕的節點，由人類決定是否允許繼續。
這種設計能確保流程在關鍵步驟不會無條件往下跑，而是交由人類確認後再執行，若被拒絕，也可以根據回覆轉換到替代路徑。
以下是一個簡單的範例：
import { Annotation, StateGraph, interrupt, Command, MemorySaver, START } from '@langchain/langgraph';
// 定義狀態
const GraphAnnotation = Annotation.Root({
action: Annotation<string>(),
result: Annotation<string>(),
});
// 定義需要人工批准的節點
function humanApproval(state: typeof GraphAnnotation.State): Command {
const result = interrupt({
request: `是否允許執行操作：${state.action}?`
});
// 根據人類回覆，決定流程走向
if (result.isApproved) {
return new Command({ goto: 'approved_node' });
} else {
return new Command({ goto: 'rejected_node' });
}
}
// 建立流程圖
const workflow = new StateGraph(GraphAnnotation)
.addNode('human_approval', humanApproval, { ends: ['approved_node', 'rejected_node'] })
.addNode('approved_node', async (state) => ({ result: `已批准執行操作：${state.action}` }))
.addNode('rejected_node', async (state) => ({ result: `已拒絕執行操作：${state.action}` }))
.addEdge(START, 'human_approval');
// 加入 checkpointer
const graph = workflow.compile({ checkpointer: new MemorySaver() });
// 指定 thread ID
const threadConfig = { configurable: { thread_id: 'approval-thread' } };
// 第一次執行，流程會停在 interrupt
const firstRun = await graph.invoke({ action: '刪除資料庫中的用戶紀錄' }, threadConfig);
console.log(firstRun);
/*
{
action: '刪除資料庫中的用戶紀錄',
__interrupt__: [
{
id: 'f39cc7ed948c3a705a1d3fc213ae7710',
value: { request: '是否允許執行操作：刪除資料庫中的用戶紀錄?' }
}
]
}
*/
// 模擬人類批准
const approvedRun = await graph.invoke(new Command({ resume: { isApproved: true } }), threadConfig);
console.log(approvedRun);
// { action: '刪除資料庫中的用戶紀錄', result: '已批准執行操作：刪除資料庫中的用戶紀錄' }
// 模擬人類拒絕
const rejectedRun = await graph.invoke(new Command({ resume: { isApproved: false } }), threadConfig);
console.log(rejectedRun);
// { action: '刪除資料庫中的用戶紀錄', result: '已拒絕執行操作：刪除資料庫中的用戶紀錄' }
在這個例子中：
humanApproval 節點 → 呼叫 interrupt()，暫停等待人類輸入。{ isApproved: true } → 流程轉向 approved_node，執行批准後的邏輯。{ isApproved: false } → 流程轉向 rejected_node，執行拒絕後的替代方案。這樣，我們就能在流程的關鍵環節中引入人工決策，確保高風險操作不會全權交由 AI 處理。
另一種常見的人機協作模式，是在流程中斷時，讓人類直接編輯 AI 產生的輸出。這樣的設計特別適合用在需要人工修正的場景，例如 LLM 自動生成的摘要、翻譯或程式碼片段可能有誤，需要人類檢查並調整。
與「批准或拒絕」不同，這裡不只是二元選擇，而是允許人類針對輸出內容做更細緻的修改，甚至可以由系統提供多個候選版本，讓人類挑選最合適的一個。
以下是一個簡單的範例：
import { Annotation, StateGraph, interrupt, Command, MemorySaver, START } from '@langchain/langgraph';
// 定義狀態
const GraphAnnotation = Annotation.Root({
summary: Annotation<string>(),
});
// 定義需要人工編輯的節點
function humanEditing(state: typeof GraphAnnotation.State) {
const result = interrupt({
task: '請審閱以下由 AI 生成的摘要，並進行必要的修改',
summary: state.summary, // 要交由人類檢查的輸出
});
// 將人類修改過的內容更新回狀態
return {
summary: result.text,
};
}
// 建立流程
const workflow = new StateGraph(GraphAnnotation)
.addNode('human_editing', humanEditing)
.addNode('reviewed_summary', async (state) => ({
message: `最終摘要：${state.summary}`,
}))
.addEdge(START, 'human_editing')
.addEdge('human_editing', 'reviewed_summary');
// 加入 checkpointer
const graph = workflow.compile({ checkpointer: new MemorySaver() });
// 指定 thread ID
const threadConfig = { configurable: { thread_id: 'edit-thread' } };
// 第一次執行，會停在 interrupt，等待人類輸入
const firstRun = await graph.invoke(
{ summary: '這是 AI 自動生成的摘要' },
threadConfig
);
console.log(firstRun);
/*
{
summary: '這是 AI 自動生成的摘要',
__interrupt__: [
{
id: 'd4994f05f4503bdbea0547e07a0ba89e',
value: {
task: '請審閱以下由 AI 生成的摘要，並進行必要的修改',
summary: '這是 AI 自動生成的摘要'
}
}
]
}
*/
// 模擬人類輸入修改後的內容
const resumedRun = await graph.invoke(
new Command({ resume: { text: '這是人類修改後的最終摘要' } }),
threadConfig
);
console.log(resumedRun);
// { summary: '這是人類修改後的最終摘要' }
在這個例子中：
humanEditing 節點 → 觸發 interrupt()，並將內容交給人類審閱。這樣，我們就能確保 AI 的輸出在最終使用前，經過人工審核與調整，讓結果更可靠、更符合需求。
在實務應用中，AI 並不總是擁有完整資訊來做出正確決策。這時候我們可以透過 interrupt() 主動請求人類補充輸入，例如：
以下是一個簡單的範例，展示如何讓 AI 與人類交替提供輸入，建立更靈活的對話流程：
import { ChatOpenAI } from "@langchain/openai";
import {
StateGraph,
MessagesAnnotation,
interrupt,
Command,
MemorySaver,
START,
} from "@langchain/langgraph";
// 建立 LLM 模型
const model = new ChatOpenAI({
model: "gpt-4o-mini",
});
// 節點：呼叫 LLM
async function callModel(state: typeof MessagesAnnotation.State) {
const response = await model.invoke(state.messages);
return { messages: [response] };
}
// 節點：請求人類補充輸入
function humanInput(state: typeof MessagesAnnotation.State): Command {
const aiMessage = state.messages[state.messages.length - 1];
// 透過 interrupt 暫停，請求人類回覆
const userInput = interrupt({ aiMessage: aiMessage.content });
const humanMessage = { role: "human", content: userInput };
// 人類輸入完成後，流程回到 call_model 繼續
return new Command({
goto: "call_model",
update: { messages: [humanMessage] },
});
}
// 建立流程
const workflow = new StateGraph(MessagesAnnotation)
.addNode("call_model", callModel)
.addNode("human_input", humanInput, { ends: ["call_model"] })
.addEdge(START, "call_model")
.addEdge("call_model", "human_input");
// 編譯流程，並使用 MemorySaver 保存中斷狀態
const graph = workflow.compile({ checkpointer: new MemorySaver() });
// 指定 thread ID
const threadConfig = { configurable: { thread_id: "chat-thread" } };
// 第一次執行：AI 先回應，然後暫停等待人類輸入
const firstRun = await graph.invoke(
{ messages: [{ role: "user", content: "Hello!" }] },
threadConfig
);
console.log(firstRun);
/*
{
messages: [...],
__interrupt__: [
{
id: 'ae867768c5769ed0b80b1cf9b96c1609',
value: { aiMessage: 'Hello! How can I assist you today?' },
resumable: true
}
]
}
*/
// 模擬人類輸入，流程繼續執行
const resumedRun = await graph.invoke(
new Command({ resume: "請用一句話介紹 Human-in-the-loop 是什麼？" }),
threadConfig
);
console.log(resumedRun);
/*
{
messages: [...],
__interrupt__: [
{
id: '043ceac3d210aad68e10a94e09173de8',
value: {
aiMessage: 'Human-in-the-loop（HITL）是一種系統設計方法，將人類專家與自動化技術結合，以提升決策過程的準確性和可靠性。'
},
resumable: true
}
]
}
*/
在這個例子中：
Hello! → LLM 產生回應。human_input 節點中斷 → 將 AI 回覆交給人類檢視，等待輸入。call_model 節點，讓 LLM 繼續處理。這樣的設計能讓 AI 在需要額外資訊時，隨時插入人類輸入，避免因資訊不足導致的錯誤判斷。
今天我們學習了如何在 AI Agent 流程中引入 人類參與流程（Human-in-the-loop, HITL），透過設置 中斷點（interrupt），讓流程在需要時暫停並等待人工確認或補充，進而建立更安全、靈活的協作機制：
interrupt() 是實現 HITL 的核心機制，能讓流程暫停、輸出提示並等待使用者回覆，再透過 Command 恢復，並需搭配 checkpointer 與 thread_id 確保狀態正確追蹤。這樣的設計讓自動化與人工判斷能互補搭配，使 AI 不再只是執行者，而能與人類真正協作，打造更安全、可靠且值得信任的智慧系統。
iThome鐵人賽
看影片追技術