参数优化可以调用 遗传算法优化函数和穷举函数。运行过程会调用cpu的所有进程。下面这段代码,没有找到max_worker 是怎么传进去的。如果需要指定进程的数量,该如何修改?
def run_ga_optimization(
evaluate_func: EVALUATE_FUNC,
optimization_setting: OptimizationSetting,
key_func: KEY_FUNC,
max_workers: int = None,
population_size: int = 100,
ngen_size: int = 30,
output: OUTPUT_FUNC = print
) -> List[Tuple]:
"""Run genetic algorithm optimization"""
# Define functions for generate parameter randomly
buf: List[Dict] = optimization_setting.generate_settings()
settings: List[Tuple] = [list(d.items()) for d in buf]
def generate_parameter() -> list:
""""""
return choice(settings)
def mutate_individual(individual: list, indpb: float) -> tuple:
""""""
size: int = len(individual)
paramlist: list = generate_parameter()
for i in range(size):
if random() < indpb:
individual[i] = paramlist[i]
return individual,
# Set up multiprocessing Pool and Manager
with Manager() as manager, Pool(max_workers) as pool:
# Create shared dict for result cache
cache: Dict[Tuple, Tuple] = manager.dict()
# Set up toolbox
toolbox: base.Toolbox = base.Toolbox()
toolbox.register("individual", tools.initIterate, creator.Individual, generate_parameter)
toolbox.register("population", tools.initRepeat, list, toolbox.individual)
toolbox.register("mate", tools.cxTwoPoint)
toolbox.register("mutate", mutate_individual, indpb=1)
toolbox.register("select", tools.selNSGA2)
toolbox.register("map", pool.map)
toolbox.register(
"evaluate",
ga_evaluate,
cache,
evaluate_func,
key_func
)
total_size: int = len(settings)
pop_size: int = population_size # number of individuals in each generation
lambda_: int = pop_size # number of children to produce at each generation
mu: int = int(pop_size * 0.8) # number of individuals to select for the next generation
cxpb: float = 0.95 # probability that an offspring is produced by crossover
mutpb: float = 1 - cxpb # probability that an offspring is produced by mutation
ngen: int = ngen_size # number of generation
pop: list = toolbox.population(pop_size)
# Run ga optimization
output("开始执行遗传算法优化")
output(f"参数优化空间:{total_size}")
output(f"每代族群总数:{pop_size}")
output(f"优良筛选个数:{mu}")
output(f"迭代次数:{ngen}")
output(f"交叉概率:{cxpb:.0%}")
output(f"突变概率:{mutpb:.0%}")
start: int = perf_counter()
algorithms.eaMuPlusLambda(
pop,
toolbox,
mu,
lambda_,
cxpb,
mutpb,
ngen,
verbose=False
)
end: int = perf_counter()
cost: int = int((end - start))
output(f"遗传算法优化完成,耗时{cost}秒")
results: list = list(cache.values())
results.sort(reverse=True, key=key_func)
return results