code: 429 - {'error': {'message': 'Request was rejected due to rate limiting. De...

2025年11月24日 09:58 状态: processing

🚨 错误信息

Error code: 429 - {'error': {'message': 'Request was rejected due to rate limiting. Details: TPM limit reached.', 'type': 'rix_api_error', 'param': '', 'code': 'bad_response_status_code'}}

🤖 AI解决方案

HTTP 429 速率限制错误分析

错误翻译

原文:Request was rejected due to rate limiting. Details: TPM limit reached.
译文:请求因速率限制而被拒绝。详情:TPM限制已达到。

原因分析

根本原因:API服务已超出每分钟令牌限制(TPM - Tokens Per Minute)

这是典型的API速率限制错误,当客户端在短时间内发送过多请求时,服务端会返回429状态码来保护服务器资源。

解决方案

根据您的开发环境选择相应的处理方式:

Python解决方案

import time
import requests
from requests.exceptions import HTTPError

def make_api_request_with_retry(api_endpoint, payload, max_retries=3):
    """
    带有重试机制的API请求函数
    """
    for attempt in range(max_retries):
        try:
            response = requests.post(api_endpoint, json=payload)
            response.raise_for_status()
            return response.json()
            
        except HTTPError as e:
            if e.response.status_code == 429:
                if attempt < max_retries - 1:
                    # 指数退避策略
                    wait_time = (2 ** attempt) + random.uniform(0, 1)
                    print(f"速率限制触发,等待 {wait_time:.2f} 秒后重试...")
                    time.sleep(wait_time)
                    continue
                else:
                    raise Exception("重试次数已达上限,请稍后再试")
            else:
                raise e

# 使用示例
response = make_api_request_with_retry(
    "https://api.example.com/endpoint",
    {"message": "your request data"}
)

HTTP Header检查方案

// 前端应用检查响应头
fetch('/api/endpoint', {
    method: 'POST',
    headers: {'Content-Type': 'application/json'},
    body: JSON.stringify(data)
})
.then(response => {
    if (response.status === 429) {
        const retryAfter = response.headers.get('Retry-After');
        console.log(`速率限制,建议等待 ${retryAfter} 秒`);
        // 实现重试逻辑
    }
    return response.json();
});

️ 预防措施

1. 速率监控

# 实现请求计数器
import time

class RateLimiter:
    def __init__(self, max_requests_per_minute):
        self.max_requests = max_requests_per_minute
        self.requests = []
    
    def can_make_request(self):
        current_time = time.time()
        # 清理1分钟前的记录
        self.requests = [t for t in self.requests if current_time - t < 60]
        
        if len(self.requests) < self.max_requests:
            self.requests.append(current_time)
            return True
        return False

2. 最佳实践建议

  • 实施指数退避:重试间隔逐渐增加(1s, 2s, 4s...)
  • 批量处理请求:减少API调用频率
  • 缓存响应结果:对相同请求使用缓存减少调用
  • 监控API用量:实时监控TPM使用情况

推荐工具:使用API监控工具如Postman、Insomnia测试速率限制阈值。