開始透過 FastAPI 端點提供你的 ML 模型預測。你將學會載入預先訓練的 ML 模型，並建立 API 端點，將預測結果以序列化回應透過 HTTP 請求送出。你會運用 Pydantic 資料模型來驗證請求與回應。

AI 的 GET 與 POST 請求

取得模型資訊的 GET 端點

用於模型註冊的 POST 端點

使用預先訓練模型進行 FastAPI 預測

載入預先訓練的模型

建立預測端點

執行 FastAPI 應用程式

請求與回應模型

為 ML 輸入建立 Pydantic 模型

驗證 ML 預測的請求與回應

使用 FastAPI 進行模型佈署入門

學習如何透過 FastAPI 端點提供機器學習模型服務。本章涵蓋建立回傳預測結果的端點、處理不同型態的輸入資料，以及實作穩健的輸入驗證。你將打造可上線的 API，能驗證多種輸入資料型態，並在伺服器啟動時載入 ML 模型且達到零停機。

在 FastAPI 中處理不同的輸入型別

處理文字型的請求資料

處理數值型請求資料

FastAPI 的輸入驗證

欄位驗證

加入自訂驗證器

測試自訂驗證器

在伺服器啟動時載入 AI 模型

模型載入的健康檢查 API

回傳結構化的預測回應

從 API 回傳結構化輸出

測試能回傳結構化結果的端點

整合 AI 模型

本章說明如何以金鑰式驗證保護 API、以自訂速率限制管理請求頻率，並透過非同步處理提升效能。你將學會保護端點、預防濫用，並有效處理耗時工作，為 API 上線做好準備。

API Key 驗證

使用金鑰驗證保護 API

import joblib
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LogisticRegression
from fastapi import Depends, HTTPException
from fastapi.security import APIKeyHeader

# Model creation
def train_and_save_model():
    data = {
        "review": [
            "I love this product, it's fantastic!",
            "Really satisfied with the quality!",
            "Terrible, I hate it.",
            "Not happy with the purchase.",
            "Absolutely amazing and wonderful!",
            "Worst experience ever.",
            "I am very pleased with my purchase.",
            "Disappointed, it didn't work as expected.",
            "The best thing I've ever bought.",
            "Totally awful, will not buy again."
        ],
        "label": [1, 1, 0, 0, 1, 0, 1, 0, 1, 0]  # 1 = Positive, 0 = Negative
    }

    df = pd.DataFrame(data)

    positive_words = ["love", "satisfied", "amazing", "fantastic", "wonderful", "pleased", "best"]
    negative_words = ["hate", "terrible", "worst", "disappointed", "awful"]

    df["num_words"] = df["review"].apply(lambda x: len(x.split()))
    df["num_positive_words"] = df["review"].apply(lambda x: sum(word in x.lower() for word in positive_words))
    df["num_complaints"] = df["review"].apply(lambda x: sum(word in x.lower() for word in negative_words))

    X = df[["num_words", "num_positive_words", "num_complaints"]]
    y = df["label"]
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=42)

    model = LogisticRegression(solver='lbfgs')
    model.fit(X_train, y_train)

    # Save model using joblib instead of pickle
    joblib.dump(model, 'sentiment_model.joblib', compress=3)

# Define a callable class
class SentimentAnalyzer:
    def __init__(self, model_path):
        # Load the model using joblib
        train_and_save_model()
        self.model = joblib.load(model_path)
        self.positive_words = ["love", "satisfied", "amazing", "fantastic", "wonderful", "pleased", "best"]
        self.negative_words = ["hate", "terrible", "worst", "disappointed", "awful"]

    def __call__(self, text):
        num_words = len(text.split())
        num_positive_words = sum(word in text.lower() for word in self.positive_words)
        num_complaints = sum(word in text.lower() for word in self.negative_words)
        features = [[num_words, num_positive_words, num_complaints]]
        
        # Get prediction and confidence score
        prediction = self.model.predict(features)
        confidence_scores = self.model.predict_proba(features)
        
        # Return dictionary directly instead of JSON string
        result = {
            "label": "Positive" if prediction[0] == 1 else "Negative",
            "confidence": float(confidence_scores[0][prediction[0]])  
        }
        return result 
      
      
api_key_header = APIKeyHeader(name="X-API-Key")
API_KEY = "your_secret_key"

# Pass the variable containing the APIKeyHeader
def verify_api_key(api_key: str = Depends(api_key_header)):  
    # Verify the API key
    if api_key != API_KEY:  
      	# Raise the HTTP exception here
        raise HTTPException(status_code=403, detail="Invalid API Key")
    return api_key

from model import SentimentAnalyzer, verify_api_key
from fastapi import FastAPI, Depends
from pydantic import BaseModel

app = FastAPI()

class SentimentRequest(BaseModel):
    text: str

@app.post("/predict")
def get_prediction(
    request: SentimentRequest,
    # Authenticate the incoming API key using verify_api_key function
    api_key: str = ____(____)
):
    sentiment_model = SentimentAnalyzer("sentiment_model.joblib")
    result = sentiment_model(request.text)
    return {
        "text": request.text,
        "sentiment": result,
        "status": "success"
    }
    
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8080)

from model import SentimentAnalyzer, verify_api_key
from fastapi import FastAPI, Depends
from pydantic import BaseModel

app = FastAPI()

class SentimentRequest(BaseModel):
    text: str
    
@app.post("/predict")
def get_prediction(
    request: SentimentRequest,
    # Authenticate the incoming API key using verify_api_key function
    api_key: str = Depends(verify_api_key)
):
    sentiment_model = SentimentAnalyzer("sentiment_model.joblib")
    result = sentiment_model(request.text)
    return {
        "text": request.text,
        "sentiment": result,
        "status": "success"
    }
    
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8080)

保護 API 端點

頻率限制

實作速率限制器

import joblib
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LogisticRegression
from datetime import datetime, timedelta
from collections import defaultdict
from fastapi.security import APIKeyHeader

api_key_header = APIKeyHeader(name="X-API-Key")
API_KEY = "your_secret_key"

# Model creation
def train_and_save_model():
    data = {
        "review": [
            "I love this product, it's fantastic!",
            "Really satisfied with the quality!",
            "Terrible, I hate it.",
            "Not happy with the purchase.",
            "Absolutely amazing and wonderful!",
            "Worst experience ever.",
            "I am very pleased with my purchase.",
            "Disappointed, it didn't work as expected.",
            "The best thing I've ever bought.",
            "Totally awful, will not buy again."
        ],
        "label": [1, 1, 0, 0, 1, 0, 1, 0, 1, 0]  # 1 = Positive, 0 = Negative
    }

    df = pd.DataFrame(data)

    positive_words = ["love", "satisfied", "amazing", "fantastic", "wonderful", "pleased", "best"]
    negative_words = ["hate", "terrible", "worst", "disappointed", "awful"]

    df["num_words"] = df["review"].apply(lambda x: len(x.split()))
    df["num_positive_words"] = df["review"].apply(lambda x: sum(word in x.lower() for word in positive_words))
    df["num_complaints"] = df["review"].apply(lambda x: sum(word in x.lower() for word in negative_words))

    X = df[["num_words", "num_positive_words", "num_complaints"]]
    y = df["label"]
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=42)

    model = LogisticRegression(solver='lbfgs')
    model.fit(X_train, y_train)

    # Save model using joblib instead of pickle
    joblib.dump(model, 'sentiment_model.joblib', compress=3)

# Define a callable class
class SentimentAnalyzer:
    def __init__(self, model_path):
        # Load the model using joblib
        train_and_save_model()
        self.model = joblib.load(model_path)
        self.positive_words = ["love", "satisfied", "amazing", "fantastic", "wonderful", "pleased", "best"]
        self.negative_words = ["hate", "terrible", "worst", "disappointed", "awful"]

    def __call__(self, text):
        num_words = len(text.split())
        num_positive_words = sum(word in text.lower() for word in self.positive_words)
        num_complaints = sum(word in text.lower() for word in self.negative_words)
        features = [[num_words, num_positive_words, num_complaints]]
        
        # Get prediction and confidence score
        prediction = self.model.predict(features)
        confidence_scores = self.model.predict_proba(features)
        
        # Return dictionary directly instead of JSON string
        result = {
            "label": "Positive" if prediction[0] == 1 else "Negative",
            "confidence": float(confidence_scores[0][prediction[0]])  
        }
        return result 

class RateLimiter:
    def __init__(self, requests_per_minute: int = 10):
        self.requests_per_minute = requests_per_minute
        self.requests = defaultdict(list)  # Store request timestamps per API key

    def is_rate_limited(self, api_key: str) -> tuple[bool, int]:
        """
        Check if the request should be rate limited
        Returns (is_limited, requests_remaining)
        """
        now = datetime.now()
        minute_ago = now - timedelta(minutes=1)
        
        # Remove requests older than 1 minute
        self.requests[api_key] = [
            req_time for req_time in self.requests[api_key]
            if req_time > minute_ago
        ]
        
        # Check if rate limit is exceeded
        recent_requests = len(self.requests[api_key])
        if recent_requests >= self.requests_per_minute:
            return True, 0
            
        # Add new request timestamp
        self.requests[api_key].append(now)
        return False, self.requests_per_minute - recent_requests - 1


from model import SentimentAnalyzer, RateLimiter, API_KEY, api_key_header
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel

app = FastAPI()
rate_limiter = RateLimiter(requests_per_minute=2)

def test_api_key(api_key: str = Depends(api_key_header)):
    if api_key != API_KEY:
        raise HTTPException(
            status_code=403, detail="Invalid API key")
    
    # Check rate limit corresponding to the input api key
    is_limited, _ = rate_limiter.is_rate_limited(____)
    # Check if returned boolean by is_rate_limited is True
    if ____:
        # Raise the http exception with status code
        raise HTTPException(status_code=____,
            detail="Rate limit exceeded. Please try again later."
        )
    return api_key

class SentimentRequest(BaseModel):
    text: str

@app.post("/predict")
def get_prediction(
    request: SentimentRequest,
    api_key: str = Depends(test_api_key)
):
    sentiment_model = SentimentAnalyzer("sentiment_model.joblib")
    result = sentiment_model(request.text)
    return {
        "text": request.text,
        "sentiment": result,
        "status": "success"
    }
    
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8080)

from model import SentimentAnalyzer, RateLimiter, API_KEY, api_key_header
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel

app = FastAPI()
rate_limiter = RateLimiter(requests_per_minute=2)

async def test_api_key(api_key: str = Depends(api_key_header)):
    if api_key != API_KEY:
        raise HTTPException(
            status_code=403, detail="Invalid API key")
        
    # Check rate limit corresponding to the input api key
    is_limited, _ = rate_limiter.is_rate_limited(api_key)
    # Check if returned boolean by is_rate_limited is True
    if is_limited:
        # Raise the http exception with status code
        raise HTTPException(status_code=429,
            detail="Rate limit exceeded. Please try again later."
        )
    return api_key

class SentimentRequest(BaseModel):
    text: str
    
@app.post("/predict")
async def get_prediction(
    request: SentimentRequest,
    api_key: str = Depends(test_api_key)
):
    sentiment_model = SentimentAnalyzer("sentiment_model.joblib")
    result = sentiment_model(request.text)

    return {
        "text": request.text,
        "sentiment": result,
        "status": "success"
    }
    
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8080)

為端點加入 rate limiting

非同步處理

建立非同步的情感分析端點

實作背景任務

import asyncio
import joblib
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LogisticRegression

# Model creation
def train_and_save_model():
    data = {
        "review": [
            "I love this product, it's fantastic!",
            "Really satisfied with the quality!",
            "Terrible, I hate it.",
            "Not happy with the purchase.",
            "Absolutely amazing and wonderful!",
            "Worst experience ever.",
            "I am very pleased with my purchase.",
            "Disappointed, it didn't work as expected.",
            "The best thing I've ever bought.",
            "Totally awful, will not buy again."
        ],
        "label": [1, 1, 0, 0, 1, 0, 1, 0, 1, 0]  # 1 = Positive, 0 = Negative
    }

    df = pd.DataFrame(data)

    positive_words = ["love", "satisfied", "amazing", "fantastic", "wonderful", "pleased", "best"]
    negative_words = ["hate", "terrible", "worst", "disappointed", "awful"]

    df["num_words"] = df["review"].apply(lambda x: len(x.split()))
    df["num_positive_words"] = df["review"].apply(lambda x: sum(word in x.lower() for word in positive_words))
    df["num_complaints"] = df["review"].apply(lambda x: sum(word in x.lower() for word in negative_words))

    X = df[["num_words", "num_positive_words", "num_complaints"]]
    y = df["label"]
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=42)

    model = LogisticRegression(solver='lbfgs')
    model.fit(X_train, y_train)

    # Save model using joblib instead of pickle
    joblib.dump(model, 'sentiment_model.joblib', compress=3)

# Define a callable class
class SentimentAnalyzer:
    def __init__(self):
        # Load the model using joblib
        train_and_save_model()
        self.model = joblib.load("sentiment_model.joblib")
        self.positive_words = ["love", "satisfied", "amazing", "fantastic", "wonderful", "pleased", "best"]
        self.negative_words = ["hate", "terrible", "worst", "disappointed", "awful"]

    async def __call__(self, text):
        # Simulate a long-running operation
        await asyncio.sleep(11)  # This will trigger the timeout since it's > 10 seconds
        
        num_words = len(text.split())
        num_positive_words = sum(word in text.lower() for word in self.positive_words)
        num_complaints = sum(word in text.lower() for word in self.negative_words)
        features = [[num_words, num_positive_words, num_complaints]]
        
        # Get prediction and confidence score
        prediction = self.model.predict(features)
        confidence_scores = self.model.predict_proba(features)
        
        # Return dictionary directly instead of JSON string
        result = {
            "label": "Positive" if prediction[0] == 1 else "Negative",
            "confidence": float(confidence_scores[0][prediction[0]])
        }
        return result

import asyncio
from fastapi import FastAPI, HTTPException
from model import SentimentAnalyzer
from pydantic import BaseModel

app = FastAPI()

class Review(BaseModel):
    text: str

@app.post("/analyze_reviews")
async def analyze_reviews(review: Review):
    try:
        sentiment_model = SentimentAnalyzer()
        # Set model input and timeout limit
        result = await asyncio.wait_for(
            sentiment_model(____),
            timeout=____
        )
        return {"sentiment": result["label"]}      
    except asyncio.TimeoutError:
        # Raise HTTP status code for timeout error
        raise HTTPException(status_code=____, detail="Analysis timed out")
    except Exception:
        # Raise HTTP status code for internal error
        raise HTTPException(status_code=____, detail="Analysis failed")
        
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8080)

import asyncio
from fastapi import FastAPI, HTTPException
from model import SentimentAnalyzer
from pydantic import BaseModel

app = FastAPI()

class Review(BaseModel):
    text: str

@app.post("/analyze_reviews")
async def analyze_reviews(review: Review):
    try:
        sentiment_model = SentimentAnalyzer()
        # Set model input and timeout limit
        result = await asyncio.wait_for(
            sentiment_model(review.text),
            timeout=10
        )
        return {"sentiment": result["label"]}      
    except asyncio.TimeoutError:
        # Raise HTTP status code for timeout error
        raise HTTPException(status_code=408, detail="Analysis timed out")
    except Exception:
        # Raise HTTP status code for internal error
        raise HTTPException(status_code=500, detail="Analysis failed")
        
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8080)

處理逾時錯誤

保護並最佳化 API

本章涵蓋進階主題，協助你在生產環境中長期維運 FastAPI 應用。內容包含 API 端點的版本控管與文件撰寫、支援更複雜輸入與輸出的進階輸入驗證，以及監控與記錄，以確保應用正常運作，並在異常時即時除錯。

API 版本管理與文件

為新端點版本分類程式碼變更

新增端點版本

加入 API 說明標頭

進階輸入驗證與錯誤處理

巢狀模型

自訂模型驗證器

全域例外處理器

監控與記錄

記錄載入模型所花費的時間

加入簡單的健康檢查端點

與監控分享模型參數

總結

API 版本控管、監控與記錄

掌握使用 FastAPI 將機器學習模型佈署到生產環境的技巧。你將學會打造穩健、可擴充的 AI 驅動 API，整合預先訓練的模型，並處理複雜的資料輸入。你會學到如何保護 API、實作驗證機制，並以非同步處理最佳化效能。進階主題包含 API 版本控管、日誌與監控策略。課程結束時，你將能開發符合真實情境需求的高效率 AI 模型服務。本課程非常適合資料科學家與開發者，協助你銜接 AI 開發與可上線應用之間的落差。

<h2>使用 FastAPI 部署 AI 模型</h2> 
FastAPI 是一款適合用於構建 API 的 Python 網頁框架，特別適合部署機器學習和 AI 模型。 FastAPI 的高速與模組化特性，使其成為資料科學家與機器學習工程師在將 AI 解決方案整合至網路應用程式時的強大選擇。<br><br> 
<h2>為生產環境中的模型建立 API</h2> 
在本課程中，您將學習如何使用 FastAPI 建立並部署用於模型服務的 API。 在課程結束時，您將能夠建立強大的 API 端點，處理資料輸入驗證、身份驗證以及錯誤管理。 實作練習將引導您建立能夠處理 AI 模型互動的完整 API。<br><br> 
<h2>保護與擴展 FastAPI 應用程式</h2> 您還將實作 API 金鑰驗證、應用自訂的速率限制以控制請求流量，並透過非同步處理提升效能。 此外，您還將學習如何管理 API 版本、提升複雜資料型別的輸入驗證，以及實作監控與日誌記錄，確保您的應用程式在生產環境中穩定運行。

Introduction to FastAPI

Large Language Models (LLMs) Concepts

學習使用FastAPI開發支持AI模型的API，以滿足現實需求。

使用 FastAPI 將 AI 佈署到生產環境

掌握使用 FastAPI 開發支援 AI 模型的 API，滿足實際應用需求。

處理逾時錯誤

使用 FastAPI 將 AI 佈署到生產環境

動手互動練習