deepseek-429-rate-limiting-exponential-backoff-circuit-breaker-example

title: "DeepSeek 429 限流錯誤：指數退避與熔斷器設定範例" description: "本教程提供 DeepSeek API 遇到 429 限流錯誤時的應對策略，透過指數退避與熔斷器模式，確保應用程式穩定性與服務可靠性。" date: 2026-07-04 generated: true tags: ["posts"] layout: "layouts/post.njk" permalink: "/posts/2026-07-04-deepseek-429-rate-limiting-exponential-backoff-circuit-breaker-example-206/index.html"

在使用 DeepSeek API 開發應用程式時，尤其是在面對高併發請求或突發流量時，開發者經常會遇到 HTTP 429 Too Many Requests 錯誤。這表示您的應用程式在短時間內向 DeepSeek API 發送了過多的請求，超出了其設定的速率限制。未能妥善處理這些限流錯誤，可能導致服務中斷、用戶體驗下降，甚至影響整個系統的穩定性。

為了解決這個問題，業界廣泛採用兩種關鍵的彈性模式：「指數退避 (Exponential Backoff)」和「熔斷器 (Circuit Breaker)」。本教程將深入探討如何將這些模式應用於 DeepSeek API 整合中，並提供實用的 Python 程式碼範例。

理解 DeepSeek API 的限流機制

DeepSeek 平台為保障服務質量與資源公平分配，會對其 API 設有不同的速率限制。這些限制可能包括：

• 每分鐘請求數 (RPM - Requests Per Minute)：在指定時間窗口內，允許發送的最大請求數量。
• 每分鐘 Token 數 (TPM - Tokens Per Minute)：對於語言模型 API，這限制了每分鐘處理的 Token 總數。
• 併發請求數 (Concurrent Requests)：同時進行的活躍請求數量。

當您的應用程式發送的請求速率超出這些限制時，DeepSeek API 便會返回 HTTP 429 狀態碼。請求的返回標頭 (response headers) 中通常會包含 Retry-After 字段，建議您在指定秒數後再進行重試。

有效處理 429 錯誤至關重要，它能：

1. 提升服務穩定性：避免因短時間內的大量重試請求而進一步加劇 DeepSeek 服務的負載。
2. 改善用戶體驗：減少因 API 錯誤導致的應用程式卡頓或功能失效。
3. 優化資源使用：避免不必要的重試浪費本地計算資源和網絡帶寬。

策略一：實作指數退避 (Exponential Backoff)

指數退避是一種智能的重試策略，當 API 請求失敗（例如收到 429 錯誤）時，它會以指數級增長的延遲時間來自動重試請求。這意味着每次重試失敗後，下一次重試的等待時間都會比前一次更長。

運作原理：

• 首次失敗：等待 X 秒後重試。
• 第二次失敗：等待 X * K 秒後重試。
• 第三次失敗：等待 X * K^2 秒後重試。
• ...依此類推，直到達到最大重試次數或最大等待時間。

其中 K 通常為 2。這種方式能有效降低 API 服務端的壓力，並為其提供恢復的時間。

Python 實作範例：使用 tenacity 庫

tenacity 是一個強大的 Python 重試庫，它支援多種退避策略，包括指數退避。

步驟 1：安裝 tenacity

pip install tenacity

步驟 2：實作帶有指數退避的 DeepSeek API 請求

以下是一個使用 tenacity 處理 DeepSeek API 請求的範例，它會針對 429 錯誤碼進行指數退避重試。

import os
import time
import requests
from tenacity import (
    retry,
    wait_exponential,
    stop_after_attempt,
    retry_if_exception_type,
    before_sleep_log,
)
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# 從環境變量獲取 DeepSeek API Key
DEEPSEEK_API_KEY = os.environ.get("DEEPSEEK_API_KEY")
if not DEEPSEEK_API_KEY:
    raise ValueError("請設定環境變數 DEEPSEEK_API_KEY")

DEEPSEEK_API_URL = "https://api.deepseek.com/v1/chat/completions"

class RateLimitError(Exception):
    """自定義異常，用於標識 429 錯誤"""
    pass

@retry(
    wait=wait_exponential(multiplier=1, min=4, max=60), # 每次重試等待時間為 2^x 秒，最小 4 秒，最大 60 秒
    stop=stop_after_attempt(5),                          # 最多重試 5 次
    retry=(retry_if_exception_type(RateLimitError)),     # 僅在 RateLimitError 發生時重試
    before_sleep=before_sleep_log(logger, logging.INFO)  # 重試前記錄日誌
)
def call_deepseek_api_with_backoff(prompt: str) -> str:
    """
    呼叫 DeepSeek API，並實作指數退避重試機制。
    """
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {DEEPSEEK_API_KEY}"
    }
    data = {
        "model": "deepseek-chat",  # 可以替換為您使用的 DeepSeek 模型
        "messages": [
            {"role": "user", "content": prompt}
        ],
        "temperature": 0.7,
        "max_tokens": 100
    }

    try:
        logger.info(f"正在發送 DeepSeek API 請求...")
        response = requests.post(DEEPSEEK_API_URL, headers=headers, json=data, timeout=30)
        response.raise_for_status() # 對於 4XX/5XX 響應拋出 HTTPError
        return response.json()['choices'][0]['message']['content']
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 429:
            retry_after = e.response.headers.get("Retry-After")
            logger.warning(f"收到 DeepSeek API 429 錯誤，建議在 {retry_after} 秒後重試。")
            raise RateLimitError(f"DeepSeek API 達到速率限制: {e.response.text}") from e
        else:
            logger.error(f"DeepSeek API 請求失敗: {e.response.status_code} - {e.response.text}")
            raise # 其他 HTTP 錯誤不重試
    except requests.exceptions.RequestException as e:
        logger.error(f"DeepSeek API 請求發生網絡錯誤: {e}")
        raise # 網絡錯誤不重試

if __name__ == "__main__":
    test_prompt = "請用一句話介紹香港的特色。"
    try:
        result = call_deepseek_api_with_backoff(test_prompt)
        print(f"\nDeepSeek API 回應: {result}")
    except (RateLimitError, requests.exceptions.RequestException, ValueError) as e:
        print(f"\n最終請求失敗: {e}")

代碼說明：

• @retry 裝飾器：這是 tenacity 的核心。
• wait_exponential(multiplier=1, min=4, max=60)：設定指數退避策略。首次等待時間至少 4 秒，每次失敗後等待時間翻倍，最長不超過 60 秒。
• stop_after_attempt(5)：最多重試 5 次。如果 5 次重試後仍然失敗，則拋出異常。
• retry_if_exception_type(RateLimitError)：指定只有在捕獲到 RateLimitError 時才進行重試。我們自定義了 RateLimitError，當 requests 捕獲到 429 狀態碼時拋出此異常。
• before_sleep_log：在每次重試等待前記錄日誌，方便追蹤。
• requests.exceptions.HTTPError：requests 庫在收到非 2xx 響應時拋出的異常。我們檢查其 status_code 是否為 429。
• Retry-After 標頭：當收到 429 錯誤時，DeepSeek API 可能會在響應頭中包含 Retry-After，指示客戶端應等待多少秒後再重試。

策略二：建構熔斷器 (Circuit Breaker) 模式

雖然指數退避能有效處理暫時性的限流，但如果 DeepSeek API 長時間處於高負載或故障狀態，持續的重試仍然會加重問題，並導致應用程式長時間阻塞。熔斷器模式旨在防止應用程式不斷嘗試執行可能失敗的操作，從而保護下游服務並允許系統快速失敗。

運作原理：

熔斷器有三種主要狀態：

1. 閉合 (Closed)：這是正常狀態。所有請求都通過。如果錯誤率達到預設閾值，熔斷器會轉換為「打開」狀態。
2. 打開 (Open)：熔斷器在此狀態下會直接拒絕所有請求，不再嘗試呼叫底層服務，而是立即返回錯誤。經過一段預設的「重置時間」後，熔斷器會轉換為「半開」狀態。
3. 半開 (Half-Open)：在半開狀態下，熔斷器會允許少數探測性請求通過。如果這些請求成功，它會恢復到「閉合」狀態；如果再次失敗，則返回「打開」狀態。

Python 實作範例：使用 pybreaker 庫

pybreaker 是一個輕量級的 Python 熔斷器庫，易於整合。

步驟 1：安裝 pybreaker

pip install pybreaker

步驟 2：實作帶有熔斷器的 DeepSeek API 請求

import os
import requests
from pybreaker import CircuitBreaker, CircuitBreakerError
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

DEEPSEEK_API_KEY = os.environ.get("DEEPSEEK_API_KEY")
if not DEEPSEEK_API_KEY:
    raise ValueError("請設定環境變數 DEEPSEEK_API_KEY")

DEEPSEEK_API_URL = "https://api.deepseek.com/v1/chat/completions"

# 配置熔斷器
# fail_max: 在閉合狀態下，失敗次數達到此值後熔斷器打開
# reset_timeout: 熔斷器打開後，等待此秒數後進入半開狀態
# exclude: 哪些異常不計入失敗 (例如，用戶輸入錯誤等)
# enable_retry_after: 如果為 True，且響應有 Retry-After 標頭，熔斷器將在該時間後重置。
deepseek_breaker = CircuitBreaker(
    fail_max=3,              # 在 3 次失敗後打開熔斷器
    reset_timeout=30,        # 熔斷器打開後，30 秒後進入半開狀態
    exclude=[               # 排除某些不應導致熔斷器打開的異常
        requests.exceptions.ConnectionError # 網絡連接錯誤可能由本地問題引起，而不是 DeepSeek 服務端問題
    ]
)

# 熔斷器事件監聽
def on_circuit_open(cb, exc):
    logger.error(f"DeepSeek API 熔斷器已打開！原因: {exc.__class__.__name__}: {exc}")

def on_circuit_half_open(cb):
    logger.warning("DeepSeek API 熔斷器進入半開狀態，將嘗試少量請求。")

def on_circuit_closed(cb):
    logger.info("DeepSeek API 熔斷器已閉合，恢復正常操作。")

deepseek_breaker.add_listener(on_circuit_open)
deepseek_breaker.add_listener(on_circuit_half_open)
deepseek_breaker.add_listener(on_circuit_closed)


@deepseek_breaker # 將熔斷器應用於 DeepSeek API 呼叫函數
def call_deepseek_api_with_breaker(prompt: str) -> str:
    """
    呼叫 DeepSeek API，並實作熔斷器模式。
    """
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {DEEPSEEK_API_KEY}"
    }
    data = {
        "model": "deepseek-chat",
        "messages": [
            {"role": "user", "content": prompt}
        ],
        "temperature": 0.7,
        "max_tokens": 100
    }

    logger.info(f"正在發送 DeepSeek API 請求 (熔斷器狀態: {deepseek_breaker.current_state.name})...")
    response = requests.post(DEEPSEEK_API_URL, headers=headers, json=data, timeout=30)
    response.raise_for_status()
    return response.json()['choices'][0]['message']['content']

if __name__ == "__main__":
    test_prompt = "香港有多少人口？"
    for i in range(10): # 模擬多次請求
        try:
            result = call_deepseek_api_with_breaker(test_prompt)
            print(f"DeepSeek API 回應 ({i+1}): {result}")
            time.sleep(1) # 模擬正常請求間隔
        except CircuitBreakerError:
            print(f"DeepSeek API 熔斷器已打開，請求 ({i+1}) 被阻止。")
            time.sleep(deepseek_breaker.reset_timeout / 2) # 等待熔斷器重置
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 429:
                print(f"收到 DeepSeek API 429 錯誤 ({i+1})，但熔斷器尚未打開。")
            else:
                print(f"DeepSeek API 請求失敗 ({i+1}): {e.response.status_code} - {e.response.text}")
            time.sleep(1)
        except requests.exceptions.RequestException as e:
            print(f"DeepSeek API 請求發生網絡錯誤 ({i+1}): {e}")
            time.sleep(1)
        except Exception as e:
            print(f"發生未知錯誤 ({i+1}): {e}")
            time.sleep(1)

代碼說明：

• CircuitBreaker(fail_max=3, reset_timeout=30)：初始化熔斷器，在連續 3 次失敗後打開，並在 30 秒後進入半開狀態。
• @deepseek_breaker：將熔斷器應用於函數上。
• CircuitBreakerError：當熔斷器處於「打開」狀態時，所有被保護的函數呼叫都會立即拋出此異常，而不會實際執行底層邏輯。
• 事件監聽器：pybreaker 允許您註冊回調函數，以在熔斷器狀態改變時執行特定操作（例如日誌記錄、發送警報等）。

整合指數退避與熔斷器

在實際應用中，將指數退避和熔斷器結合使用，可以提供最全面的彈性。熔斷器用於防止對持續故障的服務進行不必要的嘗試，而指數退避則用於處理暫時性的、可恢復的錯誤。

協同工作方式：

1. 熔斷器閉合狀態：請求正常發送。如果遇到暫時性 429 錯誤，指數退避會介入進行重試。
2. 熔斷器打開狀態：如果 429 錯誤頻繁發生，導致熔斷器打開，那麼所有請求將被熔斷器直接拒絕，不再觸發指數退避，從而立即失敗，避免向 DeepSeek API 發送更多請求。
3. 熔斷器半開狀態：在重置時間後，熔斷器允許少量請求通過。這些探測性請求如果成功，會使熔斷器閉合；如果失敗，則熔斷器再次打開。此時，指數退避仍可在半開狀態下的探測請求中發揮作用。

整合範例

import os
import time
import requests
from tenacity import (
    retry,
    wait_exponential,
    stop_after_attempt,
    retry_if_exception_type,
    before_sleep_log,
)
from pybreaker import CircuitBreaker, CircuitBreakerError
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# 從環境變量獲取 DeepSeek API Key
DEEPSEEK_API_KEY = os.environ.get("DEEPSEEK_API_KEY")
if not DEEPSEEK_API_KEY:
    raise ValueError("請設定環境變數 DEEPSEEK_API_KEY")

DEEPSEEK_API_URL = "https://api.deepseek.com/v1/chat/completions"

class RateLimitError(Exception):
    """自定義異常，用於標識 429 錯誤"""
    pass

# ---- 熔斷器配置 ----
deepseek_breaker = CircuitBreaker(
    fail_max=5,              # 5 次失敗後打開熔斷器
    reset_timeout=60,        # 熔斷器打開後 60 秒進入半開狀態
    exclude=[
        requests.exceptions.ConnectionError,
        RateLimitError # 429錯誤由指數退避處理，熔斷器不直接計數，除非退避重試最終失敗
    ]
)

def on_circuit_open(cb, exc):
    logger.critical(f"DeepSeek API 熔斷器已打開！請求將被阻止。原因: {exc.__class__.__name__}: {exc}")

def on_circuit_half_open(cb):
    logger.warning("DeepSeek API 熔斷器進入半開狀態，將允許探測性請求通過。")

def on_circuit_closed(cb):
    logger.info("DeepSeek API 熔斷器已閉合，恢復正常操作。")

deepseek_breaker.add_listener(on_circuit_open)
deepseek_breaker.add_listener(on_circuit_half_open)
deepseek_breaker.add_listener(on_circuit_closed)

# ---- 指數退避和熔斷器整合函數 ----
@deepseek_breaker # 首先應用熔斷器
@retry(
    wait=wait_exponential(multiplier=1, min=2, max=30), # 每次重試等待時間為 2^x 秒，最小 2 秒，最大 30 秒
    stop=stop_after_attempt(3),                          # 最多重試 3 次 (在熔斷器開放前)
    retry=(retry_if_exception_type(RateLimitError)),     # 僅在 RateLimitError 發生時重試
    before_sleep=before_sleep_log(logger, logging.INFO)  # 重試前記錄日誌
)
def call_deepseek_api_robustly(prompt: str) -> str:
    """
    呼叫 DeepSeek API，整合了指數退避與熔斷器模式。
    """
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {DEEPSEEK_API_KEY}"
    }
    data = {
        "model": "deepseek-chat",
        "messages": [
            {"role": "user", "content": prompt}
        ],
        "temperature": 0.7,
        "max_tokens": 100
    }

    logger.info(f"正在發送 DeepSeek API 請求 (熔斷器狀態: {deepseek_breaker.current_state.name})...")
    try:
        response = requests.post(DEEPSEEK_API_URL, headers=headers, json=data, timeout=30)
        response.raise_for_status() # 對於 4XX/5XX 響應拋出 HTTPError
        return response.json()['choices'][0]['message']['content']
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 429:
            retry_after = e.response.headers.get("Retry-After")
            logger.warning(f"收到 DeepSeek API 429 錯誤，建議在 {retry_after} 秒後重試。觸發指數退避。")
            raise RateLimitError(f"DeepSeek API 達到速率限制: {e.response.text}") from e
        else:
            logger.error(f"DeepSeek API 請求失敗: {e.response.status_code} - {e.response.text}")
            raise # 其他 HTTP 錯誤不計入熔斷器失敗計數
    except requests.exceptions.RequestException as e:
        logger.error(f"DeepSeek API 請求發生網絡錯誤: {e}")
        # 網絡錯誤會計入熔斷器的失敗計數 (如果未被 exclude)
        raise

if __name__ == "__main__":
    test_prompt = "請解釋香港金融業的未來趨勢。"
    for i in range(15): # 模擬更多次請求
        try:
            result = call_deepseek_api_robustly(test_prompt)
            print(f"DeepSeek API 回應 ({i+1}): {result}")
            time.sleep(1) # 模擬正常請求間隔
        except CircuitBreakerError:
            print(f"DeepSeek API 熔斷器已打開，請求 ({i+1}) 被阻止。當前狀態: {deepseek_breaker.current_state.name}")
            time.sleep(deepseek_breaker.reset_timeout / 5) # 稍微等待，但不要等到重置
        except RateLimitError as e:
            print(f"DeepSeek API 最終因限流而失敗 ({i+1}): {e}")
            time.sleep(2) # 如果最終重試都失敗了，多等一會兒
        except requests.exceptions.RequestException as e:
            print(f"DeepSeek API 請求發生網絡或不明HTTP錯誤 ({i+1}): {e}")
            time.sleep(1)
        except Exception as e:
            print(f"發生未知錯誤 ({i+1}): {e}")
            time.sleep(1)

這個整合範例將 @deepseek_breaker 放在 @retry 之上。這意味着在執行實際的 API 呼叫之前，熔斷器會先檢查其狀態。如果熔斷器已打開，請求會被立即阻止，tenacity 的重試邏輯甚至不會啟動。只有當熔斷器處於閉合或半開狀態時，請求才會繼續，並由 tenacity 處理潛在的 429 錯誤和指數退避重試。

最佳實踐與監控

• 日誌記錄 (Logging)：詳細記錄所有 API 請求、成功響應、429 錯誤、重試嘗試以及熔斷器狀態變更。這對於事後分析和問題排查至關重要。
• 監控 (Monitoring)：設定實時監控 DeepSeek API 的錯誤率（尤其是 429 錯誤）、請求延遲和應用程式的熔斷器狀態。當錯誤率過高或熔斷器打開時，應觸發警報，以便及時介入。
• 參數調優：指數退避的 min, max 等待時間，以及熔斷器的 fail_max, reset_timeout 應根據您的應用程式的流量模式、對延遲的容忍度以及 DeepSeek API 的實際限制進行測試和調整。沒有一勞永逸的設定，需要持續優化。
• 抖動 (Jitter) 的引入：在指數退避中，通常建議引入隨機抖動（例如在計算出的延遲時間基礎上加減一個小範圍的隨機值），以避免大量客戶端在同一時間點重試，造成「雷鳴堆疊 (Thundering Herd)」效應。tenacity 的 wait_exponential 預設就包含抖動。
• DeepSeek 平台額度管理：定期檢查您在 DeepSeek 平台上的 API 使用情況和額度限制，避免達到硬性限制而導致服務中斷。

透過實作這些彈性模式和監控策略，您的應用程式將能更穩健地應對 DeepSeek API 的限流挑戰，保障服務的高可用性與用戶的流暢體驗。