- 為了記錄Keras基本API，本博客展示一次線性回歸極簡(jiǎn)機(jī)器學(xué)習(xí)全流程。
-
建立模型
定義一個(gè)簡(jiǎn)單的線性回歸模型，使用 Keras 模塊來構(gòu)建和編譯模型。以最簡(jiǎn)單的單層網(wǎng)絡(luò)為例，設(shè)置1個(gè)輸出節(jié)點(diǎn)，輸入節(jié)點(diǎn)的數(shù)量為特征的種數(shù)。
keras.Sequential(layers=None, trainable=True, name=None)是models中的一個(gè)類，表示序貫?zāi)Ｐ停ǜ鲗觾H線性堆疊，無跨層連接）。
注意到定義時(shí)沒有層，那么如何添加層？可以用其自帶的方法Sequential.add(layer, rebuild=True)。其各個(gè)參數(shù)代表含義如下：
- layer: layer instance.
什么是layer實(shí)例？一層layer由一個(gè)IO為張量的計(jì)算函數(shù)（層的call方法）和一些狀態(tài)組成，這些狀態(tài)保存在TensorFlow變量中（層的weights）。網(wǎng)絡(luò)層通用的讀寫weights方法有
layer.get_weights(): 以含有Numpy矩陣的列表形式返回層的權(quán)重。layer.set_weights(weights): 從含有Numpy矩陣的列表中設(shè)置層的權(quán)重（與get_weights的輸出形狀相同）。
當(dāng)且僅當(dāng)它作為模型的第一層時(shí)，需傳入輸入尺寸：
-
input_shape: （整數(shù)元組，不包括樣本數(shù)的軸）
以keras.layers中的全連接層為例
Dense
keras.layers.Dense(units, activation=None, use_bias=True, kernel_initializer='glorot_uniform', bias_initializer='zeros', kernel_regularizer=None, bias_regularizer=None, activity_regularizer=None, kernel_constraint=None, bias_constraint=None)對(duì)于這種2D層，輸入尺寸傳入方式也可更為
- input_dim: 即輸入特征種類數(shù)量，是整數(shù)類型。
接下來需傳入一些重要參數(shù)：
- units: 正整數(shù)，輸出空間維度。
- activation: 激活函數(shù) (詳見 activations)。 若None，則不使用激活函數(shù) (即，線性激活:
a(x) = x)。
要實(shí)現(xiàn)生成一個(gè)指定參數(shù)的模型，既可以通過將網(wǎng)絡(luò)層實(shí)例的列表傳遞給 Sequential 的構(gòu)造器，來創(chuàng)建一個(gè) Sequential 模型：
Sequential
num_features = 1
model = Sequential([ Dense(1, input_shape=(num_feaetures,)), ])也可以簡(jiǎn)單地使用 .add() 方法將各層添加到模型中：
model = keras.models.Sequential()
# Describe the topography of the model.
# The topography of a simple linear regression model is a single node in a single layer.
model.add(keras.layers.Dense(units=1, input_shape=(num_features,)))在機(jī)器學(xué)習(xí)之前，需要把模型的topography編譯為Keras可以高效執(zhí)行的程序。使用Sequential 模型的compile 方法完成，它接收三個(gè)參數(shù)：
- 優(yōu)化器 optimizer。它可以是現(xiàn)有優(yōu)化器的字符串標(biāo)識(shí)符，如
rmsprop或adagrad；也可以是 Optimizer 類的實(shí)例，此時(shí)可以自定義傳參。
keras.optimizers類具有公共的參數(shù) clipnorm 和 clipvalue ，用于控制梯度裁剪（Gradient Clipping）。
而對(duì)于優(yōu)化器對(duì)象，以keras.optimizers.RMSprop(learning_rate=0.001, rho=0.9)為例
-
- learning_rate: float >= 0. 學(xué)習(xí)率。
- rho: float >= 0. RMSProp 梯度平方的移動(dòng)均值的衰減率。
詳見：optimizers。
- 損失函數(shù) loss，模型試圖最小化的目標(biāo)函數(shù)。它可以是現(xiàn)有損失函數(shù)的字符串標(biāo)識(shí)符，如
categorical_crossentropy或mse，也可以是一個(gè)損失函數(shù)。詳見：losses，Compile。 - 評(píng)估標(biāo)準(zhǔn) metrics：List of現(xiàn)有的標(biāo)準(zhǔn)的字符串標(biāo)識(shí)符，或
keras.metrics.Metric的實(shí)例，或自定義的評(píng)估標(biāo)準(zhǔn)函數(shù)。
對(duì)于任何分類問題，你都希望將其設(shè)置為 metrics = ['accuracy']。在下面的例子中，采用均方誤差作為評(píng)估標(biāo)準(zhǔn)：
keras.metrics.RootMeanSquaredError(name="root_mean_squared_error", dtype=None)Compile
model.compile(optimizer=keras.optimizers.RMSprop(learning_rate=my_learning_rate),
loss="mean_squared_error",
metrics=[keras.metrics.RootMeanSquaredError()])注意：Compile方法的metrics參數(shù)可接受的標(biāo)準(zhǔn)built-in評(píng)價(jià)函數(shù)詳見: metrics。上例中的均方誤差類是指標(biāo) - Keras 中文類的一個(gè)實(shí)例。
下面的程序創(chuàng)建了一個(gè)單層、輸出為一維的模型。
1 def build_model(my_learning_rate, num_features): 2 """Create and compile a simple linear regression model.""" 3 # Most simple keras models are sequential. 4 model = keras.models.Sequential() 5 6 # Describe the topography of the model. 7 # The topography of a simple linear regression model 8 # is a single node in a single layer. 9 model.add(keras.layers.Dense(units=1, 10 input_shape=(num_features,))) 11 12 # Compile the model topography into code that Keras can efficiently 13 # execute. Configure training to minimize the model's mean squared error. 14 model.compile(optimizer=keras.optimizers.RMSprop(learning_rate=my_learning_rate), 15 loss="mean_squared_error", 16 metrics=[keras.metrics.RootMeanSquaredError()]) 17 18 return model
2. 訓(xùn)練模型
Keras 模型在輸入數(shù)據(jù)和標(biāo)簽的 Numpy 矩陣上進(jìn)行訓(xùn)練。使用Sequential模型的fit方法完成：
fit
fit(x=None, y=None, batch_size=None, epochs=1, verbose=1, callbacks=None, validation_split=0.0, validation_data=None, shuffle=True, class_weight=None, sample_weight=None, initial_epoch=0, steps_per_epoch=None, validation_steps=None, validation_freq=1, max_queue_size=10, workers=1, use_multiprocessing=False)常用的參數(shù)包括：
- x: 輸入數(shù)據(jù)。可以是：
- 一個(gè) Numpy 數(shù)組（或類數(shù)組），等
- 一個(gè)將名稱映射到對(duì)應(yīng)array/tensor的字典（如果模型有已命名的輸入）
- y: 目標(biāo)數(shù)據(jù)。與輸入數(shù)據(jù)
x類似，它可以是 Numpy 數(shù)組（序列）等。 - batch_size: 整數(shù)或
None。每次梯度更新的樣本數(shù)。如果未指定，默認(rèn)為 32。未完待續(xù)。 - epochs: 整數(shù)。訓(xùn)練模型迭代輪次。一個(gè)輪次是在整個(gè)
x或y上的一輪迭代。 請(qǐng)注意，與initial_epoch一起，epochs被理解為「最終輪次」。 模型并不是訓(xùn)練了epochs輪，而是到第epochs輪停止訓(xùn)練。
該方法返回一個(gè) History 對(duì)象：其History.epoch屬性是各訓(xùn)練輪次的索引組成的列表；其 History.history屬性是一個(gè)字典，記錄連續(xù) epoch 訓(xùn)練以及驗(yàn)證集（如果適用）的損失和評(píng)估值，可以直接傳給pandas.DataFrame方法轉(zhuǎn)化。
更多信息，請(qǐng)查閱Sequential 順序模型 - 中文文檔。
訓(xùn)練結(jié)束后，可以用模型的get_weights() 方法查看模型參數(shù)（weights, bias, etc.），它返回一個(gè)flat list，元素依次為：首層的權(quán)重，首層的bias，第二層的權(quán)重，第二層的bias，…（某層的權(quán)重矩陣.shape=(該層的輸入維數(shù)，輸出維數(shù))）
3. Validate模型
使用Sequential模型的predict_on_batch(x)方法，其中
- x: 輸入數(shù)據(jù)，Numpy 數(shù)組（可用pandas的Series.values方法轉(zhuǎn)換）或列表（如果模型有多輸入）。
它返回預(yù)測(cè)值的NumPy數(shù)組。
- 為了補(bǔ)充Keras基本用法，本博客展示一次極簡(jiǎn)二元分類機(jī)器學(xué)習(xí)全流程。
1. 設(shè)置隨機(jī)數(shù)seed
為了使實(shí)驗(yàn)可復(fù)現(xiàn)，設(shè)置NumPy，backend和Python的random seed，一句就解決：
隨機(jī)種子
keras.utils.set_random_seed(42)2. 將數(shù)據(jù)貼標(biāo)簽（數(shù)值標(biāo)簽）并切分
對(duì)于二分類問題，可以把標(biāo)簽轉(zhuǎn)化為0和1
查看代碼
normalized_dataset['Class_Bool'] = (
normalized_dataset['Class'] == 'Cammeo'
).astype(int)
normalized_dataset.sample(10)上述程序的廣播計(jì)算此處不再贅述。Series.astype(dtype, copy=None, errors='raise') casts a pandas object to a specified dtype dtype, where
- dtype: str, data type, Series or Mapping of column name -> data type
Use a str, numpy.dtype, pandas.ExtensionDtype or Python type to cast entire pandas object to the same type.
-
copy: bool, return a copy when
copy=True. - errors: {‘raise’, ‘ignore’}, ’
Control raising of exceptions on invalid data for provided dtype.
-
raise: allow exceptions to be raised -
ignore: suppress exceptions. On error return original object.
-
切分?jǐn)?shù)據(jù)集
number_samples = len(normalized_dataset)
index_80th = round(number_samples * 0.8)
index_90th = index_80th + round(number_samples * 0.1)
shuffled_dataset = normalized_dataset.sample(frac=1, random_state=100)
train_data = shuffled_dataset.iloc[0:index_80th]
validation_data = shuffled_dataset.iloc[index_80th:index_90th]
test_data = shuffled_dataset.iloc[index_90th:]
test_data.head()為了防止標(biāo)簽泄露，通常的做法是分開存儲(chǔ)特征和標(biāo)簽：
查看代碼
label_columns = ['Class', 'Class_Bool']
train_features = train_data.drop(columns=label_columns)
train_labels = train_data['Class_Bool'].to_numpy()
validation_features = validation_data.drop(columns=label_columns)
validation_labels = validation_data['Class_Bool'].to_numpy()
test_features = test_data.drop(columns=label_columns)
test_labels = test_data['Class_Bool'].to_numpy()3. 訓(xùn)練
輸入特征
input_features = [
'Eccentricity',
'Major_Axis_Length',
'Area',
]可以用dataclass便捷保存實(shí)驗(yàn)參數(shù)及試驗(yàn)記錄。
數(shù)據(jù)類
import dataclasses
@dataclasses.dataclass()
class ExperimentSettings:
"""Lists the hyperparameters and input features used to train am model."""
learning_rate: float
number_epochs: int
batch_size: int
classification_threshold: float
input_features: list[str]
@dataclasses.dataclass()
class Experiment:
"""Stores the settings used for a training run and the resulting model."""
name: str
settings: ExperimentSettings
model: keras.Model
epochs: np.ndarray
metrics_history: keras.callbacks.History
def get_final_metric_value(self, metric_name: str) -> float:
"""Gets the final value of the given metric for this experiment."""
if metric_name not in self.metrics_history:
raise ValueError(
f'Unknown metric {metric_name}: available metrics are'
f' {list(self.metrics_history.columns)}'
)
return self.metrics_history[metric_name].iloc[-1]掌握注解最佳實(shí)踐和dataclasses --- 數(shù)據(jù)類可以輕松理解上述Python語法。
創(chuàng)建模型還可以通過定義模型輸入輸出的方式完成：例如，如果 a、b 和 c 是 Keras 張量，則可以執(zhí)行以下操作：model = Model(input=[a, b], output=c)。
首先，實(shí)例化輸入對(duì)象：
keras.Input(shape=None, batch_size=None, dtype=None, sparse=None, batch_shape=None, name=None, tensor=None,
)其中
- name: 層的可選名稱字符串。在模型中應(yīng)唯一（不要重復(fù)使用相同的名稱兩次）。如果未提供，它將自動(dòng)生成。
- shape: 形狀元組（整數(shù)或
None對(duì)象的元組），不包括批次大小。例如，shape=(32,)表示預(yù)期輸入將是 32 維向量的批次。此元組的元素可以是None；None元素表示形狀未知且可能變化的維度（例如，序列長(zhǎng)度）。
它返回一個(gè)Keras張量。
多個(gè)輸入對(duì)象（張量）可以拼接成一個(gè)輸入對(duì)象：keras.layers.Concatenate(axis=-1, **kwargs) 類實(shí)例化后，以多個(gè)張量的列表為輸入（這些張量除了要對(duì)齊/拼接的維度，其他維度都是同形的），返回拼接后的單個(gè)張量。
有了輸入張量，下來該確定輸出張量了。 如何得到輸出張量？你向一個(gè)層輸入，計(jì)算輸出就得到了么。
以常見的全連接神經(jīng)網(wǎng)絡(luò)層——密集層為例，Dense類的實(shí)例接受輸入為N維張量，形狀為：(batch_size, ..., input_dim)。最常見的情況是形狀為 (batch_size, input_dim) 的二維輸入。其計(jì)算前向傳播過程，輸出為N維張量，形狀為：(batch_size, ..., units)。例如，對(duì)于形狀為 (batch_size, input_dim) 的二維輸入，輸出的形狀將為 (batch_size, units)。
創(chuàng)建模型topography
def create_model(
settings: ExperimentSettings,
metrics: list[keras.metrics.Metric],
) -> keras.Model:
"""Create and compile a simple classification model."""
model_inputs = [
keras.Input(name=feature, shape=(1,))
for feature in settings.input_features
]
# Use a Concatenate layer to assemble the different inputs into a single
# tensor which will be given as input to the Dense layer.
# For example: [input_1[0][0], input_2[0][0]]
concatenated_inputs = keras.layers.Concatenate()(model_inputs)
dense = keras.layers.Dense(
units=1, input_shape=(1,), name='dense_layer', activation=keras.activations.sigmoid
)
model_output = dense(concatenated_inputs)
model = keras.Model(inputs=model_inputs, outputs=model_output)
# Call the compile method to transform the layers into a model that
# Keras can execute. Notice that we're using a different loss
# function for classification than for regression.
model.compile(
optimizer=keras.optimizers.RMSprop(
settings.learning_rate
),
loss=keras.losses.BinaryCrossentropy(),
metrics=metrics,
)
return model有細(xì)心的朋友就會(huì)問了，密集層的文檔里并未顯式給出Dense類的輸入?yún)?shù)name，為什么可以用在上段例程中呢？實(shí)際上，其基類
keras.layers.Layer(activity_regularizer=None, trainable=True, dtype=None, autocast=True, name=None, **kwargs)包含了這個(gè)輸入?yún)?shù)。
定義訓(xùn)練函數(shù)
訓(xùn)練函數(shù)
def train_model(
experiment_name: str,
model: keras.Model,
dataset: pd.DataFrame,
labels: np.ndarray,
settings: ExperimentSettings,
) -> Experiment:
"""Feed a dataset into the model in order to train it."""
# The x parameter of keras.Model.fit can be a list of arrays, where
# each array contains the data for one feature.
features = {
feature_name: np.array(dataset[feature_name])
for feature_name in settings.input_features
}
history = model.fit(
x=features,
y=labels,
batch_size=settings.batch_size,
epochs=settings.number_epochs,
)
return Experiment(
name=experiment_name,
settings=settings,
model=model,
epochs=history.epoch,
metrics_history=pd.DataFrame(history.history),
)
print('Defined the create_model and train_model functions.')開始主實(shí)驗(yàn)：給定實(shí)驗(yàn)參數(shù)
實(shí)驗(yàn)一參數(shù)
settings = ExperimentSettings(
learning_rate=0.001,
number_epochs=60,
batch_size=100,
classification_threshold=0.35,
input_features=input_features,
)
metrics = [
keras.metrics.BinaryAccuracy(
name='accuracy', threshold=settings.classification_threshold
),
keras.metrics.Precision(
name='precision', thresholds=settings.classification_threshold
),
keras.metrics.Recall(
name='recall', thresholds=settings.classification_threshold
),
keras.metrics.AUC(num_thresholds=100, name='auc'),
]上例傳遞給model.compile方法的實(shí)參metrics變成了keras.metrics.Metric實(shí)例：
keras.metrics.BinaryAccuracy(name="binary_accuracy", dtype=None, threshold=0.5)計(jì)算二分類準(zhǔn)確率，其中
- name: (Optional) metric實(shí)例對(duì)外名稱.
- dtype: (Optional) data type of the metric result.
- threshold: (Optional) Float.
keras.metrics.Precision(thresholds=None, top_k=None, class_id=None, name=None, dtype=None)計(jì)算精確度，其中
- thresholds: (Optional) A float value, or a Python list/tuple of float 介于
[0, 1].如果連同都沒有指定，默認(rèn)閾值為半。 - top_k: (Optional) An int value，缺省缺少設(shè)置。指定.如果不設(shè)定，那么只有你預(yù)測(cè)概率最高的那個(gè)類別對(duì)了，計(jì)算精確度時(shí)才會(huì)記為TP；否則就會(huì)從回歸概率最高的
top_k個(gè)類別中找有沒有標(biāo)簽類別，有的話記為TP。 - name: 同上
keras.metrics.Recall(thresholds=None, top_k=None, class_id=None, name=None, dtype=None)計(jì)算召回率，其中
- top_k: 類推同上。
- name: 同上。
- thresholds: 同上
keras.metrics.AUC( num_thresholds=200, curve="ROC", summation_method="interpolation", name=None, dtype=None, thresholds=None, multi_label=False, num_labels=None, label_weights=None, from_logits=False,
)計(jì)算ROC或者PR曲線的AUC，其中：
- num_thresholds: (Optional) 離散化曲線所使用的閾值點(diǎn)數(shù)。必須大于一
- name: (Optional) 同上。
這里需要注意的是，各個(gè)keras.metrics.Metric實(shí)例的輸入?yún)?shù)名有細(xì)微差別，如閾值，有的是threshold，有的是thresholds。
實(shí)驗(yàn)正文：
核心代碼
# Establish the model's topography.
model = create_model(settings, metrics)
# Train the model on the training set.
experiment = train_model(
'baseline', model, train_features, train_labels, settings
)可視化：定義基于matplotlib的繪畫函數(shù)：
繪圖函數(shù)
def plot_experiment_metrics(experiment: Experiment, metrics: list[str]):
"""Plot a curve of one or more metrics for different epochs."""
plt.figure(figsize=(12, 8))
for metric in metrics:
plt.plot(
experiment.epochs, experiment.metrics_history[metric], label=metric
)
plt.xlabel("Epoch")
plt.ylabel("Metric value")
plt.grid()
plt.legend()
print("Defined the plot_curve function.")作圖
作圖
# Plot metrics vs. epochs
plot_experiment_metrics(experiment, ['accuracy', 'precision', 'recall'])
plot_experiment_metrics(experiment, ['auc'])4. 測(cè)試
Model.evaluate( x=None, y=None, batch_size=None, verbose="auto", sample_weight=None, steps=None, callbacks=None, return_dict=False, **kwargs )其中，
- x: 輸入數(shù)據(jù)，它可以是：
- A NumPy array (or array-like), or a list of arrays (模型有多輸入的情況).
- A tensor, or a list of tensors (模型有多輸入的情況).
- A dict mapping input names to the corresponding array/tensors（模型有命名的輸入）.
- 等
- y: 標(biāo)簽值。It could be either NumPy array(s) or backend-native tensor(s).（未詳盡）
- batch_size: 整數(shù)或
None。每次計(jì)算的樣本數(shù)。如果未指定，默認(rèn)為 32。未完待續(xù)。 - verbose:
"auto", 0, 1, or 2. 0 = silent, 1 = progress bar, 2 = single line."auto"大多數(shù)情況下為1。 - return_dict: 如果為真，loss and metric results are returned as a dict, with each key being the name of the metric. 否則它們將返回一張列表。
它返回標(biāo)量test loss (模型僅一個(gè)輸出且沒有metrics) 或 list of 標(biāo)量 (模型有多輸出 and/or metrics)。 詳盡描述看evaluate的文檔。
據(jù)此，定義評(píng)估函數(shù)
測(cè)試函數(shù)
def evaluate_experiment( experiment: Experiment, test_dataset: pd.DataFrame, test_labels: np.array) -> dict[str, float]:
features = {
feature_name: np.array(test_dataset[feature_name])
for feature_name in experiment.settings.input_features
}
return experiment.model.evaluate( x=features, y=test_labels, batch_size=settings.batch_size, verbose=0, # Hide progress bar return_dict=True,
)進(jìn)行測(cè)試，并對(duì)比訓(xùn)練測(cè)試結(jié)果：
測(cè)試、比對(duì)
def compare_train_test(experiment: Experiment, test_metrics: dict[str, float]):
print('Comparing metrics between train and test:')
for metric, test_value in test_metrics.items():
print('------')
print(f'Train {metric}: {experiment.get_final_metric_value(metric):.4f}')
print(f'Test {metric}: {test_value:.4f}')
# Evaluate test metrics
test_metrics = evaluate_experiment(experiment, test_features, test_labels)
compare_train_test(experiment, test_metrics)
浙公網(wǎng)安備 33010602011771號(hào)