使用多进程时无法复现依赖scikit-learn与numpy的代码问询

你遇到的情况很典型：自定义了DecisionTreeClassifier的子类，在fit方法里用np.random.uniform生成随机权重，当cross_validate的n_jobs=1时结果完全可复现，但改成多进程（n_jobs=-1或2）就不行了；但Sklearn内置的比如RandomForestClassifier这类支持多进程的估计器却能正常复现。你想知道怎么在Sklearn里同时用Numpy随机调用和多进程，还能保持完全可复现性。

先看你给出的代码问题点：

import numpy as np
from sklearn.tree import DecisionTreeClassifier
from sklearn.datasets import load_iris
from sklearn.model_selection import cross_validate,RepeatedStratifiedKFold
class DecisionTree(DecisionTreeClassifier):
    def fit(self,X,Y):
        # 这里用了全局的np.random，多进程下会出问题
        weight = np.random.uniform(size=Y.shape)
        return super().fit(X,Y,sample_weight=weight)
def main():
    X,Y = load_iris(return_X_y=True)
    rks = RepeatedStratifiedKFold(n_repeats=2,n_splits=5,random_state=42)
    clf = DecisionTree(random_state=42)
    res = cross_validate(clf,X,Y,cv=rks,n_jobs=2)['test_score']*100
    return res.mean(),res.std()
if __name__=='__main__':
    np.random.seed(42)
    print(main())

解决方案 #

核心原因：全局随机状态的进程不安全问题 #

np.random.seed()设置的是全局随机状态，多进程环境下，每个子进程会继承主进程的随机状态，但进程的执行顺序是不确定的，多个进程同时调用np.random会打乱随机序列的生成顺序，导致结果不可复现。而Sklearn内置的估计器（比如RandomForest）之所以能在多进程下复现，是因为它们每个组件都有独立的随机状态，由主估计器的random_state衍生而来，不会依赖全局的np.random。

具体解决方法：给自定义估计器添加独立的随机生成器 #

你需要修改自定义类，让它创建并使用自己的numpy.random.RandomState（或新版本的numpy.random.Generator）实例，而不是依赖全局的np.random。这样每个估计器实例（包括多进程中复制的实例）都有独立的随机状态，互不干扰。

修改后的代码如下：

import numpy as np
from sklearn.tree import DecisionTreeClassifier
from sklearn.datasets import load_iris
from sklearn.model_selection import cross_validate,RepeatedStratifiedKFold

class DecisionTree(DecisionTreeClassifier):
    def __init__(self, random_state=None, **kwargs):
        # 调用父类的初始化方法，传入random_state
        super().__init__(random_state=random_state, **kwargs)
        # 创建独立的随机生成器，基于传入的random_state初始化
        if isinstance(random_state, np.random.RandomState):
            self.rng = random_state
        elif random_state is not None:
            self.rng = np.random.RandomState(random_state)
        else:
            self.rng = np.random.RandomState()

    def fit(self,X,Y):
        # 使用自己的随机生成器生成权重，而非全局的np.random
        weight = self.rng.uniform(size=Y.shape)
        return super().fit(X,Y,sample_weight=weight)

def main():
    X,Y = load_iris(return_X_y=True)
    rks = RepeatedStratifiedKFold(n_repeats=2,n_splits=5,random_state=42)
    clf = DecisionTree(random_state=42)
    res = cross_validate(clf,X,Y,cv=rks,n_jobs=2)['test_score']*100
    return res.mean(),res.std()

if __name__=='__main__':
    # 这里可以不用全局seed了，因为我们用了独立的随机生成器
    # np.random.seed(42)
    print(main())

为什么这样能解决问题？ #

当cross_validate使用多进程时，每个子进程会复制一份clf实例，而每个复制的实例都带着自己初始化好的self.rng（基于random_state=42生成）。这样每个进程里的fit方法都会使用独立的随机序列，不会因为进程间的执行顺序打乱随机状态，从而保证结果完全可复现。

额外注意事项 #

• 如果你用的是Numpy 1.17+版本，推荐使用np.random.default_rng()创建Generator实例，它比RandomState更安全、功能更全，用法类似：

  self.rng = np.random.default_rng(random_state)
  

• 不要在自定义估计器里使用全局的np.random调用，所有随机操作都应该通过自己的随机生成器实例完成。
• 确保Sklearn的交叉验证器（比如RepeatedStratifiedKFold）也设置了random_state，这样划分数据集的过程也是可复现的。

内容的提问来源于stack exchange，提问作者Shihab Shahriar Khan