速率限制應該設在哪些層面?每個層面的具體限制值怎麼定?
速率限制需要在三個層面同時設計,形成層層保護:
LLM API 調用層(防止 Token 費用爆炸): 限制 Agent 每個任務循環的最大 LLM 調用次數(通常 5-15 次,超過說明 Agent 可能陷入無限循環);限制每小時的 LLM 調用次數(根據你的 LLM API 計劃和預算設定);限制單次調用的最大 Token 上限(超長的輸入 Prompt 也需要限制,防止 Context 費用超預期)。
工具調用層(防止外部 API 被過度請求):
對每個工具函數設定速率限制,例如 get_aave_apy 最多每分鐘調用 10 次(超過說明 Agent 在異常重試);對寫入工具(鏈上操作)的速率限制更嚴格——例如 withdraw_from_protocol 最多每小時 3 次。
鏈上操作層(最重要,直接影響資金安全): 單日最大交易筆數(例如每日上限 10 筆鏈上交易);單日最大 Gas 消耗(以 ETH 或 USD 計算);單日最大操作金額(例如單日移動的 USDC 不超過總倉位的 30%);單次最大操作金額(單筆操作的絕對金額上限)。
速率限制值的設定方法: 在測試網跑 48-72 小時,記錄正常操作下每個指標的峰值,把生產環境的速率限制設定為峰值的 1.5-2 倍(留一些正常波動的空間,但不允許異常的大幅超出)。
熔斷的觸發條件怎麼設計?熔斷觸發後 Agent 應該做什麼?
熔斷觸發條件的設計需要覆蓋四類異常場景:
類型一:成本類熔斷(保護資金預算)
類型二:安全類熔斷(偵測攻擊信號)
類型三:工具失敗類熔斷(防止在服務故障時繼續操作)
類型四:市場異常類熔斷(防止在極端市場條件下執行)
熔斷觸發後的 Agent 行為: 自動停止所有新的工具調用;記錄完整的熔斷日誌(觸發原因、觸發時間、當前狀態);通過 Telegram 發送 P0 告警(帶熔斷原因);等待人工確認(不自動恢復)。注意:熔斷不是「關機」,而是「暫停等待確認」——Agent 的狀態(當前持倉、上一次任務進度)應該被完整保留,讓人工確認後能從暫停點繼續,而不是重新開始。
速率限制在 Python 代碼裡怎麼實作?有沒有現成的庫可以用?
在 Python 裡實作 Agent 速率限制有幾種方式,從最簡單到最完整:
方式一:使用 ratelimit 庫(最簡單,適合函數級速率限制)
from ratelimit import limits, sleep_and_retry → @sleep_and_retry → @limits(calls=10, period=60) → def get_aave_apy(): ...
這讓 get_aave_apy 函數每 60 秒最多只能被調用 10 次,超過時自動等待而不是報錯。sleep_and_retry 裝飾器讓超限的調用自動等到下一個窗口再執行,而不是拋出異常。
方式二:使用 Redis + 滑動窗口計數器(適合分散式場景) 如果你的 Agent 跑在多個實例上(多個 Sub-agent 共享速率限制),用 Redis 維護全局的速率計數器:每次調用前在 Redis 裡遞增計數器,過期時間設為速率窗口;如果計數器已達上限,等待後重試。這讓多個 Sub-agent 的調用次數共享一個全局限制,不會因為單個 Sub-agent 行為正常但多個加起來超限。
方式三:PostgreSQL 計數表(最適合 Onchain Agent 的持久化速率限制)
在 PostgreSQL 裡維護一個 agent_rate_counters 表,記錄每種操作今天的執行次數。每次執行前先查詢(今天的 Gas 消耗是否超過預算?今天的鏈上交易次數是否達上限?),查詢通過後執行並更新計數。這讓速率限制在 Agent 重啟後仍然有效(Redis 在 Agent 重啟後可能重置,PostgreSQL 是持久化的),特別適合「每日操作上限」類型的速率限制。
熔斷的 Python 實作:
使用 pybreaker 庫(開源,專門的 Circuit Breaker 實作)或手動維護一個 circuit_state 變量(CLOSED/OPEN/HALF-OPEN),在每次工具調用前檢查 circuit_state,OPEN 狀態下拒絕所有操作。
速率限制和熔斷怎麼測試?怎麼確認它們在生產環境裡真的有效?
速率限制和熔斷的測試需要主動觸發邊界條件,而不只是「希望它們工作」:
測試速率限制(在測試網): 寫一個測試腳本,在 1 分鐘內連續調用某個受速率限制的工具函數 20 次,確認:第 11 次(超過速率限制)是自動等待還是拋出錯誤?速率限制的計數器在 Agent 重啟後是否被正確恢復(測試 PostgreSQL 持久化是否有效)?速率限制是否正確地在跨任務循環間共享(不只是限制單個循環的調用次數)?
測試熔斷(在測試網):
手動觸發熔斷條件——直接修改 agent_rate_counters 表裡的日 Gas 消耗值,讓它超過設定閾值,確認熔斷是否被觸發;模擬 Validation BLOCKED——在後端驗證代碼裡臨時讓所有操作返回 BLOCKED,確認 3 次 BLOCKED 後熔斷是否觸發;測試熔斷後的 Agent 行為:確認熔斷觸發後,Agent 確實停止了所有新的工具調用;確認 Telegram 告警被正確發送;確認 Agent 狀態被正確保存(讓你能在確認後從暫停點繼續)。
生產環境的持續驗證: 在監控系統裡加一個「速率限制健康檢查」——定期(例如每天)查詢速率計數器,確認計數器在正常範圍內增長(說明 Agent 在正常操作,速率限制沒有不必要地被觸發或完全沒有被記錄)。對熔斷:記錄所有熔斷觸發的歷史,每次熔斷後做根因分析——是正確的安全攔截,還是誤觸發(說明閾值需要調整)?
一個完整的速率限制 + 熔斷設計例子
以下是一個 DeFi 利率優化 Agent 的速率限制和熔斷設計(Python 偽代碼風格):
# 速率限制配置
RATE_LIMITS = {
'lmm_calls_per_cycle': 10, # 每個任務循環最多 10 次 LLM 調用
'apy_queries_per_min': 5, # 每分鐘最多 5 次 APY 查詢
'write_ops_per_hour': 3, # 每小時最多 3 次鏈上寫入
'daily_gas_budget_usd': 20, # 每日 Gas 預算 $20
'daily_tx_count': 8 # 每日最多 8 筆鏈上交易
}
# 熔斷觸發條件
CIRCUIT_BREAKERS = {
'gas_exceeded_200pct': True, # 日 Gas 超 200% → 立即熔斷
'blocked_3x_30min': True, # 30 分鐘內 3 次 BLOCKED → 立即熔斷
'tvl_drop_20pct_1hr': True, # 協議 TVL 1 小時內降 20% → 熔斷
'tool_fail_5x': True # 關鍵工具連續失敗 5 次 → 熔斷
}
# 每次工具調用前的檢查流程(代碼層面強制,不依賴 LLM)
def pre_execute_check(operation_type: str) -> bool:
check_rate_limit(operation_type) # 速率限制
check_circuit_breaker() # 熔斷狀態
check_daily_budget() # 每日預算
return True # 通過所有檢查才執行
這個設計的關鍵:pre_execute_check() 在工具調用函數的最外層執行,不可被 LLM 的 Thought 推理繞過——即使 LLM 產生了幻覺或被 Prompt Injection 攻擊,在代碼層面的速率限制和熔斷檢查仍然有效。
速率限制越嚴格(上限越低)→ Agent 越安全(異常時影響範圍小),但在 Gas 費低谷期或高收益機會窗口,Agent 可能因為達到速率上限而錯過最優執行時機(策略效率降低)。速率限制越寬鬆(上限越高)→ 策略執行效率更高,但異常時的潛在影響更大。建議:生產初期嚴格(保護本金),在確認 Agent 行為穩定後根據實際數據適度放寬。熔斷的觸發閾值也一樣:太敏感(頻繁誤觸發)會讓 Agent 頻繁暫停影響策略效率;太遲鈍(閾值太高)則失去保護作用。每個月基於運行數據重新校準一次。