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

NNI 自动化机器学习工具包

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

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

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 

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

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

NNI 运行机制

此外，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 代码文件的目录。

实际上，必须存在的是 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 可视化面板。

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

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

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