使用多处理从另一个类调用类方法的实例

Question

我需要在一个类中使用多处理技术来调用另一个类的方法。解释如下。

我正在编写多代理策略。因此，我的代码分为 3 个类：

1. 代理类
2. 环境类
3. 主要。

在代理类中，我正在运行一个返回决策变量字典的 MILP 问题：

class Agent:
    def __init__():
        ....
    def Optimisation():
       .....
       return decision_agent

类环境接收所有代理的输出以计算“社区”输出。由于每个代理都是一个独立的实体，代理将并行计算他们的优化问题。因此，我想在环境类中使用多处理来获取所有代理的决策

import Agent as Agent

class Env:
    def __init__():
         self.agents = #list of agents in my environment

    def community_output():
        #**start the part of code I want to multiprocess**
        for agent in self.agents:
            agents_output.append(agent.Optimisation) 
        #**end the part of code I want to multiprocess**

        """compute community_output based on agents_output """

        return community_output

然后在main中，调用方法community_output()：

import Env as Env

if __name__ == '__main__': 
    env = Env()
    result = env.community_output()

请问我该怎么做？

由于我有一个实例方法列表作为可迭代对象，因此我尝试了以下方法：

def job(instance):
    instance.Optimisation()
    return instance

if __name__ == '__main__':
   pool = mp.Pool(processes=4)
   agents_output= pool.map(job, agents)
   pool.close() 
   pool.join()

这仅在我删除 Env 类并将其所有处理放在 def job(instance) 之前才有效。

当我将它集成到Env 类中时：

import Agent as Agent

class Env:
    def __init__():
         self.agents = #list of agents in my environment

    def job(instance):
        instance.Optimisation()
        return instance

    def community_output():
        pool = mp.Pool(processes=4)
        agents_output = pool.map(self.job, self.agents)
        pool.close() 
        pool.join()

        """ Compute community_output based on agents_output """

        return community_output

Import Env as Env
if __name__ == '__main__': 
    env = Env()
    result = env.community_output()

我得到了错误：

  File "C:\Users\chaim\...\Env.py", line 576, in community_output
    results = pool.map(self.job, self.agents)
  File "C:\Users\chaim\anaconda3\envs\picos\lib\multiprocessing\pool.py", line 268, in map
    return self._map_async(func, iterable, mapstar, chunksize).get()
  File "C:\Users\chaim\anaconda3\envs\picos\lib\multiprocessing\pool.py", line 657, in get
    raise self._value
multiprocessing.pool.MaybeEncodingError: Error sending result: '[{'opt_prob': <Mixed-Integer Linear Program>, 'q_us_val': <190x1 matrix, tc='d'>, 'q_ds_val': 0.0, 'p_net_exp_val': <48x1 matrix, tc='d'>, 'q_sup_sell_val': <48x1 matrix, tc='d'>, 'q_sup_buy_val': <48x1 matrix, tc='d'>, 'p_es_ch_val': <48x1 matrix, tc='d'>, 'p_es_dis_val': <48x1 matrix, tc='d'>, 'q_ds_0': <1×1 Real Constant: 0>, 'q_us_0': <1×1 Real Constant: q_usᵀ·[190×1]>}]'. Reason: 'PicklingError("Can't pickle <function <lambda> at 0x000001B8B9C31288>: attribute lookup <lambda> on picos.modeling.options failed")'

我注意到：

{'opt_prob': <Mixed-Integer Linear Program>, '... q_usᵀ·[190×1]>}

是agent.Optimisation方法返回的字典，picos是使用的优化API。

有什么办法可以解决这个问题吗？

Answer 1

您是否尝试从代理模块调用 Optimisation()？如果是，那么不应该

def job(instance):
    instance.Optimisation()
    return instance

改为

def job():
    return Agent.Optimisation()

Answer 2

我对 PICOS 本身并不熟悉，但在对它做了一些假设之后，我认为以下是关于如何做你想做的事情的大致概述。请注意，传递给pool.map() 的函数参数是unbound Agent 类方法optimisation()。它将使用您在它们列表中的Agent 实例之一调用。

import multiprocessing as mp
from random import randrange  # For testing purposes.


class Agent:
    def __init__(self):
        ...
    def optimisation(self):
       decision_agent = {'var1': randrange(10), 'var2': randrange(10), 'var3': randrange(10)}
       return decision_agent


class Env:
    def __init__(self, num_agents):
         self.agents = [Agent() for _ in range(num_agents)]

    def community_output(self):
        with mp.Pool(processes=4, maxtasksperchild=1) as pool:
            agents_output = pool.map(Agent.optimisation, self.agents)
        return agents_output


if __name__ == '__main__':
    env = Env(3)
    result = env.community_output()
    print(result)