Grok API 429 Rate Limit 与配额熔断深度排查指南

import time
from collections import deque

class TokenRateLimiter:
    def __init__(self, max_tpm=200000):
        self.max_tpm = max_tpm
        self.history = deque() # 存储 (timestamp, token_count)

    def acquire(self, estimated_tokens):
        now = time.time()
        # 清理 60 秒之前的历史记录
        while self.history and now - self.history[0][0] > 60:
            self.history.popleft()

        current_usage = sum(tokens for _, tokens in self.history)
        if current_usage + estimated_tokens > self.max_tpm:
            # 计算需要休眠的时间
            oldest_time, _ = self.history[0]
            sleep_time = 60 - (now - oldest_time) + 0.5
            print(f"[主动整形] 预估本次将超出 TPM 上限，客户端主动排队避让: 等待 {sleep_time:.2f} 秒...")
            time.sleep(sleep_time)
            return self.acquire(estimated_tokens)

        self.history.append((now, estimated_tokens))
        return True

import os
import requests

def probe_grok_rate_limits():
    api_key = os.getenv("GROK_API_KEY")
    url = "https://api.gpt345.com/v1/chat/completions"
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    payload = {
        "model": "grok-4.6",
        "messages": [{"role": "user", "content": "Ping"}],
        "max_tokens": 1
    }

    print("[探针] 正在向网关发送探测包...")
    resp = requests.post(url, headers=headers, json=payload, timeout=10)
    print(f"HTTP 状态码: {resp.status_code}")

    # 提取速率限制相关的核心响应头
    rl_headers = {k: v for k, v in resp.headers.items() if "ratelimit" in k.lower() or "retry" in k.lower()}
    print("\n[速率相关响应头详情]:")
    for k, v in rl_headers.items():
        print(f"  {k}: {v}")

    if resp.status_code == 429:
        print("\n[诊断结果] 当前请求被限流！详细错误正文:")
        print(resp.text)
    elif resp.status_code == 200:
        print("\n[诊断结果] 链路畅通，配额充裕！")

if __name__ == "__main__":
    probe_grok_rate_limits()