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

AI 的 GET 与 POST 请求

用于模型信息的 GET 端点

用于模型注册的 POST 端点

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

加载预训练模型

创建预测端点

运行 FastAPI 应用

请求与响应模型

为 ML 输入创建一个 Pydantic 模型

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

使用 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)

为端点添加限流

异步处理

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

实现后台任务

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>用于 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，以满足真实世界需求。

为端点添加限流

使用 FastAPI 在生产环境中部署 AI

动手互动练习