基于门控残差和变量选择网络的分类
作者: Khalid Salama
创建日期: 2021/02/10
最后修改: 2021/02/10
描述: 使用门控残差和变量选择网络进行收入水平预测。
简介
此示例演示了使用门控残差网络（GRN）和变量选择网络（VSN），由Bryan Lim等人在 可解释的多时间跨度时间序列预测的时间融合变换器 (TFT)中提出，进行结构化数据分类。GRN使模型能够在需要的地方应用非线性处理。VSN允许模型软性去除任何可能对性能产生负面影响的不必要噪声输入。这些技术共同帮助提高深度神经网络模型的学习能力。
请注意，此示例仅实现了论文中描述的GRN和VSN组件，而不是整个TFT模型，因为GRN和VSN在结构化数据学习任务中可以单独使用。
要运行代码，您需要使用TensorFlow 2.3或更高版本。
数据集
此示例使用由加州大学欧文分校机器学习库提供的 美国人口普查收入数据集。 任务是进行二元分类，以确定一个人是否年收入超过50K美元。
数据集包含约300K个实例，具有41个输入特征：7个数值特征和34个分类特征。
设置
import math
import numpy as np
import pandas as pd
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers
准备数据
首先，我们从UCI机器学习库加载数据到Pandas DataFrame中。
# 列名。
CSV_HEADER = [
"age",
"class_of_worker",
"detailed_industry_recode",
"detailed_occupation_recode",
"education",
"wage_per_hour",
"enroll_in_edu_inst_last_wk",
"marital_stat",
"major_industry_code",
"major_occupation_code",
"race",
"hispanic_origin",
"sex",
"member_of_a_labor_union",
"reason_for_unemployment",
"full_or_part_time_employment_stat",
"capital_gains",
"capital_losses",
"dividends_from_stocks",
"tax_filer_stat",
"region_of_previous_residence",
"state_of_previous_residence",
"detailed_household_and_family_stat",
"detailed_household_summary_in_household",
"instance_weight",
"migration_code-change_in_msa",
"migration_code-change_in_reg",
"migration_code-move_within_reg",
"live_in_this_house_1_year_ago",
"migration_prev_res_in_sunbelt",
"num_persons_worked_for_employer",
"family_members_under_18",
"country_of_birth_father",
"country_of_birth_mother",
"country_of_birth_self",
"citizenship",
"own_business_or_self_employed",
"fill_inc_questionnaire_for_veterans_admin",
"veterans_benefits",
"weeks_worked_in_year",
"year",
"income_level",
]
data_url = "https://archive.ics.uci.edu/ml/machine-learning-databases/census-income-mld/census-income.data.gz"
data = pd.read_csv(data_url, header=None, names=CSV_HEADER)
test_data_url = "https://archive.ics.uci.edu/ml/machine-learning-databases/census-income-mld/census-income.test.gz"
test_data = pd.read_csv(test_data_url, header=None, names=CSV_HEADER)
print(f"数据形状: {data.shape}")
print(f"测试数据形状: {test_data.shape}")
数据形状: (199523, 42)
测试数据形状: (99762, 42)
我们将目标列从字符串转换为整数。
data["income_level"] = data["income_level"].apply(
lambda x: 0 if x == " - 50000." else 1
)
test_data["income_level"] = test_data["income_level"].apply(
lambda x: 0 if x == " - 50000." else 1
)
然后，我们将数据集分割为训练集和验证集。
random_selection = np.random.rand(len(data.index)) <= 0.85
train_data = data[random_selection]
valid_data = data[~random_selection]
最后，我们将训练和测试数据分割本地存储为CSV文件。
train_data_file = "train_data.csv"
valid_data_file = "valid_data.csv"
test_data_file = "test_data.csv"
train_data.to_csv(train_data_file, index=False, header=False) # 将训练数据保存为CSV文件
valid_data.to_csv(valid_data_file, index=False, header=False) # 将验证数据保存为CSV文件
test_data.to_csv(test_data_file, index=False, header=False) # 将测试数据保存为CSV文件
定义数据集元数据
在这里，我们定义数据集的元数据，这将有助于读取和解析数据为输入特征，并根据它们的类型对输入特征进行编码。
# 目标特征名称。
TARGET_FEATURE_NAME = "income_level"
# 权重列名称。
WEIGHT_COLUMN_NAME = "instance_weight"
# 数值特征名称。
NUMERIC_FEATURE_NAMES = [
"age",
"wage_per_hour",
"capital_gains",
"capital_losses",
"dividends_from_stocks",
"num_persons_worked_for_employer",
"weeks_worked_in_year",
]
# 分类特征及其词汇表。
# 请注意，我们在所有分类特征值前添加“v=”作为前缀，以确保
# 它们被视为字符串。
CATEGORICAL_FEATURES_WITH_VOCABULARY = {
feature_name: sorted([str(value) for value in list(data[feature_name].unique())])
for feature_name in CSV_HEADER
if feature_name
not in list(NUMERIC_FEATURE_NAMES + [WEIGHT_COLUMN_NAME, TARGET_FEATURE_NAME])
}
# 所有特征名称。
FEATURE_NAMES = NUMERIC_FEATURE_NAMES + list(
CATEGORICAL_FEATURES_WITH_VOCABULARY.keys()
)
# 特征默认值。
COLUMN_DEFAULTS = [
[0.0]
if feature_name in NUMERIC_FEATURE_NAMES + [TARGET_FEATURE_NAME, WEIGHT_COLUMN_NAME]
else ["NA"]
for feature_name in CSV_HEADER
]
为训练和评估创建一个 tf.data.Dataset
我们创建一个输入函数来读取和解析文件，并将特征和标签转换为
[tf.data.Dataset](https://www.tensorflow.org/api_docs/python/tf/data/Dataset) 以进行训练和评估。
from tensorflow.keras.layers import StringLookup
def process(features, target):
for feature_name in features:
if feature_name in CATEGORICAL_FEATURES_WITH_VOCABULARY:
# 将分类特征值转换为字符串。
features[feature_name] = tf.cast(features[feature_name], tf.dtypes.string)
# 获取实例权重。
weight = features.pop(WEIGHT_COLUMN_NAME)
return features, target, weight
def get_dataset_from_csv(csv_file_path, shuffle=False, batch_size=128):
dataset = tf.data.experimental.make_csv_dataset(
csv_file_path,
batch_size=batch_size,
column_names=CSV_HEADER,
column_defaults=COLUMN_DEFAULTS,
label_name=TARGET_FEATURE_NAME,
num_epochs=1,
header=False,
shuffle=shuffle,
).map(process)
return dataset
创建模型输入
def create_model_inputs():
inputs = {}
for feature_name in FEATURE_NAMES:
if feature_name in NUMERIC_FEATURE_NAMES:
inputs[feature_name] = layers.Input(
name=feature_name, shape=(), dtype=tf.float32
)
else:
inputs[feature_name] = layers.Input(
name=feature_name, shape=(), dtype=tf.string
)
return inputs
编码输入特征
对于分类特征，我们使用 layers.Embedding 进行编码，使用
encoding_size 作为嵌入维度。对于数值特征，
我们使用 layers.Dense 应用线性变换，将每个特征映射到
encoding_size 维向量中。因此，所有编码特征将具有相同的维数。
def encode_inputs(inputs, encoding_size):
encoded_features = []
for feature_name in inputs:
if feature_name in CATEGORICAL_FEATURES_WITH_VOCABULARY:
vocabulary = CATEGORICAL_FEATURES_WITH_VOCABULARY[feature_name]
# 创建查找，将字符串值转换为整数索引。
# 由于我们不使用掩码标记，也不期望任何超出词汇表
# (oov) 的令牌，因此我们将 mask_token 设置为 None，num_oov_indices 设置为 0。
index = StringLookup(
vocabulary=vocabulary, mask_token=None, num_oov_indices=0
)
# 将字符串输入值转换为整数索引。
value_index = index(inputs[feature_name])
# 创建一个具有指定维度的嵌入层
embedding_ecoder = layers.Embedding(
input_dim=len(vocabulary), output_dim=encoding_size
)
# 将索引值转换为嵌入表示。
encoded_feature = embedding_ecoder(value_index)
else:
# 通过线性变换将数值特征投影到 encoding_size。
encoded_feature = tf.expand_dims(inputs[feature_name], -1)
encoded_feature = layers.Dense(units=encoding_size)(encoded_feature)
encoded_features.append(encoded_feature)
return encoded_features
实现门控线性单元
门控线性单元 (GLUs) 提供了灵活性，以抑制与特定任务无关的输入。
class GatedLinearUnit(layers.Layer):
def __init__(self, units):
super().__init__()
self.linear = layers.Dense(units)
self.sigmoid = layers.Dense(units, activation="sigmoid")
def call(self, inputs):
return self.linear(inputs) * self.sigmoid(inputs) # 计算线性和sigmoid的乘积
实现门控残差网络
门控残差网络 (GRN) 的工作原理如下：
- 对输入应用非线性 ELU 转换。
- 应用线性转换，随后进行 dropout。
- 应用 GLU 并将原始输入加到 GLU 的输出上，以执行跳过（残差）连接。
- 应用层归一化并生成输出。
class GatedResidualNetwork(layers.Layer):
def __init__(self, units, dropout_rate):
super().__init__()
self.units = units
self.elu_dense = layers.Dense(units, activation="elu")
self.linear_dense = layers.Dense(units)
self.dropout = layers.Dropout(dropout_rate)
self.gated_linear_unit = GatedLinearUnit(units)
self.layer_norm = layers.LayerNormalization()
self.project = layers.Dense(units)
def call(self, inputs):
x = self.elu_dense(inputs)
x = self.linear_dense(x)
x = self.dropout(x)
if inputs.shape[-1] != self.units:
inputs = self.project(inputs)
x = inputs + self.gated_linear_unit(x)
x = self.layer_norm(x)
return x
实现变量选择网络
变量选择网络 (VSN) 的工作原理如下：
- 对每个特征单独应用 GRN。
- 在所有特征的拼接上应用 GRN，随后进行 softmax 以生成特征权重。
- 生成单独 GRN 输出的加权和。
注意，VSN 的输出为 [batch_size, encoding_size]，与输入特征的数量无关。
class VariableSelection(layers.Layer):
def __init__(self, num_features, units, dropout_rate):
super().__init__()
self.grns = list()
# 为每个特征独立创建一个 GRN
for idx in range(num_features):
grn = GatedResidualNetwork(units, dropout_rate)
self.grns.append(grn)
# 为所有特征的拼接创建一个 GRN
self.grn_concat = GatedResidualNetwork(units, dropout_rate)
self.softmax = layers.Dense(units=num_features, activation="softmax")
def call(self, inputs):
v = layers.concatenate(inputs)
v = self.grn_concat(v)
v = tf.expand_dims(self.softmax(v), axis=-1)
x = []
for idx, input in enumerate(inputs):
x.append(self.grns[idx](input))
x = tf.stack(x, axis=1)
outputs = tf.squeeze(tf.matmul(v, x, transpose_a=True), axis=1)
return outputs
创建门控残差和变量选择网络模型
def create_model(encoding_size):
inputs = create_model_inputs()
feature_list = encode_inputs(inputs, encoding_size)
num_features = len(feature_list)
features = VariableSelection(num_features, encoding_size, dropout_rate)(
feature_list
)
outputs = layers.Dense(units=1, activation="sigmoid")(features)
model = keras.Model(inputs=inputs, outputs=outputs)
return model
编译、训练和评估模型
learning_rate = 0.001
dropout_rate = 0.15
batch_size = 265
num_epochs = 20
encoding_size = 16
model = create_model(encoding_size)
model.compile(
optimizer=keras.optimizers.Adam(learning_rate=learning_rate),
loss=keras.losses.BinaryCrossentropy(),
metrics=[keras.metrics.BinaryAccuracy(name="accuracy")],
)
# 创建早期停止回调。
early_stopping = tf.keras.callbacks.EarlyStopping(
monitor="val_loss", patience=5, restore_best_weights=True
)
print("开始训练模型...")
train_dataset = get_dataset_from_csv(
train_data_file, shuffle=True, batch_size=batch_size
)
valid_dataset = get_dataset_from_csv(valid_data_file, batch_size=batch_size)
model.fit(
train_dataset,
epochs=num_epochs,
validation_data=valid_dataset,
callbacks=[early_stopping],
)
print("模型训练完成。")
print("评估模型性能...")
test_dataset = get_dataset_from_csv(test_data_file, batch_size=batch_size)
_, accuracy = model.evaluate(test_dataset)
print(f"测试准确率: {round(accuracy * 100, 2)}%")
开始训练模型...
第 1 轮/20
640/640 [==============================] - 31s 29ms/步 - 损失: 253.8570 - 精度: 0.9468 - 验证损失: 229.4024 - 验证精度: 0.9495
第 2 轮/20
640/640 [==============================] - 17s 25ms/步 - 损失: 229.9359 - 精度: 0.9497 - 验证损失: 223.4970 - 验证精度: 0.9505
第 3 轮/20
640/640 [==============================] - 17s 25ms/步 - 损失: 225.5644 - 精度: 0.9504 - 验证损失: 222.0078 - 验证精度: 0.9515
第 4 轮/20
640/640 [==============================] - 16s 25ms/步 - 损失: 222.2086 - 精度: 0.9512 - 验证损失: 218.2707 - 验证精度: 0.9522
第 5 轮/20
640/640 [==============================] - 17s 25ms/步 - 损失: 218.0359 - 精度: 0.9523 - 验证损失: 217.3721 - 验证精度: 0.9528
第 6 轮/20
640/640 [==============================] - 17s 26ms/步 - 损失: 214.8348 - 精度: 0.9529 - 验证损失: 210.3546 - 验证精度: 0.9543
第 7 轮/20
640/640 [==============================] - 17s 26ms/步 - 损失: 213.0984 - 精度: 0.9534 - 验证损失: 210.2881 - 验证精度: 0.9544
第 8 轮/20
640/640 [==============================] - 17s 26ms/步 - 损失: 211.6379 - 精度: 0.9538 - 验证损失: 209.3327 - 验证精度: 0.9550
第 9 轮/20
640/640 [==============================] - 17s 26ms/步 - 损失: 210.7283 - 精度: 0.9541 - 验证损失: 209.5862 - 验证精度: 0.9543
第 10 轮/20
640/640 [==============================] - 17s 26ms/步 - 损失: 209.9062 - 精度: 0.9538 - 验证损失: 210.1662 - 验证精度: 0.9537
第 11 轮/20
640/640 [==============================] - 16s 25ms/步 - 损失: 209.6323 - 精度: 0.9540 - 验证损失: 207.9528 - 验证精度: 0.9552
第 12 轮/20
640/640 [==============================] - 16s 25ms/步 - 损失: 208.7843 - 精度: 0.9544 - 验证损失: 207.5303 - 验证精度: 0.9550
第 13 轮/20
640/640 [==============================] - 21s 32ms/步 - 损失: 207.9983 - 精度: 0.9544 - 验证损失: 206.8800 - 验证精度: 0.9557
第 14 轮/20
640/640 [==============================] - 18s 28ms/步 - 损失: 207.2104 - 精度: 0.9544 - 验证损失: 216.0859 - 验证精度: 0.9535
第 15 轮/20
640/640 [==============================] - 16s 25ms/步 - 损失: 207.2254 - 精度: 0.9543 - 验证损失: 206.7765 - 验证精度: 0.9555
第 16 轮/20
640/640 [==============================] - 16s 25ms/步 - 损失: 206.6704 - 精度: 0.9546 - 验证损失: 206.7508 - 验证精度: 0.9560
第 17 轮/20
640/640 [==============================] - 19s 30ms/步 - 损失: 206.1322 - 精度: 0.9545 - 验证损失: 205.9638 - 验证精度: 0.9562
第 18 轮/20
640/640 [==============================] - 21s 31ms/步 - 损失: 205.4764 - 精度: 0.9545 - 验证损失: 206.0258 - 验证精度: 0.9561
第 19 轮/20
640/640 [==============================] - 16s 25ms/步 - 损失: 204.3614 - 精度: 0.9550 - 验证损失: 207.1424 - 验证精度: 0.9560
第 20 轮/20
640/640 [==============================] - 16s 25ms/步 - 损失: 203.9543 - 精度: 0.9550 - 验证损失: 206.4697 - 验证精度: 0.9554
模型训练完成。
评估模型性能...
377/377 [==============================] - 4s 11ms/步 - 损失: 204.5099 - 精度: 0.9547
测试精度: 95.47%
为了提高模型的学习能力，您可以尝试增加encoding_size值，或者在VSN层上堆叠多个GRN层。这可能需要同时增加dropout_rate值以避免过拟合。
示例可在HuggingFace上找到
| 训练模型 | 演示 |
|---|---|
 |
 |