如何使用FastAPI进行机器学习

FastAPI 提供了一种使用 Python 构建后端服务的快速方法。使用一些装饰器，您可以将 Python 函数转换为 API 应用程序。

它被许多公司广泛使用，包括 Microsoft、Uber 和 Netflix。根据 Python 开发者调查，FastAPI 的使用率已从 2021 年的 21% 增长到 2023 年的 29%。对于数据科学家来说，它是第二受欢迎的框架，有 31% 的人使用它。

在这篇博文中，我们将为可能想要为他们的项目构建快速原型的数据科学家介绍 FastAPI 的基础知识。

什么是 FastAPI？

FastAPI 是一种流行的 Web 框架，用于使用标准 Python 类型提示使用 Python 构建 API。它直观且易于使用，并且可以在短时间内提供可用于生产的应用程序。它与 OpenAPI 和 JSON Schema 完全兼容。

为什么使用 FastAPI 进行机器学习？

大多数从事机器学习项目的团队由数据科学家组成，他们的专业领域和职业背景偏向于统计学。他们可能缺乏将机器学习项目转化为软件或应用程序的开发经验。FastAPI使数据科学家能够轻松地为以下项目创建API：

部署预测模型

数据科学团队可能已经训练了一个模型，用于预测仓库中的销售需求。为了使其有用，他们必须提供一个 API 接口，以便股票管理系统的其他部分可以使用这种新的预测功能。

建议引擎

机器学习的一个非常常见的用途是作为一个根据用户的选择提供建议的系统。例如，如果有人将某些产品放入购物车，则可以向该用户推荐更多商品。此类电子商务系统需要对采用输入参数的建议引擎进行 API 调用。

动态仪表板和报告系统

有时，数据科学项目的报告需要以控制面板的形式呈现，以便用户可以自行检查结果。一种可能的方法是让数据模型提供 API。前端开发人员可以使用此 API 创建允许用户与数据交互的应用程序。

使用 FastAPI 的优势

与其他 Python Web 框架相比，FastAPI 简单但功能齐全。它主要使用装饰器和类型提示，允许您构建 Web 应用程序，而无需构建整个 ORM（对象关系映射）模型的复杂性，并且可以灵活地使用任何数据库，包括任何 SQL 和 NoSQL 数据库。FastAPI 还提供自动文档生成、对查询参数的附加信息和验证的支持，以及良好的异步支持。

快速开发

在 FastAPI 中创建 API 调用就像在 Python 代码中添加装饰器一样简单。对于任何想要将 Python 函数转换为响应 API 调用的应用程序的人来说，几乎不需要后端经验。

快速记录

FastAPI 使用 Swagger UI 提供自动交互式 API 文档，这是行业标准。无需额外努力即可使用 API 调用示例构建清晰的文档。这为忙碌的数据科学团队创造了优势，他们可能没有精力和专业知识来编写技术规范和文档。

易于测试

编写测试是软件开发中最重要的步骤之一，但也可能是最乏味的步骤之一，尤其是当数据科学团队的时间很宝贵时。多亏了 Starlette 和 HTTPX，测试 FastAPI 变得简单。大多数时候不需要 monkey 补丁，测试很容易编写和理解。

快速部署

自带一个 CLI 工具，可以顺利地桥接开发和部署。它允许您在开发模式和生产模式之间轻松切换。开发完成后，可以使用带有预构建 Python 的映像的 Docker 容器轻松部署代码。

如何将 FastAPI 用于机器学习项目

在此示例中，我们将一个分类预测模型转换为后端应用程序，该模型使用 Nearest Neighbors 算法根据企鹅的喙和脚蹼长度来预测各种企鹅的种类。我们将提供一个 API，该 API 从 URL 的查询参数中获取参数并返回预测结果。这表明任何没有后端开发经验的数据科学家都可以快速制作原型。

我们将使用一个简单的 ‘KNeighborsClassifier’ 来描述企鹅数据集。在下面的教程中，我们将重点介绍 FastAPI 的使用，并解释一些基本概念。我们将构建一个原型来执行此操作。

1. 使用 PyCharm
启动 FastAPI 项目

在这篇博文中，我们将使用 PyCharm Professional 2024.1。开始使用 FastAPI 的最佳方式是使用 PyCharm 创建一个 FastAPI 项目。当您在 PyCharm 中单击 New Project （新建项目） 时，您将看到大量可供选择的项目。选择 FastAPI 选项卡：

在此处，您可以输入项目名称并利用其他选项，例如初始化 Git 和要使用的虚拟环境。

完成此操作后，您将看到为您设置的 FastAPI 项目的基本结构

此外，还设置了一个 ” 文件，用于快速测试所有端点。test_main.http

2. 设置环境依赖项

接下来，通过在 PyCarm 的 Tool （工具） 菜单下选择 Sync Python Requirements （同步 Python 要求） 来使用 ” 设置我们的环境依赖项。

然后您可以选择要使用的 ” 文件。

您可以复制并使用此 ‘requirements.txt’ 文件。我们将在项目的机器学习部分使用 pandas 和 scikit-learn。此外，将 ‘penguins.csv’ 文件添加到项目目录中。requirements.txtrequirements.txt

3. 设置机器学习模型

将机器学习代码排列在 ” 文件中。我们将从一个训练模型的脚本开始：main.py

将 pandas 导入为 PD从 sklearn.model_selection 导入 train_test_split从 sklearn 导入预处理from sklearn.neighbors import KNeighborsClassifierfrom sklearn.pipeline 导入管道from sklearn.preprocessing import StandardScaler数据 = PD。read_csv（'penguins.csv')data = 数据。滴()le = 预处理。标签编码器()X = data[[“bill_length_mm”， “flipper_length_mm”]]乐。fit（data[“物种”])y = le。transform（data[“物种”])X_train、X_test、y_train、y_test = train_test_split（X， y， stratify=y， random_state=0)clf = 管道( steps=[（“scaler”， StandardScaler（）））， （“knn”， KNeighborsClassifier（n_neighbors=11))])CLF 的set_params（） 中。fit（X_train， y_train)

我们可以将上面的代码放在 ” 之后。当我们启动应用程序时，所有这些都将运行。

但是，有一种更好的方法来运行我们用于设置模型的启动代码。我们将在博客文章的后面部分介绍这一点。app = FastAPI()

4. 请求回复

接下来，我们将看看如何将我们的模型添加到 FastAPI 功能中。作为第一步，我们将向 URL 的根添加响应，然后简单地以 JSON 格式返回有关模型的消息。将 ” 中的代码从 “Hello world” 更改为我们的消息，如下所示：async def root():

@app。获取("/")  async def root（） 来执行：    返回 {        “name”： “企鹅预测”，        “description”： “这是一个基于鸟的喙长和鳍状肢长度的企鹅预测模型。”、    }

现在，测试我们的应用程序。首先，我们将启动我们的应用程序，这在 PyCharm 中很容易。只需按顶部项目名称旁边的箭头按钮 即可。

如果您使用的是默认设置，则应用程序将在 http://127.0.0.1:8000 上运行。您可以通过查看 Run （运行） 窗口中的提示来仔细检查这一点。

该过程开始后，让我们转到 ” 并按下 ” 旁边的第一个箭头按钮 。

在 Services 窗口的 HTTP Client 中，您将看到我们放入的响应消息。

响应 JSON 文件也会被保存以供将来检查。test_main.httpGET

5. 使用查询参数的请求

接下来，我们希望通过在 URL 中提供查询参数来让用户进行预测。让我们在 ” 函数之后添加下面的代码。root

@app。get（“/预测/”)  async def predict（bill_length_mm： float = 0.0， flipper_length_mm： float = 0.0）：参数 ={                “bill_length_mm”：bill_length_mm、                “flipper_length_mm”：flipper_length_mm            }    如果 bill_length_mm <=0.0 或 flipper_length_mm <=0.0：        返回 {            “parameters”： 参数，            “错误消息”： “输入值无效”，        }    否则： 结果 = CLF。预测（[[bill_length_mm， flipper_length_mm]])        返回 {            “parameters”： 参数，            “result”： le.inverse_transform（结果）[0]，        }

这里，如果用户没有输入值，我们将 ” 和 ” 的默认值设置为 0。我们还添加了一个检查以查看任何一个值是否为 0 并返回错误消息，而不是尝试预测输入引用的企鹅。

如果输入不为 0，我们将使用模型进行预测，并使用编码器进行逆变换以获取预测目标的标签，即企鹅物种的名称。

这不是验证输入的唯一方法。你也可以考虑使用 Pydantic 进行输入验证。bill_length_mmflipper_length_mm

如果您使用的是与 ” 中所述的相同版本的 FastAPI，FastAPI 会自动刷新服务并在保存时应用更改。现在在 ” 中放入一个新的 URL 进行测试（与之前的 URL 用 ### 分隔）：requirements.txttest_main.http

###获取 http:// 127.0。0.1：8000/预测/？bill_length_mm=40.3&flipper_length_mm=195接受：application/json

按新 URL 旁边的箭头按钮 并查看输出。

接下来，您可以尝试删除了一个或两个参数的 URL，以查看错误消息：

###获取 http:// 127.0。0.1：8000/预测/？bill_length_mm=40.3接受：application/json

6. 设置包含生命周期事件的机器学习模型

最后，让我们看看如何使用 FastAPI 生命周期事件设置我们的模型。这样做的好处是，我们可以确保在模型仍在设置时不会接受任何请求，并且之后会清理使用的内存。为此，我们将使用 ‘asynccontextmanager’。在 ” 之前，我们将添加：app = FastAPI()

从 contextlib import asynccontextmanagerml_models ={}@asynccontextmanager  async def lifespan（app： FastAPI）的调用：    # 在此处设置 ML 模型        屈服    # 清理模型并释放资源 ml_models。清楚()

现在我们将 pandas 和 scikit-learn 的导入移动到其他导入的旁边。我们还会将设置代码移动到 ” 函数中，在 ” 中设置机器学习模型和 LabelEncoder，如下所示：lifespanml_models

from fastapi import FastAPI从 contextlib import asynccontextmanager将 pandas 导入为 PD从 sklearn.model_selection 导入 train_test_split从 sklearn 导入预处理from sklearn.neighbors import KNeighborsClassifierfrom sklearn.pipeline 导入管道from sklearn.preprocessing import StandardScalerml_models ={}@asynccontextmanager  async def lifespan（app： FastAPI）的调用：    # 在此处设置 ML 模型 数据 = PD。read_csv（'penguins.csv') data = 数据。滴() le = 预处理。标签编码器() X = data[[“bill_length_mm”， “flipper_length_mm”]] 乐。fit（data[“物种”]) y = le。transform（data[“物种”]) X_train、X_test、y_train、y_test = train_test_split（X， y， stratify=y， random_state=0) clf = 管道( steps=[（“scaler”， StandardScaler（）））， （“knn”， KNeighborsClassifier（n_neighbors=11))]    ) CLF 的set_params（） 中。fit（X_train， y_train) ml_models[“clf”] = clf ml_models[“le”] = le    屈服    # 清理模型并释放资源 ml_models。清楚()

之后，我们将在 ” 中添加 ” 参数：lifespan=lifespanapp = FastAPI()

应用程序 = FastAPI（lifespan=lifespan)

现在保存并再次测试。一切都应该正常工作，我们应该会看到和以前一样的结果。

事后思考：何时训练模型？

从我们的示例中，您可能想知道模型是何时训练的。由于 ” 是在一开始训练的，即服务启动时，你可能会想为什么我们不是每次有人做出预测时都训练模型。clf

我们不希望每次有人拨打电话时都训练模型，因为重新训练所有内容会花费更多的资源。此外，它可能会导致竞争条件，因为我们的 FastAPI 应用程序正在并发工作。如果我们使用不断变化的实时数据，则尤其如此。

从技术上讲，我们可以设置一个 API 来收集数据并重新训练模型（我们将在下一个示例中演示）。其他选项是安排在收集了一定数量新数据的特定时间进行重新训练，或者让超级用户上传新数据并触发重新训练。

到目前为止，我们的目标是构建一个在本地运行的原型。

什么是并发？

简单地说，并发就像你在厨房做饭，在等待水沸腾时，你继续切菜。由于，在 Web 服务世界中，服务器正在与许多终端通信，而服务器与终端之间的通信比大多数内部应用程序都要慢，因此服务器不会逐个与终端通信和服务。相反，它将在满足他们的请求的同时与他们中的许多人交谈并为他们提供服务。

在 Python 中，这是通过使用异步代码来实现的。在我们的 FastAPI 代码中，使用 ” 而不是 ” 是 FastAPI 并发工作的明显证据。Python 异步代码中还使用了其他关键字，例如 ” 和 ”，但我们无法在这篇博文中介绍它们。async defdefawaitasyncio.get_event_loop

如何将 FastAPI 用于图像分类项目

为了发现更多的 FastAPI 功能，我们还将基于 Keras 中的 MNIST 示例的图像分类模型添加到我们的应用程序中（我们使用的是 TensorFlow 后端）。如果您安装了提供的 ”，则应安装 Keras 和 Pillow 以进行图像处理和构建卷积神经网络 （CNN）。requirements.txt

1. 重构

在开始之前，让我们重构代码。为了使代码更有条理，我们将企鹅预测的模型设置放在一个函数中：

 def penguins_pipeline（） ： 数据 = PD。read_csv（'penguins.csv') data = 数据。滴() le = 预处理。标签编码器() X = data[[“bill_length_mm”， “flipper_length_mm”]] 乐。fit（data[“物种”]) y = le。transform（data[“物种”]) X_train、X_test、y_train、y_test = train_test_split（X， y， stratify=y， random_state=0) clf = 管道( steps=[（“scaler”， StandardScaler（）））， （“knn”， KNeighborsClassifier（n_neighbors=11))]    ) CLF 的set_params（） 中。fit（X_train， y_train)    返回 CLF、LE

然后我们重写 lifespan 函数。在 PyCharm 中完成全行代码，非常简单：

2. 设置 CNN 模型进行 MNIST 预测

与企鹅预测模型类似，我们创建了一个用于 MNIST 预测的函数（我们将全局存储元参数）：

# MNIST 模型元参数num_classes = 10input_shape = （28， 28， 1)batch_size = 128纪元 = 15  def mnist_pipeline（） 的：    # 加载数据并将其拆分到训练集和测试集之间    (x_train， y_train）， _ = keras.datasets.mnist.load_data()    # 将图像缩放到 [0， 1] 范围 x_train = x_train。astype（“float32”） / 255    # 确保图片的形状为 （28， 28， 1） x_train = np。expand_dims（x_train， -1)    # 将类向量转换为二进制类矩阵 y_train = keras.utils。to_categorical（y_train， num_classes) 模型 = Keras。顺序(        [ 凯拉斯。输入（shape=input_shape）， 层。Conv2D（32， kernel_size=（3， 3）， activation=“relu”）， 层。MaxPooling2D（pool_size=（2， 2）））、 层。Conv2D（64， kernel_size=（3， 3）， activation=“relu”）， 层。MaxPooling2D（pool_size=（2， 2）））、 层。Flatten（）、 层。Dropout（0.5）//辍学 层。密集（num_classes， activation=“softmax”）），        ]    ) 型。compile（loss=“categorical_crossentropy”， optimizer=“Adam”， metrics=[“准确性”]) 型。fit（x_train， y_train， batch_size=batch_size， epochs=epochs， validation_split=0.1)    返回模型

然后在 lifespan 函数中添加模型设置：ml_models[“cnn”] = mnist_pipeline()

请注意，由于已添加此项，因此每次更改 ” 并保存时，模型都会再次被训练。这可能需要一些时间。因此，在开发过程中，您可能希望使用完全不需要训练时间的虚拟模型或预先训练的模型。训练后，CNN 模型将准备就绪。main.py

3. 设置 POST 终端节点，用于上传图像文件进行预测

要设置一个接受上传文件的端点，我们必须在 FastAPI 中使用 UploadFile：

@app。post（“/预测图像/”)  async def predicct_upload_file（file： UploadFile）： img = await 文件。读()        # 处理图像进行预测 img = 图像。open（BytesIO（img）） 的 API 中。convert（'L') img = NP 的。array（img） 的astype（“float32”） / 255 img = NP 的。expand_dims（img， （0， -1))        # 预测结果 result = ml_models[“cnn”] 的predict（img） 的argmax（axis=-1）[0]    返回 {“filename”： file.filename，            “result”： str（结果)}

请注意，这是一个终端节点（到目前为止，我们只设置了终端节点）。POSTGET

不要忘记从 ” 导入 ”：UploadFilefastapi

from fastapi import FastAPI， UploadFile

还有 ” 来自 Pillow。我们还使用了 ” 模块中的 ”：ImageBytesIOio

from PIL import 图片from io import BytesIO

要使用带有测试图像文件的 PyCharm HTTP 客户端对此进行测试，我们将使用 ” 编码。这是您将放入 ”文件的内容：multipart/form-datatest_in.http

###帖子 http:// 127.0。0.1：8000/预测图像/ HTTP/1.1内容类型： multipart/form-data;boundary=边界--边界内容处置： form-data; name=“file”; filename=“test_img0.png”<./test_img0.png--边界–

4. 添加 API 以收集数据并触发重新训练

现在，再培训来了。我们设置了一个如上所述的终端节点，以接受包含训练图像和标签的 zip 文件。然后，将处理 zip 文件并准备训练数据。之后，我们将再次拟合 CNN 模型：POST

@app。post（“/上传图片/”)  async def retrain_upload_file（file： UploadFile）：img_files =[] labels_file = 无 train_img = 无    替换为 ZipFile（BytesIO（await file.read（））， 'r'） 指定为 zfile：        for fname 在 zfile 中。namelist（） 中：            if fname[-4：] == '.txt' 和 fname[：2] ！= '__'：labels_file = fname            elif fname[-4：] == '.png'： img_files。append（fname)        如果 len（img_files） == 0：            return {“error”： “未找到训练图像（png 文件）。”}        否则：            对于 sorted（img_files） 中的  fname：                与 zfile 一起使用。open（fname） 作为 img_file： img = img_file。读()                    # 进程镜像 img = 图像。open（BytesIO（img）） 的 API 中。convert（'L') img = NP 的。array（img） 的astype（“float32”） / 255 img = NP 的。expand_dims（img， （0， -1))                    如果 train_img 为 None：train_img = 图像                    否则： train_img = np。vstack（（train_img， IMG))        如果 labels_file 为 None：            return {“error”： “未找到训练标签文件（txt 文件）。”}        否则：            与 zfile 一起使用。open（labels_file） 作为标签： labels_data = 标签。读() labels_data = labels_data。decode（“utf-8”） 的分裂() labels_data = np。array（labels_data） 的astype（“int”) labels_data = keras.utils。to_categorical（labels_data， num_classes)    # 重新训练模型 ml_models[“cnn”]。fit（train_img， labels_data， batch_size=batch_size， epochs=epochs， validation_split=0.1)    return {“message”： “模型训练成功。”}

请记得导入 ”：ZipFile

from zipfile import ZipFile

如果我们现在使用这个包含 1000 个重新训练图像和标签的 zip 文件尝试终端节点，您将看到响应需要一些时间才能出现，因为训练需要一段时间：

POST http：127.0.0.1：8000/upload-images/ HTTP/1.1内容类型： multipart/form-data;boundary=边界--边界内容处置： form-data; name=“file”; filename=“training_data.zip”< ./retrain_img。邮编--边界--

假设 zip 文件包含更多的训练数据，或者您正在重新训练更复杂的模型。然后，用户将不得不等待很长时间，并且似乎事情对他们不起作用。

5. 使用 BackgroundTasks 重新训练模型

处理重新训练的更好方法是，在收到训练数据后，我们对其进行处理并检查数据的格式是否正确，然后给出响应，指出重新训练已重新启动，并在“BackgroundTasks”中训练模型。这是如何做到的。首先，我们将 ” 添加到我们的 ‘upload-images’ 端点：BackgroundTasks

@app。post（“/上传图片/”)  async def retrain_upload_file（文件： UploadFile， background_tasks： BackgroundTasks）：...

记得从 ” 导入它：fastapi

从 fastapi 导入 FastAPI、UploadFile、BackgroundTasks

然后，我们将模型的拟合放入 ”：background_tasks

# 重新训练模型 background_tasks。add_task( ml_models[“cnn”].fit，train_img，labels_data，batch_size=batch_size，epochs=纪元，        validation_split=0.1    )

此外，我们还将更新响应中的消息：return {“message”： “数据接收成功，模型训练已开始。”

}

现在再次测试终端节点。您将看到响应到达的速度要快得多，如果您查看 Run 窗口，您将看到训练在响应到达后正在运行。

此时，可以添加更多功能，例如，在培训结束时（例如通过电子邮件）通知用户，或者在构建完整应用程序时在仪表板中跟踪培训进度的选项。

使用 PyCharm 开发 ML FastAPI 应用程序

FastAPI 提供了一种简单的方法，可以通过几个简单的步骤将数据科学项目转换为工作应用程序。它非常适合希望为其机器学习模型提供应用程序原型的数据科学团队，如果需要，该原型可以进一步开发为专业的 Web 应用程序。

PyCharm Professional 是一款 Python IDE，它允许您通过 FastAPI 的预配置项目、编码辅助、定制的运行/调试配置以及用于高效管理 API 端点的端点工具窗口，更轻松地开发 FastAPI 应用程序。

原文来源：https://blog.jetbrains.com/pycharm/2024/09/how-to-use-fastapi-for-machine-learning/