运行Keras模型以在多线程中进行预测 [英] Running Keras model for prediction in multiple threads

问题描述

类似于这个问题一种异步强化学习算法，需要在多个线程中运行模型预测才能更快地获取训练数据.我的代码基于GitHub上的 DDPG-keras ，其神经网络构建于凯拉斯& Tensorflow.我的代码片段如下所示:

similar to this question I was running an asynchronous reinforcement learning algorithm and need to run model prediction in multiple threads to get training data more quickly. My code is based on DDPG-keras on GitHub, whose Neural Network was build on top of Keras & Tensorflow. Pieces of my code are shown below:

• 异步线程的创建和加入:

• Asynchronous Thread creation and join:

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 Code

"""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异常:

While running these codes, I encountered a Tensorflow Exception:

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作为后端)在多个线程中同时进行预测?

So how can I use a trained Keras model (using Tensorflow as backend) to concurrently predict in multiple threads?

4月2日更新: 我尝试应对体重过重的模型，但是没有用:

Update on April 2nd: I tried coping model over weight, but didn't work:

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

see this jaara/AI-blog on Github, seems

model._make_predict_function()  # have to initialize before threading

有效.

作者在此问题上做了一些解释.有关进一步的讨论，请参见关于Keras的问题

The author explained a little on this issue. For further discussion, see this issue on Keras

推荐答案

python中的多线程未必能更好地利用您的资源，因为python使用全局解释器锁定，一次只能运行一个本机线程.

multi threading in python doesn't necessarily make a better use of your resources since python uses global interpreter lock and only one native thread can run at a time.

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

in python, usually you should use multi processing to utilize your resources, but since we're talking about keras models, I'm not sure even that is the right thing to do. loading several models in several processes has its own overhead, and you could simply increase the batch size as others have already pointed out.

或者，如果您有一个繁重的预处理阶段，则可以在一个过程中对数据进行预处理，然后在另一个过程中进行预测(尽管我怀疑这是否有必要).

OR if you have a heavy pre-processing stage you could preprocess your data in one process and predict them in another (although I doubt that would be necessary either).

这篇关于运行Keras模型以在多线程中进行预测的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！