iT邦幫忙

2026 iThome 鐵人賽

DAY 5
0
AI Engineering

30天從零打造 AI 中台自學之路系列 第 5

Day 05 - 中台核心 Gateway 基礎骨架:基於 FastAPI 實作路由、API Key 驗證與請求標準化

  • 分享至 

  • xImage
  •  

在完成地端推論環境部署(Day 03)與多雲端 Provider 抽象層(Day 04)後,今天我們要動手打造中台的統一入口:API Gateway 核心骨架

Gateway 是整個 AI 中台的神經樞紐,所有來自內部開發者、IDE 外掛或自動化腳本的請求都必須先經過此處。今天我們將使用高效能的 FastAPI,建立一個相容於 OpenAI 規格的 /v1/chat/completions 端點,並實作 API Key 權限驗證與全域例外處理機制。


一、Gateway 核心設計目標

  1. 介面標準化:遵循業界通用的 OpenAI Chat Completion API 結構,降低客戶端串接成本。
  2. 安全鑑權(Authentication):透過 HTTP Header 攜帶的 API Key 識別請求來源與部門別。
  3. 解耦業務邏輯:Gateway 僅負責請求驗證與轉發,具體模型調用交由 Day 04 實作的 FallbackLLMManager 執行。
  4. 標準錯誤響應:攔截未捕捉例外,輸出語義清晰且一致的 JSON 錯誤結構。

二、專案目錄結構

ai-gateway/
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI 核心入口與端點
│   ├── config.py            # 設定檔與 Key 管理
│   ├── auth.py              # API Key 驗證依賴項
│   ├── models.py            # Pydantic 資料模型 (Day 04 定義)
│   ├── providers.py         # 多模型 Client (Day 04 定義)
│   └── manager.py           # Fallback 管理器 (Day 04 定義)
├── requirements.txt
└── test_gateway.py          # 端點測試腳本

三、模組實作

1. 套件依賴 (requirements.txt)

fastapi>=0.110.0
uvicorn>=0.29.0
pydantic>=2.6.0
python-dotenv>=1.0.0

2. 金鑰設定管理 (app/config.py)

在生產環境中,API Key 通常儲存於資料庫或快取。此處先以環境變數或記憶體字典模擬企業各部門的 Key 配置:

import os
from typing import Dict

# 模擬企業內部合法 Key 及其所屬部門
VALID_API_KEYS: Dict[str, str] = {
    os.getenv("GATEWAY_KEY_RD", "sk-corp-rd-team-1001"): "Research & Development",
    os.getenv("GATEWAY_KEY_QA", "sk-corp-qa-team-2002"): "Quality Assurance"
}

3. API Key 驗證依賴項 (app/auth.py)

使用 FastAPI 的 Security 與 HTTPBearer 抽取請求中的 Bearer Token:
from fastapi import Security, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from app.config import VALID_API_KEYS

security = HTTPBearer()

def verify_api_key(credentials: HTTPAuthorizationCredentials = Security(security)) -> str:
    token = credentials.credentials
    if token not in VALID_API_KEYS:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="無效的 API Key 或未經授權的訪問請求",
            headers={"WWW-Authenticate": "Bearer"},
        )
    # 回傳該 Key 所屬部門名稱供 Context 使用
    return VALID_API_KEYS[token]

4. FastAPI Gateway 入口 (app/main.py)

建立 /v1/chat/completions 端點,注入鑑權依賴並串接 Provider 管理器:
from fastapi import FastAPI, Depends, HTTPException, Request
from fastapi.responses import JSONResponse
import logging

from app.models import ChatCompletionRequest, ChatCompletionResponse
from app.auth import verify_api_key
from app.providers import OpenAIProvider, ClaudeProvider, GeminiProvider
from app.manager import FallbackLLMManager

# 初始化日誌
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("AIGateway")

app = FastAPI(
    title="Enterprise AI Gateway",
    version="1.0.0",
    description="企業 AI 中台統一核心網關"
)

# 初始化多模型降級管理器
llm_manager = FallbackLLMManager([
    OpenAIProvider(),
    ClaudeProvider(),
    GeminiProvider()
])

# 全域例外處理器
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    logger.error(f"Gateway 處理未預期錯誤: {str(exc)}")
    return JSONResponse(
        status_code=500,
        content={
            "error": {
                "message": "中台服務內部錯誤,請聯繫系統管理員",
                "type": "internal_server_error",
                "detail": str(exc)
            }
        }
    )

# 健康檢查端點
@app.get("/health")
async def health_check():
    return {"status": "ok", "service": "Enterprise AI Gateway"}

# 核心端點:Chat Completions
@app.post(
    "/v1/chat/completions",
    response_model=ChatCompletionResponse,
    summary="統一對話生成接口"
)
async def create_chat_completion(
    request: ChatCompletionRequest,
    department: str = Depends(verify_api_key)
):
    logger.info(f"收到來自 [{department}] 的生成請求,模型指向: {request.model}")
    
    try:
        # 交由 Fallback 管理器執行調用
        response = llm_manager.execute_with_fallback(request)
        return response
    except Exception as e:
        logger.error(f"請求處理失敗: {str(e)}")
        raise HTTPException(status_code=502, detail=f"上游模型服務調用失敗: {str(e)}")

四、啟動與測試驗證

1. 啟動 Gateway 服務

uvicorn app.main:app --host 0.0.0.0 --port 8080 --reload

2. 使用 cURL 測試端點

測試一:未帶金鑰(預期回傳 403 / 401 錯誤)

curl -X POST http://localhost:8080/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{
    "messages": [{"role": "user", "content": "Hello"}]
  }'

回應:

{"detail": "Not authenticated"}

測試二:攜帶合法金鑰請求

測試二:攜帶合法金鑰請求

curl -X POST http://localhost:8080/v1/chat/completions \
  -H "Authorization: Bearer sk-corp-rd-team-1001" \
  -H "Content-Type: application/json" \
  -d '{
    "messages": [
      {"role": "user", "content": "請列出三個 Clean Code 的核心原則"}
    ],
    "temperature": 0.2
  }'

回應:

{
  "provider": "openai",
  "model": "gpt-4o-mini",
  "content": "Clean Code 的三個核心原則包括:1. 單一職責原則... 2. 有意義的命名... 3. 避免重複 (DRY)...",
  "usage": {
    "prompt_tokens": 18,
    "completion_tokens": 120,
    "total_tokens": 138
  }
}

五、架構演進重點

  1. 無狀態設計:Gateway 保持 Stateless,利於後續水平擴展(Horizontal Scaling)至 K8s 集群。
  2. 多租戶識別基礎:在 auth.py 中解析出的 department 資訊,將是後續章節注入限流統計與 Token 計費的重要標籤。
  3. 統一路徑注入:未來要加入的 RAG 檢索管線與敏感詞檢測,都能以 Middleware 或 FastAPI Dependency 的形式無縫插入此架構中。

明日進度

中台 Gateway 基礎框架已經建立。明天 Day 06 我們將進入向量存儲的核心環節:向量資料庫選型與建置,實際搭建本地 Qdrant 向量資料庫,並規劃企業規範知識庫的 Collections 與 Payload 結構。


上一篇
Day 04 - 多雲端模型介面抽象化:封裝 OpenAI、Claude 與 Gemini 統一調用與 Fallback 機制
下一篇
Day 06 - 向量資料庫選型與建置:Qdrant 環境搭建、集合規劃與 Payload 結構設計
系列文
30天從零打造 AI 中台自學之路8
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言