從零開始：用 FastAPI 打造一個專業級 AI 服務骨架
摘要
厭倦了從頭寫 API、驗證、限流和模型載入嗎？本文將解析一個完整的 FastAPI AI 服務骨架 (fastapi-ai-starter) 的核心程式碼，讓您在最短的時間內掌握如何建構一個具備：API Key 驗證、速率限制、Mock/Real 模型切換、異步批次推理 等功能的專業級 AI 服務。
您可以透過以下 GitHub 連結檢閱本專案的原始碼：https://github.com/BpsEason/fastapi-ai-starter.git
一、 服務的入口與門戶：app/main.py
這是整個應用程式的啟動點。
from fastapi import FastAPI
from app.api import predict
# ... 略 ...
def create_app() -> FastAPI:
"""【核心 1：服務啟動入口】初始化 FastAPI 應用程式，並註冊所有路由模組。"""
app = FastAPI(title="FastAPI AI Starter")
# 註冊 /v1 路由，包含核心的 /v1/predict
app.include_router(predict.router, prefix="/v1")
# ... 略 ...
return app
app = create_app()
@app.get("/ping")
async def ping():
"""標準健康檢查：確認服務是否運行 (Status: OK)"""
return {"status": "ok"}
學習重點： 您只需要知道
uvicorn會運行這個檔案來啟動服務。路由模線（如predict.router）都是在這裡註冊進來的。初學建議： 無需修改。
二、 請求與回應的規格：app/schemas.py
這個檔案使用 Pydantic 來定義 API 輸入 (Request) 和輸出 (Response) 的 JSON 資料結構，確保了資料的型別和格式正確性。
from pydantic import BaseModel, Field
from typing import Optional
class PredictRequest(BaseModel):
"""【核心 3：API 請求的輸入格式】"""
prompt: str = Field(..., min_length=1, max_length=2000, description="使用者輸入的文字或問題。")
max_tokens: Optional[int] = Field(128, ge=1, le=2048, description="AI 回應的最大長度。")
class PredictResponse(BaseModel):
"""【核心 3：API 回應的輸出格式】"""
text: str = Field(..., description="AI 模型生成的文字回應。")
學習重點： Pydantic 提供了自動的資料驗證。
如何修改： 如果您的 AI 模型需要使用者傳入新的參數（例如
temperature: float或language: str），請直接在PredictRequest中新增欄位。
三、 API 請求的處理器：app/api/predict.py
這是處理 /v1/predict 請求的函式。它的主要職責是驗證、轉發請求、接收結果並返回。
from fastapi import APIRouter, Depends, HTTPException
# ... 略 ...
router = APIRouter()
@router.post("/predict", response_model=List[PredictResponse], dependencies=[Depends(require_api_key)])
async def predict_endpoint(payload: List[PredictRequest], ai_client = Depends(get_ai_client)):
"""【核心 2：AI 預測 API 處理器】這是接收所有 AI 推理請求的主要路由。"""
results = []
try:
# 1. 提取所有 prompt
texts = [p.prompt for p in payload]
# 2. 呼叫 AI 核心模組進行批次預測 (關鍵：將所有 AI 邏輯抽象化)
responses = await ai_client.predict_batch(texts)
# 3. 格式化結果
for r in responses:
results.append(PredictResponse(text=r))
except Exception:
raise HTTPException(status_code=500, detail="Inference failed")
return results
學習重點：
dependencies=[Depends(require_api_key)]：這是啟用 API Key 驗證的方式。ai_client = Depends(get_ai_client)：FastAPI 會自動注入正確的 AI 服務實例（可能是 Mock 或 Real）。ai_client.predict_batch(texts)：所有的 AI 複雜性都被抽象化到這個調用中。
四、 身份驗證與限流：app/middleware.py
這個模組提供了企業級的 API 安全保障，處理金鑰驗證和速率限制，支持內存模式（測試）和 Redis 模式（生產環境）。
# ... 略 ...
RATE_LIMIT = int(os.getenv("API_RATE_LIMIT_PER_MIN", "120")) # per key per minute
def require_api_key(request: Request):
"""
【核心驗證函式】API Key 驗證與速率限制 (Rate Limiting)
檢查 x-api-key 是否存在且有效，並依據 Redis 或內存限制每分鐘請求次數。
"""
key = request.headers.get("x-api-key")
if not key:
raise HTTPException(status_code=401, detail="Missing x-api-key header")
if USE_REDIS:
# --- Redis 驗證與限流邏輯 (分佈式) ---
# ... 略 ...
pass
else:
# --- 內存驗證與限流邏輯 (單一程序) ---
if key not in _KEY_STORE:
raise HTTPException(status_code=401, detail="Invalid API key")
minute = int(time.time()//60)
k = (key, minute)
_RATE_STORE.setdefault(k, 0)
_RATE_STORE[k] += 1
if _RATE_STORE[k] > RATE_LIMIT:
raise HTTPException(status_code=429, detail="Rate limit exceeded")
return key
學習重點： 所有的安全檢查都在請求到達 AI 核心邏輯之前完成。
初學建議： 無需修改。 直接使用它提供的強大安全功能。
五、 AI 邏輯的核心：app/ai_client.py
這是您專注於 AI 實作的地方。它巧妙地將模擬模型和真實模型進行了分離。
# ... 略 ...
class MockModel:
"""【AI 核心-模擬模型】這是您入門 AI 服務的第一個修改點！"""
def generate(self, prompt: str, max_tokens: int = 128) -> str:
"""
**初學任務：** 將回傳的字串修改為你喜歡的內容！
例如：return f"AI 說：收到您的訊息：{s}，請稍候！"
"""
s = prompt.strip()
# ... 略 ...
return f"MockReply: {s}" # <--- 修改此行來改變 AI 的回應
class RealAIClient:
"""【AI 核心-真實模型】這是整合 PyTorch/TensorFlow/Hugging Face 等模型的地方。"""
# ... 載入模型邏輯 ...
async def predict_batch(self, prompts: List[str], max_tokens: int = 128) -> List[str]:
# 優化：利用執行緒池並行執行多個推理任務
loop = asyncio.get_event_loop()
# ... 略 ...
return await asyncio.gather(*tasks)
def get_ai_client():
"""依據環境變數 USE_REAL_MODEL 動態決定返回 RealAIClient 或 MockAIClient。"""
# ... 略 ...
return MockAIClient() # 預設返回 MockAIClient
學習重點：
Mock 模式 (起步)：修改
MockModel.generate即可控制 AI 回覆。Real 模式 (進階)：在
RealAIClient中載入您的模型，並修改predict_batch進行實際的推理計算。異步處理：藉由
asyncio和loop.run_in_executor確保 AI 推理的阻塞操作不會卡住整個 API 服務（這是 FastAPI 服務高併發的關鍵技巧）。
六、 應用程式設定：app/core/config.py
集中的環境變數管理，讓部署和測試變得容易。
import os
from pydantic import BaseSettings
class Settings(BaseSettings):
"""【應用程式核心設定】所有環境變數的集中管理點。"""
# ... API 安全與驗證設定 ...
APP_API_KEY: str = os.getenv("APP_API_KEY", "changeme")
# ... 模型控制設定 ...
USE_REAL_MODEL: bool = os.getenv("USE_REAL_MODEL", "0") == "1"
"""'1' (True) 啟用真實模型，'0' (False) 啟用 Mock 模型。"""
MODEL_PATH: str = os.getenv("MODEL_PATH", "models/real")
settings = Settings()
七、 模型載入的預留點：app/models/model_loader.py
雖然目前只有 MockModel，但這個檔案標示了未來擴展真實模型載入的乾淨位置。
class MockModel:
"""【AI 核心邏輯 - 模擬模型】這是未來 RealModel 載入邏輯的結構參考。"""
def generate(self, prompt: str, max_tokens: int = 128) -> str:
# ... 略 ...
return f"MockReply: {summary}"
八、 測試與質量保證：tests/test_predict.py
自動化測試是專業開發的標準。這個檔案確保您的 API 始終按預期運行。
# ... 略 ...
def test_predict_missing_key():
"""【測試案例 2】測試 API Key 遺失時的錯誤處理 (預期 401 錯誤)。"""
r = client.post("/v1/predict", json=[{"prompt": "hello"}])
assert r.status_code == 401
assert r.json()["detail"] == "Missing x-api-key header"
def test_predict_success():
"""【測試案例 3】測試 AI 預測成功並批次處理的能力 (預期 200 成功)。"""
headers = {"x-api-key": "test-key"}
# ... 略 ...
r = client.post("/v1/predict", json=payload, headers=headers)
assert r.status_code == 200
assert len(r.json()) == 2
assert r.json()[0]["text"].startswith("MockReply")
學習重點： 運行
pytest來驗證 API 路由、金鑰驗證和回應格式。
結語
這個 fastapi-ai-starter 專案骨架為您提供了一個功能強大且結構清晰的起點。您無需擔心基礎架構，只需專注於一個檔案：app/ai_client.py，開始將您的 AI 模型整合進去！
現在，您已經擁有了所有程式碼和詳細的教學文件，立即開始您的 AI 服務開發之旅吧！
沒有留言:
張貼留言