在多线程中运行 Keras 模型进行预测答案

【问题标题】：Running Keras model for prediction in multiple threads在多线程中运行 Keras 模型进行预测
【发布时间】：2017-03-31 08:44:16
【问题描述】：

类似于this question 我正在运行异步强化学习算法，需要在多个线程中运行模型预测以更快地获取训练数据。我的代码基于 GitHub 上的DDPG-keras，其神经网络构建在 Keras 和 Tensorflow 之上。我的代码片段如下所示：

异步线程创建和加入：

for roundNo in xrange(self.param['max_round']):
    AgentPool = [AgentThread(self.getEnv(), self.actor, self.critic, eps, self.param['n_step'], self.param['gamma'])]
    for agent in AgentPool:
        agent.start()
    for agent in AgentPool:
        agent.join()

代理线程代码

"""Agent Thread for collecting data"""
def __init__(self, env_, actor_, critic_, eps_, n_step_, gamma_):
    super(AgentThread, self).__init__()
    self.env = env_         # type: Environment
    self.actor = actor_     # type: ActorNetwork
    # TODO: use Q(s,a)
    self.critic = critic_   # type: CriticNetwork
    self.eps = eps_         # type: float
    self.n_step = n_step_   # type: int
    self.gamma = gamma_
    self.data = {}

def run(self):
    """run behavior policy self.actor to collect experience data in self.data"""
    state = self.env.get_state()
    action = self.actor.model.predict(state[np.newaxis, :])[0]
    action = np.maximum(np.random.normal(action, self.eps, action.shape), np.ones_like(action) * 1e-3)

在运行这些代码时，我遇到了 Tensorflow 异常：

Using TensorFlow backend.
create_actor_network
Exception in thread Thread-1:
Traceback (most recent call last):
  File "/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/threading.py", line 801, in __bootstrap_inner
    self.run()
  File "/Users/niyan/code/routerRL/A3C.py", line 26, in run
    action = self.actor.model.predict(state[np.newaxis, :])[0]
  File "/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/keras/engine/training.py", line 1269, in predict
    self._make_predict_function()
  File "/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/keras/engine/training.py", line 798, in _make_predict_function
    **kwargs)
  File "/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/keras/backend/tensorflow_backend.py", line 1961, in function
    return Function(inputs, outputs, updates=updates)
  File "/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/keras/backend/tensorflow_backend.py", line 1919, in __init__
    with tf.control_dependencies(self.outputs):
  File "/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/tensorflow/python/framework/ops.py", line 3583, in control_dependencies
    return get_default_graph().control_dependencies(control_inputs)
  File "/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/tensorflow/python/framework/ops.py", line 3314, in control_dependencies
    c = self.as_graph_element(c)
  File "/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/tensorflow/python/framework/ops.py", line 2405, in as_graph_element
    return self._as_graph_element_locked(obj, allow_tensor, allow_operation)
  File "/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/tensorflow/python/framework/ops.py", line 2484, in _as_graph_element_locked
    raise ValueError("Tensor %s is not an element of this graph." % obj)
ValueError: Tensor Tensor("concat:0", shape=(?, 4), dtype=float32) is not an element of this graph.

那么如何使用经过训练的 Keras 模型（使用 Tensorflow 作为后端）在多个线程中同时进行预测？

4 月 2 日更新：我尝试过超重的应对模型，但没有奏效：

for roundNo in xrange(self.param['max_round']):
    for agent in self.AgentPool:
        agent.syncModel(self.getEnv(), self.actor, self.critic, eps)
        agent.start()
    for agent in self.AgentPool:
        agent.join()

def syncModel(self, env_, actor_, critic_, eps_):
    """synchronize A-C models before collecting data"""
    # TODO copy env, actor, critic
    self.env = env_     # shallow copy
    self.actor.model.set_weights(actor_.model.get_weights())        # deep copy, by weights
    self.critic.model.set_weights(critic_.model.get_weights())      # deep copy, by weights
    self.eps = eps_     # shallow copy
    self.data = {}

编辑：在 Github 上看到这个 jaara/AI-blog，似乎

model._make_predict_function()  # have to initialize before threading

有效。

作者在this issue上稍作解释。更多讨论请见this issue on Keras

【问题讨论】：

请添加您的编辑作为答案，我们只是有一个骗子，因此我无法链接。
因为这不是一个明确的问题并且附带了许多信息，所以我能提供的唯一答案就是查看这些网页。如果您能找到另一个好的解决方案，请将您的部分代码与错误进行比较。您在this webpage 上有一个很好的代理示例。在 Keras GitHub 和 keras-multi-threaded-model-fitting 上查看此类似问题

标签： python tensorflow keras thread-safety reinforcement-learning

【解决方案1】：

python 中的多线程不一定能更好地利用您的资源，因为 python 使用 global interpreter lock 并且一次只能运行一个本机线程。

在 python 中，通常您应该使用多处理来利用您的资源，但由于我们正在谈论 keras 模型，我不确定这是否是正确的做法。在多个进程中加载多个模型有其自身的开销，您可以像其他人已经指出的那样简单地增加批量大小。

或者，如果您有一个繁重的预处理阶段，您可以在一个过程中预处理您的数据并在另一个过程中预测它们（尽管我怀疑这也是必要的）。

【讨论】：

【解决方案2】：

在多个线程中预测数据是个坏主意。离线预测数据时可以在model.predict中使用更大的batch_size，在线预测数据时可以使用tensorflow serving。

【讨论】：

【解决方案3】：

Keras 不是线程安全的，为了预测大批量，您可以使用 batch_size 设置最大限制。如果您要部署到生产环境，那么理想的做法是转换模型权重 tensorflow protobuf，而不是使用 tensorflow serving。

你可以关注这个博客 http://machinelearningmechanic.com/keras/2019/06/26/keras-serving-keras-model-quickly-with-tensorflow-serving-and-docker-md.html

【讨论】：

【解决方案4】：

由于 Python 的 Global Interpreter Lock，您应该考虑使用多处理而不是线程。 Ray 是一个很棒的 API，可以用 Python 构建分布式应用程序，他们已经有了一个名为 RLlib 的强化学习框架。我强烈建议您看看 Ray，尤其是对于强化学习应用程序。

【讨论】：