使用 Keras 创建自定义预测例程

Colab 徽标 在 Colab 中以笔记本的形式运行本教程 GitHub 徽标 在 GitHub 上查看笔记本

本教程介绍如何使用自定义预测例程将经过训练的 Keras 模型部署到 AI Platform Prediction 并执行预测。这一方法让您可以自定义 AI Platform Prediction 如何响应每个预测请求。

在此示例中，您将使用自定义预测例程，通过缩放对预测输入进行预处理，并通过将 softmax 概率输出转换为标签字符串对预测输出进行后处理。

本教程将完成以下多个步骤：

  • 在本地训练一个简单的 Keras 模型
  • 创建自定义预测例程并将其部署到 AI Platform Prediction
  • 从该部署中处理预测请求

数据集

本教程使用 R.A. Fisher 的鸢尾数据集，这是一个小型数据集，非常适合用于试用机器学习技术。每个实例具有四个数值特征（这些特征是花朵的不同测量值）和一个目标标签（将实例标记为三种类型的鸢尾之一：山鸢尾、变色鸢尾或维吉尼亚鸢尾）。

本教程使用 scikit-learn 库中包含的鸢尾数据集副本。

目标

目标是训练一个模型，该模型使用花朵的测量值作为输入，以预测此花是什么类型的鸢尾。

本教程重点介绍如何将此模型与 AI Platform Prediction 结合使用，而不是仅仅介绍模型本身的设计。

费用

本教程使用 Google Cloud 的以下收费组件：

  • AI Platform Prediction
  • Cloud Storage

了解 AI Platform Prediction 价格和 Cloud Storage 价格，并使用价格计算器根据您的预计使用情况来估算费用。

准备工作

在 AI Platform Prediction 中训练和部署模型之前，您必须先完成以下事项：

  • 设置本地开发环境。
  • 设置 Google Cloud 项目，并启用结算功能和必要的 API。
  • 创建 Cloud Storage 存储桶以存储您的训练软件包和经过训练的模型。

设置本地开发环境

您需要以下资源才能完成本教程：

  • Python 3
  • virtualenv
  • Google Cloud SDK

有关设置 Python 开发环境的 Google Cloud 指南详细说明了如何满足这些要求。以下步骤提供了一系列简要的说明：

  1. 安装 Python 3。

  2. 安装 virtualenv 并创建一个使用 Python 3 的虚拟环境。

  3. 激活该环境。

  4. 完成以下部分中的步骤以安装 Google Cloud SDK。

设置您的 Google Cloud 项目

  1. 登录您的 Google Cloud 账号。如果您是 Google Cloud 新手，请创建一个账号来评估我们的产品在实际场景中的表现。新客户还可获享 $300 赠金，用于运行、测试和部署工作负载。

  2. 在 Google Cloud Console 中的项目选择器页面上，选择或创建一个 Google Cloud 项目。

    转到“项目选择器”

  3. 确保您的 Google Cloud 项目已启用结算功能。

  4. 启用 AI Platform Training & Prediction and Compute Engine API。

    启用 API

  5. 安装 Google Cloud CLI。

  6. 如需初始化 gcloud CLI，请运行以下命令： 

    gcloud init
    7.

  7. 在 Google Cloud Console 中的项目选择器页面上，选择或创建一个 Google Cloud 项目。

    转到“项目选择器”

  8. 确保您的 Google Cloud 项目已启用结算功能。

  9. 启用 AI Platform Training & Prediction and Compute Engine API。

    启用 API

  10. 安装 Google Cloud CLI。

  11. 如需初始化 gcloud CLI，请运行以下命令： 

    gcloud init
    12.

验证您的 GCP 账号

要设置身份验证，您需要创建服务账号密钥并为服务账号密钥的文件路径设置环境变量。

  1. 创建服务账号：

    1. 在 Google Cloud 控制台中，前往创建服务帐号页面。

      转到“创建服务账号”

    2. 在服务账号名称字段中，输入一个名称。
    3. 可选：在服务账号说明字段中，输入说明。
    4. 点击创建。
    5. 点击选择角色字段。在所有角色下，选择 AI Platform > AI Platform Admin。
    6. 点击添加其他角色。

    7. 点击选择角色字段。在所有角色下，选择存储 > Storage Object Admin。

    8. 点击完成以创建服务账号。

      不要关闭浏览器窗口。您将在下一步骤中用到它。

  2. 创建用于身份验证的服务账号密钥：

    1. 在 Google Cloud 控制台中，点击您创建的服务账号的电子邮件地址。
    2. 点击密钥。
    3. 依次点击添加密钥和创建新密钥。
    4. 点击创建。JSON 密钥文件将下载到您的计算机上。
    5. 点击关闭。

  3. 将环境变量 GOOGLE_APPLICATION_CREDENTIALS 设置为包含服务账号密钥的 JSON 文件的文件路径。此变量仅适用于当前的 shell 会话，因此，如果您打开新的会话，请重新设置该变量。

创建 Cloud Storage 存储桶

如需部署自定义预测例程，您必须将经过训练的模型工件和您的自定义代码上传到 Cloud Storage。

将 Cloud Storage 存储桶的名称设置为环境变量。它在所有 Cloud Storage 存储分区中必须是唯一的：

BUCKET_NAME="your-bucket-name"

选择一个提供 AI Platform Prediction 的区域，然后另外创建一个环境变量。

REGION="us-central1"

在您选择的区域中创建 Cloud Storage 存储桶，稍后使用同一区域进行训练和预测。如果存储桶尚不存在，请运行以下命令创建一个：

gsutil mb -l $REGION gs://$BUCKET_NAME

构建和训练 Keras 模型

通常，您不能使用原始形式的数据来训练机器学习模型。即便可以，在用于训练之前对数据进行预处理有时可改善您的模型。

假设您希望预测输入与训练数据采用相同的格式，您必须在训练和预测过程中应用相同的预处理，以确保您的模型做出一致的预测。

在本部分中，创建一个预处理模块，并在训练过程中使用它。然后，导出具备在训练期间获知的特征的预处理器，以便稍后在自定义预测例程中使用。

安装用于本地训练的依赖项

在本地训练需要多个依赖项：

pip install numpy scikit-learn 'tensorflow=2.3'

编写预处理器

缩放训练数据，使每个数值特征列的平均值为 0，标准差为 1，以便改善您的模型。

创建 preprocess.py，其中包含用于执行此缩放操作的类：

import numpy as np

class MySimpleScaler(object):
  def __init__(self):
    self._means = None
    self._stds = None

  def preprocess(self, data):
    if self._means is None: # during training only
      self._means = np.mean(data, axis=0)

    if self._stds is None: # during training only
      self._stds = np.std(data, axis=0)
      if not self._stds.all():
        raise ValueError('At least one column has standard deviation of 0.')

    return (data - self._means) / self._stds

请注意，MySimpleScaler 的实例会在首次使用时保存每个特征列的均值和标准偏差。然后，它使用这些总结统计信息来缩放之后出现的数据。

这样一来，您可以存储训练分布的特征，并在预测时使用这些特征实现相同的预处理。

训练模型

接下来，使用 preprocess.MySimpleScaler 对鸢尾数据进行预处理，然后使用 Keras 训练简单的神经网络。

最后，将经过训练的 Keras 模型导出为 HDF5 (.h5) 文件，并将您的 MySimpleScaler 实例导出为 pickle (.pkl) 文件：

import pickle

from sklearn.datasets import load_iris
import tensorflow as tf

from preprocess import MySimpleScaler

iris = load_iris()
scaler = MySimpleScaler()
num_classes = len(iris.target_names)
X = scaler.preprocess(iris.data)
y = tf.keras.utils.to_categorical(iris.target, num_classes=num_classes)

model = tf.keras.Sequential()
model.add(tf.keras.layers.Dense(25, activation=tf.nn.relu))
model.add(tf.keras.layers.Dense(25, activation=tf.nn.relu))
model.add(tf.keras.layers.Dense(num_classes, activation=tf.nn.softmax))
model.compile(
  optimizer='adam', loss='categorical_crossentropy', metrics=['accuracy'])
model.fit(X, y, epochs=10, batch_size=1)

model.save('model.h5')
with open ('preprocessor.pkl', 'wb') as f:
  pickle.dump(scaler, f)

部署自定义预测例程

要部署自定义预测例程以便根据经过训练的模型执行预测，请执行以下操作：

  • 创建自定义预测器来处理请求
  • 封装预测器和预处理模块
  • 将您的模型工件和自定义代码上传到 Cloud Storage
  • 将自定义预测例程部署到 AI Platform Prediction

创建自定义预测器

如需部署自定义预测例程，您必须创建一个实现 Predictor 接口的类。这会告知 AI Platform Prediction 如何加载您的模型以及如何处理预测请求。

将以下代码写入 predictor.py：

import os
import pickle

import numpy as np
from sklearn.datasets import load_iris
import tensorflow as tf

class MyPredictor(object):
  def __init__(self, model, preprocessor):
    self._model = model
    self._preprocessor = preprocessor
    self._class_names = load_iris().target_names

  def predict(self, instances, **kwargs):
    inputs = np.asarray(instances)
    preprocessed_inputs = self._preprocessor.preprocess(inputs)
    outputs = self._model.predict(preprocessed_inputs)
    if kwargs.get('probabilities'):
      return outputs.tolist()
    else:
      return [self._class_names[index] for index in np.argmax(outputs, axis=1)]

  @classmethod
  def from_path(cls, model_dir):
    model_path = os.path.join(model_dir, 'model.h5')
    model = tf.keras.models.load_model(model_path)

    preprocessor_path = os.path.join(model_dir, 'preprocessor.pkl')
    with open(preprocessor_path, 'rb') as f:
      preprocessor = pickle.load(f)

    return cls(model, preprocessor)

请注意，除了使用您在训练期间定义的预处理器，该预测器还会执行后处理步骤，将神经网络的 softmax 输出（指示每个标签正确概率的数组）转换为具有最高概率的标签。

但是，如果预测器收到值为 True 的 probabilities 关键字参数，则会返回概率数组。本教程的最后一部分介绍如何提供该关键字参数。

封装您的自定义代码

您必须将 predictor.py 和 preprocess.py 封装为 .tar.gz 源分发软件包，并将该软件包提供给 AI Platform Prediction，以使其能够使用您的自定义代码来执行预测。

编写以下 setup.py 以定义您的软件包：

from setuptools import setup

setup(
    name='my_custom_code',
    version='0.1',
    scripts=['predictor.py', 'preprocess.py'])

然后，运行以下命令以创建 dist/my_custom_code-0.1.tar.gz：

python setup.py sdist --formats=gztar

将模型工件和自定义代码上传到 Cloud Storage

在部署您的模型以提供服务之前，AI Platform Prediction 需要 Cloud Storage 中以下文件的访问权限：

  • model.h5（模型工件）
  • preprocessor.pkl（模型工件）
  • my_custom_code-0.1.tar.gz（自定义代码）

模型工件必须一起存储在模型目录中，您的 Predictor 可以将该目录作为 from_path 类方法中的 model_dir 参数进行访问。自定义代码不需要位于同一目录中。您可以运行以下命令上传您的文件：

gsutil cp ./dist/my_custom_code-0.1.tar.gz gs://$BUCKET_NAME/custom_prediction_routine_tutorial/my_custom_code-0.1.tar.gz
gsutil cp model.h5 preprocessor.pkl gs://$BUCKET_NAME/custom_prediction_routine_tutorial/model/

部署自定义预测例程

创建model模型资源和model版本资源可以部署您的自定义预测例程。首先，使用您的资源名称定义环境变量：

MODEL_NAME='IrisPredictor'
VERSION_NAME='v1'

然后创建模型：

gcloud ai-platform models create $MODEL_NAME \
  --regions $REGION

接下来，创建版本。在该步骤中，提供您上传到 Cloud Storage 的工件和自定义代码的路径：

gcloud components install beta

gcloud beta ai-platform versions create $VERSION_NAME \
  --model $MODEL_NAME \
  --runtime-version 2.3 \
  --python-version 3.7 \
  --origin gs://$BUCKET_NAME/custom_prediction_routine_tutorial/model/ \
  --package-uris gs://$BUCKET_NAME/custom_prediction_routine_tutorial/my_custom_code-0.1.tar.gz \
  --prediction-class predictor.MyPredictor

详细了解您在部署自定义预测例程时必须指定的选项。

执行在线预测

通过发送在线预测请求试用您的部署。首先，安装 Python 版 Google API 客户端库：

pip install --upgrade google-api-python-client

然后，通过运行以下 Python 代码，将鸢尾数据的两个实例发送到已部署的版本：

import googleapiclient.discovery

instances = [
  [6.7, 3.1, 4.7, 1.5],
  [4.6, 3.1, 1.5, 0.2],
]

service = googleapiclient.discovery.build('ml', 'v1')
name = 'projects/{}/models/{}/versions/{}'.format(PROJECT_ID, MODEL_NAME, VERSION_NAME)

response = service.projects().predict(
    name=name,
    body={'instances': instances}
).execute()

if 'error' in response:
    raise RuntimeError(response['error'])
else:
  print(response['predictions'])

['versicolor', 'setosa']

发送关键字参数

将预测请求发送到自定义预测例程时，您可以在请求正文中提供其他字段。Predictor 的 predict 方法将这些字段作为 **kwargs 字典的字段接收。

以下代码发送的请求与之前相同，但这一次，它在请求正文中添加了 probabilities 字段：

response = service.projects().predict(
    name=name,
    body={'instances': instances, 'probabilities': True}
).execute()

if 'error' in response:
    raise RuntimeError(response['error'])
else:
  print(response['predictions'])

[[0.0019204545533284545, 0.8623144626617432, 0.13576509058475494], [0.999488353729248, 0.000511515187099576, 1.293626752385535e-07]]

清理

要清除此项目中使用的所有 GCP 资源，您可以删除用于本教程的 GCP 项目。

您也可以运行以下命令，清理各个资源：

# Delete version resource
gcloud ai-platform versions delete $VERSION_NAME --quiet --model $MODEL_NAME

# Delete model resource
gcloud ai-platform models delete $MODEL_NAME --quiet

# Delete Cloud Storage objects that were created
gsutil -m rm -r gs://$BUCKET_NAME/custom_prediction_routine_tutorial

后续步骤