Paddle-NEAT——飛槳進化神經網絡元件
目錄
- Paddle-NEAT——飛槳進化神經網絡元件
-
- 寫在前面：
- NEAT 簡介
-
- 基因組的表示
- 基因組的變異
-
- 節點變異
- 連接配接變異
- 基因組的交叉
- 最後稍微介紹一下 NEAT 的兩種改進形式：
-
- HyperNEAT
- Adaptive HyperNEAT
- 運作Paddle-NEAT
-
- 安裝
- 繼續來立我們的棍子吧
- 運作一下
- 當然走一下迷宮也是可以滴
- 運作一下
寫在前面：
最近自己寫了個把 neat-python 和 paddlepaddle 深度學習架構相結合的套件，取名叫 Paddle-NEAT。連結會在下面的安裝方式處給出。
該套件涉及到了機器學習的另一個方向——遺傳算法與進化政策。但本文的重點是介紹其中的一個分支——NEAT(NeuroEvolution of Augmenting Topologies)
遺傳算法與進化政策可以跳脫出梯度的限制（因為不以梯度作為更新的依據），在高并行的計算條件下，說不定會有超越深度學習的效果。
噢對了，現在已經有将遺傳算法與 NAS 相結合的文章了——Genetic CNN。
注意：本文的代碼環境基于 Jupyter Notebook
NEAT 簡介
遺傳算法是個通用的架構，是以需要根據具體的問題來定義遺傳算法
整體架構可分為三部分：交叉、變異與适應度
我們的 NEAT 将神經元及其連接配接定義成基因組
基因組的表示
從上圖可以看出，節點變量分為節點變量與連接配接變量兩種。其中節點變量包括了節點屬性等變量，連接配接變量則包括了輸入節點、輸出節點、權重與連接配接是否可行等屬性。
通過将節點與連接配接各列成一張連結清單，網絡計算時查詢即可得到輸出。
基因組的變異
同樣的，對應節點與連接配接兩種基因組組成，變異類型也有結點變異和連結變異兩種
節點變異
在連接配接清單中随機選擇一個連接配接，然後中間加入一個節點
連接配接變異
随機選擇兩個不同的節點，增加一段連接配接
基因組的交叉
我們的基因組如何交叉呢？
首先，我們要把父母雙方的基因先按順序鋪開對齊，如果是雙方都有的連接配接id（稱為 innovation），那就随機選擇一個；
如果有不比對的基因，那麼繼承具有更好 fitness 的一方；
如果雙方的 fitness 相同，那就随便選了。
最後稍微介紹一下 NEAT 的兩種改進形式：
HyperNEAT
HyperNEAT是對NEAT的擴充，它用一個單獨的網絡（稱為CPPN，用于合成圖案生成網絡）間接地編碼網絡（稱為襯底）的權重
Adaptive HyperNEAT
Adaptive HyperNEAT是HyperNEAT的一個擴充，它間接地對初始權值和權值的更新規則進行編碼，以便在網絡的“生命周期”内進行一些學習
運作Paddle-NEAT
安裝
!pip install git+https://github.com/AgentMaker/Paddle-NEAT.git
!pip install neat-python
本套件底層仍然是 neat-python，但某些算子可以借由 paddle2.0 實作加速
網絡配置與節點資訊參照下面的 neat-python 中的介紹
neat-python 連結：
https://github.com/CodeReclaimers/neat-python
繼續來立我們的棍子吧
%%writefile cart_pole.py
import os
import click
import gym
import neat
from paddle_neat.multi_env_eval import MultiEnvEvaluator
from paddle_neat.neat_reporter import LogReporter
from paddle_neat.recurrent_net import RecurrentNet
max_env_steps = 200
# 建立環境
def make_env():
return gym.make("CartPole-v0")
# 建立網絡
def make_net(genome, config, bs):
return RecurrentNet.create(genome, config, bs)
# 定義輸出層
def activate_net(net, states):
outputs = net.activate(states).numpy()
return outputs[:, 0] > 0.5
@click.command()
@click.option("--n_generations", type=int, default=100)
def run(n_generations):
config_path = os.path.join(os.path.dirname(__file__), "neat_cart_pole.cfg")
config = neat.Config(
neat.DefaultGenome,
neat.DefaultReproduction,
neat.DefaultSpeciesSet,
neat.DefaultStagnation,
config_path,
)
evaluator = MultiEnvEvaluator(
make_net, activate_net, make_env=make_env, max_env_steps=max_env_steps
)
def eval_genomes(genomes, config):
for _, genome in genomes:
genome.fitness = evaluator.eval_genome(genome, config)
pop = neat.Population(config)
stats = neat.StatisticsReporter()
pop.add_reporter(stats)
reporter = neat.StdOutReporter(True)
pop.add_reporter(reporter)
logger = LogReporter("neat.log", evaluator.eval_genome)
pop.add_reporter(logger)
pop.run(eval_genomes, n_generations)
if __name__ == "__main__":
run()
運作一下
!python cart_pole.py
當然走一下迷宮也是可以滴
%%writefile maze.py
import multiprocessing
import os
import click
import neat
import paddle
import numpy as np
import sys
sys.path.append("..")
from paddle_neat import t_maze
from paddle_neat.activations import tanh_activation
from paddle_neat.adaptive_linear_net import AdaptiveLinearNet
from paddle_neat.multi_env_eval import MultiEnvEvaluator
from paddle_neat.neat_reporter import LogReporter
batch_size = 4
DEBUG = True
def make_net(genome, config, _batch_size):
input_coords = [[-1.0, 0.0], [0.0, 0.0], [1.0, 0.0], [0.0, -1.0]]
output_coords = [[-1.0, 0.0], [0.0, 0.0], [1.0, 0.0]]
return AdaptiveLinearNet.create(
genome,
config,
input_coords=input_coords,
output_coords=output_coords,
weight_threshold=0.4,
batch_size=batch_size,
activation=tanh_activation,
output_activation=tanh_activation,
)
def activate_net(net, states, debug=False, step_num=0):
if debug and step_num == 1:
print("\n" + "=" * 20 + " DEBUG " + "=" * 20)
print(net.delta_w_node)
print("W init: ", net.input_to_output.numpy()[0])
outputs = net.activate(states).numpy()
if debug and (step_num - 1) % 100 == 0:
print("\nStep {}".format(step_num - 1))
print("Outputs: ", outputs[0])
print("Delta W: ", net.delta_w.numpy()[0])
print("W: ", net.input_to_output.numpy()[0])
return np.argmax(outputs, axis=1)
@click.command()
@click.option("--n_generations", type=int, default=10000)
@click.option("--n_processes", type=int, default=1)
def run(n_generations, n_processes):
config_path = os.path.join(os.path.dirname(__file__), "neat_maze.cfg")
config = neat.Config(
neat.DefaultGenome,
neat.DefaultReproduction,
neat.DefaultSpeciesSet,
neat.DefaultStagnation,
config_path,
)
envs = [t_maze.TMazeEnv(init_reward_side=i, n_trials=100) for i in [1, 0, 1, 0]]
evaluator = MultiEnvEvaluator(
make_net, activate_net, envs=envs, batch_size=batch_size, max_env_steps=1000
)
if n_processes > 1:
pool = multiprocessing.Pool(processes=n_processes)
def eval_genomes(genomes, config):
fitnesses = pool.starmap(
evaluator.eval_genome, ((genome, config) for _, genome in genomes)
)
for (_, genome), fitness in zip(genomes, fitnesses):
genome.fitness = fitness
else:
def eval_genomes(genomes, config):
for i, (_, genome) in enumerate(genomes):
try:
genome.fitness = evaluator.eval_genome(
genome, config, debug=DEBUG and i % 100 == 0
)
except Exception as e:
print(genome)
raise e
pop = neat.Population(config)
stats = neat.StatisticsReporter()
pop.add_reporter(stats)
reporter = neat.StdOutReporter(True)
pop.add_reporter(reporter)
logger = LogReporter("log.json", evaluator.eval_genome)
pop.add_reporter(logger)
winner = pop.run(eval_genomes, n_generations)
print(winner)
final_performance = evaluator.eval_genome(winner, config)
print("Final performance: {}".format(final_performance))
generations = reporter.generation + 1
return generations
if __name__ == "__main__":
run()
運作一下
!python maze.py