「在免費 API 與生產級維運之間，強健的 API Gateway 與 Model Fallback 自動降級是能讓 AI Agent 穩定運作的橋樑。」

在先前的文章中，我們建置了 RHEL 上的 Rootless Podman 容器環境，並透過 Ansible 實現了自動化部署與巡檢 !

從今天開始我們正式進入 Angelina AI Agent 的 Python 服務架構層！我們將深入探討 app/main.py 作為 FastAPI 的中心樞紐、app/services/gemini_gateway.py 如何實現 Gemini 2.5 Flash API 整合與流量控管，以及自動化任務（如 daily_analysis.py）中如何設計 Model Fallback 自動降級備援機制！

本篇重點摘要

1. 拆解 app/main.py 的 FastAPI 服務架構與 Lifespan 模組生命週期。
2. 解析 app/services/gemini_gateway.py 中的限流保護（15 RPM）與 429 配額重置。
3. 實作多模型自動降級（Model Fallback）機制，確保自動化分析腳本不中斷。
4. 實作 /chat 核心 API 流程與特殊指令 (/learn, /fetch-url, /learning-stats) 處理。

一、系統入口與單例生命週期 (app/main.py)

在 app/main.py 中，我們使用 FastAPI 的 @asynccontextmanager（Lifespan 機制）來集中管理四大單例服務（Singletons）的初始化與 Graceful Shutdown：

Python
# app/main.py 服務初始化摘要
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
    global _memory, _rag_engine, _gemini_gateway, _learning_module

    logger.info("application_startup", message="Initializing services...")
    _memory = ConversationMemory()
    await _memory.initialize()

    _rag_engine = RAGEngine()
    await _rag_engine.initialise()

    _gemini_gateway = GeminiGateway()
    _learning_module = LearningModule(rag_engine=_rag_engine, gemini_gateway=_gemini_gateway)
    yield

    if _gemini_gateway:
        await _gemini_gateway.close()

二、Gemini API 閘道器實務 (app/services/gemini_gateway.py)

在 Free Tier 環境中，最常遇到的是 Rate Limit (15 RPM) 或 Daily Quota 耗盡 (HTTP 429 RESOURCE_EXHAUSTED)。
為此，我們在 GeminiGateway 中設計了完整的防禦機制：
1. Sliding Window 滑動視窗算法：利用 time.monotonic() 紀錄 60 秒內的請求時間戳，精準控管每分鐘流量。
2. 容量保護佇列 (GeminiQueueFullError)：當排隊請求超過 10 個時立即拋出異常，防止系統無限制耗盡 CPU 與記憶體。
3. UTC 零點自動重置（Quota Auto-Reset）：當偵測到 daily quota 耗盡時標記狀態，並在跨越 UTC 00:00 時自動解除鎖定恢復服務。

滑動視窗與流量控制邏輯

Python
async def _acquire_rate_limit(self) -> int:
    async with self._lock:
        self._cleanup_old_timestamps()

        # 若在每分鐘限額內，立即取得槽位
        if len(self._request_timestamps) < self._max_rpm:
            self._request_timestamps.append(time.monotonic())
            return 0

        # 若超過排隊容量限制 (>10)，直接拒絕請求
        if self._queued_count >= self._queue_capacity:
            raise GeminiQueueFullError()

        self._queued_count += 1
        wait_time = ceil(self._queued_count / self._max_rpm)

    await asyncio.sleep(wait_time)
    
    async with self._lock:
        self._request_timestamps.append(time.monotonic())
    return wait_time

三、Model Fallback 自動降級備援機制 (daily_analysis.py)

當主要模型（例如 gemini-2.5-flash）因為超額或服務異常無法回應時，系統必須具備自動降級（Fallback）機制，自動輪詢次要備援模型（如 gemini-flash-lite-latest），並搭配 Exponential Backoff 重試邏輯：

Python
# daily_analysis.py 中的 Model Fallback 實作範例
GEMINI_MODELS = ['gemini-2.5-flash', 'gemini-flash-lite-latest']

async def generate_report(market_data, knowledge):
    async with httpx.AsyncClient() as client:
        # 依次嘗試清單中的模型 (Fallback 機制)
        for model_name in GEMINI_MODELS:
            url = f"https://generativelanguage.googleapis.com/v1beta/models/{model_name}:generateContent?key={GEMINI_API_KEY}"
            
            # 單一模型重試 3 次
            for attempt in range(3):
                resp = await client.post(url, json=payload, timeout=120)
                if resp.status_code == 200:
                    data = resp.json()
                    return data['candidates'][0]['content']['parts'][0]['text']
                elif resp.status_code in (503, 429):
                    await asyncio.sleep(20 * (attempt + 1))
                else:
                    break  # 切換至下一個備援模型
                    
        print("[ERROR] All fallback models failed.")
        return None

四、/chat 核心對話流程與內建指令

在 POST /chat 端點中，除了常規的對話流程之外，我們還實現了特殊的快捷指令：
1. /learning-stats：查詢目前向量資料庫累積的總知識點與當前 Session 學習量。
2. /learn ：允許使用者或系統自動腳本直接將文本輸入注入 RAG 知識庫。
3. /fetch-url ：抓取指定網頁內容，自動剔除 HTML/CSS/JS 標籤並分塊存入知識庫。

Python
# app/main.py 中的 /chat 端點核心流轉
@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest) -> ChatResponse:
    # 1. 載入當前 Session 最近 20 筆歷史對話 (SQLite)
    context_turns = await _memory.load_context(session_id, limit=20)

    # 2. 進行 RAG 語意檢索 (ChromaDB top_k=5, similarity >= 0.5)
    knowledge_chunks = await _rag_engine.search(query=message, top_k=5)

    # 3. 呼叫 Gemini Gateway 生成回應
    gemini_response = await _gemini_gateway.generate(...)

    # 4. 將 user 與 assistant 的對話寫回 SQLite
    await _memory.save_turn(session_id, "user", message)
    await _memory.save_turn(session_id, "assistant", reply_text)

    # 5. 觸發非同步背景學習任務 (Non-blocking)
    asyncio.create_task(_learning_module.extract_and_store(assistant_turn, session_id))

    return response

五、今日總結與架構亮點

今天我們完成了 AI Agent 核心推理層與 API 閘道的解析：
1. 高可用保護：透過 GeminiGateway 阻擋短時間高併發衝擊，解決免費 API 的 429 限流問題。
2. 降級容錯 (Fallback)：透過模型備援清單，確保主要模型過載時自動降級至 Lite 模型完成分析任務。
3. 非同步無阻塞設計：對話生成後立即回應使用者，背景學習（extract_and_store）透過 asyncio.create_task 在背景平行處理。

明天（Day 05）我們將深入 app/services/rag_engine.py，探討如何利用 ChromaDB 與 sentence-transformers 打造本地端零成本的 RAG 知識檢索增強引擎！

明日預告：【Day 05】知識檢索增強：ChromaDB 與 sentence-transformers 打造本地端 RAG 引擎