使用 LangGraph 或 State Machine 整合 Agent 交接、Checkpoint、失敗重試及 Human-in-the-loop,保存完整狀態與稽核紀錄,並以任務成功率與人工介入率評估流程穩定度。
讀完能做到:把一條脆弱的 Agent 串接改造成中斷後可恢復、重試不重複執行副作用、關鍵動作會等待人工核准的工作流。
實作狀態:狀態轉移與流程指標為【本機核心已測試】;LangGraph 與 Gemini Spark 整合為【Spark 設計藍圖】,尚未連接正式資料庫或外部工具。
一份供應商報告依序經過資料蒐集、分析、Reviewer 與寄送。若寄送前程序逾時,整條流程從頭再跑,可能重複抓資料、重算成本,甚至再次建立採購單。可靠工作流的目標不是永不失敗,而是知道停在哪裡、能否重試,以及哪些步驟不能自動重做。
created → collecting → analyzing → reviewing → awaiting_approval
↘ retrying ↗ ↓
executing → completed
↘ failed / cancelled
每次轉移都保存 run_id、task_id、目前狀態、輸入/輸出 Schema 版本、Evidence、錯誤、重試次數與時間。狀態機禁止 created → completed 這類跳躍,也禁止完成後再執行工具。
Checkpoint 保存安全恢復點;重試處理暫時性錯誤;冪等鍵防止副作用重複。三者不能互相替代。
def call_tool(task_id, action, payload):
idempotency_key = sha256(f"{task_id}:{action}:{canonical(payload)}")
if action_store.exists(idempotency_key):
return action_store.result(idempotency_key)
result = tool.execute(payload)
action_store.save(idempotency_key, result)
return result
只對可重試錯誤做指數退避,例如 429、暫時性 5xx 或網路逾時;Schema 錯誤、權限不足與政策阻擋不能靠多試幾次解決。超過上限後進死信佇列或 blocked,交由人處理。
LangGraph 的 checkpointer 能保存 thread 狀態,用於中斷恢復與 fault tolerance;官方也提醒 InMemorySaver 在程序重啟後會遺失,正式環境應使用持久化儲存[1]。
遇到寄信、付款、刪除、正式報價或低信心決策時,流程進入 awaiting_approval,保存完整狀態後停止。LangGraph interrupt() 可暫停 Graph,之後以相同 thread_id 和 Command(resume=...) 恢復[2]。
核准資料至少包含:動作、參數、Evidence、風險、外部影響與回復方式。人可以批准、修改或拒絕;模型不能產生假的 Approval 物件繞過等待。
{
"approval_id": "AP-029",
"status": "pending",
"action": "send_daily_briefing",
"evidence_ids": ["EV-1", "EV-2"],
"requested_by": "workflow",
"approved_by": null
}
任務成功率為 completed 任務數 ÷ 全部終態任務數;人工介入率為 曾由人處理的任務數 ÷ 全部任務數。本機十筆虛構執行中九筆完成、兩筆曾介入,因此成功率 90%、介入率 20%。
介入率不是愈低愈好。高風險任務零介入可能表示核准被繞過;低風險任務介入過高,則表示自動化沒有創造價值。儀表板應依風險級別分層,再加上重試成功率、重複副作用數、P95 完成時間與 checkpoint 恢復率。
併發測試也不可少:同一冪等鍵同時抵達只能執行一次;兩個平行 Agent 更新狀態要有版本檢查;取消訊號必須阻止尚未開始的工具呼叫。Gemini Spark 可作任務入口,但企業持久化、冪等與核准仍須由受控工作流承擔[3]。
本機測試已涵蓋合法/非法狀態轉移與流程指標:
python outputs/day25_30_metrics.py
python -m unittest work/test_day25_30_metrics.py -v
可上線的 Agent 不是每次都成功,而是失敗後能從正確位置繼續、重試不製造第二次副作用,並在關鍵節點確實停下來等人決定。
[3] Google:Use Gemini Spark to manage tasks and workflows