Keras 预测不会在 celery 任务中返回

Posted 2023-02-24

【中文标题】Keras 预测不会在 celery 任务中返回

【英文标题】：Keras predict not returning inside celery task

【发布时间】：2018-01-09 13:59:13

【问题描述】：

以下 Keras 函数（预测）在同步调用时起作用

pred = model.predict(x)

但是从异步任务队列 (Celery) 中调用时它不起作用。 Keras predict 函数在异步调用时不返回任何输出。

堆栈是：Django、Celery、Redis、Keras、TensorFlow

【问题讨论】：

事实证明 - 用 Theano 替换 TensorFlow 后端，奇怪但很好，没有这个问题，并从异步任务队列中返回输出。尽管如此，有一个适用于 TensorFlow 后端的解决方案仍然很棒。

CELERY_ALWAYS_EAGER = True 时是否运行？仔细查看 celery 设置以检查它是否配置正确，然后设置一个可以向您指示正在运行的进程的任务（发送电子邮件，触摸文件）。请记住，Celery 独立于 Web 服务器运行 django 应用程序，并且每当您更新任何内容时都需要重新启动 Celery。

它也不适用于 CELERY_ALWAYS_EAGER 为 True。

***.com/questions/47295025/… 任何建议

就像我提到的，如果可能，请考虑使用 Theano。

【参考方案1】：

我遇到了同样的问题，真是个兔子洞。想在这里发布我的解决方案，因为它可能会为某人节省一天的工作：

TensorFlow 线程特定的数据结构

在 TensorFlow 中，当您调用 model.predict（或 keras.models.load_model，或 keras.backend.clear_session，或几乎任何其他与 TensorFlow 后端交互的函数）时，有两个关键数据结构在幕后工作：

TensorFlow graph，表示您的 Keras 模型的结构 TensorFlow session，它是当前图形和 TensorFlow 运行时之间的连接

如果没有深入研究，文档中没有明确说明的是，会话和图表都是当前线程的属性。请参阅 API 文档 here 和 here。

在不同线程中使用 TensorFlow 模型

想加载你的模型一次然后多次调用.predict()是很自然的：

from keras.models import load_model

MY_MODEL = load_model('path/to/model/file')

def some_worker_function(inputs):
    return MY_MODEL.predict(inputs)

在像 Celery 这样的网络服务器或工作池上下文中，这意味着您将在导入包含 load_model 行的模块时加载模型，然后另一个线程将执行 some_worker_function，在全局上运行预测包含 Keras 模型的变量。但是，尝试在不同线程中加载的模型上运行预测会产生“张量不是此图的元素”错误。感谢涉及该主题的几篇 SO 帖子，例如 ValueError: Tensor Tensor(...) is not an element of this graph. When using global variable keras model。为了让它工作，你需要坚持使用的 TensorFlow 图——正如我们之前看到的，图是当前线程的一个属性。更新后的代码如下所示：

from keras.models import load_model
import tensorflow as tf

MY_MODEL = load_model('path/to/model/file')
MY_GRAPH = tf.get_default_graph()

def some_worker_function(inputs):
    with MY_GRAPH.as_default():
        return MY_MODEL.predict(inputs)

这里有点令人惊讶的转折是：如果您使用Threads，上面的代码就足够了，但如果您使用Processes，则无限期挂起。默认情况下，Celery 使用进程管理其所有工作人员池。所以在这一点上，事情在 Celery 上仍然不起作用。

为什么这只适用于Threads？

在 Python 中，Threads 与父进程共享相同的全局执行上下文。来自Python _thread docs：

此模块提供用于处理多个线程（也称为轻量级进程或任务）的低级原语——多个控制线程共享它们的全局数据空间。

因为线程不是真正的独立进程，它们使用相同的 python 解释器，因此受到臭名昭著的 Global Interpeter Lock (GIL) 的约束。对于本次调查而言，或许更重要的是，它们与父级共享全局数据空间。

与此相反，Processes 是程序产生的实际新进程。这意味着：

新的 Python 解释器实例（并且没有 GIL） 全局地址空间

重复

注意这里的区别。 Threads 可以访问共享的单个全局 Session 变量（内部存储在 Keras 的 tensorflow_backend 模块中），Processes 具有 Session 变量的副本。

我对这个问题最好的理解是 Session 变量应该代表客户端（进程）和 TensorFlow 运行时之间的唯一连接，但是由于在分叉过程中被复制，这个连接信息不正确调整。这会导致 TensorFlow 在尝试使用在不同进程中创建的 Session 时挂起。如果有人对 TensorFlow 的底层工作原理有更深入的了解，我很想听听！

解决方案/解决方法

我对 Celery 进行了调整，使其使用 Threads 而不是 Processes 进行池化。这种方法有一些缺点（参见上面的 GIL 注释），但这允许我们只加载一次模型。无论如何，我们并没有真正受 CPU 限制，因为 TensorFlow 运行时最大化了所有 CPU 内核（它可以回避 GIL，因为它不是用 Python 编写的）。你必须为 Celery 提供一个单独的库来进行基于线程的池化；文档建议两个选项：gevent 或 eventlet。然后将您选择的库传递给工作人员via the --pool command line argument。

另外，似乎（正如您已经发现 @pX0r）其他 Keras 后端（例如 Theano）没有这个问题。这是有道理的，因为这些问题与 TensorFlow 实现细节密切相关。我个人还没有尝试过 Theano，所以你的里程可能会有所不同。

我知道这个问题是不久前发布的，但问题仍然存在，所以希望这会对某人有所帮助！

【参考方案2】：

我从这个Blog得到了参考

Tensorflow 是线程特定的数据结构，当您调用

model.predict

时，它在幕后工作

GRAPH = tf.get_default_graph()
with GRAPH.as_default():
    pred = model.predict
return pred

但 Celery 使用进程来管理其所有工作池。所以在这一点上，Celery 仍然无法正常工作，因为您需要使用 gevent 或 eventlet 库

pip 安装 gevent

现在运行 celery 为：

celery -A mysite worker --pool gevent -l info

【讨论】：

刚刚发布的 Celery 4.4.0 将“线程”并发添加到列表中。我想这也行。

@DejanLekic 使用相同的 4.4.0，但它没有用！