AG Day 7 錯誤處理:逾時、重試與工具失敗
執行需求:CPU+API key。在上一篇 AG Day 6(原文連結)中,我們以純 Python 親手打造了自主代理迴圈(Agent Loop),透過五階段狀態機成功驅動模型自主完成多步驟的資料檢索與運算收斂。然而,當 AI Agent 走出平穩的本機測試、進入變幻莫測的真實網路環境時,各種非預期的分散式系統故障將接踵而至:第三方 API 可能因為瞬間尖峰負載而無預警逾時(Timeout)、網路節點可能突然抖動斷線、模型也可能因為幻覺產出帶有型別錯誤或不存在欄位的工具引數。如果在面對這些偶發或語意故障時,系統唯一的反應就是拋出未捕獲的例外並導致整條研調流程崩潰,那麼無論 Agent 表面具備多高的推論智力,它在生產環境中都將脆弱不堪。今天,我們將深入剖析生產級 Agent 的防禦性架構,全面實作嚴格的逾時控制、基於指數退避(Exponential Backoff)與抖動(Jitter)的智慧重試機制,並將工具失敗轉化為模型自我診斷與自我修復(Self-healing)的認知回饋,打造真正具備工業級容錯能力的強韌研究助理。
引言
在傳統軟體工程中,例外處理通常遵循嚴格的確定性邏輯:當發生錯誤時,捕捉特定型別的 Exception,記錄錯誤日誌,然後回傳錯誤碼或將請求中止。但在由大型語言模型驅動的自主代理系統中,故障的本質發生了深刻的轉變:錯誤不再只是系統終止的句點,而是推動模型自我修正的全新線索。
試想一個典型的場景:模型試圖呼叫資料庫查詢工具,但不小心傳入了非法的日期格式 2026/02/30。在缺乏容錯設計的系統中,這行程式碼會拋出 ValueError,隨即引發整個應用程式崩潰,長達數分鐘的研調脈絡瞬間灰飛煙滅。但在強韌的 Agent 架構中,系統會捕獲這個例外,將錯誤詳情清晰地包裝成 role="tool" 的回饋訊息送回上下文:「執行失敗:無效日期格式,請使用 YYYY-MM-DD 標準格式」。模型讀取這段錯誤後,會在下一輪推理中主動意識到問題所在,調整引數為 2026-02-28 並重新發起呼叫。這種將「程式例外」轉化為「認知回饋」的自癒閉環,正是生產級 Agent 能在開放世界中穩定運作的終極關鍵。
原理/觀念
代理系統的錯誤分類學(Error Taxonomy)
要建立高可用的容錯體系,第一步是精確區分錯誤的根本成因。在 AI Agent 生命週期中,錯誤大致可歸納為三大維度:
- 暫態故障(Transient Failures):包含網路傳輸暫態抖動、供應商 API 速率限制(HTTP 429 Too Many Requests)、以及後端閘道伺服器短暫過載(HTTP 502/503/504)。這類錯誤具備高度的隨機性與暫時性,程式碼本身並沒有任何錯誤。對於暫態故障,最佳應對策略是透過指數退避與隨機抖動進行自動重試。
- 語意與契約故障(Semantic / Contract Failures):包含模型傳入不存在的工具名稱、遺漏必填引數、數值超出業務範圍、或是查詢無效的資料格式。這類錯誤是由於模型推理瑕疵引起的,單純在底層機械式重試一百次也毫無意義。最佳應對策略是捕捉錯誤,提煉結構化提示文字,交由模型進行自我修復。
- 致命永久故障(Fatal / Permanent Failures):包含未設定有效的 API 金鑰(HTTP 401 Unauthorized)、目標資料庫毀損、磁碟空間耗竭等硬體或授權錯誤。這類錯誤必須立即中斷流程並發出最高級別告警,絕對不能無效重試,以免徒耗運算資源與成本。
指數退避(Exponential Backoff)與隨機抖動(Full Jitter)
當遭遇 API 速率限制或暫態網路斷線時,若客戶端在失敗後立即以固定頻率重試,極易引發「驚群效應(Thundering Herd Problem)」,導致原本已經負載過重的伺服器被瞬間湧入的重試請求徹底擊潰。工業級標準的重試演算法採用指數退避結合隨機抖動:
每次重試的等待時間 $T_{wait}$ 隨嘗試次數呈幾何級數增長,並加入均勻分佈的隨機微擾值:
$$T_{wait} = \text{random}(0, \; \min(T_{max}, \; T_{base} \times 2^{\text{attempt}}))$$
透過隨機抖動,分散在同一個時間點失敗的大量客戶端會被均勻錯開到不同的時間區間重試,從根本上平滑了尖峰流量,保護後端伺服器體系。
分層逾時防護(Layered Timeouts)
沒有逾時保護的網路呼叫是系統掛死的最大元凶。在 Agent 體系中,我們必須實施雙層逾時隔離:
- 模型推理逾時:針對大語言模型的 HTTP 請求設定連線逾時(Connect Timeout, 如 5 秒)與讀取逾時(Read Timeout, 如 30 秒)。若模型遲遲未產出回覆,及時熔斷並進行重試。
- 工具執行逾時:針對代理所調度的本地或外部工具(如爬蟲、資料庫查詢、繁重運算)設定獨立的執行緒逾時(如 10 秒)。嚴防因為某個外部網頁掛起而導致整個研究助理無限期停滯。
完整實作
今天我們將在 src/research_agent/resilience.py 中實作一套生產級的彈性容錯模組。本實作採用 Python 業界權威重試庫 tenacity,結合執行緒池逾時控制與語意自癒機制,為我們的 research-agent 穿上一層無懈可擊的防彈衣。
第一步:建立具備指數退避與隨機抖動的模型重試裝飾器 src/research_agent/resilience.py:
# research-agent/src/research_agent/resilience.py
import time
import random
import logging
from typing import Callable, Any
from tenacity import (
retry,
stop_after_attempt,
wait_exponential_jitter,
retry_if_exception_type,
before_sleep_log
)
logger = logging.getLogger("research_agent.resilience")
class TransientAPIError(Exception):
"""定義可安全重試的暫態網路或速率限制錯誤"""
pass
class PermanentAPIError(Exception):
"""定義不可重試的致命授權或系統錯誤"""
pass
def create_model_retry_decorator(max_attempts: int = 3) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
"""建立專為大型語言模型 API 量身打造的重試裝飾器"""
return retry(
reraise=True,
stop=stop_after_attempt(max_attempts),
wait=wait_exponential_jitter(initial=1.0, max=8.0, jitter=1.0),
retry=retry_if_exception_type(TransientAPIError),
before_sleep=before_sleep_log(logger, logging.WARNING)
)
第二步:實作具備硬性逾時限制的執行緒安全工具調度器 src/research_agent/timeout_executor.py:
# research-agent/src/research_agent/timeout_executor.py
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeoutError
from typing import Callable, Any
def execute_with_timeout(
func: Callable[..., str],
timeout_seconds: float,
*args,
**kwargs
) -> str:
"""在獨立執行緒中執行本地工具,強制在指定秒數內中斷逾時操作"""
with ThreadPoolExecutor(max_workers=1) as executor:
future = executor.submit(func, *args, **kwargs)
try:
return future.result(timeout=timeout_seconds)
except FuturesTimeoutError:
return f"執行逾時錯誤:工具 [{func.__name__}] 執行超過上限 {timeout_seconds} 秒,系統已強制截斷。"
except Exception as exc:
return f"執行內部錯誤:工具 [{func.__name__}] 拋出例外:{exc}"
第三步:實作具備自動診斷與自癒引導的強韌工具集 src/research_agent/robust_tools.py:
# research-agent/src/research_agent/robust_tools.py
import time
from research_agent.tools import ToolRegistry
from research_agent.timeout_executor import execute_with_timeout
robust_registry = ToolRegistry()
@robust_registry.register(
description="查詢特定晶片架構的功耗資料。注意:年份必須為西元年(如 2026),且晶片名稱不能為空。"
)
def fetch_chip_power_data(chip_name: str, year: int) -> str:
"""查詢晶片功耗並具備嚴格的業務引數驗證"""
# 語意驗證防護
if not chip_name or len(chip_name.strip()) == 0:
raise ValueError("引數不合法:chip_name 不得為空字串。")
if year < 2020 or year > 2030:
raise ValueError(f"引數超出合理研調範圍:年份 {year} 不在 2020 至 2030 之間。")
return f"【功耗資料庫】:晶片 [{chip_name}] 在 {year} 年的典型功耗為 18.5W,支援動態休眠模式。"
@robust_registry.register(
description="連線至遠端技術指標伺服器進行即時壓力測試模擬(此操作較耗時)。"
)
def slow_stress_simulation(duration_seconds: float) -> str:
"""模擬耗時外部呼叫以測試逾時截斷防護"""
# 內部透過逾時保護器包裝
def _inner_work():
time.sleep(duration_seconds)
return f"壓力測試順利完成,耗時 {duration_seconds} 秒。"
return execute_with_timeout(_inner_work, timeout_seconds=1.5)
第四步:實作模擬多輪自我修復的離線推理引擎 src/research_agent/resilient_simulator.py。展示模型在初次出錯後,如何讀取回饋並完成自我修正:
# research-agent/src/research_agent/resilient_simulator.py
import json
from typing import List
from research_agent.llm_types import ChatMessage
def simulate_self_healing_agent(
messages: List[ChatMessage],
step_count: int
) -> ChatMessage:
"""模擬具備自癒除錯特性的 Agent 在遭遇失敗時的應變歷程"""
last_msg = messages[-1]
# 第一輪:模型犯錯,傳入了超出範圍的年份(1999 年)
if step_count == 1:
return ChatMessage(
role="assistant",
content="我將查詢歷史與未來的硬體功耗數據,預計檢索 1999 年的基準指標。",
tool_calls=[{
"id": "call_err_01",
"type": "function",
"function": {
"name": "fetch_chip_power_data",
"arguments": json.dumps({"chip_name": "Edge-TPU", "year": 1999})
}
}]
)
# 第二輪:模型讀取到上一輪的錯誤提示,主動進行引數修正
elif step_count == 2:
return ChatMessage(
role="assistant",
content="偵測到引數年份超出範圍。依據系統回饋,我將年份修正為 2026 年重新查詢。",
tool_calls=[{
"id": "call_repair_02",
"type": "function",
"function": {
"name": "fetch_chip_power_data",
"arguments": json.dumps({"chip_name": "Edge-TPU", "year": 2026})
}
}]
)
# 第三輪:成功取得資料,整合出正確結論
else:
return ChatMessage(
role="assistant",
content=(
"【研調自癒報告】:經過引數校正與重新檢索,成功取得資料。\n"
"Edge-TPU 在 2026 年的典型功耗為 18.5W,功耗控制符合預期標準。"
),
tool_calls=None
)
第五步:撰寫暫態重試與逾時防護單元測試腳本 test_resilience_mechanics.py:
# research-agent/test_resilience_mechanics.py
import time
from research_agent.resilience import create_model_retry_decorator, TransientAPIError
from research_agent.robust_tools import robust_registry
def test_retry_mechanism():
print("=== 1. 測試指數退避重試裝飾器 ===")
attempt_counter = 0
retryer = create_model_retry_decorator(max_attempts=3)
@retryer
def flaky_api_call():
nonlocal attempt_counter
attempt_counter += 1
print(f" [API 連線嘗試第 {attempt_counter} 次]...")
if attempt_counter < 3:
raise TransientAPIError("模擬 503 伺服器短暫過載")
return "連線成功!資料已收斂。"
result = flaky_api_call()
assert result == "連線成功!資料已收斂。"
assert attempt_counter == 3
print(" -> 成功驗證:經歷 2 次重試後第 3 次成功返回!\n")
def test_timeout_mechanism():
print("=== 2. 測試工具執行逾時強制截斷 ===")
# 呼叫耗時 3.0 秒的任務,預期觸發 1.5 秒硬性逾時
res = robust_registry.dispatch("slow_stress_simulation", '{"duration_seconds": 3.0}')
print(f" 執行結果:{res}")
assert "逾時錯誤" in res
print(" -> 成功驗證:逾時截斷防護有效攔截耗時工具!\n")
if __name__ == "__main__":
test_retry_mechanism()
test_timeout_mechanism()
第六步:撰寫端到端自癒迴圈示範腳本 demo_self_healing.py,完整展示「模型出錯 — 工具回報 — 自我修復 — 任務成功」的自我修復閉環:
# research-agent/demo_self_healing.py
from research_agent.llm_types import ChatMessage
from research_agent.robust_tools import robust_registry
from research_agent.resilient_simulator import simulate_self_healing_agent
def run_self_healing_demo():
print("=== 啟動具備自我修復(Self-healing)能力之 Agent 迴圈示範 ===\n")
messages: list[ChatMessage] = [
ChatMessage(role="system", content="你是具備高度自癒能力的調研助理。"),
ChatMessage(role="user", content="請查詢 Edge-TPU 的最新功耗指標。")
]
print(f"[任務指派]: {messages[-1].content}\n")
step_count = 0
max_steps = 4
while step_count < max_steps:
step_count += 1
print(f"--- [輪次 {step_count}] 模型決策中 ---")
reply = simulate_self_healing_agent(messages, step_count)
messages.append(reply)
print(f"[助理思考]: {reply.content}")
# 檢查任務是否終止
if not reply.tool_calls:
print("\n[任務達成]:模型輸出最終研調結論!")
break
# 執行工具並回傳結果
for tool_call in reply.tool_calls:
fn_name = tool_call["function"]["name"]
fn_args = tool_call["function"]["arguments"]
print(f" -> 嘗試呼叫工具:{fn_name}({fn_args})")
tool_output = robust_registry.dispatch(fn_name, fn_args)
print(f" -> 工具執行回饋:{tool_output}")
# 將工具輸出(無論成功或拋錯)作為 tool 訊息送回對話佇列
messages.append(
ChatMessage(role="tool", content=tool_output, tool_call_id=tool_call["id"])
)
print()
print("=== 自我修復流程順利完成閉環! ===")
if __name__ == "__main__":
run_self_healing_demo()
第七步:在終端機中執行自我修復示範腳本:
python demo_self_healing.py
此時在終端機中將呈現完整的「錯誤發生 — 工具回饋 — 參數自癒」的動態軌跡(示範輸出):
=== 啟動具備自我修復(Self-healing)能力之 Agent 迴圈示範 ===
[任務指派]: 請查詢 Edge-TPU 的最新功耗指標。
--- [輪次 1] 模型決策中 ---
[助理思考]: 我將查詢歷史與未來的硬體功耗數據,預計檢索 1999 年的基準指標。
-> 嘗試呼叫工具:fetch_chip_power_data({"chip_name": "Edge-TPU", "year": 1999})
-> 工具執行回饋:工具內部執行異常:引數超出合理研調範圍:年份 1999 不在 2020 至 2030 之間。
--- [輪次 2] 模型決策中 ---
[助理思考]: 偵測到引數年份超出範圍。依據系統回饋,我將年份修正為 2026 年重新查詢。
-> 嘗試呼叫工具:fetch_chip_power_data({"chip_name": "Edge-TPU", "year": 2026})
-> 工具執行回饋:【功耗資料庫】:晶片 [Edge-TPU] 在 2026 年的典型功耗為 18.5W,支援動態休眠模式。
--- [輪次 3] 模型決策中 ---
[助理思考]: 【研調自癒報告】:經過引數校正與重新檢索,成功取得資料。
Edge-TPU 在 2026 年的典型功耗為 18.5W,功耗控制符合預期標準。
[任務達成]:模型輸出最終研調結論!
=== 自我修復流程順利完成閉環! ===
常見錯誤與踩雷
在規劃 Agent 系統的錯誤防護層時,有四個極為普遍且危險的架構反模式(Anti-patterns):
- 對 4xx 客戶端錯誤進行盲目重試:最常見的疏失是在重試裝飾器中設定了
retry_if_exception_type(Exception)。這會導致當模型傳入錯誤引數或金鑰失效(HTTP 401/400)時,系統仍進行 3 到 5 次無意義的退避重試,既浪費了寶貴的時間,又無故耗損了系統資源。必須嚴格限定只有暫態異常(如 429、502、503、連線逾時)才觸發重試。 - 將落落長且具資安疑慮的 Traceback 直接塞給模型:當工具發生例外時,若直接把完整的 Python 崩潰呼叫堆疊(Traceback)回傳給模型,不僅會消耗數千輸入 Token,更可能洩漏伺服器內部路徑與資料庫連線細節;更糟的是,雜亂的程式碼行號會嚴重干擾模型的注意力焦點。應將例外清洗為一句簡明扼要的業務診斷語句。
- 未對工具執行緒進行資源隔離與清理:使用
ThreadPoolExecutor進行逾時攔截時,若底層 Python 函式正在執行不可中斷的 C 擴充模組或阻塞式 I/O,即使主執行緒拋出TimeoutError,子執行緒仍可能在背景持續運作,造成執行緒池洩漏與記憶體耗盡。對於極耗時或高風險的外部操作,應考慮採用獨立行程(Process)或容器進行沙箱隔離。 - 自癒嘗試次數缺乏獨立計數器:若模型在同一個工具呼叫中反覆修復失敗(例如連續 3 輪給出的年份都超出範圍),代理會迅速吃光所有步數配額。實務上應針對「單一工具的修復次數」設定局部上限(如最多允許自我修復 2 次),一旦超過則強制要求模型更換備用方案或標記該資料不可用。
效能與實務提醒
在生產級架構中,彈性與延遲永遠是一場精密的權衡:
第一,整體延遲預算(Latency Budget)的全局配置:重試與退避時間不是孤立存在的。如果終端使用者在前端等待回應的逾時門檻是 30 秒,而代理的重試策略累計等待時間達到 20 秒,再加上模型推理本身的時間,請求極可能在客戶端被無情切斷。因此,重試裝飾器的最大等待上限與嘗試次數,必須受到全局延遲預算的嚴格約束。
第二,斷路器模式(Circuit Breaker)避免系統雪崩:如果某個外部檢索工具在短時間內連續失敗 5 次,代表該遠端服務極可能已經完全宕機。此時斷路器應當立即進入「開啟(Open)」狀態,在接下來的數分鐘內直接拒絕所有針對該工具的呼叫並回傳降級提示,直到服務恢復健康。這能有效防止失敗骨牌效應蔓延至整體系統。
第三,嚴格記錄重試與修復造成的額外 Token 開銷:每一次自我修復都會額外產生一輪對話往返,增加對應的輸入與輸出 Token(費用計量請依各廠商官方文件為準)。在日誌與觀測模組中,應將「自癒額外開銷」作為一項獨立的健康度指標進行追蹤,評估提示詞與 Schema 是否需要針對性最佳化。
小結
今天我們為自主研究代理注入了生產環境最不可或缺的生存能力:工業級容錯與自我修復。我們建立了精確的三維錯誤分類法,利用 tenacity 實作了具備指數退避與隨機抖動的高階重試裝飾器,運用執行緒池建立了嚴格的工具逾時防護網,並透過「錯誤即回饋」機制讓模型學會了在遭受挫折時主動調整決策、完成自我修正。
以下整理本章節的核心概念與台灣用語對照表:
- 暫態故障(Transient Failure):因網路波動或負載過重而短暫出現、隨後可自行恢復的非語意錯誤。
- 指數退避(Exponential Backoff):隨重試次數呈幾何級數增加等待時間的退避演算法。
- 隨機抖動(Full Jitter):在退避時間中引入均勻隨機微擾以化解並行衝擊的保護策略。
- 執行逾時(Execution Timeout):為防止任務無限期卡死而設定的硬性時間中斷限制。
- 自我修復(Self-healing):將系統例外訊息轉化為模型上下文,引導其主動修正錯誤行為的認知回饋機制。
結語
經過這幾天的深耕,我們已經從環境建置、對話底層、提示詞工程、函式呼叫、自主迴圈一路推進到了強韌的錯誤防禦。現在,我們的代理已經具備了強大的推理能力與極高的生存彈性。但是,回顧我們目前所得到的結論與中間狀態,它們依然是以鬆散的字串形式在系統中穿梭。如果要將這些研調成果對接到後續的關聯式資料庫、報表生成器或是向量搜尋引擎,字串的不可預測性依然是潛在的隱患。
明天,我們會進入「AG Day 8 結構化輸出:Pydantic 契約與 JSON Schema」,深入探索現代大型語言模型最重要的工業轉折點:如何使用 Pydantic v2 建立剛性的資料合約,讓模型百分之百保證產出合法、強型別的結構化研調物件!
延伸資源
- Tenacity 官方文件:Retrying Code Execution in Python。掌握進階重試策略與回呼機制。
- AWS 架構最佳實務:Exponential Backoff and Jitter。深入解析隨機抖動解決並行尖峰的數學原理。
- OpenAI 官方文件:Handling Rate Limits and HTTP Errors。大型模型 API 的標準錯誤碼處理指南。
- Martin Fowler 架構模式論文:Circuit Breaker Pattern。了解分散式容錯斷路器架構的設計精髓。
留言
張貼留言