iT邦幫忙

2026 iThome 鐵人賽

DAY 15
0
Build on Google AI

打造零成本企業級 AI Agent:以 Gemini 2.5 Flash 構建金融分析助手與維運實戰系列 第 23

【Day 23】效能與成本優化:Gemini API Token 消耗分析與 Rate Limit 應對實務

  • 分享至 

  • xImage
  •  

「在 AI Agent 的生產級運維中,Token 消耗不只是金錢成本,更直接影響 API 的 Rate Limit 額度與整體回應延遲。透過階段性 Context 剪裁與精準的 Request Rate 防禦,能大幅提高 Agent 在高併發與盤後尖峰時刻的穩定度。」

先前,我們完成了一系列的架構重構、Telegram Byte-Safe 推播、Google Sheets 自動追蹤、systemd 服務託管、資安縱深防禦與離線評測資料沉澱。
當系統每天自動化運行時,我們開始關注另一個核心運維指標:API 的效能與 Token 成本。
特別是在使用 Gemini API 進行盤後數據與大量新聞分析時,若一股腦將所有 raw data 與全量歷史新聞帶入 Prompt,不僅容易引發 API 的 429 Too Many Requests (Rate Limit) 限制,更會造成不必要的 Token 浪費與 Latency 增加。
今天我們將解析我們如何在程式碼中實現 Token 消耗控制、自動化 Rate Limit 避讓與請求重試機制!

本篇重點摘要

  1. 剖析透過三階段分析(Stage 1-3)控制 Context Window 的節省策略。
  2. 拆解 API 呼叫間隔控制(Rate Limit Safeguard)與 asyncio.sleep(1) 避讓實務。
  3. 實作防護型 Try-Except 降級機制,確保尖峰時刻服務不中斷。

一、Token 消耗優化:三階段管道的 Context 剪裁策略

在早期設計中,若將所有市場 raw JSON、三大法人數據與數十條盤後新聞直接塞入單一 Prompt,單次呼叫的 Token 消耗高達 8,000~15,000 Tokens。
在先前重構的三階段管道(Stage 1-3 Pipeline)中,我們實作了精準的 Context 剪裁機制:
1. Stage 1(盤態精簡分類)
策略:僅餵給當日市場結構數據,輸出結構化短 JSON(如 market_sentiment, key_drivers)。
Token 效益:輸入約 800 Tokens,輸出小於 150 Tokens,極速完成大盤結構定調。

  2. Stage 2(ChromaDB 向量檢索過濾)
     策略:不將整份知識庫帶入,而是透過語意相似度檢索(Top-K),僅提取最相關的 3-5 條投資框架與歷史邏輯。
     Token 效益:節省超過 70% 不必要的背景 Context 消耗。

  3. Stage 3(最終生成控制)
     策略:Prompt 明確規定「報告長度:總輸出字數嚴格控制在 800-1000 字」,從模型輸出端直接鎖定 Max Output Tokens。
     Token 效益:將單次全套運算的 Token 總消耗穩定控制在最平價且高效的區間內。

二、API Rate Limit 應對與 asyncio.sleep 避讓實務

Gemini API 與 Telegram Bot API 對於 Requests Per Minute 皆有硬性限制。若短時間內連續觸發多次呼叫或發送多段推播訊息,極易踩到 429 Rate Limit。
我們在 daily_analysis.py 的 send_telegram() 中實作了主動式 Rate Limit 避讓與請求間隔(Throttling):

### Telegram 訊息多段發送間隔 (send_telegram)
     在 Telegram 發送長篇報告切分訊息時,為避免觸發 Telegram Bot API Rate Limit,每次發送之間加入硬性 1 秒延遲:
Python
# daily_analysis.py 中的 Telegram 推播 Rate Limit 避讓真實程式碼
for i, part in enumerate(messages):
    payload = {
        'chat_id': TELEGRAM_CHAT_ID,
        'text': part,
        'parse_mode': 'Markdown'
    }
    resp = await client.post(url, json=payload, timeout=30)

    # 多段訊息發送之間加入 1 秒延遲,遵守 Telegram API Rate Limit
    if i < len(messages) - 1:
        await asyncio.sleep(1)

三、例外捕捉與安全降級(Exception Handling & Fallback)

當 Gemini API 出現暫時性網路抖動或 Stage 1 JSON 回傳格式不符合預期時,系統不會直接放任程式潰散,而是透過 try...except 捕捉例外並給予標準預設值,完成 Graceful Degradation:

Python
# daily_analysis.py 中的 Stage 1 API 呼叫與 JSON 解析降級實作
try:
    stage1_data = json.loads(cleaned_response)
except json.JSONDecodeError:
    print("  [WARN] Stage 1 JSON parse failed, using fallback market state.")
    # 解析失敗時降級為安全預設值,確保 Stage 3 依然能正常順暢運作
    stage1_data = {
        "market_sentiment": "中性",
        "key_drivers": ["盤面數據解析異常,採用保守觀察策略"],
        "risk_level": "中"
    }

營運效益:

 高可用性保證:即便 Gemini API 在盤後尖峰時刻回傳格式稍有瑕疵,系統依然能透過安全 Fallback 完成整個三階段流程,確保 Telegram 推播管線 100% 成功執行。

四、今日總結

透過 daily_analysis.py 中的 Token 與效能優化設計,我們達成了:
1. Context 精準控管:三階段管道有效將單次分析的 Token 消耗鎖定在最佳區間。
2. 主動式 Rate Limit 避讓:透過 asyncio.sleep(1) 建立請求間隔,徹底告別 API 429 錯誤。
3. 高韌性 Fallback 降級:完善的 Try-Except 解析降級機制,確保自動化推播鏈結的高可用性。

明日預告:敬請期待期待~


上一篇
【Day 22】離線評測體系:如何為 LLM 金融 Agent 建立客觀的 Evaluation 指標
系列文
打造零成本企業級 AI Agent:以 Gemini 2.5 Flash 構建金融分析助手與維運實戰23
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言