多向交叉中的Keras预测

Question

我正在寻找一种在子进程中使用KerasModel.predict()函数的方法。 我正在使用keras2.3.1和TensorFlow 2.0.0。（我试过keras2.25和TensorFlow 1.14）

下面的代码抛出该错误。你知道吗

import itertools
import random
from abc import ABC
from multiprocessing import Pool as Pool

import numpy as np
from keras.engine.saving import load_model
from keras.models import Sequential
from keras.layers import Dense, Activation


class Pre(ABC):
    pass


class Prediction(Pre):

    def __init__(self):
        model = Sequential([
            Dense(32, input_shape=(2,)),
            Activation('relu'),
            Dense(2),
            Activation('softmax'),
        ])
        model.compile(optimizer='rmsprop',
                      loss='binary_crossentropy',
                      metrics=['accuracy'])
        model.summary()
        x = np.array([[0, 0], [0, 1], [1, 0], [1, 1]])
        y = np.array([[0, 1], [1, 0], [1, 0], [0, 1]])
        model.fit(x, y, epochs=20)
        model.save("temp")
        self.model = load_model('temp')
        self.modifier = 2

    def predict(self, input_array):
        prediction = self.model.predict(np.array([input_array]))[0]
        prediction += self.modifier
        return prediction[0]


class B:
    def __init__(self):
        self.pred = Prediction()


def calculate_something(pred_inner: B, modifier: int):
    pred_inner.modifier = modifier
    sum_all = sum(pred_inner.pred.predict(np.array([random.choice([0, 1]), random.choice([0, 1])])) for _ in range(100))

    # do some modifi
    return (pred_inner,
            sum_all)


if __name__ == '__main__':
    probe_size = 100
    pred = B()
    for i in range(1000):
        with Pool() as pool:
            results = pool.starmap(calculate_something, zip(itertools.repeat(pred),
                                                            [probe_size for _ in range(i)]))
            for r in results:
                print(r[1])

因为我在子进程中调用了predict函数，所以它会与自己的子进程发生冲突。 我的网络非常小，所以我认为多处理不是严格必要的。有没有什么方法可以停用Keras和TensorFlow中的多处理？ 或者有没有其他API可以替代Keras/TensorFlow？你知道吗

    Exception in thread Thread-24:
    Traceback (most recent call last):
      File "C:\Python37\lib\threading.py", line 926, in _bootstrap_inner
        self.run()
      File "C:\Python37\lib\threading.py", line 870, in run
        self._target(*self._args, **self._kwargs)
      File "C:\Python37\lib\multiprocessing\pool.py", line 470, in _handle_results
        task = get()
      File "C:\Python37\lib\multiprocessing\connection.py", line 251, in recv
        return _ForkingPickler.loads(buf.getbuffer())
      File "C:\Users\phhor\PycharmProjects\py_doku\venv37\lib\site-packages\keras\engine\network.py", line 1334, in __setstate__
        model = saving.unpickle_model(state)
      File "C:\Users\phhor\PycharmProjects\py_doku\venv37\lib\site-packages\keras\engine\saving.py", line 604, in unpickle_model
        return _deserialize_model(h5dict)
      File "C:\Users\phhor\PycharmProjects\py_doku\venv37\lib\site-packages\keras\engine\saving.py", line 274, in _deserialize_model
        model = model_from_config(model_config, custom_objects=custom_objects)
      File "C:\Users\phhor\PycharmProjects\py_doku\venv37\lib\site-packages\keras\engine\saving.py", line 627, in model_from_config
        return deserialize(config, custom_objects=custom_objects)
      File "C:\Users\phhor\PycharmProjects\py_doku\venv37\lib\site-packages\keras\layers\__init__.py", line 168, in deserialize
        printable_module_name='layer')
      File "C:\Users\phhor\PycharmProjects\py_doku\venv37\lib\site-packages\keras\utils\generic_utils.py", line 147, in deserialize_keras_object
        list(custom_objects.items())))
      File "C:\Users\phhor\PycharmProjects\py_doku\venv37\lib\site-packages\keras\engine\sequential.py", line 302, in from_config
        model.add(layer)
      File "C:\Users\phhor\PycharmProjects\py_doku\venv37\lib\site-packages\keras\engine\sequential.py", line 162, in add
        name=layer.name + '_input')
      File "C:\Users\phhor\PycharmProjects\py_doku\venv37\lib\site-packages\keras\engine\input_layer.py", line 178, in Input
        input_tensor=tensor)
      File "C:\Users\phhor\PycharmProjects\py_doku\venv37\lib\site-packages\keras\legacy\interfaces.py", line 91, in wrapper
        return func(*args, **kwargs)
      File "C:\Users\phhor\PycharmProjects\py_doku\venv37\lib\site-packages\keras\engine\input_layer.py", line 87, in __init__
        name=self.name)
      File "C:\Users\phhor\PycharmProjects\py_doku\venv37\lib\site-packages\keras\backend\tensorflow_backend.py", line 73, in symbolic_fn_wrapper
        if _SYMBOLIC_SCOPE.value:
    Attrib

uteError: '_thread._local' object has no attribute 'value'

Answer 1

我发现了问题！ 如果您创建一个对象来创建Keras模型，并将其从子流程中返回，则会出现此错误。 如果在返回对象之前删除模型，一切正常。你知道吗

def calculate_something(pred_inner: B, modifier: int):
    pred_inner.modifier = modifier
    sum_all = sum(pred_inner.pred.predict(np.array([random.choice([0, 1]), random.choice([0, 1])])) for _ in range(100))
    del pred_inner.pred
    # do some modifi
    return (pred_inner,
            sum_all)