AI Agent 設計：降級處理——多供應商 Fallback、限流重試與 Token 成本估算實作
在設計Agent系統時必須考慮降級處理，遇到服務終端、額度耗盡時必須能夠自動轉換服務商，這其中就包含了以下幾項要點
- 多供應商抽象
- Token / 成本估算
- Streaming
- 限流重試
- 多模型 fallback
供應商偵測與優先序
python llm_api_lab.py providers輸出結果：
供應商偵測（呼叫時依此優先序選用第一個可用者）：
1. Kimi/Moonshot ✗ 未設定 ← KIMI_API_KEY 或 課程根目錄 kimi.json
2. Anthropic ✗ 未設定 ← ANTHROPIC_API_KEY
3. OpenAI ✗ 未設定 ← OPENAI_API_KEY
Token 與成本估算
python llm_api_lab.py tokens --text "把這段話估算 token 與成本"
輸出結果：
def cmd_tokens(args):
n, how = count_tokens(args.text)
print(f"輸入文字：{args.text!r}")
print(f"token 估算：{n}（{how}）\n")
print(f"假設輸出 200 tokens，各模型單次成本：")
print(f" {'模型':16s}{'input$':>10}{'output$':>10}{'單次總計$':>12}")
for model, (pin, pout) in PRICING.items():
cin = n / 1e6 * pin
cout = 200 / 1e6 * pout
print(f" {model:16s}{cin:>10.6f}{cout:>10.6f}{cin + cout:>12.6f}")
print("\n洞察：成本主要由 output 與模型檔次決定。生產上常用「小模型分流 + 大模型兜底」省成本。")輸出結果：
輸入文字：'把這段話估算 token 與成本'
token 估算：10（近似估算（未裝 tiktoken））
假設輸出 200 tokens，各模型單次成本：
模型 input$ output$ 單次總計$
gpt-4o 0.000025 0.002000 0.002025
claude-opus-4-8 0.000050 0.005000 0.005050
kimi-k2.6 0.000006 0.000500 0.000506
gpt-4o-mini 0.000002 0.000120 0.000121
成本主要由 output 與模型檔次決定。生產上常用「小模型分流 + 大模型兜底」省成本。
Streaming 串流輸出
python llm_api_lab.py stream --prompt "用一句話介紹 RAG" --offline
輸出結果：
def cmd_stream(args):
if args.offline or not any(ok for _, ok, _ in detect_providers()):
text = f"（mock）關於「{args.prompt}」：這是一段逐字串流輸出的示範回覆。"
for ch in text:
sys.stdout.write(ch); sys.stdout.flush(); time.sleep(0.01)
print("\n\n[stream 結束] 真實 API 用 stream=True，逐 chunk 取 delta，邊收邊顯示降低首字延遲。")
return
# 線上：以 Kimi/OpenAI 相容介面示範真實 streaming
kimi = _kimi_cfg()
from openai import OpenAI
cli = OpenAI(api_key=kimi["api_key"], base_url=kimi["base_url"]) if kimi else OpenAI()
model = kimi["model"] if kimi else "gpt-4o-mini"
stream = cli.chat.completions.create(
model=model, messages=[{"role": "user", "content": args.prompt}], stream=True)
for chunk in stream:
delta = chunk.choices[0].delta.content or ""
sys.stdout.write(delta); sys.stdout.flush()
print("\n[stream 結束]")輸出結果：
（mock）關於「用一句話介紹 RAG」：這是一段逐字串流輸出的示範回覆。
[stream 結束] 真實 API 用 stream=True，逐 chunk 取 delta，邊收邊顯示降低首字延遲。當使用者等整段生成完要好幾秒，串流讓首字在數百毫秒內出現，體驗天差地別。聊天、Copilot 類產品幾乎都必須串流。
限流重試（指數退避 + jitter）
python llm_api_lab.py retry --fail-n 2
輸出結果：
def call_with_retry(fn, max_retries=5, base=0.5):
"""指數退避 + jitter 的通用重試包裝。fn 應在失敗時 raise，成功回傳結果。"""
for attempt in range(max_retries + 1):
try:
return fn()
except Exception as e:
if attempt == max_retries:
print(f"✗ 重試 {max_retries} 次仍失敗：{e}")
raise
wait = base * (2 ** attempt) + random.uniform(0, 0.3) # jitter 防驚群
print(f" ⚠ 第 {attempt + 1} 次失敗（{e}），{wait:.2f}s 後重試…")
time.sleep(wait)
def cmd_retry(args):
state = {"calls": 0}
def flaky(): # 模擬前 N 次回 429，之後成功
state["calls"] += 1
if state["calls"] <= args.fail_n:
raise RuntimeError("429 Too Many Requests")
return "✓ 第 %d 次呼叫成功取得回覆" % state["calls"]
print(f"模擬：前 {args.fail_n} 次觸發 429，重試機制應自動退避後成功\n")
print(call_with_retry(flaky))
print("\n要點：只對可重試錯誤（429/5xx/timeout）重試；4xx 參數錯誤不該重試。jitter 避免同時重試。")輸出結果：
模擬：前 2 次觸發 429，重試機制應自動退避後成功
⚠ 第 1 次失敗（429 Too Many Requests），0.66s 後重試…
⚠ 第 2 次失敗（429 Too Many Requests），1.27s 後重試…
✓ 第 3 次呼叫成功取得回覆
要點：只對可重試錯誤（429/5xx/timeout）重試；4xx 參數錯誤不該重試。jitter 避免同時重試。- 指數退避：等待時間
base * 2^attempt，避免一直猛打加重過載 - jitter：加隨機抖動，防止大量 client 同時重試造成「驚群」
- 只對可重試錯誤重試：429 / 5xx / timeout 重試；4xx 參數錯誤直接放棄
多模型 Fallback
python llm_api_lab.py fallback
輸出結果：
def cmd_fallback(args):
# 模擬一條「主模型故障 → 自動降級到備援」的鏈路
chain = [("primary: claude-opus-4-8", True), # True = 故障
("secondary: gpt-4o", True),
("fallback: kimi-k2.6", False)] # False = 正常
print("多模型 fallback 鏈路（任一成功即停）：\n")
for name, broken in chain:
if broken:
print(f" ✗ {name} 不可用，切換下一個…")
continue
print(f" ✓ {name} 回應成功 → 採用此結果")
break
else:
print(" ✗ 全鏈路皆失敗 — 應回傳降級訊息或排隊重試")
print("\n要點：fallback 要考慮成本/品質排序、逾時門檻、以及記錄每次降級以便監控告警。")輸出結果：
多模型 fallback 鏈路（任一成功即停）：
✗ primary: claude-opus-4-8 不可用，切換下一個…
✗ secondary: gpt-4o 不可用，切換下一個…
✓ fallback: kimi-k2.6 回應成功 → 採用此結果fallback 要考慮成本/品質排序、逾時門檻、以及記錄每次降級以便監控告警。