在python中使用多处理生成数组列表

Question

我很难实现并行化以生成数组列表。在这种情况下，每个数组都是独立生成的，然后附加到一个列表中。当我给它提供复杂的参数时，不知何故 multiprocessing.apply_asynch() 正在输出一个空数组。

更具体地说，只是为了给出上下文，我正在尝试使用并行化实现机器学习算法。这个想法如下：我有一个“系统”和一个在系统上执行操作的“代理”。为了教代理（在这种情况下是神经网络）如何以最佳方式表现（关于我在这里省略的某个奖励方案），代理需要通过对其应用动作来生成系统的轨迹。从执行动作获得的奖励中，智能体然后学习做什么和不做什么。请注意，代码中可能的操作被称为整数：

    possible_actions = [0,1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20]
    

所以我在这里尝试使用多处理生成许多这样的轨迹（抱歉，代码在这里无法运行，因为它需要许多其他文件，但我希望有人能发现这个问题）：

    from quantum_simulator_EC import system
    from reinforce_keras_EC import Agent
    import multiprocessing as mp


    s = system(1200, N=3)
    s.set_initial_state([0,0,1])  
    agent = Agent(alpha=0.0003,  gamma=0.95, n_actions=len( s.actions ))


    def get_result(result):
        global action_batch
        action_batch.append(result)

    def generate_trajectory(s, agent):

        sequence_of_actions = []

        for k in range( 5 ):

            net_input = s.generate_net_input_FULL(6)
            action = agent.choose_action( net_input )
    
            sequence_of_actions.append(action)
    

        return sequence_of_actions
    
    action_batch = []

    pool = mp.Pool(2)
    for i in range(0, batch_size):
        pool.apply_async(generate_trajectory, args=(s,agent), callback=get_result)
    pool.close()
    pool.join()

    print(action_batch)

问题是代码返回一个空数组[]。有人可以向我解释问题是什么吗？我可以传递给 apply_asynch 的参数类型是否有限制？在这个例子中，我传递了我的系统's'和我的'代理'，这两个都是复杂的对象。我之所以提到这一点，是因为当我使用整数或矩阵等简单参数而不是代理和系统测试我的代码时，它可以正常工作。如果没有明显的原因导致它不工作，如果有人有一些调试代码的技巧也会有帮助。

请注意，如果我不使用多处理，将最后一部分替换为：

    action_batch = []

    for i in range(0, batch_size):
        get_result( generate_sequence(s,agent) )

    print(action_batch)

在这种情况下，这里的输出与预期的一样，是 5 个动作的序列列表：

    [[4, 2, 1, 1, 7], [8, 2, 2, 12, 1], [8, 1, 9, 11, 9], [7, 10, 6, 1, 0]]

Answer 1

最终结果可以直接附加到主进程中的列表中，无需创建回调函数。然后你可以close和join这个pool，最后用get检索所有结果。

请参阅以下两个示例，使用apply_async 和starmap_async，（请参阅this 帖子了解区别）。

解决方案apply

import multiprocessing as mp
import time


def func(s, agent):
    print(f"Working on task {agent}")
    time.sleep(0.1)  # some task
    return (s, s, s)


if __name__ == '__main__':
    agent = "My awesome agent"
    with mp.Pool(2) as pool:
        results = []
        for s in range(5):
            results.append(pool.apply_async(func, args=(s, agent)))
        pool.close()
        pool.join()

    print([result.get() for result in results])

解决方案starmap

import multiprocessing as mp
import time


def func(s, agent):
    print(f"Working on task {agent}")
    time.sleep(0.1)  # some task
    return (s, s, s)


if __name__ == '__main__':
    agent = "My awesome agent"
    with mp.Pool(2) as pool:
        result = pool.starmap_async(func, [(s, agent) for s in range(5)])
        pool.close()
        pool.join()

    print(result.get())

输出

Working on task My awesome agent
Working on task My awesome agent
Working on task My awesome agent
Working on task My awesome agent
Working on task My awesome agent
[(0, 0, 0), (1, 1, 1), (2, 2, 2), (3, 3, 3), (4, 4, 4)]