【Bug已解决】openclaw: "rate limit exceeded" / 429 Too Many Requests — OpenClaw 请求频率限制解决方案
1. 问题描述
在使用 OpenClaw 频繁调用外部 API 或 AI 模型时，系统报出请求频率限制或 429 错误：
# API 频率限制 $ openclaw "批量调用AI API" Error: rate limit exceeded HTTP 429: Too Many Requests Rate limit: 60 requests/minute Retry after: 45s Current window: 60/60 requests used # 并发请求过多 $ openclaw --parallel 20 "并行处理" Error: 429 Too Many Requests Concurrent request limit: 10 Active requests: 20 Exceeded by: 10 # Token 速率限制 $ openclaw "分析大型文档" Error: token rate limit exceeded Token limit: 150000 TPM (tokens per minute) Used: 150000 TPM Retry after: 30s # 日配额耗尽 $ openclaw "继续执行任务" Error: daily quota exceeded Daily limit: 1000000 requests Used: 1000000 requests Resets at: 2024-07-08T00:00:00Z这个问题在以下场景中特别常见：
- 批量处理大量文件或任务
- 高并发并行调用 API
- 脚本循环调用未加限流
- 多个进程共享同一 API Key
- 免费计划配额较低
- 突发流量超过限制
2. 原因分析
OpenClaw发起请求 ↓ API网关检查频率 ←──── 滑动窗口/令牌桶 ↓ 超过限制 ←──── RPM/TPM/并发数 ↓ 返回429 ←──── "Too Many Requests" ↓ 需要等待/重试| 原因分类 | 具体表现 | 占比 |
|---|---|---|
| RPM超限 | 每分钟请求过多 | 约 35% |
| 并发超限 | 同时请求过多 | 约 25% |
| TPM超限 | 每分钟Token过多 | 约 20% |
| 日配额耗尽 | 每日上限 | 约 10% |
| 多Key冲突 | 共享Key | 约 5% |
| 突发流量 | 瞬间高并发 | 约 5% |
深层原理
API 限流通常使用三种算法：固定窗口计数器（在固定时间窗口内计数请求）、滑动窗口（在滚动时间窗口内计数，更精确）、令牌桶（以固定速率生成令牌，请求消耗令牌）。429 状态码是 HTTP 标准的"Too Many Requests"响应，通常附带Retry-After头告知客户端等待时间。API 供应商通常设置多个维度的限制：RPM（Requests Per Minute，每分钟请求数）、TPM（Tokens Per Minute，每分钟 Token 数）、并发请求数（同时进行的请求）、日配额（每日总请求量）。当任何一个维度超限时，API 返回 429 错误。
3. 解决方案
方案一：实现请求限流器（最推荐）
# 创建令牌桶限流器 import time import threading from collections import deque from functools import wraps class TokenBucketRateLimiter: """令牌桶限流器""" def __init__(self, rate=60, capacity=60): """ rate: 每秒生成的令牌数（60/分钟 = 1/秒） capacity: 桶容量（允许突发） """ self.rate = rate / 60 # 转换为每秒 self.capacity = capacity self.tokens = capacity self.last_refill = time.time() self.lock = threading.Lock() def acquire(self, timeout=None): """获取一个令牌""" start_time = time.time() while True: with self.lock: # 补充令牌 now = time.time() elapsed = now - self.last_refill self.tokens = min(self.capacity, self.tokens + elapsed * self.rate) self.last_refill = now if self.tokens >= 1: self.tokens -= 1 return True # 计算需要等待的时间 wait_time = (1 - self.tokens) / self.rate # 检查超时 if timeout and (time.time() - start_time + wait_time) > timeout: return False time.sleep(min(wait_time, 0.1)) class SlidingWindowRateLimiter: """滑动窗口限流器""" def __init__(self, max_requests=60, window_seconds=60): self.max_requests = max_requests self.window = window_seconds self.requests = deque() self.lock = threading.Lock() def acquire(self, timeout=None): """尝试发送请求""" start_time = time.time() while True: with self.lock: now = time.time() # 移除过期请求 while self.requests and self.requests[0] < now - self.window: self.requests.popleft() if len(self.requests) < self.max_requests: self.requests.append(now) return True # 计算需要等待的时间 wait_time = self.requests[0] + self.window - now if timeout and (time.time() - start_time + wait_time) > timeout: return False time.sleep(min(wait_time, 0.1)) # 并发限制器 class ConcurrencyLimiter: """并发请求限制器""" def __init__(self, max_concurrent=10): self.max_concurrent = max_concurrent self.current = 0 self.lock = threading.Lock() self.condition = threading.Condition(self.lock) def acquire(self, timeout=None): with self.condition: start_time = time.time() while self.current >= self.max_concurrent: remaining = timeout - (time.time() - start_time) if timeout else None if remaining is not None and remaining <= 0: return False self.condition.wait(timeout=remaining) self.current += 1 return True def release(self): with self.condition: self.current -= 1 self.condition.notify() # 组合限流器 class RateController: """组合限流管理器""" def __init__(self, rpm=60, tpm=150000, max_concurrent=10): self.rpm_limiter = SlidingWindowRateLimiter(rpm, 60) self.tpm_limiter = TokenBucketRateLimiter(tpm / 1000, tpm / 1000) self.concurrent_limiter = ConcurrencyLimiter(max_concurrent) def acquire(self, estimated_tokens=1000, timeout=60): """获取请求许可""" # 获取并发许可 if not self.concurrent_limiter.acquire(timeout): return False, "并发限制超时" # 获取 RPM 许可 if not self.rpm_limiter.acquire(timeout): self.concurrent_limiter.release() return False, "RPM限制超时" # 获取 TPM 许可 tokens_needed = estimated_tokens / 1000 for _ in range(int(tokens_needed)): if not self.tpm_limiter.acquire(timeout): self.concurrent_limiter.release() return False, "TPM限制超时" return True, "OK" def release(self): """释放并发许可""" self.concurrent_limiter.release() if __name__ == "__main__": manager = RateController(rpm=60, max_concurrent=10) for i in range(20): success, msg = manager.acquire(estimated_tokens=500) if success: print(f" 请求 {i+1}: ✅ {msg}") # 模拟请求 time.sleep(0.1) manager.release() else: print(f" 请求 {i+1}: ❌ {msg}") time.sleep(1)方案二：配置自动重试和退避
# 配置 OpenClaw 的重试策略 python3 -c " import json with open('.openclaw/config.json', 'r') as f: config = json.load(f) config['rateLimit'] = { 'retryOn429': True, # 429时自动重试 'maxRetries': 5, # 最大重试5次 'retryStrategy': 'exponential', # 指数退避 'baseDelay': 1000, # 基础延迟1秒 'maxDelay': 60000, # 最大延迟60秒 'jitter': True, # 添加随机抖动 'jitterRange': 0.3, # 30%抖动范围 'respectRetryAfter': True, # 遵守Retry-After头 'failOnMaxRetries': True, # 超过重试次数则失败 'logRetries': True, # 记录重试日志 'circuitBreaker': { 'enabled': True, # 断路器 'threshold': 10, # 10次429触发 'resetTime': 60000, # 60秒后重试 'halfOpenRequests': 3 # 半开状态允许3个请求 } } with open('.openclaw/config.json', 'w') as f: json.dump(config, f, indent=2) print('重试策略: 指数退避+抖动+断路器+遵守Retry-After') " # 指数退避重试实现 cat > .openclaw/retry_handler.js << 'JEOF' // 429 重试处理器 class RetryHandler { constructor(options = {}) { this.maxRetries = options.maxRetries || 5; this.baseDelay = options.baseDelay || 1000; this.maxDelay = options.maxDelay || 60000; this.jitter = options.jitter !== false; } async execute(requestFn) { let lastError; for (let attempt = 0; attempt <= this.maxRetries; attempt++) { try { const result = await requestFn(); return result; } catch (error) { if (error.response?.status !== 429) { throw error; // 非429错误直接抛出 } lastError = error; if (attempt === this.maxRetries) { throw new Error(`达到最大重试次数: ${this.maxRetries}`); } // 计算429后的等待时间 const retryAfter = error.response.headers['retry-after']; let delay; if (retryAfter) { // 遵守 Retry-After 头 delay = parseInt(retryAfter) * 1000; } else { // 指数退避 delay = Math.min( this.baseDelay * Math.pow(2, attempt), this.maxDelay ); // 添加抖动 if (this.jitter) { delay = delay * (1 + Math.random() * 0.3 - 0.15); } } console.warn( `429 限流，${delay.toFixed(0)}ms 后重试 ` + `(尝试 ${attempt + 1}/${this.maxRetries})` ); await new Promise(resolve => setTimeout(resolve, delay)); } } throw lastError; } } module.exports = RetryHandler; JEOF方案三：配置请求队列和批处理
# 配置请求队列 python3 -c " import json with open('.openclaw/config.json', 'r') as f: config = json.load(f) config['requestQueue'] = { 'enabled': True, 'maxQueueSize': 1000, # 最大队列1000 'maxWaitTime': 300000, # 最大等待5分钟 'priority': { 'levels': 3, # 3个优先级 'preempt': False # 不抢占 }, 'batching': { 'enabled': True, # 启用批处理 'maxBatchSize': 20, # 每批最多20个 'batchTimeout': 1000, # 1秒超时 'maxBatchTokens': 50000 # 每批最大Token }, 'scheduling': { 'strategy': 'fifo', # 先进先出 'fairShare': True, # 公平共享 'perUserLimit': 10 # 每用户最多10个/分钟 } } with open('.openclaw/config.json', 'w') as f: json.dump(config, f, indent=2) print('请求队列: 最大1000+批处理20+公平共享') " # 使用批处理减少请求数 openclaw --batch "批量处理文件" --batch-size 20 --batch-delay 1000 # 查看队列状态 openclaw --queue-status # 输出: # 队列长度: 15 # 处理中: 8 # 已完成: 142 # 平均等待: 2.3s # 预计完成: 45s方案四：多 API Key 轮换
# 创建 API Key 轮换管理器 import time import threading from collections import deque class APIKeyRotationManager: """API Key 轮换管理器""" def __init__(self, keys): self.keys = keys self.key_status = {} self.lock = threading.Lock() for key in keys: self.key_status[key] = { 'requests': 0, 'last_used': 0, 'rate_limited_until': 0, 'daily_count': 0, 'daily_reset': time.time() + 86400 } self.key_queue = deque(keys) def get_available_key(self): """获取可用的 API Key""" with self.lock: now = time.time() # 重置每日计数 for key in self.keys: status = self.key_status[key] if now > status['daily_reset']: status['daily_count'] = 0 status['daily_reset'] = now + 86400 # 尝试找到可用的 Key tried = 0 while tried < len(self.keys): key = self.key_queue[0] self.key_queue.rotate(-1) # 轮换 tried += 1 status = self.key_status[key] # 检查是否被限流 if now < status['rate_limited_until']: continue # 检查每日配额 if status['daily_count'] >= 1000000: # 假设每日100万 continue # 使用此 Key status['requests'] += 1 status['daily_count'] += 1 status['last_used'] = now return key # 所有 Key 都不可用 # 计算最短等待时间 min_wait = min( self.key_status[k]['rate_limited_until'] for k in self.keys ) wait_seconds = max(0, min_wait - now) return None, wait_seconds def mark_rate_limited(self, key, retry_after=60): """标记 Key 被限流""" with self.lock: self.key_status[key]['rate_limited_until'] = time.time() + retry_after print(f" ⚠️ Key {key[:8]}... 被限流，{retry_after}s 后恢复") def get_stats(self): """获取统计""" with self.lock: now = time.time() stats = {} for key in self.keys: s = self.key_status[key] stats[key[:8] + '...'] = { 'total_requests': s['requests'], 'daily_count': s['daily_count'], 'rate_limited': now < s['rate_limited_until'], 'last_used': time.ctime(s['last_used']) if s['last_used'] else 'never' } return stats # 使用示例 if __name__ == "__main__": import os # 从环境变量加载多个 Key keys = [ os.environ.get('OPENAI_API_KEY_1', 'key1'), os.environ.get('OPENAI_API_KEY_2', 'key2'), os.environ.get('OPENAI_API_KEY_3', 'key3'), ] manager = APIKeyRotationManager(keys) # 模拟请求 for i in range(20): result = manager.get_available_key() if isinstance(result, tuple): key, wait = result if key is None: print(f" 请求 {i+1}: ❌ 所有Key不可用，等待 {wait:.0f}s") time.sleep(min(wait, 5)) continue else: key = result print(f" 请求 {i+1}: 使用 Key {key[:8]}...") # 模拟429 if i == 10: manager.mark_rate_limited(key, 60) print("\n统计:") for k, v in manager.get_stats().items(): print(f" {k}: {v}")方案五：自适应限流
# 创建自适应限流器 import time import threading class AdaptiveRateLimiter: """自适应限流器 - 根据API响应动态调整""" def __init__(self, initial_rpm=60): self.current_rpm = initial_rpm self.max_rpm = 200 self.min_rpm = 10 self.success_streak = 0 self.error_streak = 0 self.lock = threading.Lock() self.last_adjust = time.time() # 请求历史 self.recent_requests = [] self.window_size = 60 # 60秒窗口 def can_send(self): """检查是否可以发送请求""" with self.lock: now = time.time() # 清理过期记录 self.recent_requests = [ t for t in self.recent_requests if t > now - self.window_size ] # 检查当前窗口内的请求数 if len(self.recent_requests) >= self.current_rpm: return False self.recent_requests.append(now) return True def on_success(self): """请求成功时调用""" with self.lock: self.success_streak += 1 self.error_streak = 0 # 连续成功后逐步提高限制 if self.success_streak >= 10: self._increase_limit() self.success_streak = 0 def on_rate_limit(self, retry_after=None): """遇到429时调用""" with self.lock: self.error_streak += 1 self.success_streak = 0 # 降低限制 self._decrease_limit(retry_after) def _increase_limit(self): """提高限制""" old = self.current_rpm self.current_rpm = min(self.max_rpm, int(self.current_rpm * 1.2)) if self.current_rpm != old: print(f" 📈 限流提高: {old} -> {self.current_rpm} RPM") def _decrease_limit(self, retry_after=None): """降低限制""" old = self.current_rpm factor = 0.5 if self.error_streak >= 3 else 0.8 self.current_rpm = max(self.min_rpm, int(self.current_rpm * factor)) print(f" 📉 限流降低: {old} -> {self.current_rpm} RPM") if retry_after: print(f" Retry-After: {retry_after}s") def get_current_limit(self): """获取当前限制""" with self.lock: return self.current_rpm # 使用示例 if __name__ == "__main__": limiter = AdaptiveRateLimiter(initial_rpm=60) for i in range(100): if limiter.can_send(): # 模拟请求 if i % 15 == 0 and i > 0: # 模拟429 limiter.on_rate_limit(retry_after=30) print(f" 请求 {i}: ❌ 429") else: limiter.on_success() print(f" 请求 {i}: ✅ (限制: {limiter.get_current_limit()} RPM)") else: print(f" 请求 {i}: ⏳ 等待") time.sleep(1)方案六：监控和告警
# 配置限流监控 python3 -c " import json with open('.openclaw/config.json', 'r') as f: config = json.load(f) config['rateLimit']['monitoring'] = { 'enabled': True, 'trackByEndpoint': True, # 按端点跟踪 'trackByKey': True, # 按Key跟踪 'logFile': '.openclaw/logs/rate_limit.json', 'alertThreshold': 0.8, # 80%使用率告警 'criticalThreshold': 0.95, # 95%严重告警 'metrics': { 'totalRequests': True, 'rateLimitedCount': True, 'retryCount': True, 'averageLatency': True, 'circuitBreakerTrips': True }, 'dailyReport': True, 'reportFile': '.openclaw/logs/rate_daily.json' } with open('.openclaw/config.json', 'w') as f: json.dump(config, f, indent=2) print('限流监控: 按端点/Key跟踪, 80%告警, 每日报告') " # 查看限流统计 openclaw --rate-limit-stats # 输出: # === 限流统计 === # 总请求: 5000 # 被限流: 150 (3%) # 平均重试: 1.2次 # 断路器触发: 2次 # # 按端点: # /api/chat: 3000请求, 100限流 (3.3%) # /api/analyze: 1500请求, 40限流 (2.7%) # /api/search: 500请求, 10限流 (2%) # # 当前限制: 55 RPM (已降低) # 建议限制: 60 RPM # 设置告警通知 python3 -c " import json with open('.openclaw/config.json', 'r') as f: config = json.load(f) config['rateLimit']['alerts'] = { 'webhook': os.getenv('ALERT_WEBHOOK', ''), 'email': os.getenv('ALERT_EMAIL', ''), 'onRateLimited': True, 'onCircuitBreaker': True, 'onQuotaExceeded': True, 'cooldownMinutes': 30 # 30分钟冷却 } import os with open('.openclaw/config.json', 'w') as f: json.dump(config, f, indent=2) print('告警通知已配置') "4. 各方案对比总结
| 方案 | 适用场景 | 推荐指数 |
|---|---|---|
| 方案一：限流器 | 通用防护 | ⭐⭐⭐⭐⭐ |
| 方案二：自动重试 | 429恢复 | ⭐⭐⭐⭐⭐ |
| 方案三：队列批处理 | 批量任务 | ⭐⭐⭐⭐⭐ |
| 方案四：Key轮换 | 多Key环境 | ⭐⭐⭐⭐ |
| 方案五：自适应 | 优化吞吐 | ⭐⭐⭐⭐ |
| 方案六：监控告警 | 运维 | ⭐⭐⭐⭐ |
5. 常见问题 FAQ
5.1 Windows 上定时器精度问题
Windows 定时器精度较低可能影响限流：
# Windows 默认定时器精度约15ms # 提高精度 python3 -c " import ctypes # 设置高精度定时器 winmm = ctypes.windll.winmm winmm.timeBeginPeriod(1) # 1ms精度 print('定时器精度: 1ms') " # 配置限流器使用更高精度 python3 -c " import json with open('.openclaw/config.json', 'r') as f: config = json.load(f) config['rateLimit']['timerPrecision'] = 'high' # high | normal config['rateLimit']['minInterval'] = 50 # 最小间隔50ms with open('.openclaw/config.json', 'w') as f: json.dump(config, f, indent=2) "5.2 Docker 中限流不共享
多个容器各自限流导致超限：
# 使用 Redis 共享限流状态 python3 -c " import json with open('.openclaw/config.json', 'r') as f: config = json.load(f) config['rateLimit']['sharedState'] = { 'backend': 'redis', 'url': 'redis://redis:6379', 'keyPrefix': 'openclaw:ratelimit:', 'syncInterval': 1000 # 1秒同步 } with open('.openclaw/config.json', 'w') as f: json.dump(config, f, indent=2) print('Redis 共享限流已配置') " # Docker Compose services: openclaw: environment: - REDIS_URL=redis://redis:6379 - RATE_LIMIT_SHARED=true depends_on: - redis redis: image: redis:alpine5.3 CI/CD 中限流问题
CI 中突发请求可能触发限流：
# CI 中配置保守的限流 env: OPENCLAW_RPM: 30 # 保守30 RPM OPENCLAW_MAX_CONCURRENT: 5 steps: - name: Run with rate limiting run: | openclaw --rpm 30 --max-concurrent 5 "任务" - name: Handle 429 run: | # 429时等待重试 openclaw --retry-on-429 --max-retries 3 --backoff exponential "任务"5.4 不同API限制差异
不同供应商限制不同：
# 配置多供应商限流 python3 -c " import json with open('.openclaw/config.json', 'r') as f: config = json.load(f) config['rateLimit']['providers'] = { 'openai': { 'rpm': 60, 'tpm': 150000, 'maxConcurrent': 10, 'dailyQuota': 1000000 }, 'anthropic': { 'rpm': 50, 'tpm': 100000, 'maxConcurrent': 5, 'dailyQuota': 500000 }, 'local': { 'rpm': 1000, 'tpm': 999999999, 'maxConcurrent': 50, 'dailyQuota': 999999999 } } config['rateLimit']['autoDetect'] = True # 自动检测供应商限制 with open('.openclaw/config.json', 'w') as f: json.dump(config, f, indent=2) print('多供应商限流: OpenAI 60RPM, Anthropic 50RPM') "5.5 免费计划配额过低
免费API配额很快耗尽：
# 配置配额管理 python3 -c " import json with open('.openclaw/config.json', 'r') as f: config = json.load(f) config['quotaManagement'] = { 'dailyQuota': 1000, # 每日1000请求 'monthlyQuota': 30000, # 每月3万 'warnAt': 0.8, # 80%警告 'blockAt': 0.95, # 95%阻止 'reserveForCritical': 100, # 保留100给关键任务 'priorityAccess': { 'critical': True, # 关键任务优先 'normal': True, 'low': False # 低优先级在配额紧张时阻止 } } with open('.openclaw/config.json', 'w') as f: json.dump(config, f, indent=2) print('配额管理: 每日1000+月3万+保留100给关键') " # 使用优先级 openclaw --priority critical "紧急任务" openclaw --priority normal "普通任务" openclaw --priority low "低优先级任务"5.6 限流后任务积压
限流导致任务排队堆积：
# 配置任务积压处理 python3 -c " import json with open('.openclaw/config.json', 'r') as f: config = json.load(f) config['rateLimit']['backlog'] = { 'maxBacklog': 500, # 最大积压500 'dropPolicy': 'oldest', # 满时丢弃最旧的 'persistBacklog': True, # 持久化积压 'persistFile': '.openclaw/backlog.json', 'processOnRecover': True, # 恢复后处理 'maxAge': 3600000, # 最长1小时 'compressBacklog': True # 压缩积压 } with open('.openclaw/config.json', 'w') as f: json.dump(config, f, indent=2) print('积压处理: 最大500+持久化+1小时过期') " # 查看积压 openclaw --backlog-status openclaw --backlog-process # 手动处理积压 openclaw --backlog-clear # 清除积压5.7 WebSocket 长连接限流
WebSocket 连接的限流方式不同：
# WebSocket 限流配置 python3 -c " import json with open('.openclaw/config.json', 'r') as f: config = json.load(f) config['rateLimit']['websocket'] = { 'messagesPerMinute': 100, # 每分钟100条消息 'bytesPerMinute': 1048576, # 每分钟1MB 'maxConnections': 10, # 最大10连接 'idleTimeout': 300000, # 5分钟空闲超时 'pingInterval': 30000 # 30秒ping } with open('.openclaw/config.json', 'w') as f: json.dump(config, f, indent=2) print('WebSocket限流: 100msg/min+1MB/min+10连接') "5.8 限流恢复后突发请求
恢复后一次性发送大量请求再次触发限流：
# 渐进恢复策略 class ProgressiveRecovery: """渐进恢复策略""" def __init__(self, target_rpm=60): self.target_rpm = target_rpm self.current_rpm = 0 self.recovery_step = target_rpm // 10 # 每次增加10% self.recovery_interval = 5 # 每5秒增加 self.last_recovery = 0 def can_send(self): """检查是否可以发送""" now = time.time() # 渐进恢复 if self.current_rpm < self.target_rpm: if now - self.last_recovery >= self.recovery_interval: self.current_rpm = min( self.target_rpm, self.current_rpm + self.recovery_step ) self.last_recovery = now print(f" 📈 恢复中: {self.current_rpm}/{self.target_rpm} RPM") return self.current_rpm > 0 def on_rate_limit(self): """再次限流时重置""" self.current_rpm = self.current_rpm // 2 # 减半 print(f" 📉 再次限流，降至: {self.current_rpm} RPM")排查清单速查表
□ 1. 检查 API 限制文档: RPM/TPM/并发/日配额 □ 2. 实现限流器: 令牌桶/滑动窗口 □ 3. 配置自动重试: 指数退避+抖动 □ 4. 遵守 Retry-After 头 □ 5. 使用请求队列和批处理 □ 6. 多 Key 轮换分散负载 □ 7. 配置断路器防止雪崩 □ 8. 部署自适应限流优化吞吐 □ 9. 监控限流统计和告警 □ 10. 配额管理: 保留给关键任务6. 总结
- 最常见原因：每分钟请求数（RPM）超限（35%）和并发请求数超限（25%）
- 限流器：实现令牌桶+滑动窗口+并发限制三重限流，确保请求速率在限制内
- 自动重试：配置指数退避重试（1s→2s→4s→8s→16s），添加 30% 随机抖动避免雷同效应
- 请求队列：使用 FIFO 队列+批处理（每批20个）减少请求次数，支持优先级调度
- 最佳实践建议：部署自适应限流器根据 API 响应动态调整速率，多 API Key 轮换分散负载，配置断路器防止限流雪崩，监控使用率并在 80% 时告警
故障排查流程图
flowchart TD A[429限流错误] --> B[检查限制类型] B --> C[RPM/TPM/并发/日配额] C --> D{RPM超限?} D -->|是| E[实现限流器] D -->|否| F{并发超限?} E --> G[滑动窗口60RPM] G --> H[配置自动重试] F -->|是| I[并发限制器] F -->|否| J{TPM超限?} I --> H J -->|是| K[令牌桶TPM] J -->|否| L{日配额耗尽?} K --> H L -->|是| M[多Key轮换] L -->|否| N[检查突发流量] M --> H N --> O[渐进恢复] O --> H H --> P[指数退避+抖动] P --> Q[遵守Retry-After] Q --> R[openclaw测试] R --> S{成功?} S -->|是| T[✅ 问题解决] S -->|否| U[配置请求队列] U --> V[批处理20个/批] V --> W[自适应限流] W --> X[部署监控告警] X --> T
