API Rate Limiting429 Error HandlingProduction Reliability

如何在生產環境處理 LLM API 速率限制與 429 錯誤

閱讀 1 分鐘

429 Too Many Requests 是你的 LLM 供應商送出的最有用的回應。比 JSON body 有用。比錯誤碼更有行動價值。因為藏在它的標頭裡的是即時容量儀表——x-ratelimit-remainingretry-after——而大多數生產程式碼都忽略了它。工程師把速率限制當成一面要撞上去的牆。它是該讀的儀表。

撞上牆是這樣子的。你的 App 下午兩點跑得好好的。三點流量增加。三點十五分,每個請求都回 429。你的重試邏輯啟動——固定一秒間隔——於是製造出驚群效應。每一次重試都撞進同一個速率限制視窗。十分鐘的連鎖故障。使用者看到錯誤。值班人員被呼叫。解法從來不是「更用力重試」。永遠不該是更用力重試。

這兩種結果的差距——撞牆 vs 讀儀表——是三層程式碼。這篇文章涵蓋全部三層。反應層:帶抖動的指數退避。主動層:讀取剩餘預算、在撞牆之前就慢下來的標頭感知節流。預測層:在下一次呼叫會耗盡配額之前,先為 Agent 狀態做檢查點的呼叫前暫停。每一層都有可運作的 Python 程式碼,都是生產環境驗證過的寫法。

作為速率限制處理基礎的 API 認證與金鑰管理基本概念,請見API 金鑰安全指南

速率限制為什麼存在——以及實際上怎麼運作

速率限制不是懲罰,而是基礎設施保護。每一個 API 請求都消耗 GPU 記憶體與運算。一個不節流的用戶端能在幾秒內灌爆供應商的推理叢集。速率限制確保所有用戶之間公平分配。

你需要關心的三個限制:

  • RPM(每分鐘請求數): 你能發出多少次 API 呼叫。隨用隨付方案:通常 500–3,000 RPM。免費方案:10–50 RPM。企業:客製。
  • TPM(每分鐘 Token 數): 所有請求的 Token 總和——輸入+輸出。一個 100K Token 的提示詞消耗的配額等同 500 個普通請求。TPM 限制防的就是這個。
  • 並發請求數: 同時在途的請求上限。超過後新請求排隊或被拒絕。這個限制常常沒有文件,是靠慘痛經驗學到的。

供應商各自的方案把這些限制變成具體數字。 OpenAI 的 Tier 5(最高隨用隨付等級)對 GPT-4.x 模型給到 10,000 RPM 和 30,000,000 TPM——但 Tier 1 只有 500 RPM 和 200,000 TPM。Anthropic 的 Claude API 標準方案給 1,000 RPM,Claude Opus 是 80,000 TPM、Claude Sonnet 是 400,000 TPM,反映各模型不同的推理成本。

Google 的 Gemini API 隨用隨付給 1,500 RPM、上限 2,000,000 TPM。每家供應商還會套用模型級的覆寫——Claude Opus 4 的 TPM 限制比 Claude Sonnet 4 嚴,因為更大的模型消耗比例更多的運算。在 OpenAI 從 Tier 1 升到 Tier 5,既需要消費歷史(每月 $250 以上),也需要 30 天以上無濫用紀錄的實績證明。

高限制不是客氣請求得來的——是靠持續的生產流量模式證明你不會灌爆叢集來贏得的。知道你的確切等級與限制,是建構後面三層的第一步。

怎麼讀你目前的限制。 每個 API 回應都帶速率限制標頭——但幾乎沒人讀。OpenAI 的速率限制文件說明了標頭格式與等級結構:

x-ratelimit-remaining-requests: 487
x-ratelimit-remaining-tokens: 823000
x-ratelimit-reset-requests: 12s

這些在 200 回應上就有,不只 429。它們精確告訴你被節流之前還剩多少預算。把它們變成監控儀表板上的儀表。剩餘低於 20% 就警示。「我們撞上速率限制」和「我們看到它要來了、繞過去」的差別,就是讀這些標頭。

速率限制恐怖故事:兩個你不想重蹈的案例

一個歐洲電商平台推出搭載 GPT-4.5 的黑色星期五 AI 購物助理。QA 用 50 個並發用戶測試——生產環境第一個小時就衝到 2,300。固定間隔的重試把 429 變成一場 47 分鐘的故障。

營收損失:那個時段追蹤到的棄購車 $180,000。根本原因不是流量規模——而是重試邏輯把高峰放大而不是吸收。

一家 SaaS 分析公司沒有讀速率限制變更日誌就做了 OpenAI API 版本遷移。新版本把他們等級的 RPM 從 3,000 砍半到 1,500。他們既有的節流程式碼預設的是舊限制。

生產環境順順跑了六天——直到月報週期把請求量翻了三倍。所有報表任務同時撞上 429。偵測花了 22 分鐘,因為他們的監控只追 5xx 錯誤,沒追 429。

修復只有三行:更新 RPM 常數。教訓是永久的:每一次 API 版本遷移,都是一次速率限制遷移。

第 1 層:用戶端節流

最簡單的一層。用 token bucket 或 semaphore 防止你的應用程式超過供應商聲明的限制。

import asyncio
import time

class RateLimiter:
    """Token bucket rate limiter for LLM API calls."""

    def __init__(self, max_rpm: int):
        self.max_rpm = max_rpm
        self.tokens = max_rpm
        self.last_refill = time.monotonic()
        self.semaphore = asyncio.Semaphore(max_rpm // 6)  # Concurrency cap

    async def acquire(self):
        """Wait until a request can be sent without exceeding RPM."""
        # Refill tokens based on elapsed time
        now = time.monotonic()
        elapsed = now - self.last_refill
        refill = elapsed * (self.max_rpm / 60)
        self.tokens = min(self.max_rpm, self.tokens + refill)
        self.last_refill = now

        if self.tokens < 1:
            wait_time = (1 - self.tokens) / (self.max_rpm / 60)
            await asyncio.sleep(wait_time)
            self.tokens = 1

        self.tokens -= 1

limiter = RateLimiter(max_rpm=500)

async def rate_limited_api_call(model: str, messages: list):
    await limiter.acquire()
    # Make the API call

重要的順序: 永遠先取得 RPM token,再拿並發 semaphore。順序反過來會造成隊頭阻塞——並發槽被發不出去的請求塞滿,讓發得出去的請求挨餓。

這一層防止最常見的自找速率限制傷口:因為沒追蹤呼叫速度而超過自己等級的聲明限制。對多數中低流量的應用程式,這樣就夠了。

第 2 層:感知標頭的重試退避

第 1 層防止你超過已知限制。第 2 層處理供應商實際容量波動時的情況——而它隨時都在波動,視整個叢集負載而定。

import random
from openai import OpenAI

client = OpenAI(
    base_url="https://api.tokspan.com/v1",
    api_key="ts-your-key-here"
)

def chat_with_backoff(messages, model="claude-opus-4-8", max_retries=4):
    """Exponential backoff with jitter + header awareness."""
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model=model,
                messages=messages,
                timeout=30
            )
            # Read remaining budget from headers —even on success
            remaining = response.headers.get("x-ratelimit-remaining-requests")
            if remaining and int(remaining) < 50:
                print(f"Rate limit low: {remaining} requests remaining. Slow down.")
            return response

        except Exception as e:
            if "429" in str(e) or "rate_limit" in str(e).lower():
                if attempt == max_retries - 1:
                    raise  # Out of retries
                # Exponential backoff: 1s —2s —4s —8s
                wait = (2 ** attempt) + random.uniform(0, 1)
                print(f"Rate limited. Retrying in {wait:.1f}s (attempt {attempt + 1}/{max_retries})")
                import time
                time.sleep(wait)
            else:
                raise  # Not a rate limit error —don't retry

退避的三條鐵律:

  1. 絕不用固定間隔重試。它會製造驚群。每個在 t+1 秒重試的用戶端都會撞進同一個速率限制視窗。
  2. 永遠加抖動。+ random.uniform(0, 1) 把重試分散到整個視窗。光這一步就能擋掉多數連鎖故障。
  3. 絕不重試 401、403、400。重試打錯的 API key 或格式錯誤的請求不會修好它。只重試 429 和 5xx。

別做的事。 這個寫法——在生產程式碼裡意外地常見——等於對一個聽不懂你語言的人吼得更大聲:

# DO NOT DO THIS
while True:
    try:
        response = client.chat.completions.create(...)
        break
    except:
        time.sleep(1)  # Fixed interval, no jitter, infinite retry

這會製造驚群,並保證你一直卡在被限速。每一次重試都落在速率限制視窗的同一個點。供應商的基礎設施看到一堆相同請求,把他們全部節流,你的 App 進入死亡螺旋。

第 3 層:預測性暫停

第 1、2 層是反應式的——撞上(或接近)限制之後才回應。第 3 層是預測式的——呼叫前先讀預算,決定:繼續、短暫等待、還是檢查點後暫停。

def predict_rate_limit(response_headers: dict) -> str:
    """Three-valued decision based on remaining budget."""
    remaining_req = int(response_headers.get("x-ratelimit-remaining-requests", 1000))
    remaining_tok = int(response_headers.get("x-ratelimit-remaining-tokens", 1000000))

    if remaining_req > 100 and remaining_tok > 200000:
        return "continue"      # Plenty of budget
    elif remaining_req > 20:
        return "wait"          # Budget running low —short pause
    else:
        return "checkpoint"    # Budget nearly exhausted —suspend

# Usage in an agent loop:
for step in agent_steps:
    response = call_llm(current_state)
    decision = predict_rate_limit(response.headers)

    if decision == "continue":
        process(response)
    elif decision == "wait":
        time.sleep(5)  # Short pause, let budget recover
        process(response)
    else:  # checkpoint
        save_agent_state(current_state)  # Save progress
        time.sleep(60)  # Wait for rate-limit window reset
        resume_agent_from_checkpoint()  # Resume without losing work

2026 年的最新做法:agentpause 一個 Python 函式庫,在每個回應上讀取速率限制標頭、在下一次呼叫前預測耗盡、並在暫停前為 Agent 狀態做檢查點。實測結果:崩潰率 0% 對比反應式基準的 100%。429 錯誤歸零。失敗重試浪費的 Token 少 80%。如果你在跑高吞吐的生產 Agent,agentpause 或同等預測邏輯已經不是選配——它是「我們的 Agent 很可靠」和「流量一衝高我們的 Agent 就隨機當掉」的分界線。

多供應商路由:最好的速率限制解法

上面每一種策略都假設你只對一家供應商講話。最有效的速率限制策略是對好幾家講話。

核心洞察: 與其等一家供應商的速率限制重置,不如把請求送去另一家供應商。你的有效速率限制變成你能路由到的所有供應商的總和。

一個帶每供應商 token bucket 與自動冷卻的輪詢路由器:

PROVIDERS = {
    "openai": {"rpm": 2000, "cooldown_until": 0},
    "anthropic": {"rpm": 1500, "cooldown_until": 0},
    "google": {"rpm": 1000, "cooldown_until": 0},
}

def route_request(messages):
    now = time.time()
    available = [
        p for p, cfg in PROVIDERS.items()
        if now > cfg["cooldown_until"]
    ]
    if not available:
        raise RuntimeError("All providers in cooldown")

    # Round-robin among available providers
    provider = available[hash(str(messages)) % len(available)]
    try:
        return call_provider(provider, messages)
    except RateLimitError:
        PROVIDERS[provider]["cooldown_until"] = now + 30  # Cooldown for 30s
        return route_request(messages)  # Retry with a different provider

聚合平台在基礎設施層處理這件事——你的單一端點路由到所有供應商,帶自動冷卻、故障轉移與速率限制監控。你設定想要的吞吐量,平台管理每家供應商的速率限制。完整的路由實作與感知延遲的故障轉移,請見我們的生產多模型建置指南自訂路由文件

常見問題

最常見的速率限制錯誤是什麼?

固定間隔重試。所有用戶端的一秒間隔重試製造驚群,保證更多 429。永遠用帶隨機抖動的指數退避。光抖動——在等待時間加上random.uniform(0, 1)——就能擋掉多數連鎖故障。

怎麼知道我的速率限制是多少?

查供應商儀表板看你等級的聲明限制。然後讀每個 API 回應上的x-ratelimit-remaining-*標頭——它們即時告訴你實際剩餘預算。監控它們。你通常要到撞上才知道自己的限制——而這正是多數團隊的運作方式。

用多個供應商真的能解決速率限制嗎?

可以——而且很有效。每家 500 RPM 的限制,透過輪詢路由器在四家供應商上變成 2,000 RPM。帶多供應商路由的聚合平台讓這件事透明化:一個端點、供應商層級的自動速率限制管理。把多供應商路由和成本優化策略配對,在不加倍帳單的前提下提高吞吐——非關鍵請求走便宜模型,把貴的留給需要的任務。

今天就能做的最簡單修復是什麼?

把固定間隔重試換成指數退避+抖動。五行程式碼。擋掉把單一 429 變成連鎖停機的驚群效應。這篇文章第 2 層的程式碼可以直接複製貼上。

聚合平台能幫我處理速率限制嗎?

可以。多供應商路由、被節流供應商的自動冷卻、統一的速率限制儀表板都是標準功能。你設定想要的吞吐量,平台處理供應商層級的配額管理、標頭監控與自動故障轉移。一個端點。沒有 429。

這裡的三層——節流、退避、預測——把速率限制從可靠度威脅變成已解決的工程問題。但隨著多供應商路由成為基本盤、供應商在吞吐保證上互相競爭,問題變了:「速率限制超過」會不會像「磁碟已滿」和「記憶體不足」一樣,變成現代基礎設施直接讓它過時的錯誤?就目前而言,這篇文章的程式碼讓你的系統繼續跑。更長的弧線指向更有趣的地方。

第 1 層和第 2 層,你用上面的程式碼一下午就能實作。第 3 層——預測性暫停——需要更多投入。聚合平台把三層全部內建進請求路徑:多供應商路由吸收供應商層級的節流、標頭監控餵養共用的速率限制儀表板、自動冷卻讓健康的供應商在某一家劣化時繼續輪替。沒有任何架構能完全消滅 429,但把流量分散到多家供應商、在撞牆前讀標頭,會把它們從週週都發生的事故變成罕見的邊緣案例。