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

AI 的 GET 與 POST 請求

取得模型資訊的 GET 端點

用於模型註冊的 POST 端點

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

載入預先訓練的模型

建立預測端點

執行 FastAPI 應用程式

請求與回應模型

為 ML 輸入建立 Pydantic 模型

驗證 ML 預測的請求與回應

使用 FastAPI 進行模型佈署入門

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

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

處理文字型的請求資料

import numpy as np
from scorer import CommentMetrics, CommentScorer
from fastapi import FastAPI

app = FastAPI()
model = CommentScorer()

@app.post("/predict_trust")
def predict_trust(comment: CommentMetrics):
    # Convert input and extract comment metrics
    features = ____.____([[
        comment.____,
        comment.____,
        comment.____
    ]])
    # Get prediction from model 
    score = model.predict(____)
    return {
        "trust_score": round(score, 2),
        "comment_metrics": comment.dict()
    }
    
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8080)

import numpy as np
from pydantic import BaseModel

class CommentMetrics(BaseModel):
    length: int
    user_reputation: int
    report_count: int

class CommentScorer:
    def predict(self, features: np.ndarray) -> float:
        """
        Predict trust score based on comment metrics
        features: [[length, user_reputation, report_count]]
        """
        # Unpack features
        length, reputation, reports = features[0]
        
        # Calculate trust score
        score = (0.3 * (length/500) +        # Normalize length
                 0.5 * (reputation/100) +    # Normalize reputation
                 -0.2 * reports)             # Reports reduce score
        
        return float(max(min(score * 100, 100), 0))  # Scale to 0-100

import numpy as np
from scorer import CommentMetrics, CommentScorer
from fastapi import FastAPI

app = FastAPI()
model = CommentScorer()

@app.post("/predict_trust")
def predict_trust(comment: CommentMetrics):
    # Convert input and extract comment metrics
    features = np.array([[
        comment.length,
        comment.user_reputation,
        comment.report_count
    ]])
    # Get prediction from model
    score = model.predict(features)
    return {
        "trust_score": round(score, 2),
        "comment_metrics": comment.dict()
    }
    
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8080)

import numpy as np
from pydantic import BaseModel

class CommentMetrics(BaseModel):
    length: int
    user_reputation: int
    report_count: int

class CommentScorer:
    def predict(self, features: np.ndarray) -> float:
        """
        Predict trust score based on comment metrics
        features: [[length, user_reputation, report_count]]
        """
        # Unpack features
        length, reputation, reports = features[0]
        
        # Calculate trust score
        score = (0.3 * (length/500) +      # Normalize length
                0.5 * (reputation/100) +    # Normalize reputation
                -0.2 * reports)            # Reports reduce score
        
        return float(max(min(score * 100, 100), 0))  # Scale to 0-100

處理數值型請求資料

FastAPI 的輸入驗證

欄位驗證

加入自訂驗證器

from fastapi import FastAPI
from pydantic import BaseModel, Field, validator

app = FastAPI()

class User(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: str
    age: int
    
    @validator('email')
    def email_must_be_example_domain(cls, user_email):
        if not user_email.endswith("@mode360.com"):
            raise ValueError('Email must be from the mode360.com domain')
        return user_email

# Create a post request endpoint
@app.____("/register")
# Validate incoming user data with a pydantic model
def register_user(user: ____):
    return {"status": "success", "user": user.dict()}
  
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8080)

from fastapi import FastAPI
from pydantic import BaseModel, Field, validator

app = FastAPI()

class User(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: str
    age: int

    @validator('email')
    def email_must_be_example_domain(cls, user_email):
        if not user_email.endswith("@mode360.com"):
            raise ValueError('Email must be from the mode360.com domain')
        return user_email

# Create a post request endpoint
@app.post("/register")
# Validate incoming user data with a pydantic model
def register_user(user: User):
    return {"status": "success", "user": user.dict()}
  
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8080)

測試自訂驗證器

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

from model import SentimentAnalyzer
from fastapi import FastAPI
from contextlib import asynccontextmanager

def load_model():
    global sentiment_model
    sentiment_model = SentimentAnalyzer("sentiment_model.joblib")

@asynccontextmanager
async def lifespan(app: FastAPI):
    load_model()
    yield

app = FastAPI(title="Sentiment Analysis API", lifespan=lifespan)

# Define a GET endpoint at route "/health"
@app.____("/health")
def health_check():
  	# Check whether sentiment_model is loaded or not.
    if ____ is not None:
        return {
          	# Mark status as healthy and loaded boolean to True
            "status": "____",
            "model_loaded": ____
        }
    # Mark status as unhealthy and loaded boolean to False
    return {
        "status": "____",
        "model_loaded": ____
    }

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8080)

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, 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 


from model import SentimentAnalyzer
from fastapi import FastAPI
from contextlib import asynccontextmanager

def load_model():
    global sentiment_model
    sentiment_model = SentimentAnalyzer("sentiment_model.joblib")

@asynccontextmanager
async def lifespan(app: FastAPI):
    load_model()
    yield

app = FastAPI(title="Sentiment Analysis API", lifespan=lifespan)
    
# Define a GET endpoint at route "/health"
@app.get("/health")
def health_check():
  	# Check whether sentiment_model is loaded or not.
    if sentiment_model is not None:
        return {
          	# Mark status as healthy and loaded boolean to True
            "status": "healthy",
            "model_loaded": True
        }
    # Mark status as unhealthy and loaded boolean to False
    return {
        "status": "unhealthy",
        "model_loaded": False
    }

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8080)

模型載入的健康檢查 API

回傳結構化的預測回應

從 API 回傳結構化輸出

# File: sentiment_api.py
import joblib
import json
import pandas as pd
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from contextlib import asynccontextmanager
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)

# Train and save the model
train_and_save_model()

sentiment_model = None

# Define a callable class
class SentimentAnalyzer:
    def __init__(self, model_path):
        # Load the model using joblib
        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 

      # Define model loading function
def load_model():
    global sentiment_model
    sentiment_model = SentimentAnalyzer("sentiment_model.joblib")

# Initialize FastAPI app with lifespan
@asynccontextmanager
async def lifespan(app: FastAPI):
    # Initialize our mock model on startup
    global sentiment_model
    load_model()
    yield


# Initialize FastAPI app
app = FastAPI(title="Sentiment Analysis API", lifespan=lifespan)

# Initialize model variable
sentiment_model = None

# Define request/response models
class CommentRequest(BaseModel):
    text: str

class CommentResponse(BaseModel):
    text: str
    sentiment: str
    confidence: float
    

@app.post("/analyze")
def analyze_comment(request: CommentRequest):
    if sentiment_model is None:
        raise HTTPException(
            status_code=503,
            detail="Model not loaded"
        )
    
    if not request.text.strip():
        raise HTTPException(
            status_code=400,
            detail="Empty text provided"
        )
        
    result = sentiment_model(request.text)
    return CommentResponse(
        text=request.text,
        sentiment=result["label"],
        confidence=result["confidence"]
    )

@app.get("/health")
def health_check():
    """Check if model is loaded and ready"""
    return {
        "status": "healthy" if sentiment_model is not None else "unhealthy",
        "model_loaded": sentiment_model is not None
    }

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8080)

import ____

url = "http://localhost:8080/analyze"
data = {"text": "This is great, I can totally relate."}

# Send post request and pass the sample request data
response = requests.____(url, json=____)

# Print prediction response
print(response.____())

import requests

url = "http://localhost:8080/analyze"
data = {"text": "This is great, I can totally relate."}

# Send post request and pass the sample request data
response = requests.post(url, json=data)

# Print prediction response
print(response.json())

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

整合 AI 模型

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

API Key 驗證

使用金鑰驗證保護 API

保護 API 端點

頻率限制

實作速率限制器

為端點加入 rate limiting

非同步處理

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

實作背景任務

處理逾時錯誤

保護並最佳化 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，滿足實際應用需求。

回傳結構化的預測回應

Create Your Free Account