一、LangChain 重試機制深度解析
當構建生產級AI應用時,with_retry()
機制可有效提升系統容錯性,典型應用場景包括:
- API調用頻率限制時的自動恢復
- 模型服務臨時不可用的故障轉移
- 網絡波動導致的瞬時異常處理
參數詳解與配置策略
1. 參數配置表
參數名 | 類型 | 默認值 | 作用說明 | 示例值 |
---|---|---|---|---|
retry_if_exception_type | Tuple[Exception] | None | 指定需要重試的異常類型 | (APIError, TimeoutError) |
wait_exponential_jitter | bool | True | 啟用指數退避+隨機抖動策略 | False |
stop_after_attempt | int | 3 | 最大重試次數(含初始調用) | 5 |
2. 代碼示例
想要讓一個 Runnable
組件最多重試 2
次,只需在 with_retry()
函數中傳遞 stop_after_attempt
參數即可,代碼:
from langchain_core.runnables import RunnableLambdacounter = -1def func(x):global countercounter += 1print(f"當前的值為 {counter=}")return x / counterchain = RunnableLambda(func).with_retry(stop_after_attempt=2)resp = chain.invoke(2)print(resp)
3. 資料推薦
- 🔗 官方文檔參考
- 💡大模型中轉API推薦
- ?中轉使用教程
二、with_retry() 函數的運行原理
構建一個新的 Runnable,在執行調用類的函數時,循環特定次數,直到組件能正常執行結束即暫停,并且在每次循環的過程中,休眠特定的時間,運行流程圖如下: