99AutoML 自动化机器学习实践--NNI 自动化机器学习工具包

NNI 自动化机器学习工具包

NNI 是 Neural Network Intelligence 的缩写，可以译作：智能神经网络。名字听起来陌生，但 NNI 实际上就是一个自动化机器学习工具包。它通过多种调优的算法来搜索最好的神经网络结构和超参数，并支持单机、本地多机、云等不同的运行环境。
NNI 由微软主导开发，背景雄厚。目前已经成为了支持框架和库最多，支持调优算法最全，支持训练平台最广的开源 AutoML 工具包。

image.png

NNI 官方给出了如下所示的组件结构图。实际上 NNI 总共包含 3 部分：Python 接口，NNICTL 命令行工具，以及 NNI Board 可视化面板。

image.png

NNI 使用时，我们会使用其提供的 Python 库来对接训练代码，然后通过 NNICTL 命令行工具读取配置文件并开始训练。训练过程中，可以使用 NNI Board 实时查看训练情况。这就是一个典型的 NNI 自动化学习过程。
你会发现 NNI 与 auto-sklearn 和 Auto-Keras 的巨大不同。后两者是单独的库，而 NNI 相当于是接口。NNI 的优势是显而易见的，你只需要对现有的代码稍作修改即可开始自动化训练过程。而如果使用 auto-sklearn 和 Auto-Keras 则基本上是重写代码。此外，auto-sklearn 和 Auto-Keras 本身开发质量并不好，使用起来也是一言难尽。

NNI 环境搭建

本次实验中，我们将利用实验楼提供的 WebIDE 来搭建 NNI 开发环境。原因在于 NNI 本身提供了基于 Web 技术的可视化看板，WebIDE 提供了更好的使用体验。
一般情况下，你可以在本地尝试使用 pip 安装 NNI，但线上环境需要通过直接编译源码完成安装。
我们首先克隆源码仓库：

git clone -b v1.0 https://github.com/Microsoft/nni.git --depth=1

接下来，编译源码并安装 NNI：

# 切换到目录下方
cd nni/
# 更新 pip 组件
python3 -m pip install --upgrade pip
# 编译安装 NNI
source install.sh

NNI 安装时间较长，需要等待 5 ～ 10 分钟。安装完成之后会看到 Complete 的提示。
安装完成之后，添加环境变量以便于在终端中使用 NNICTL 命令行工具：

export PATH="$PATH:/home/shiyanlou/.local/bin"

此时，你可以在终端中输入 nnictl -h，如果正确返回了 NNICTL 命令行工具的使用介绍，则表面一切安装就绪。注意，打开新终端或重启环境后都需要重新执行上方添加环境变量的语句，否则将无法调用 NNICTL 命令行工具。

image.png

NNI 运行机制

在正式了解 NNI 的使用之前，我们需要先知晓其中的一些基本概念以及运行机制。
NNI 的构成核心是 Experiment，实验是一次找到模型的最佳超参组合，或最好的神经网络架构的任务，它由 Trial 和 Tuner 所组成。其中，Trial 是一次尝试，它会使用某组配置或者特定的神经网络架构完成执行，Trial 会基于提供的配置来运行。Tuner 是一个自动机器学习算法，会为下一个 Trial 生成新的配置，新的 Trial 会使用这组配置来运行。

此外，NNI 还会涉及到其他的一些重要概念：
Search Space：搜索空间是模型调优的范围。例如，超参的取值范围。
Configuration：配置是来自搜索空间的一个参数实例，每个超参都会有一个特定的值。
Assessor：Assessor 分析 Trial 的中间结果，来确定 Trial 是否应该被提前终止。
Training Platform：训练平台是 Trial 的执行环境。根据 Experiment 的配置，可以是本机，远程服务器组，或其它大规模训练集群。

Experiment 的运行过程为：Tuner 接收搜索空间并生成配置，这些配置将被提交到训练平台，如本机，远程服务器组或训练集群。执行的性能结果会被返回给 Tuner。然后，再生成并提交新的配置。重复训练过程，直到 Assessor 确认终止。

想要使用 NNI 来完成一次实验，一般会有以下几个步骤：
定义模型训练和测试代码。
定义 NNI 搜索空间参数。
基于 NNI 接口改动模型代码。
定义 NNI Experiment 配置。
使用 NNICTL 工具完成训练。

接下来，我们就以 scikit-learn 为例，使用 NNI 来完成一次自动机器学习训练过程。

NNI 使用示例

NNI 支持很多机器学习相关的库和框架，选择以 scikit-learn 举例是因为其相对简单，很适合用作示例。之后，只需要同理类推，就可以很快速的迁移到 TensorFlow，PyTorch 等深度学习框架中使用。
我们按照上面所示的 5 个步骤来完成 NNI 的使用。首先，构建一个示例训练过程，你需要在 IDE 左侧新建一个名为 digits 的文件夹用于存放代码。后续的所有代码及配置文件都存放在该目录下方。
定义模型训练和测试代码
本次实验使用 DIGITS 手写字符数据集，并使用 SVM 完成分类。首先，新建 svm_before.py 文件用于存储代码：

from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.datasets import load_digits
from sklearn.svm import SVC

def load_data():
    '''加载数据函数'''
    digits = load_digits()  # DIGITS 数据集
    # 切分数据，20% 用于测试
    X_train, X_test, y_train, y_test = train_test_split(
        digits.data, digits.target, random_state=99, test_size=0.2)
    # 标准化数据
    ss = StandardScaler()
    X_train = ss.fit_transform(X_train)
    X_test = ss.transform(X_test)

    return X_train, X_test, y_train, y_test

if __name__ == '__main__':
    X_train, X_test, y_train, y_test = load_data()
    model = SVC()
    model.fit(X_train, y_train)  # 训练模型
    score = model.score(X_test, y_test)
    print(score)

代码非常简单。加载 DIGITS 数据集并完成归一化，使用默认参数定义 SVM 模型，然后训练并获得准确度。这其实是一个非常标准的训练过程。
定义 NNI 搜索空间参数
scikit-learn 结合自动化机器学习应用，实际上就是自动完成超参数搜索。所以，我们需要定义 NNI 搜索空间参数。在 NNI 中，Tuner 会根据搜索空间来取样生成参数和网络架构。搜索空间通过 JSON 文件来定义，需要变量名称、采样策略的类型及其参数。
你需要建立一个 search_space.json 的 JSON 文件。然后，选择部分 SVM 支持的超参数，并添加搜索空间。

{
  "C": { "_type": "uniform", "_value": [0.1, 1] },
  "keral": {
    "_type": "choice",
    "_value": ["linear", "rbf", "poly", "sigmoid"]
  },
  "degree": { "_type": "choice", "_value": [1, 2, 3, 4] },
  "gamma": { "_type": "uniform", "_value": [0.01, 0.1] },
  "coef0 ": { "_type": "uniform", "_value": [0.01, 0.1] }
}

_type 实际上就是定义以何种方式从 _value 后续参数中取值。你可以阅读官方文档详细了解，这里就不再罗列了。
基于 NNI 接口改动模型代码
接下来，我们需要基于 NNI 提供的 Python 接口来修改之前定义好的代码，并使用 svm.py 新文件存储。

from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.datasets import load_digits
from sklearn.svm import SVC
import nni

def load_data():
    '''加载数据函数'''
    digits = load_digits()  # DIGITS 数据集
    # 切分数据，20% 用于测试
    X_train, X_test, y_train, y_test = train_test_split(
        digits.data, digits.target, random_state=99, test_size=0.2)
    # 标准化数据
    ss = StandardScaler()
    X_train = ss.fit_transform(X_train)
    X_test = ss.transform(X_test)

    return X_train, X_test, y_train, y_test

if __name__ == '__main__':
    X_train, X_test, y_train, y_test = load_data()

    # 默认超参数
    PARAMS = {'C': 1.0, 'kernel': 'linear', 'degree': 3, 'gamma': 0.01, 'coef0': 0.01}
    # 从 Tuner 接收搜索空间生成的超参数
    RECEIVED_PARAMS = nni.get_next_parameter()
    PARAMS.update(RECEIVED_PARAMS) # 更新超参数

    # 传入超参数
    model = SVC(C=PARAMS.get('C'), kernel=PARAMS.get('kernel'),
                degree=PARAMS.get('degree'), gamma=PARAMS.get('gamma'),
                coef0=PARAMS.get('coef0'))

    model.fit(X_train, y_train)  # 训练模型
    score = model.score(X_test, y_test)  # 准确度

    # 最后将 score 发送给可视化看板
    nni.report_final_result(score)

对比 svm_before.py，需要补充的代码非常简单。首先使用 import nni 导入 NNI 库，然后定义默认参数字典 PARAMS。接下来，使用 nni.get_next_parameter() 接收到新的参数，并更新默认参数后传入模型。最后，使用 nni.report_final_result 将需要可视化的指标发送给可视化看板，一般会选择分类准确度。
定义 NNI Experiment 配置
按照步骤，接下来定义 NNI Experiment 配置文件。配置文件必须为 YAML 格式，我们新建 config.yml 用于保存配置。一般情况下，我们会从官方文档复制基础配置信息，并按照需求进行修改。

authorName: shiyanlou
experimentName: digits-sklearn-nni
trialConcurrency: 3
maxExecDuration: 1h
maxTrialNum: 100
trainingServicePlatform: local
searchSpacePath: search_space.json
useAnnotation: false
tuner:
  builtinTunerName: TPE
  classArgs:
    optimize_mode: maximize
trial:
  command: python3 svm.py
  codeDir: .
  gpuNum: 0

配置文件中比较关键的字段有：
trialConcurrency：并发尝试任务的最大数量，根据机器配置而定。
maxExecDuration：Experiment 最大运行时长。
maxTrialNum：Experiment 最大运行 Trial 数量。
trainingServicePlatform：local 表示本地，可选择 remote，pai，kubeflow 等平台。
searchSpacePath：搜索空间参数文件。
builtinTunerName：指定优化算法，例如：TPE, Random, Anneal, Evolution, BatchTuner, GridSearch 等。
optimize_mode：根据优化算法设置，TPE 默认为 maximize。
command：运行 Trial 进程的命令行。
codeDir：指定了 Trial 代码文件的目录。

至此，你的环境目录下方应该存在这些文件：

image.png

实际上，必须存在的是 config.yml，search_space.json 和 svm.py 三个文件。svm_before.py 是实验为了对比代码，当你对 NNI 足够熟悉时，往往可以直接开始写最终的训练代码脚本。
使用 NNICTL 工具完成训练
一切就绪，接下来就可以使用 NNICTL 工具完成训练。我们在终端中使用命令行加载 NNI Experiment 配置。

nnictl create --config digits/config.yml

上面代表加载 digits/config.yml 路径下方的配置文件，并完成训练。默认情况下，NNI 运行在 8080 端口。此时，你可以通过实验环境右侧的 Web 服务菜单打开 8080 端口兼听的进程，即为 NNI 可视化面板。

image.png

你可以面板顶部的菜单切换到 Trails Detail，即为每次 Trail 的详情信息。这这里，可以非常直观看出不同超参数的选择对最终准确度的影响。

image.png

选择下方 Trail Job 中的一个事件，你可以通过 Parameters 看到本次 Trail 所使用到的超参数。

image.png

NNI 的可视化面板非常简洁，相信你自己通过尝试，在很短的实验里就能够了解不同选项的作用了。等待搜索结束，你可以按照降序排列 Trail Job，以便于找出最优参数组合。当然，通过面板首页 Top10 trials 一栏也可以很清晰看出最优的 10 次搜索结果。

MNIST 使用 NNI 训练模型

前面的挑战中，我们已经介绍过 MNIST 手写字符数据集，并尝试使用 auto-sklearn 完成了自动化机器学习训练过程。本次挑战中，同样使用该数据集，并结合一个自己最熟悉的机器学习或深度学习框架完成训练。
NNI 支持的框架非常多，并且给出了大量的官方参考使用示例。
题目：结合你最熟悉的机器学习框架，使用 NNI 完成针对 MNIST 的训练过程。
我们推荐你使用 TensorFlow 或者 PyTorch 深度学习框架完成基础训练代码的书写，神经网络的结构不定，建议使用一些经典且表现不错的网络。
为了便于挑战的统一性，我们仍然使用 Digit Recognizer 比赛提供的 MNIST 数据集。
数据集下载地址：

# 本地复制链接粘贴到浏览器下载
https://labfile.oss.aliyuncs.com/courses/1357/digit-recognizer.zip
# WebIDE 环境内终端下载
wget https://labfile.oss.aliyuncs.com/courses/1357/digit-recognizer.zip

完成训练之后，仍然可以将结构提交到 Kaggle Digit Recognizer 比赛中，对比与 auto-sklearn 的效果。你可以在实验楼 WebIDE 线上环境中完成，也可以到 Kaggle Notebook 环境中完成该挑战。

import argparse
import logging
import keras
import numpy as np
from keras import backend as K
from keras.datasets import mnist
from keras.layers import Conv2D, Dense, Flatten, MaxPooling2D
from keras.models import Sequential

K.set_image_data_format('channels_last')

H, W = 28, 28
NUM_CLASSES = 10

def create_mnist_model(hyper_params, input_shape=(H, W, 1), num_classes=NUM_CLASSES):
    layers = [
        Conv2D(32, kernel_size=(3, 3), activation='relu', input_shape=input_shape),
        Conv2D(64, (3, 3), activation='relu'),
        MaxPooling2D(pool_size=(2, 2)),
        Flatten(),
        Dense(100, activation='relu'),
        Dense(num_classes, activation='softmax')
    ]

    model = Sequential(layers)

    if hyper_params['optimizer'] == 'Adam':
        optimizer = keras.optimizers.Adam(lr=hyper_params['learning_rate'])
    else:
        optimizer = keras.optimizers.SGD(lr=hyper_params['learning_rate'], momentum=0.9)
    model.compile(loss=keras.losses.categorical_crossentropy, optimizer=optimizer, metrics=['accuracy'])

    return model

def load_mnist_data(args):
    (x_train, y_train), (x_test, y_test) = mnist.load_data()

    x_train = (np.expand_dims(x_train, -1).astype(np.float) / 255.)[:args.num_train]
    x_test = (np.expand_dims(x_test, -1).astype(np.float) / 255.)[:args.num_test]
    y_train = keras.utils.to_categorical(y_train, NUM_CLASSES)[:args.num_train]
    y_test = keras.utils.to_categorical(y_test, NUM_CLASSES)[:args.num_test]

    return x_train, y_train, x_test, y_test

class SendMetrics(keras.callbacks.Callback):
    def on_epoch_end(self, epoch, logs={}):
        pass

def train(args, params):
    x_train, y_train, x_test, y_test = load_mnist_data(args)
    model = create_mnist_model(params)

    model.fit(x_train, y_train, batch_size=args.batch_size, epochs=args.epochs, verbose=1,
        validation_data=(x_test, y_test), callbacks=[SendMetrics()])

    _, acc = model.evaluate(x_test, y_test, verbose=0)

def generate_default_params():
    return {
        'optimizer': 'Adam',
        'learning_rate': 0.001
    }

if __name__ == '__main__':
    PARSER = argparse.ArgumentParser()
    PARSER.add_argument("--batch_size", type=int, default=200, help="batch size", required=False)
    PARSER.add_argument("--epochs", type=int, default=10, help="Train epochs", required=False)
    PARSER.add_argument("--num_train", type=int, default=1000, help="Number of train samples to be used, maximum 60000", required=False)
    PARSER.add_argument("--num_test", type=int, default=1000, help="Number of test samples to be used, maximum 10000", required=False)

    ARGS, UNKNOWN = PARSER.parse_known_args()
    PARAMS = generate_default_params()
    train(ARGS, PARAMS)

import argparse
import logging
import keras
import numpy as np
from keras import backend as K
from keras.datasets import mnist
from keras.layers import Conv2D, Dense, Flatten, MaxPooling2D
from keras.models import Sequential

import nni

...

if __name__ == '__main__':
    PARSER = argparse.ArgumentParser()
    PARSER.add_argument("--batch_size", type=int, default=200, help="batch size", required=False)
    PARSER.add_argument("--epochs", type=int, default=10, help="Train epochs", required=False)
    PARSER.add_argument("--num_train", type=int, default=1000, help="Number of train samples to be used, maximum 60000", required=False)
    PARSER.add_argument("--num_test", type=int, default=1000, help="Number of test samples to be used, maximum 10000", required=False)

    ARGS, UNKNOWN = PARSER.parse_known_args()

    PARAMS = generate_default_params()
    RECEIVED_PARAMS = nni.get_next_parameter()
    PARAMS.update(RECEIVED_PARAMS)
    train(ARGS, PARAMS)

class SendMetrics(keras.callbacks.Callback):
    def on_epoch_end(self, epoch, logs={}):
        nni.report_intermediate_result(logs)

def train(args, params):
    x_train, y_train, x_test, y_test = load_mnist_data(args)
    model = create_mnist_model(params)

    model.fit(x_train, y_train, batch_size=args.batch_size, epochs=args.epochs, verbose=1,
        validation_data=(x_test, y_test), callbacks=[SendMetrics()])

    _, acc = model.evaluate(x_test, y_test, verbose=0)

class SendMetrics(keras.callbacks.Callback):
    def on_epoch_end(self, epoch, logs={}):
        nni.report_intermediate_result(logs)

def train(args, params):
    x_train, y_train, x_test, y_test = load_mnist_data(args)
    model = create_mnist_model(params)

    model.fit(x_train, y_train, batch_size=args.batch_size, epochs=args.epochs, verbose=1,
        validation_data=(x_test, y_test), callbacks=[SendMetrics()])

    _, acc = model.evaluate(x_test, y_test, verbose=0)
    nni.report_final_result(acc)

import argparse
import logging

import keras
import numpy as np
from keras import backend as K
from keras.datasets import mnist
from keras.layers import Conv2D, Dense, Flatten, MaxPooling2D
from keras.models import Sequential

import nni

LOG = logging.getLogger('mnist_keras')
K.set_image_data_format('channels_last')

H, W = 28, 28
NUM_CLASSES = 10

def create_mnist_model(hyper_params, input_shape=(H, W, 1), num_classes=NUM_CLASSES):
    '''
    Create simple convolutional model
    '''
    layers = [
        Conv2D(32, kernel_size=(3, 3), activation='relu', input_shape=input_shape),
        Conv2D(64, (3, 3), activation='relu'),
        MaxPooling2D(pool_size=(2, 2)),
        Flatten(),
        Dense(100, activation='relu'),
        Dense(num_classes, activation='softmax')
    ]

    model = Sequential(layers)

    if hyper_params['optimizer'] == 'Adam':
        optimizer = keras.optimizers.Adam(lr=hyper_params['learning_rate'])
    else:
        optimizer = keras.optimizers.SGD(lr=hyper_params['learning_rate'], momentum=0.9)
    model.compile(loss=keras.losses.categorical_crossentropy, optimizer=optimizer, metrics=['accuracy'])

    return model

def load_mnist_data(args):
    '''
    Load MNIST dataset
    '''
    (x_train, y_train), (x_test, y_test) = mnist.load_data()

    x_train = (np.expand_dims(x_train, -1).astype(np.float) / 255.)[:args.num_train]
    x_test = (np.expand_dims(x_test, -1).astype(np.float) / 255.)[:args.num_test]
    y_train = keras.utils.to_categorical(y_train, NUM_CLASSES)[:args.num_train]
    y_test = keras.utils.to_categorical(y_test, NUM_CLASSES)[:args.num_test]

    LOG.debug('x_train shape: %s', (x_train.shape,))
    LOG.debug('x_test shape: %s', (x_test.shape,))

    return x_train, y_train, x_test, y_test

class SendMetrics(keras.callbacks.Callback):
    '''
    Keras callback to send metrics to NNI framework
    '''
    def on_epoch_end(self, epoch, logs={}):
        '''
        Run on end of each epoch
        '''
        LOG.debug(logs)
        nni.report_intermediate_result(logs)

def train(args, params):
    '''
    Train model
    '''
    x_train, y_train, x_test, y_test = load_mnist_data(args)
    model = create_mnist_model(params)

    model.fit(x_train, y_train, batch_size=args.batch_size, epochs=args.epochs, verbose=1,
        validation_data=(x_test, y_test), callbacks=[SendMetrics()])

    _, acc = model.evaluate(x_test, y_test, verbose=0)
    LOG.debug('Final result is: %d', acc)
    nni.report_final_result(acc)

def generate_default_params():
    '''
    Generate default hyper parameters
    '''
    return {
        'optimizer': 'Adam',
        'learning_rate': 0.001
    }

if __name__ == '__main__':
    PARSER = argparse.ArgumentParser()
    PARSER.add_argument("--batch_size", type=int, default=200, help="batch size", required=False)
    PARSER.add_argument("--epochs", type=int, default=10, help="Train epochs", required=False)
    PARSER.add_argument("--num_train", type=int, default=1000, help="Number of train samples to be used, maximum 60000", required=False)
    PARSER.add_argument("--num_test", type=int, default=1000, help="Number of test samples to be used, maximum 10000", required=False)

    ARGS, UNKNOWN = PARSER.parse_known_args()

    try:
        # get parameters from tuner
        RECEIVED_PARAMS = nni.get_next_parameter()
        LOG.debug(RECEIVED_PARAMS)
        PARAMS = generate_default_params()
        PARAMS.update(RECEIVED_PARAMS)
        # train
        train(ARGS, PARAMS)
    except Exception as e:
        LOG.exception(e)
        raise

https://github.com/microsoft/nni/tree/master/examples/trials

最后编辑于：2024-09-09 20:06:09