Pydantic AI 請求重試與錯誤處理
在與外部的大型語言模型 (LLM) API 互動時,網路不穩定、伺服器暫時性的 502 Bad Gateway 錯誤、或是觸發了速率限制 (Rate Limit / 429 Too Many Requests) 都是極為常見的情況。如果沒有妥善的錯誤處理機制,你的 AI 應用程式會變得非常脆弱且容易崩潰。
此外,當我們要求模型進行結構化輸出 (Structured Output) 時,模型偶爾也會犯錯,產生不符合 JSON Schema 規則的回覆。
Pydantic AI 內建了優雅且強大的重試機制 (Retries),能自動處理上述這些問題。
基本重試設定 (Retries)
在建立 Agent 時,你可以透過 retries 參數來設定當發生錯誤時,框架應該自動重試幾次。預設情況下,Pydantic AI 已經具有基礎的重試次數,但你可以根據需求覆寫它。
from pydantic_ai import Agent
# 設定這個 Agent 遇到可恢復的錯誤時,最多自動重試 3 次
agent = Agent(
'openai:gpt-4o',
retries=3
)
這個 retries 參數保護了兩層不同的錯誤:
- 底層網路與 API 錯誤:如連線逾時、伺服器端錯誤。
- 模型驗證錯誤:如模型回傳的資料不符合 Pydantic 定義的結構。
處理結構化輸出驗證錯誤
這是 Pydantic AI 重試機制中最具智慧的部分。
當你設定了 result_type 要求特定的輸出格式,如果 AI 產出的結果在經過 Pydantic 驗證時失敗(例如:我們要求一個 int,但模型回傳了字串 "unknown"),Pydantic AI 不會直接拋出錯誤讓程式掛掉。
相反地,框架會捕獲這個 Pydantic 驗證錯誤,並將其轉化為一個友善的提示訊息,再次發送給模型。
AI 會收到類似這樣的訊息:「你的上一次回覆在 age 欄位發生了型別錯誤,請回傳整數而不是字串,請修正並重試。」
這個過程在 retries 允許的次數內會自動進行,對開發者完全透明,大幅度提升了結構化資料提取的成功率。
自定義工具函數的重試 (ModelRetry)
有時候,錯誤不是發生在模型端,而是發生在你寫的 @agent.tool 工具函數內部。
例如,你的工具呼叫了某個不穩定的外部 API,剛好該 API 回傳了錯誤。這時如果你直接使用 Python 內建的 raise Exception,整個 Agent 的執行流程就會中斷崩潰。
為了解決這個問題,Pydantic AI 提供了 ModelRetry 異常類別。當你在工具內捕獲到預期內的錯誤時,你可以拋出 ModelRetry。這會告訴 Pydantic AI:「這個工具執行失敗了,請將失敗原因告訴模型,讓模型自己決定要不要換個參數再試一次,或是直接向使用者回報無法取得資料。」
import httpx
from pydantic_ai import Agent, ModelRetry
agent = Agent('gemini-1.5-flash', retries=2)
@agent.tool
def get_user_profile(user_id: str) -> str:
"""透過 ID 取得使用者檔案"""
try:
# 假設這裡呼叫外部 API
response = httpx.get(f"https://api.example.com/users/{user_id}")
response.raise_for_status()
return response.text
except httpx.HTTPStatusError as e:
# 假如找不到使用者 (404),我們可以拋出 ModelRetry
# 模型收到這個文字後,可能會決定請使用者重新確認 ID 是否正確
raise ModelRetry(f"找不到 ID 為 {user_id} 的使用者,伺服器回傳:{e}")
except httpx.RequestError as e:
# 網路連線錯誤,也可以拋出 ModelRetry,讓模型知道工具暫時失效
raise ModelRetry(f"外部 API 連線異常,請稍後再試。錯誤細節:{e}")
result = agent.run_sync("請幫我查詢 ID 為 12345 的使用者資料。")
print(result.data)
透過拋出 ModelRetry,你等於是把錯誤處理的決策權交還給了 AI 代理人,讓它表現得更像一個遇到阻礙會想辦法解決的智慧助手,而不是一遇到 Bug 就當機的傳統程式碼。