设置n_jobs≥2时keras.wrappers报无法pickle _thread.lock对象错误

这个问题我之前碰到过，本质是Keras模型内部包含无法被pickle序列化的线程锁对象，而当StackedRegressor设置n_jobs>1时，底层的joblib会默认用多进程来并行训练基模型——多进程需要把模型对象序列化后传递给子进程，这就触发了报错。n_jobs=1是单进程模式，不需要序列化，所以能正常运行。

下面是几个可行的解决方案，按简单到复杂排序：

方案1：切换到多线程后端（最推荐） #

多线程模式下，子线程共享父进程的内存空间，不需要序列化模型，完美避开pickle问题。你只需要在代码开头添加几行，强制joblib使用线程后端：

import joblib
# 全局设置joblib用threading后端，n_jobs设为你需要的数量
joblib.parallel_backend('threading', n_jobs=2)

如果civismlext的StackedRegressor支持直接传递joblib参数，也可以在初始化时指定：

stacked_reg = StackedRegressor(
    base_estimators=[...],  # 你的基模型列表（包括Keras模型）
    final_estimator=NonNegativeLinearRegression(),
    n_jobs=2,
    joblib_kwargs={'backend': 'threading'}  # 直接给joblib传参数
)

这个方案几乎不需要修改你的模型代码，适合大多数场景。

方案2：包装Keras模型为可Pickle对象 #

如果你的训练是CPU密集型，多线程受GIL限制导致性能不佳，需要用多进程的话，可以自定义一个包装类，绕过Keras模型的序列化限制：

import tensorflow as tf

class PicklableKerasRegressor:
    def __init__(self, model_builder):
        self.model_builder = model_builder
        self.model = None

    def fit(self, X, y):
        # 构建并训练模型
        self.model = self.model_builder()
        self.model.fit(X, y, epochs=10, batch_size=32)  # 按需调整训练参数
        return self

    def predict(self, X):
        return self.model.predict(X)

    def __getstate__(self):
        # 序列化时只保存模型构建函数，不保存带锁的模型对象
        return {'model_builder': self.model_builder}

    def __setstate__(self, state):
        # 反序列化时重新构建模型
        self.model_builder = state['model_builder']
        self.model = self.model_builder()

然后用这个类包装你的Keras模型：

# 保留你原来的create_model函数
def create_model():
    model = tf.keras.Sequential()
    model.add(tf.keras.layers.Dense(150, activation='softmax', kernel_initializer='VarianceScaling', input_dim=456, name='HL1'))
    model.add(tf.keras.layers.Dropout(0.2))  # 补全你的Dropout层参数
    # 添加其他层...
    model.compile(optimizer='adam', loss='mean_squared_error')  # 记得编译模型
    return model

# 包装成可pickle的模型对象
wrapped_keras_model = PicklableKerasRegressor(create_model)

# 传入StackedRegressor
stacked_reg = StackedRegressor(
    base_estimators=[('keras_reg', wrapped_keras_model), ...],  # 其他基模型
    final_estimator=NonNegativeLinearRegression(),
    n_jobs=2
)

如果需要保存训练后的权重，可以在__getstate__里把权重存到临时文件，__setstate__时加载，这样子进程能拿到训练好的模型。

方案3：改用TensorFlow分布式训练 #

如果你的模型是GPU密集型，也可以考虑用TensorFlow原生的分布式训练来实现并行，替代StackedRegressor的n_jobs参数。不过这个方案复杂度较高，适合大规模训练场景，这里就不展开细节了。

内容的提问来源于stack exchange，提问作者Coderji