开始通过 FastAPI 端点提供您的 ML 模型预测。您将学习加载预训练 ML 模型，并创建 API 端点，通过 HTTP 请求以序列化响应返回预测结果。您还将利用 Pydantic 数据模型验证请求与响应。

AI 的 GET 与 POST 请求

用于模型信息的 GET 端点

用于模型注册的 POST 端点

使用预训练模型进行 FastAPI 预测

加载预训练模型

创建预测端点

运行 FastAPI 应用

请求与响应模型

为 ML 输入创建一个 Pydantic 模型

为机器学习预测验证请求与响应

使用 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 端点

速率限制

实现限流器

为端点添加限流

异步处理

创建一个异步情感分析端点

实现后台任务

处理超时错误

API 的安全与优化

本章涵盖支持 FastAPI 应用长期在生产环境稳定运行的高级主题。内容包括 API 端点的版本管理与文档化、支持更复杂输入与输出的高级输入校验，以及监控与日志以确保应用正常运行，并在异常时进行在线排障。

API 版本管理与文档

为新端点版本分类代码变更

添加一个新的端点版本

为 API 添加描述性头部

高级输入校验与错误处理

嵌套模型

自定义模型验证器

全局异常处理器

监控与日志记录

记录模型加载耗时

添加一个简单的健康检查端点

与监控共享模型参数

总结

API 版本管理、监控与日志

掌握使用 FastAPI 将机器学习模型部署到生产环境的实用方法。本课程将指导您构建稳健、可扩展的 AI 驱动 API，集成预训练模型，并处理复杂数据输入。您将学习如何保护 API、实现认证机制，并通过异步处理优化性能。高级主题包括 API 版本管理、日志记录与监控策略。完成后，您将具备开发面向真实业务需求的高效 AI 模型与应用的能力。本课程适合数据科学家和开发者，帮助您打通从 AI 开发到生产级应用的关键环节。

<h2>用于 AI 模型部署的 FastAPI</h2> 
FastAPI 是一个 Python Web 框架，非常适合构建 API，尤其适用于部署机器学习和 AI 模型。 FastAPI 的速度和模块化使其成为数据科学家和机器学习工程师将 AI 解决方案集成到 Web 应用程序中的强大选择。<br><br> 
<h2>为生产环境中的模型构建 API</h2> 
在本课程中，您将学习如何使用 FastAPI 构建和部署用于模型服务的 API。 最终，您将创建健壮的 API 端点，用于处理数据输入验证、身份验证和错误管理。 动手练习将引导你构建完整的 API，用于处理 AI 模型交互。<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