只有在一个线程上运行时才可能实现 Keras 模型的重现性？

Posted 2023-03-12

【中文标题】只有在一个线程上运行时才可能实现 Keras 模型的重现性？

【英文标题】：Keras Model reproducability only possible when running on one Thread?

【发布时间】：2020-10-28 09:56:05

【问题描述】：

在过去的几个小时里，我一直在编写我的 keras/tensorflow 代码，通过播种每个使用的随机生成器来获得可重现的结果。现在我的解决方案有效，但奇怪的是，只有当我使用以下方法在单个线程上运行代码时：

from keras import backend as K
config = tf.ConfigProto(intra_op_parallelism_threads=1, inter_op_parallelism_threads=1)
sess = tf.Session(graph=tf.get_default_graph(), config=config)
K.set_session(sess)

我无法解释这种行为，所以我想知道您对此的看法。另外为了进一步了解，我将在下面发布我的完整代码：

import tensorflow as tf
random.seed(seed_value)
np.random.seed(seed_value)
tf.set_random_seed(seed_value)

from keras import backend as K
config = tf.ConfigProto(intra_op_parallelism_threads=1, inter_op_parallelism_threads=1)
sess = tf.Session(graph=tf.get_default_graph(), config=config)
K.set_session(sess)
""""""
import pandas as pd # data processing, CSV file I/O (e.g. pd.read_csv)

import matplotlib.pyplot as plt
from sklearn import preprocessing, model_selection
from sklearn.decomposition import PCA
from keras.models import load_model
from keras.utils import np_utils
from sklearn.ensemble import RandomForestClassifier
from sklearn.preprocessing import LabelEncoder, StandardScaler
from keras.utils.np_utils import to_categorical
from sklearn.utils import shuffle
from sklearn.metrics import confusion_matrix

from TimingCallback import TimeHistory


def train():
    files = ['RAW_combined_shuffled.csv']
    # select = ['html_tag_script', 'js_max_value_assignments', 'url_found_scripttags', 'url_param_count_"', 'label']
    # read_files = (pd.read_csv(f, usecols=select) for f in files)      # Nur ausgewählte features einlesen
    read_files = (pd.read_csv(f) for f in files)                        # Alle Features einlesen
    data = pd.concat(read_files, ignore_index=True)
    data = data.drop(['data'], axis=1)  # Einzelne columns rauswerfen // durch KNIME erledigt
    # data = shuffle(data)  # Reihenfolge der Datensätze randomisieren  //durch KNIME erledigt

    i = 100
    data_to_predict = data[:i].reset_index(drop=True)  # Daten für Testen des Models raussplitten (anfang bis i)
    real_label = data_to_predict.label
    real_label = np.array(real_label)
    prediction = np.array(data_to_predict.drop(['label'], axis=1))

    data = data[i:].reset_index(drop=True)  # Daten für Training und Test raussplitten (i bis Ende)

    X = data.drop(['label'], axis=1)  # x sind alle columns außer output label
    X = np.array(X)
    Y = data['label']  # y ist column mit output label

    # Transform name species into numerical values
    encoder = LabelEncoder()
    encoder.fit(Y)
    Y = encoder.transform(Y)
    Y = np_utils.to_categorical(Y)

    # We have  classes : the output looks like:
    # 0,1 : Class 1
    # 1,0 : Class 2

    # Trainings- und Testdaten aufteilen; random_state ist seed für Zufallsgenerierung, welche Datensätze train und welche test sind
    train_x, test_x, train_y, test_y = model_selection.train_test_split(X, Y, test_size=0.3, random_state=5, shuffle=False)

    input_dim = len(data.columns) - 1
    print(input_dim)

    callback = tf.keras.callbacks.EarlyStopping(monitor='val_loss', patience=10)
    time_callback = TimeHistory()

    model = tf.keras.Sequential()
    model.add(tf.keras.layers.Dense(8, input_dim=input_dim, activation='sigmoid'))
    model.add(tf.keras.layers.Dense(10, activation='sigmoid'))
    model.add(tf.keras.layers.Dense(10, activation='sigmoid'))
    model.add(tf.keras.layers.Dense(10, activation='sigmoid'))
    model.add(tf.keras.layers.Dense(2, activation='softmax'))

    model.compile(loss='binary_crossentropy', optimizer='adam', metrics=['accuracy'])
    # Start the Training for number of epochs
    # validation_split nimmt Anteil von Traindaten und nutzt diesen Teil zum validieren (bestimmung von val_acc und val_loss) am Ende jeder Epoche
    # verbose gibt anzeige an; 0= ohne anzeige, 1 = mit einem fortschrittsbalken, 2 = fortschrittsbalken pro epoche
    history = model.fit(train_x, train_y, validation_split=0.33, epochs=1000, batch_size=1000, verbose=1, callbacks=[callback, time_callback], shuffle=False)

    scores = model.evaluate(test_x, test_y)

如您所见，我还在 model.fit 方法和 train_test_split 中禁用了改组选项。由于我想进一步提高训练性能，我通常希望使用多个线程并因此使用 CPU 内核。

【参考方案1】：

是的，这是有道理的，问题由两件事组成：

浮点数只是实数的近似值，特别是因为并非所有数字都可以表示，而且加法不是关联的。例如 (a + b) + c != a + (b + c)，因为每次加法都可以舍入到最接近的浮点数，这会产生略有不同的结果。例如在python中：

(0.1 + 0.2) + 0.3 结果为 0.6000000000000001。

0.1 + (0.2 + 0.3) 结果为 0.6。

使用多线程并行计算会在进程中引入更多随机性，因为您现在涉及调度程序和其他进程。当从多个线程添加变量时会出现问题，例如在组合来自多个线程的结果时，您通常使用锁并写入同一个变量，但每个线程执行此操作的顺序未定义并更改结果。

这也发生在 GPU 内部，所以不幸的是，如果您想要可重现的结果，您需要尽量减少跨线程的并行使用（因此没有多线程）。

【讨论】：

但是当浮点数在逗号后四舍五入到相同的数字时，它们的加法不应该是累积的吗？我看不出使用多个线程会如何引发不同的舍入。我可以看到死锁或不一致的窗口，但舍入错误我无法绕着 atm 转

@marcels93 我的意思是关联的，而不是可交换的，不是线程引发不同的舍入，是操作的顺序没有定义，并且在浮动中不是关联的（当添加三个或更多数字时）点算术。