限流说明

概述

为保证服务质量和公平使用，平台对 API 调用实施了多重限制机制：

RPM 限制：限制单位时间（每分钟）内的请求次数
TPM 限制：限制单位时间（每分钟）内的请求 Tokens 数
额度限制：限制账户可用的总调用额度
并发限制：限制同时进行的请求数量

频率限制

限制级别

级别	RPM 限制	TPM 限制	说明
普通用户	100 RPM	50000000 TPM	默认限制
高级用户	300 RPM	50000000 TPM	可申请提升
企业用户	5000 RPM	50000000 TPM	更高额度可联系商务

限制响应头

API 响应中包含以下头信息，帮助您了解当前限制状态：

X-RateLimit-Limit: 60
X-RateLimit-Remaining: 45
X-RateLimit-Reset: 1698751200

头信息	说明
`X-RateLimit-Limit`	每分钟请求上限
`X-RateLimit-Remaining`	当前窗口剩余请求数
`X-RateLimit-Reset`	限制重置时间戳（Unix 时间戳）

超限处理

当请求超过频率限制时，API 将返回 429 Too Many Requests 错误：

{
  "error": {
    "message": "Rate limit exceeded. Please try again later.",
    "type": "rate_limit_error",
    "code": "rate_limit_exceeded"
  }
}

响应头中会包含重试等待时间：

Retry-After: 15

额度限制

额度类型

类型	说明
账户额度	账户总可用额度
密钥额度	单个 API 密钥的额度限制
日限额	每日可消耗的额度上限

额度耗尽响应

当额度不足时，API 将返回 403 Forbidden 错误：

{
  "error": {
    "message": "Insufficient quota. Please recharge your account.",
    "type": "insufficient_quota",
    "code": "insufficient_quota"
  }
}

并发限制

限制规则

普通用户：最多 5 个并发请求
高级用户：最多 10 个并发请求
企业用户：最多 20 个并发请求

超出并发限制

当并发请求超过限制时，后续请求将返回 429 错误：

{
  "error": {
    "message": "Too many concurrent requests. Please wait for some requests to complete.",
    "type": "rate_limit_error",
    "code": "too_many_concurrent_requests"
  }
}

限流处理最佳实践

1. 客户端限流

在客户端实现请求队列，控制请求频率：

import time
from collections import deque

class RateLimiter:
    def __init__(self, max_requests=60, window=60):
        self.max_requests = max_requests
        self.window = window
        self.requests = deque()

    def wait_if_needed(self):
        now = time.time()

        while self.requests and self.requests[0] < now - self.window:
            self.requests.popleft()

        if len(self.requests) >= self.max_requests:
            sleep_time = self.requests[0] - (now - self.window) + 0.1
            time.sleep(sleep_time)

        self.requests.append(now)

2. 指数退避重试

当收到 429 错误时，使用指数退避策略重试：

import time
import random

def exponential_backoff_retry(func, max_retries=5):
    for attempt in range(max_retries):
        try:
            return func()
        except RateLimitError as e:
            if attempt == max_retries - 1:
                raise
            wait_time = (2 ** attempt) + random.uniform(0, 1)
            print(f"Rate limited. Retrying in {wait_time:.2f} seconds...")
            time.sleep(wait_time)

3. 监控响应头

始终检查响应头中的限流信息，动态调整请求频率：

def make_request_with_rate_limit_check():
    response = make_api_request()

    limit = int(response.headers.get('X-RateLimit-Limit', 60))
    remaining = int(response.headers.get('X-RateLimit-Remaining', 0))
    reset_time = int(response.headers.get('X-RateLimit-Reset', 0))

    if remaining < 10:
        adjust_request_frequency(lower=True)

    return response

4. 批量处理

如果需要处理大量请求，使用批量处理减少请求次数：

# 不推荐：逐个处理
for text in texts:
    result = api_call(text)  # 每个文本一个请求

# 推荐：批量处理（如果 API 支持）
batch_size = 10
for i in range(0, len(texts), batch_size):
    batch = texts[i:i+batch_size]
    results = api_call_batch(batch)  # 批量请求

提升限流额度

如果您需要更高的限流额度，可以：

升级账户：联系客服升级为企业用户
申请临时提升：说明业务需求，申请临时提升额度
优化请求：通过缓存、批量处理等方式减少对 API 的调用