Day 27：重試、熔斷與配額防禦：打造穩健的 Exponential Backoff 與降級機制

在 Day 20 打造 Google Sheets 戰情室與後續串接無人值守巡航時，我們遭遇了一場震撼教育：明明在網頁端測試順暢的 Prompt，一放進 GitHub Actions 定時排程，**整排待處理情資瞬間全部變成紅色 ERROR**！

點開 Log 一看，充斥著 503 Service Unavailable 與 429 Resource Exhausted。

許多開發者在此常產生一個巨大誤區：「我明明訂閱了 Google AI Pro（或 Gemini Advanced），為什麼還會被限流？」

今天我們將揭開消費者訂閱與開發者 API 的配額壁壘，並透過實戰驗證過的 Tenacity 指數退避重試、BaseAgent 多模型自動降級、均勻冷卻節奏 與 CI/CD 防死鎖機制，徹底解決外部 API 的抖動與超載問題！

一、迷思破解：Google AI Pro 會員 vs. AI Studio API 配額

很多剛接觸 LLM 開發的朋友常問：「我買了 Google AI Pro 方案（每月 650 元），每天可以在網頁或 App 狂用 2.5/3.0 Pro，為什麼 Python 程式跑幾筆就噴 429/503？」

這是兩套完全獨立的計費與配額體系：

1. 消費者端產品（gemini.google.com / Gemini App）：

• 綁定 Google One AI Premium / Google AI Pro 訂閱。
• 計費核心是提供給終端人類使用的網頁介面、上下文窗口與各項附加功能。

2. 開發者端產品（Google AI Studio / Vertex AI）：

• 走的是後端 API 呼叫（使用 GEMINI_API_KEY）。

• 預設為 AI Studio Free Tier：

• 速率限制（Rate Limit）：通常嚴格限制在 5 RPM 或 15 RPM（Requests Per Minute）。

• 伺服器排隊優先權：Free Tier 在尖峰時段無 SLA 保證，Google 伺服器會動態丟棄連線，拋出 503 UNAVAILABLE。

• 連鎖反應：我們的一筆情報需要依序經過 Reviewer ➔ Domain Expert ➔ Synthesizer ➔ Editor 四大代理人！如果一次處理 2 筆情資，瞬間就發出 8 次請求，若無節奏控制，必定撞毀在 5 RPM 的天花板上。

二、第一道防線：Pipeline 階段的 Tenacity 指數退避重試

當伺服器拋出 503 或網路暫時性抖動時，盲目重試只會加劇伺服器負載，而過長的等待又會拖死 CI/CD。

我們使用 tenacity 函式庫，在 src/services/pipeline.py 中替每個代理人階段打造具備指數退避的執行器：

# src/services/pipeline.py (節錄)
from tenacity import retry, stop_after_attempt, wait_exponential

class IntelligencePipeline:
    ...
    @retry(
        # 最多重試 3 次（含初次共 3 次）
        stop=stop_after_attempt(3),
        # 指數退避公式：Wait = min(max, multiplier * 2^(attempt))
        # 初始等待約 3 秒，隨後 6 秒、12 秒，上限 15 秒
        wait=wait_exponential(multiplier=2, min=3, max=15),
        # 遇到未捕獲之重大例外時向上拋出
        reraise=True,
    )
    def _execute_stage_with_retry(self, stage_fn, item):
        """防禦性單一階段執行器"""
        return stage_fn(item)

• multiplier=2, min=3, max=15：
• 第 1 次失敗：等待約 3~4 秒。
• 第 2 次失敗：等待約 6~8 秒。
• 第 3 次失敗：等待約 12~15 秒。
  給予 Google 後端足夠的時間釋放負載，成功率大幅提升。

三、第二道防線：BaseAgent 多模型自動降級鏈路（Model Fallback）

如果主模型（如 gemini-3.8-flash）遭遇不可逆的服務中斷，重試 3 次依然是 503 怎麼辦？

我們不應讓整條管線掛掉，而是應該自動無縫降級至負載較低、吞吐量更大的輕量模型（gemini-3.8-flash-lite）。

我們將此機制直接封裝進 src/agents/base.py，使所有的子代理人（Reviewer、Domain Expert 等）完全零感知：

# src/agents/base.py
import json
import logging
import time
from typing import Any, List, Optional, Tuple
from google import genai
from google.genai import types

logger = logging.getLogger(__name__)

class BaseAgent:
    def __init__(
        self,
        client: genai.Client,
        primary_model: str = "gemini-3.8-flash",
        fallback_models: Optional[List[str]] = None,
    ):
        self.client = client
        self.primary_model = primary_model
        # 備援清單：若主力 503，自動降級至 gemini-3.8-flash-lite
        self.models_to_try = [primary_model] + (
            fallback_models or ["gemini-3.8-flash-lite"]
        )

    def _generate_with_fallback(
        self, prompt: str, schema: Any, system_instruction: Optional[str] = None
    ) -> Tuple[Any, str]:
        """統一封裝呼叫邏輯：支援 503 自動降級與 JSON 解析"""
        last_exception = None

        for model_name in self.models_to_try:
            try:
                config_kwargs = {
                    "response_mime_type": "application/json",
                    "response_schema": schema,
                    "temperature": 0.1,
                }
                if system_instruction:
                    config_kwargs["system_instruction"] = system_instruction

                response = self.client.models.generate_content(
                    model=model_name,
                    contents=prompt,
                    config=types.GenerateContentConfig(**config_kwargs),
                )
                
                # 支援 Pydantic 模型驗證與一般 JSON
                if hasattr(schema, "model_validate_json"):
                    result = schema.model_validate_json(response.text)
                else:
                    result = json.loads(response.text)
                return result, model_name

            except Exception as e:
                last_exception = e
                err_msg = str(e)
                # 只有在遇到 503 / UNAVAILABLE 且還有備援模型時才降級
                if (
                    "503" in err_msg or "UNAVAILABLE" in err_msg
                ) and model_name != self.models_to_try[-1]:
                    logger.warning(
                        f"模型 [{model_name}] 遭遇 503 尖峰，自動降級切換至備援模型嘗試..."
                    )
                    time.sleep(2)
                    continue
                # 若是 429 配額或其他邏輯錯誤，立即向上拋給 retry 或 pipeline
                raise e

        raise last_exception

子代理人（如 ReviewerAgent）呼叫時，只要維持一行：

result, used_model = self._generate_with_fallback(
    prompt, ReviewResult, system_instruction=REVIEWER_SYSTEM_INSTRUCTION
)
item.metadata["analyzed_by_model"] = used_model

上層業務不需要寫任何 try...except，就能在日誌中看到降級運作的軌跡！

四、第三道防線：節奏冷卻與均勻控速（勻速 12 秒）

在單條情資需要連續呼叫 4 個 Agent 的鏈路下，若執行過快，極易撞上 5 RPM 配額。

最簡單且最有效的防禦策略，是在每個代理人交接之間插入均勻的冷卻時間：

$$\text{間隔時間} = \frac{60 \text{ 秒}}{5 \text{ RPM}} = 12 \text{ 秒}$$

在 src/services/pipeline.py 的主流程中落實：

        # Step 1: Reviewer 初審
        item, should_continue = self._execute_stage_with_retry(self.reviewer.evaluate, item)
        if not should_continue:
            return item
        time.sleep(12)  # 冷卻確保不超過 5 RPM

        # Step 2: Domain Expert 專家提煉
        item = self._execute_stage_with_retry(self.expert.analyze, item)
        time.sleep(12)

        # Step 3: Synthesizer 跨領域推演
        item = self._execute_stage_with_retry(self.synthesizer.synthesize, item)
        time.sleep(12)

        # Step 4: Editor 總編定稿
        item = self._execute_stage_with_retry(self.editor.edit, item)

透過精準的 12 秒節奏控制，我們在完全零成本的 Free Tier API 下，達成了 100% 不撞 429 的穩健運行！

五、第四道防線：管線防死鎖與 GitHub Actions 超時熔斷

在雲端環境無人值守運行時，還有最後一個常見的隱形殺手：Socket 掛起（Hang）。
有時 API 伺服器沒有拋出 503，而是連線中斷後連線池未正確超時，導致 Python 程式永久卡在 client.models.generate_content。

實戰防護配置：

在 .github/workflows/patrol.yml（或 obsidian_sync.yml）中，在 Job 層級強制設定：

jobs:
  patrol-and-sync:
    name: Run Patrol and Sync to Obsidian Vault
    runs-on: ubuntu-latest
    timeout-minutes: 5 # 超過 5 分鐘強制中斷，防止死鎖掛起扣光 Actions 分鐘數

這能保證即便遭遇任何未知的網路死鎖，工作流程也能在 5 分鐘內安全強制熔斷，避免耗盡 GitHub Actions 每月的免費運算時間！

總結與下一步

今天我們結合了這幾天真實的除錯血淚，為情報雷達構築了完整的韌性體系：

1. 認清配額邊界：釐清了消費者 Google AI Pro 與開發者 AI Studio 的獨立額度。

2. 三級防禦機制：tenacity 指數退避 ➔ BaseAgent 自動模型降級（flash ➔ flash-lite）➔ time.sleep(12) 均勻控速。

3. CI/CD 熔斷防護：timeout-minutes: 5 杜絕無人值守時的連線死鎖。

現在，系統的四道防禦層已經全面就緒。然而，我們的代碼庫涵蓋了多個適配器（Sheets, Obsidian, Notion）與外部服務。如果每次測試都要呼叫真實的第三方 API，不僅浪費額度且測試極度緩慢。

在 Day 28，我們將進入專業工程品質把關：「防禦型合約測試：針對 Obsidian / Notion 適配器的自動化單元測試」，使用 Mock 技術確保所有外掛端口在斷網環境下也能以毫秒級速度全數綠燈通過！