🔹 前言

經過前面系列的文章，我們已經從零搭建出一條完整的 RAG Pipeline ，目前具備以下能力：

1. 文件清洗與 Chunking 把原始文件整理成乾淨的段落。
2. Embedding 與向量庫 把段落轉換成向量，存到 Weaviate / FAISS。
3. Pipeline 自動化 用 Prefect / Airflow 排程，確保每天自動更新知識庫。
4. LLM 協助以人性化的方式回答
5. Client (Web / Slack Bot / Line Bot) 提供好用的介面

如下圖所示：

• 黃色的是我們已經完成的部分（Day04 ~ Day16）
• 綠色區塊則是今天（Day18）要探討的 API Gateway 與接下來幾篇要探討的主題

在 Day17 的混合架構我們定義了 Routing 策略，今天（Day18）我們會開始把這套策略部分落實到 API Gateway，並延伸介紹我們可以在 API Gateway 上面做什麼事情。

🔹 為什麼需要 API Gateway？

RAG Pipeline 打好基礎後，我們就擁有了一個 企業知識庫 + LLM 驅動的 Q&A 能力。
然而目前的架構還有一個明顯限制：
部分前端或新接入的應用（Web、Slack Bot、Line Bot、Discord Bot）仍是直接呼叫 LLM Provider（如 OpenAI API），而即便已有簡易後端，各應用也常各自直連 LLM、缺少一個統一出入口。

這會帶來三個問題：

1. 安全性：API Key 暴露在前端，容易外洩或被濫用，如果不同的部門有多個 API Key 管理上也有困難
2. 缺乏抽象化：之後如果要換成別的雲端模型或本地模型，其他的前端也要改
3. 無法集中管理：監控、快取、成本計算都很分散，沒有一致的端口可以控管

因此今天我們引入 API Gateway：讓所有前端只需呼叫同一個 /ask 出入口，在這一層集中處理 Auth/Rate Limit、Log、Cache、觀測與模型抽象化/路由，安全且一致地存取 LLM 與檢索能力。

🌉 API Gateway 的價值

[註1] Personally Identifiable Information，可識別個資

🔐 補充：前端請求驗證與金鑰治理

即使把 Provider 的 API Key 放到後端，仍需確認「誰在呼叫、能不能呼叫、可用多少」，以下是幾個常見做法：

• API Key per Client：每個前端/服務一把金鑰，便於配額與限流。
• JWT（存活時間短）：驗 iss/aud/scope/exp，適合 Web/Mobile 與 RBAC。
• OAuth2 / OIDC：與第三方整合時使用，由 IdP 發 token、Gateway 驗證。
• Secrets 與 Rotation：金鑰放 Secret Manager/KMS；加 kid 支援雙簽重疊期；30/60/90 天輪替；依環境/租戶分權並記審計日誌。

🔹Demo: 用 FastAPI 建立一個簡單的 LLM Proxy

完整的程式碼放在 GitHub Repo。

還記得在 Day07 - Demo 最小可行的 RAG QA Bot (Web 版)我們做了一份專案，功能是「先查詢內部知識庫，再交由 LLM 回答」，然後再把專案裡的後端程式讓前端頁面呼叫嗎 ? 我們現在要把這個後端程式抽出來，包裝成一個 Gateway。

📁 目錄結構

day18_LLM_Gateway/
├─ gateway/
│  └─ main.py              # LLM Gateway：統一入口 /ask
├─ services/
│  └─ retrieval_service.py # Day07 改寫的檢索邏輯（向量嵌入 + FAISS 搜尋）
├─ frontend/
│  └─ index.html           # Day07 前端
├─ .env
└─ requirements.txt

Gateway 資料夾底下的 main.py 只做「統一入口 + 呼叫模型」。
它會：

• 先呼叫 retrieval_service 拿到 context
• 再呼叫 OpenAI，之後也能在這層改爲使用 Anthropic / 本地模型
• 這裡就能逐步加上 Rate Limit / Logging / Cache / Metrics 等元件 ，Day16 之後的文章我們會逐步提到

先寫一個簡單的 gateway 版本：

# gateway/main.py
from fastapi import FastAPI, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, FileResponse
from fastapi.staticfiles import StaticFiles
from pathlib import Path
from typing import Dict, Any
import os, time, traceback

from dotenv import load_dotenv
from openai import OpenAI
from services.retrieval_service import retrieve_best

load_dotenv()

app = FastAPI(title="LLM Gateway (Day18)")

# CORS（同源存取時也能留著，上線請改白名單）
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

# === 前端頁面設定 ===
BACKEND_DIR = Path(__file__).resolve().parent
FRONTEND_DIR = BACKEND_DIR.parent / "frontend"

@app.get("/")                    # 讓根路徑回首頁
def root():
    idx = FRONTEND_DIR / "index.html"
    if idx.exists():
        return FileResponse(idx)
    return JSONResponse({"error": "frontend/index.html not found"}, status_code=404)

# 如需放 JS/CSS/圖，掛到 /_static（可選）
app.mount("/_static", StaticFiles(directory=FRONTEND_DIR, html=True), name="frontend")

# === 健康檢查 ===
@app.get("/healthz")
def healthz():
    return {"ok": True}

# === OpenAI Client ===
api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
    raise ValueError("沒有找到 OPENAI_API_KEY，請檢查環境變數！")

OPENAI_TIMEOUT = float(os.getenv("OPENAI_TIMEOUT", "20"))
CHAT_MODEL = os.getenv("CHAT_MODEL", "gpt-4o-mini")
client = OpenAI(timeout=OPENAI_TIMEOUT, max_retries=0)

# === 主要 API ===
@app.post("/ask")
def ask(payload: Dict[str, Any]):
    """
    輸入例：
    { "question": "加班規則？", "temperature": 0.2 }
    """
    start = time.time()
    try:
        question = (payload.get("question") or "").strip()
        if not question:
            raise HTTPException(status_code=400, detail="question is required")

        # 1) 先查知識庫（沿用 Day07/RAG）
        context, dist = retrieve_best(question, k=3)

        # 2) 再交由 LLM 回答
        messages = [
            {"role": "system", "content": "你是一個企業 FAQ 助理。"},
            {"role": "user", "content": f"根據以下知識庫內容回答：\n{context}\n\n問題：{question}"},
        ]
        resp = client.chat.completions.create(
            model=CHAT_MODEL,
            messages=messages,
            temperature=float(payload.get("temperature", 0.2)),
        )
        answer = resp.choices[0].message.content
        return {"question": question, "context": context, "answer": answer, "debug": {"l2": dist, "model": CHAT_MODEL}}
    except HTTPException:
        raise
    except Exception as e:
        traceback.print_exc()
        return JSONResponse(status_code=500, content={"error": str(e)})
    finally:
        print(f"[gateway:/ask] {(time.time()-start):.2f}s")

▶️ 啟動方式

# 安裝套件
pip install -r requirements.txt

# 啟動 Gateway
uvicorn gateway.main:app --reload --port 8000

測試（POST JSON）：

❯ curl -X POST "http://localhost:8000/ask" \
  -H "Content-Type: application/json" \
  -d '{"question":"加班規則是什麼？"}'
{"question":"加班規則是什麼？","context":"加班申請：需事先提出，加班工時可折換補休。","answer":"加班規則是：員工需事先提出加班申請，並且加班工時可以折換為補休。","debug":{"l2":0.9121904969215393,"model":"gpt-4o-mini"}}%

UI 介面測試：

🔹進階功能：加上 Rate Limit 、 Logging、Metrics

在真實環境，我們會用 Redis 或雲端 API Gateway 做限流，但為了 Demo，我們只要在程式裡加一個「記憶體限流」＋「簡單 Logging」+「Metrics」就好。

目標：

• 限流：同一個 IP 一分鐘最多 10 次請求，超過會回 429 Too Many Requests。
• Logging：每次請求都會在後端 log 記錄 IP 與問題。

⏰ Rate Limit :

⚠️ Demo 僅示範用途；生產請改 Redis 或 Edge 限流 :
Redis + Sliding Window/Token Bucket（慢限流 + 配額）
NGINX / Envoy 邊界限流
雲端：AWS API Gateway / Cloudflare Rate Limiting

限制每分鐘最多 10 次請求，並且把 access 紀錄只存在程式的記憶體裡（重啟就會清空）：

# ============================================================
# 🟢 記憶體限流 (In-Memory Rate Limit)
#   - Demo 用：每 IP 每分鐘最多 10 次
#   - 程式重啟會清空；多機不共享
#   - 生產環境請換 Redis / 雲端 API Gateway
# ============================================================
_WINDOW_SEC = int(os.getenv("RATE_LIMIT_WINDOW_SEC", "60"))
_LIMIT = int(os.getenv("RATE_LIMIT_MAX_PER_IP", "10"))
_BUCKETS: dict[str, deque] = defaultdict(deque)

def in_memory_rate_limit(request: Request):
    ip = request.client.host
    now = time.time()
    dq = _BUCKETS[ip]

    # 移除視窗外的請求
    while dq and now - dq[0] > _WINDOW_SEC:
        dq.popleft()

    if len(dq) >= _LIMIT:
        raise HTTPException(status_code=429, detail="Too Many Requests (demo limiter)")
    dq.append(now)

測試結果：

❯ curl -sS -X POST "http://localhost:8000/ask" \
  -H "Content-Type: application/json" \
  -d '{"question":"加班規則是什麼？"}'

{"detail":"Too Many Requests (demo limiter)"}

補充：實務常見做法

📋 Logging :

把 Query log 輸出到後端伺服器的 console / stdout，同時也寫一份到 gateway.log 檔案：

# ============================================================
# 🟠 Logging：同時輸出 console + 檔案 gateway.log
# ============================================================
root_logger = logging.getLogger()
if not root_logger.handlers:
    root_logger.setLevel(logging.INFO)
    fmt = logging.Formatter("%(asctime)s %(levelname)s %(message)s")

    # Console Handler
    ch = logging.StreamHandler()
    ch.setFormatter(fmt)
    root_logger.addHandler(ch)

    # File Handler
    fh = logging.FileHandler("gateway.log", encoding="utf-8")
    fh.setFormatter(fmt)
    root_logger.addHandler(fh)

log = logging.getLogger(__name__)

gateway.log:

2025-09-10 12:42:30,072 INFO HTTP Request: POST https://api.openai.com/v1/embeddings "HTTP/1.1 200 OK"
2025-09-10 12:42:30,481 INFO HTTP Request: POST https://api.openai.com/v1/embeddings "HTTP/1.1 200 OK"
2025-09-10 12:43:28,443 INFO HTTP Request: POST https://api.openai.com/v1/chat/completions "HTTP/1.1 200 OK"
2025-09-10 12:43:28,446 INFO [ASK] success in 1.94s, answer_len=32
2025-09-10 12:43:36,389 INFO [ASK] ip=127.0.0.1 q='加班規則是什麼？' temp=0.2
2025-09-10 12:43:36,735 INFO HTTP Request: POST https://api.openai.com/v1/embeddings "HTTP/1.1 200 OK"
2025-09-10 12:43:38,272 INFO HTTP Request: POST https://api.openai.com/v1/chat/completions "HTTP/1.1 200 OK"
2025-09-10 12:43:38,275 INFO [ASK] success in 1.89s, answer_len=32
2025-09-10 12:43:38,837 INFO [ASK] ip=127.0.0.1 q='加班規則是什麼？' temp=0.2
2025-09-10 12:43:39,193 INFO HTTP Request: POST https://api.openai.com/v1/embeddings "HTTP/1.1 200 OK"
2025-09-10 12:43:40,420 INFO HTTP Request: POST https://api.openai.com/v1/chat/completions "HTTP/1.1 200 OK"
2025-09-10 12:43:40,423 INFO [ASK] success in 1.59s, answer_len=32

⚠️ 這個版本只是 教學用 Demo，因為程式一重啟，限流的記錄就會消失。

📊 Metrics

最後在 Gateway 加上 最小可用的觀測性，在 main.py 加上相關設定：

from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST


REQS = Counter(
    "gateway_requests_total", "Total HTTP requests",
    labelnames=["route", "method", "status"]
)
LAT = Histogram(
    "gateway_request_duration_seconds", "Request latency (seconds)",
    labelnames=["route", "method"]
)
ERRS = Counter(
    "gateway_errors_total", "Errors by type",
    labelnames=["route", "type"]
)

• /metrics：暴露 Prometheus 指標（text/plain exposition）。
• HTTP Middleware：自動量測所有路由的 請求數、延遲、錯誤，並排除 /metrics 自身。

備註：Middleware 會自動涵蓋 /ask、/healthz 等所有路由；/metrics 已排除不計，避免自我干擾。

🔹 實務 LLM Gateway 指南：這些方案可以直接用

快速選擇指南

詳細功能對比

決策樹狀圖（依 SLO／成本／維運／合規）

🔹 小結

今天我們完成了一個 最小可行的 API Gateway：
它提供統一的 /ask API，隱藏 API Key，讓任何 Client 都能安全地呼叫 LLM。

雖然目前功能還很精簡，但它已經是整個系統的 核心樞紐。
接下來，所有的請求都會經過這個 Gateway，我們就能逐步加入：

• 觀測：延遲、錯誤率、Token 與成本監控
• 快取：減少重複查詢，降低成本
• 模型管理：根據情境動態切換雲端或本地模型

隨著這些能力逐步加上去，Gateway 會從「單純的代理」演化成真正的 企業級 LLM 中控點。

明天（Day 19），我們就會把焦點放在 Observability（可觀測性），看看如何監控 Latency、Token 與成本，避免月底帳單嚇死人。

📚 引用 / 延伸閱讀

• OWASP Top 10 API Security Risks – 2023
• JWT.io — Introduction to JSON Web Tokens (2024)
• IETF RFC 6749 — OAuth 2.0 Authorization Framework (2012)
• Cloudflare — Counting Things: A Lot of Different Things (2019)
• How to build a Rate Limiter using Redis
• NGINX — Rate Limiting with NGINX (2023)
• Google SRE — Monitoring Distributed Systems (2016)
• Prometheus — Best Practices: Naming (2024)
• OpenTelemetry — Observability Primer (2024)
• FastAPI — Middleware (2024)
• Prometheus Python Client — GitHub (2024)
• OpenRouter
• LiteLLM
• Helicone