Scikit-Learn Wrappers for TensorFlow搭配Celery-Redis运行报错如何解决

这个报错的核心是Celery的消息序列化组件kombu在处理TensorFlow张量对象时，尝试调用.numpy()方法将张量转为可序列化的数值类型，但此时TensorFlow的eager执行模式未生效，同时Celery默认的JSON序列化器不支持直接处理TensorFlow张量类型。

• 第一步：全局开启TensorFlow eager执行
  在Celery任务入口文件（即neural_networks_task.py）的最开头，导入TensorFlow后立即添加全局开启eager执行的配置，避免worker进程初始化的TF上下文默认关闭eager：

import tensorflow as tf
tf.config.run_functions_eagerly(True)

• 第二步：更换Celery序列化器
  将Celery默认的JSON序列化器替换为支持numpy、Tensor对象的pickle序列化器，修改Celery实例的配置：

app = Celery('neural_networks_task', broker='redis://你的Redis地址')
app.conf.update(
    task_serializer='pickle',
    accept_content=['pickle'],
    result_serializer='pickle',
)

注意：pickle序列化存在安全风险，仅可在受信任的内网环境中使用。

• 第三步：手动转换张量为原生类型
  所有需要作为Celery任务参数、返回值传递的数值，提前手动转为numpy数组或Python原生数值，从根源避免序列化时接触TF张量：

# 示例：任务返回前先转换类型
def train_task():
    # 训练逻辑
    pred = model.predict(x_test)
    # 手动转成Python列表再返回
    return pred.numpy().tolist()

• 第四步：验证Celery worker池兼容性
  你当前使用的gevent池是IO密集型场景的协程池，和TensorFlow的异步执行逻辑可能存在猴子补丁冲突，可以先切换为prefork进程池验证问题是否解决：

celery -A neural_networks_task worker --pool prefork -l info

如果prefork池运行无报错，说明是gevent兼容性问题，模型训练属于计算密集型任务，直接使用prefork池即可。

• 第五步：配置Scikit-Learn Wrapper强制eager执行
  初始化KerasClassifier/KerasRegressor时添加run_eagerly=True参数，强制wrapper所有方法运行在eager模式下，返回结果直接为numpy类型：

from tensorflow.keras.wrappers.scikit_learn import KerasClassifier

def build_your_model():
    # 你的模型定义逻辑
    pass

model = KerasClassifier(build_fn=build_your_model, epochs=10, batch_size=32, run_eagerly=True)

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