Agent、工具與人工審核Agents, tools and approvals
讓模型呼叫工具,等於讓一個會犯錯、可能被操弄的使用者操作你的系統。這一章談怎麼設計工具、保存狀態,以及在對的地方放上人工審核與權限限制。
Letting a model call tools means letting a fallible, manipulable user operate your systems. This chapter covers designing tools, persisting state, and putting human review and permission limits in the right places.
讀完這章你能夠After this chapter you can
- 判斷一個任務該用固定的 workflow 還是自主的 agent。
- 設計範圍明確、回傳錯誤、可安全重試的工具。
- 實作 checkpoint、中斷與恢復,讓 agent 能等人核准。
- 把風險標在工具上,並用資料庫權限而不是關鍵字比對來保護資料。
- Decide whether a task needs a fixed workflow or an autonomous agent.
- Design tools that are narrow, return errors, and are safe to retry.
- Implement checkpoints and interrupt/resume so an agent can wait for approval.
- Declare risk on tools and protect data with database permissions, not keyword matching.
Workflow 還是 agent?Workflow or agent?
Anthropic 在〈Building effective agents〉一文中做了很實用的區分:workflow 是由程式決定步驟、在固定的位置呼叫 LLM;agent 則由模型自己決定下一步要做什麼、用哪個工具、何時結束。agent 比較彈性,但也比較貴、比較慢、比較難預測與除錯。
Anthropic's "Building effective agents" draws a useful line: in a workflow, code decides the steps and calls the LLM at fixed points; in an agent, the model decides what to do next, which tool to use and when to stop. Agents are more flexible, but also costlier, slower, and harder to predict and debug.
用 workflowUse a workflow
步驟可以事先寫出來,例如:分類 → 檢索 → 回答 → 驗證。大多數企業流程屬於這一類。lab 的 SRE 診斷就是 workflow。
The steps can be written down in advance: classify → retrieve → answer → validate. Most enterprise processes fit here. The lab's SRE diagnosis is a workflow.
用 agentUse an agent
步驟取決於中間結果、事先無法列舉,而且錯誤可以被偵測並復原。例如:在陌生的程式庫裡找出 bug 的原因。
The steps depend on intermediate results, can't be enumerated up front, and errors can be detected and recovered from. For example: finding the cause of a bug in an unfamiliar codebase.
先從最簡單、能滿足需求的做法開始。一次 LLM 呼叫能解決的,就不需要 workflow;workflow 能解決的,就不需要 agent。
Start with the simplest thing that meets the need. If one LLM call solves it, you don't need a workflow; if a workflow solves it, you don't need an agent.
Agent 迴圈The agent loop
設計工具Designing tools
工具是模型和你的系統之間的介面。好的工具設計比 prompt 技巧更能決定 agent 的可靠性。
Tools are the interface between the model and your systems. Good tool design matters more for agent reliability than prompt tricks.
| 原則Principle | 不好Bad | 好Good |
|---|---|---|
| 範圍窄Narrow scope | run_sql(query) |
get_order(order_id), refund_order(order_id, amount) |
| 錯誤當結果回傳Errors as results | 拋出例外讓整個迴圈崩潰Throw and crash the loop | 回傳「找不到 order 123,請確認編號」,讓模型修正Return "order 123 not found; check the id" so the model can adjust |
| 可安全重試Safe to retry | 重試會重複扣款A retry charges twice | 寫入類工具接受 idempotency keyWrite tools accept an idempotency key |
| 輸出有上限Bounded output | 回傳十萬列資料塞爆 contextReturn 100k rows and flood the context | 分頁、摘要、明確的筆數限制Pagination, summaries, explicit limits |
| 宣告風險Declared risk | 掃描參數字串找 DROP、DELETEScan argument strings for DROP or DELETE | 每個工具標註 read / write / destructiveEach tool is tagged read / write / destructive |
舊版教材用「SQL 字串裡有沒有 DELETE、DROP」決定要不要人工核准。這很容易被繞過:WITH x AS (DELETE ...) SELECT、大小寫與註解變化、呼叫有副作用的函式,或乾脆用 UPDATE ... SET deleted_at = now()。正確的做法是兩層:工具本身宣告風險,加上資料庫層級的權限(唯讀工具用唯讀連線)。
Older material decided on approval by checking whether a SQL string contains DELETE or DROP. That is easy to bypass: WITH x AS (DELETE ...) SELECT, case and comment tricks, side-effecting functions, or simply UPDATE ... SET deleted_at = now(). Do it in two layers instead: the tool declares its risk, and the database enforces permissions (read tools use a read-only connection).
def make_database():
uri = f"file:lab-{uuid.uuid4().hex}?mode=memory&cache=shared"
rw = sqlite3.connect(uri, uri=True)
...
ro = sqlite3.connect(uri, uri=True)
ro.execute("PRAGMA query_only = ON") # 資料庫層級的唯讀 / enforced by the database
return rw, ro
tools = [
Tool("run_select", "Run a read-only SQL query.", "read", run_select), # 使用 ro
Tool("delete_expired_tokens", "Delete tokens expired more than N days ago.",
"destructive", delete_expired_tokens), # 使用 rw,需核准
]
狀態、checkpoint 與恢復State, checkpoints and resume
人工核准可能要等幾個小時,伺服器可能在中間重新部署。所以 agent 的狀態必須能序列化並存到外部,每一步都存一次。恢復時用同一個 thread ID 讀回最後的狀態,從中斷的地方繼續。
A human approval can take hours and the server may be redeployed meanwhile. So agent state must be serializable and stored externally, saved after every step. On resume, load the latest state for the same thread ID and continue where it stopped.
lab 用不到一百行實作這個機制,好讓你看清楚框架在背後做什麼:
The lab implements this in under a hundred lines so you can see what frameworks do behind the scenes:
def _loop(self, state: AgentState) -> AgentState:
while state.status == "running":
if state.steps >= state.max_steps:
state.status = "halted" # 預算用完 / budget exhausted
break
call = self.planner.next_action(state)
if call is None:
state.status = "done"
break
tool = self.tools.get(call.tool)
if tool is None:
state.history.append({"tool": call.tool, "args": call.args, "error": "unknown tool"})
state.steps += 1
elif tool.risk != "read":
state.pending = asdict(call)
state.status = "waiting_approval" # interrupt
else:
self._execute(state, call)
self.store.save(state) # checkpoint(存 JSON 字串)
self.store.save(state)
return state
def resume(self, thread_id: str, decision: Decision) -> AgentState:
state = self.store.latest(thread_id) # 可能在另一個程序裡執行
...
state.approvals.append({"tool": call.tool, "args": call.args, "approved": decision.approved,
"approver": decision.approver, "reason": decision.reason, "at": ...})
動手試:中斷與恢復Try it: interrupt and resume
逐步執行 agentStep through the agent
與 labs/core/agent.py 相同的流程。選「清理過期 token」,一步步執行到中斷,然後試試「模擬程序重啟」:狀態從 checkpoint 還原,仍然停在等待核准。
The same flow as labs/core/agent.py. Choose "clean up expired tokens", step until it interrupts, then try "simulate restart": the state is restored from the checkpoint and is still waiting for approval.
steps:
checkpoints: 資料表筆數rows in table:
核准稽核紀錄Approval audit trail
把人工核准做對Getting human approval right
- 核准的是「確切的參數」。畫面要顯示實際會執行的內容(哪些 ID、多少筆),而不是模型寫的摘要。如果參數在核准後改變,就要重新核准。
- 記錄誰、何時、核准了什麼、為什麼。這是稽核的需求,也是事後檢討的依據。
- 核准要有期限。等了三天才按核准,資料可能已經不一樣了。過期就重新規劃。
- 不要讓人疲勞。每一步都要核准,使用者很快就會不看內容直接按。只對真正有風險的操作要求核准,其餘靠權限與可回滾設計保護。
- Approve the exact arguments. Show what will actually run (which IDs, how many rows), not the model's summary. If arguments change after approval, approve again.
- Record who approved what, when and why. Audits need it, and so do post-incident reviews.
- Expire approvals. If someone clicks approve three days later, the data may have changed. Re-plan instead.
- Avoid approval fatigue. If every step needs approval, people stop reading and click through. Require it only for genuinely risky operations; protect the rest with permissions and reversible design.
用框架實作:LangGraphWith a framework: LangGraph
LangGraph 提供同樣的機制:狀態圖、checkpointer、interrupt() 與 Command(resume=...)。注意這個例子讓核准畫面顯示確切要刪除的 ID,刪除時也只刪這些 ID。
LangGraph provides the same mechanics: a state graph, a checkpointer, interrupt() and Command(resume=...). Note that the approval payload shows the exact IDs to delete, and the delete touches only those IDs.
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver # 正式環境用 PostgresSaver
from langgraph.types import interrupt, Command
class State(TypedDict, total=False):
expired_ids: list[int]
deleted: int
def find_expired(state: State) -> State:
rows = ro_db.execute("SELECT id FROM api_tokens WHERE expires_days_ago > 30").fetchall()
return {"expired_ids": [r[0] for r in rows]}
def delete_expired(state: State) -> State:
# 恢復時,這個節點會從頭重新執行;interrupt() 之前的程式碼必須沒有副作用
decision = interrupt({"action": "delete_tokens", "ids": state["expired_ids"]})
if not decision.get("approved"):
return {"deleted": 0}
cur = rw_db.executemany("DELETE FROM api_tokens WHERE id = ?", [(i,) for i in state["expired_ids"]])
rw_db.commit()
return {"deleted": cur.rowcount}
builder = StateGraph(State)
builder.add_node("find_expired", find_expired)
builder.add_node("delete_expired", delete_expired)
builder.add_edge(START, "find_expired")
builder.add_edge("find_expired", "delete_expired")
builder.add_edge("delete_expired", END)
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "token-cleanup-2026-10-13"}}
first = graph.invoke({}, config) # 停在 interrupt;回傳值含 __interrupt__ 內容
# ……數小時後,可能在另一個程序……
graph.invoke(Command(resume={"approved": True, "approver": "alice@example.com"}), config)
在 LangGraph 中,恢復執行時,含有 interrupt() 的節點會從節點開頭重新執行。如果你在 interrupt() 之前寄了信或寫了資料,恢復時會再做一次。把副作用放在 interrupt 之後,或拆成獨立的節點。
In LangGraph, on resume the node containing interrupt() re-runs from the start of the node. If you sent an email or wrote data before interrupt(), it happens again. Put side effects after the interrupt, or in a separate node.
MCP:把工具標準化MCP: standardising tools
Model Context Protocol(MCP)是一個開放協定,讓工具以統一的方式提供給各種 AI 應用:server 宣告有哪些工具與參數 schema,client(例如 IDE 或聊天應用)負責呼叫。它建立在 JSON-RPC 2.0 上,常見的傳輸方式是 stdio 與 HTTP。
The Model Context Protocol (MCP) is an open protocol for exposing tools to AI applications in a uniform way: a server declares tools and their argument schemas; a client (an IDE, a chat app) calls them. It runs on JSON-RPC 2.0, typically over stdio or HTTP.
對 FDE 來說,MCP 的價值在於:同一個「查客戶工單系統」的 server,可以給 Claude Desktop、IDE 裡的 agent 或你自己寫的 workflow 共用。工具可以附上 annotations(例如 readOnlyHint、destructiveHint),讓 client 知道哪些操作需要確認。
For an FDE, MCP's value is reuse: the same "customer ticketing" server can serve Claude Desktop, an IDE agent or your own workflow. Tools can carry annotations (such as readOnlyHint and destructiveHint) so clients know which operations need confirmation.
{
"name": "update_cloud_run_scaling",
"description": "Change scaling settings of a Cloud Run service. Changes production; needs human approval.",
"inputSchema": {
"type": "object",
"properties": {
"name": {"type": "string"},
"min_instances": {"type": "integer"},
"max_instances": {"type": "integer"}
},
"required": ["name", "min_instances", "max_instances"],
"additionalProperties": false
},
"annotations": {"readOnlyHint": false, "destructiveHint": false, "idempotentHint": true, "openWorldHint": true}
}
annotations 是 server 自己宣稱的。來自不可信 server 的「我是唯讀的」不能當成安全依據。只對你信任的 server 依 annotations 做決策,其餘一律要求確認。
Annotations are self-declared by the server. "I'm read-only" from an untrusted server is no basis for safety. Act on annotations only for servers you trust; require confirmation for everything else.
labs/mcp_sre 實作了一個最小的 MCP server 與 client:握手與版本協商、notification 不回應、參數驗證、工具錯誤以 isError 回傳,以及「修改正式環境前必須核准、修改後重新讀取狀態驗證」的診斷流程。測試會真的以子行程啟動 server,透過 stdio 溝通。
labs/mcp_sre implements a minimal MCP server and client: handshake and version negotiation, no responses to notifications, argument validation, tool errors returned via isError, and a diagnosis workflow that requires approval before changing production and re-reads status afterwards to verify. The tests launch the server as a real subprocess over stdio.
預算與停止條件Budgets and stop conditions
- 步數上限:防止在兩個工具之間無限來回。
- token 或金額上限:每次任務、每位使用者、每天。
- 時間上限:使用者在等的請求,要有明確的逾時與降級回應。
- 停下來時要說清楚:「已達步數上限,目前完成了 A、B,C 尚未處理」,而不是靜靜失敗。
- Step limit: stops endless ping-pong between two tools.
- Token or spend limit: per task, per user, per day.
- Time limit: requests a user is waiting on need explicit timeouts and a degraded response.
- Stop loudly: "step budget reached; A and B are done, C is not", rather than failing silently.