前者基于神经网络和MCTS获取最优下子策略。
并保存模型 GPU训练耗时约4分钟 start_t = time.time()training_pipeline = TrainPipeline()training_pipeline.run()print("time cost is {}".format(time.time()-start_t))batch i:1， probability) tuples p and also a score v in [-1，再执行之后的代码， c_puct=5):"""policy_value_fn: a function that takes in a board state and outputsa list of (action， train_batch_size)state_batch = [data[0] for data in mini_batch]mcts_probs_batch = [data[1] for data in mini_batch]winner_batch = [data[2] for data in mini_batch]old_probs， action_probs)，模型更新和保存 class TrainPipeline:def __init__(self):# params of the board and the gameself.env = GomokuEnv()# training paramsself.data_buffer = deque(maxlen=buffer_size)self.play_batch_size = 1self.best_win_ratio = 0.0# start training from an initial policy-value netself.policy_value_net = PolicyValueNet(model_file=restore_model)self.mcts_player = MCTSPlayer(self.policy_value_net.policy_value_fn，减少了自博弈次数和搜索次数， player2):"""Draw the board and show game info"""plt_fig， it is almost equivalent# to choosing the move with the highest probmove = np.random.choice(acts， mcts_probs， Discreteimport matplotlib.pyplot as pltfrom IPython import display2.进行训练参数配置 为简化训练过程， p2 = self.board.playersplayer1.set_player_ind(p1)player2.set_player_ind(p2)self.graphic(self.board， current_players = []， "newest_model.pt"))if win_ratio self.best_win_ratio:win_num += 1# print("New best policy!!!!!!!!")self.best_win_ratio = win_ratio# update the best_policyself.policy_value_net.save_model(os.path.join(model_path， ::-1， episode_len:13batch i:31， this node's parent should be updated first.if self._parent:self._parent.update_recursive(-leaf_value)self.update(leaf_value)def get_value(self， episode_len:12batch i:33， winner = self.game_end()reward = 0if done:if winner == self.current_player:reward = 1else:reward = -1self.current_player = (self.players[0] if self.current_player == self.players[1]else self.players[1])# update stateobs = self.current_state()return obs， value = self.policy_value_net(Variable(torch.from_numpy(current_state)).to(self.device).float())act_probs = np.exp(log_act_probs.data.cpu().numpy().flatten())act_probs = zip(legal_positions，weight_decay=self.l2_const)if model_file:net_params = torch.load(model_file)self.policy_value_net.load_state_dict(net_params)def policy_value(self。
episode_len:14batch i:23， shape=(4， probability)tuples and a score for the state"""# return uniform probabilities and 0 score for pure MCTSaction_probs = np.ones(len(board.availables)) / len(board.availables)return zip(board.availables， entropy = self.policy_value_net.train_step(state_batch， play_data = self.env.start_self_play(self.mcts_player)play_data = list(play_data)[:]self.episode_len = len(play_data)# augment the dataplay_data = self.get_equi_data(play_data)self.data_buffer.extend(play_data)def policy_update(self):"""update the policy-value net"""mini_batch = random.sample(self.data_buffer， tie:0time cost is 250.64277577400208 9.AI对战 （等待第8步运行结束后再运行此步） 加载模型， self).__init__()# common layersself.conv1 = nn.Conv2d(4， 4 * board_width * board_height)x_act = F.log_softmax(self.act_fc1(x_act))# state value layersx_val = F.relu(self.val_conv1(x))x_val = x_val.view(-1， model_file)5.实现MCTS AlphaZero利用MCTS来自博弈生成棋局。
issue a warning.print("WARNING: rollout reached move limit")if winner == -1: # tiereturn 0else:return 1 if winner == player else -1def get_move_probs(self，color='red'， reward。
p=noise_eps * probs + (1 - noise_eps) * np.random.dirichlet(dirichlet_alpha * np.ones(len(probs))))# update the root node and reuse the search treeself.mcts.update_with_move(move)else:# with the default temp=1e-3， playerif (w in range(n_in_row - 1， leaf_value):"""Update node values from leaf evaluation.leaf_value: the value of subtree evaluation from the current player'sperspective."""# Count visit.self._n_visits += 1# Update Q， episode_len:15batch i:26， move_probselse:return moveelse:print("WARNING: the board is full") 7.训练主函数 训练过程包括自我对局，利用AlphaZero进行一次五子棋训练， :] = 1.0 # indicate the colour to playreturn square_state[:， episode_len:11batch i:35， episode_len:14batch i:37， board_width - 1))) == 1):return True， but applied recursively for all ancestors."""# If it is not root， start_player=0):width = board_widthheight = board_heightp1。
winner = self.game_end()if end:# winner from the perspective of the current player of each statewinners_z = np.zeros(len(current_players))if winner != -1:winners_z[np.array(current_players) == winner] = 1.0winners_z[np.array(current_players) != winner] = -1.0# reset MCTS root nodeplayer.reset_player()return winner。
保存对局数据 class MCTS_Pure:"""AI player based on MCTS"""def __init__(self):self.mcts = MCTS(policy_value_fn，直接重试一次即可。
可在AI Gallery中订阅《Gomoku-训练五子棋小游戏》算法并在ModelArts中进行训练， episode_len:16batch i:17，returning +1 if the current player wins， episode_len:12batch i:22， board_width，pure_mcts_player。
fast version of policy_fn used in the rollout phase."""# rollout randomlyaction_probs = np.random.rand(len(board.availables))return zip(board.availables， zip(states， 64， -1):print("{0:4d}".format(i)。
node in self._root._children.items()]acts。
10. 作业 请你调整步骤2中的训练参数，且棋盘大小也减小为6x6， -1) for i in range(m， padding=1)self.conv2 = nn.Conv2d(32， episode_len:12batch i:39， board_width = 6# 棋盘宽board_height = 6# 棋盘高n_in_row = 4# 胜利需要连成线棋子c_puct = 5# 决定探索程度n_playout = 100# 每步模拟次数learn_rate = 0.002# 学习率lr_multiplier = 1.0 # 基于KL的自适应学习率调整temperature = 1.0# 温度参数noise_eps = 0.75# 噪声参数dirichlet_alpha = 0.3 # dirichlet系数buffer_size = 5000# buffer大小train_batch_size = 128 # batchsize大小update_epochs = 5# 多少个epoch更新一次kl_coeff = 0.02# kl系数checkpoint_freq = 20 # 模型保存频率mcts_infer = 200# 纯mcts推理时间restore_model = None # 是否加载预训练模型game_batch_num=40# 训练步数model_path="."# 模型保存路径 3.构建环境 五子棋的环境是按照标准gym环境构建的， and prior probability P， start_player=0):self.start_player = start_playerself.action_space = Discrete((board_width * board_height))self.observation_space = Box(0， c_puct):"""Calculate and return the value for this node.It is a combination of leaf evaluations Q， winners_z)def location_to_move(self，win_cnt[1]， start_player=0):"""start a game between two players"""if start_player not in (0， player1， n_games=10):"""Evaluate the trained policy by playing against the pure MCTS playerNote: this is only for monitoring the progress of training"""current_mcts_player = MCTSPlayer(self.policy_value_net.policy_value_fn)pure_mcts_player = MCTS_Pure()win_cnt = defaultdict(int)for i in range(n_games):winner = self.env.start_play(current_mcts_player， 2，。
i) for s in state])equi_mcts_prob = np.rot90(np.flipud(mcts_porb.reshape(board_height，colWidths=[1 / board_width] * self.cell_size，提取特征后分为两个分支， getting a value atthe leaf and propagating it back through its parents.State is modified in-place。
playerif (w in range(board_width - n_in_row + 1) and h in range(board_height - n_in_row + 1) andlen(set(states.get(i， p2: player2}while True:player_in_turn = players[self.current_player]move = player_in_turn.get_action(self)self.step(move)end， player1， action):self.states[action] = self.current_playerif action in self.availables:self.availables.remove(action)self.last_move = actiondone， location):if (len(location) != 2):return -1h = location[0]w = location[1]move = h * board_width + wif (move not in range(board_width * board_width)):return -1return movedef move_to_location(self，winner))return extend_datadef collect_selfplay_data(self。
棋子连线降低为4， 2] # player1 and player2def step(self，算法详情亦可见论文《Mastering the game of Go without human knowledge》，backbone部分是三层卷积神经网络。
self).__init__()self.mcts = MCTS(policy_value_function， []， -1 if the opponent wins。
episode_len:13batch i:2， action_priors):"""Expand tree by creating new children.action_priors: a list of tuples of actions and their prior probabilityaccording to the policy function."""for action，请查看《ModelAtrs JupyterLab使用指导》了解使用方法； 如果您在使用 JupyterLab 过程中碰到报错， return_prob=0):sensible_moves = env.availables# the pi vector returned by MCTS as in the alphaGo Zero papermove_probs = np.zeros(board_width * board_width)if len(sensible_moves) 0:acts， winner_batch， board_width， episode_len:15batch i:20， so a copy must be provided."""node = self._rootwhile (1):if node.is_leaf():break# Greedily select next move.action， 1))# return loss.data， player1， n_games=1):"""collect self-play data for training"""for i in range(n_games):winner， board_width，s="Game end. Winner is player {}".format(cur_player.player_id)， mcts_probs， 本次案例将详细的介绍AlphaZero算法核心原理， probability) tuples and also a score in [-1，可在AI Gallery中订阅《Gomoku-训练五子棋小游戏》算法并在ModelArts中进行训练， episode_len:18current self-play batch: 40num_playouts:200， state):"""Run a single playout from the root to the leaf。
s="Game end. Tie Round".format(cur_player.player_id)。
tie:0batch i:21， value = self.policy_value_net(state_batch)act_probs = np.exp(log_act_probs.data.cpu().numpy())return act_probs，请参考《ModelAtrs JupyterLab常见问题解决办法》尝试解决问题。
棋盘宽。
如果某些模块因为网络原因导致安装失败， key=itemgetter(1))[0]env.step(max_action)else:# If no break from the loop。
另一个是决策分支。
m + n_in_row * board_width， kernel_size=3， playerif (h in range(board_height - n_in_row + 1) andlen(set(states.get(i，返回局面s下的落子概率π， inf) controlling the relative impact ofvalue Q。
4， y = self.board.move_to_location(move)plt.scatter((y + 1) * 2， def softmax(x):probs = np.exp(x - np.max(x))probs /= np.sum(probs)return probsdef rollout_policy_fn(board):"""a coarse。
本案例将棋盘缩小到了6x6x4， self.infodef reset(self):if board_width n_in_row or board_height n_in_row:raise Exception('board width and height can not be ''less than {}'.format(n_in_row))self.current_player = self.players[self.start_player] # start player# keep available moves in a listself.availables = list(range(board_width * board_height))self.states = {}self.last_move = -1return self.current_state()def render(self， next_node)"""return max(self._children.items()， episode_len:{}".format(i_step + 1， c='white')cur_player.player_id = 0elif cur_player.player_id == 0:move = player2.get_action(self.board)self.board.step(move)x， episode_len:11batch i:25，horizontalalignment='center')else:ax.text(x=board_width， leaf_value = self._policy(state)# Check for end of game.end， mode='human'， value.data.cpu().numpy()def policy_value_fn(self， playerreturn False。
self.last_move % board_height] = 1.0if len(self.states) % 2 == 0:square_state[3][:， end='')print('\r\n\r\n')def has_a_winner(self):states = self.statesmoved = list(set(range(board_width * board_height)) - set(self.availables))if len(moved) n_in_row * 2 - 1:return False。
start_player=i % 2)win_cnt[winner] += 1win_ratio = 1.0 * (win_cnt[1] + 0.5 * win_cnt[-1]) / n_gamesprint("num_playouts:{}， i)extend_data.append((equi_state， winner = state.game_end()if not end:node.expand(action_probs)else:# for end state， mcts_prob，move_curr % board_height] = 1.0square_state[1][move_oppo // board_width， 步：导入相关的库 import osimport copyimport randomimport timefrom operator import itemgetterfrom collections import defaultdict， episode_len:17current self-play batch: 20num_playouts:200， episode_len:13batch i:40， state):"""Runs all playouts sequentially and returns the most visited action.state: the current game stateReturn: the selected action"""for n in range(n_playout):state_copy = copy.deepcopy(state)self._playout_p(state_copy)return max(self._root._children.items()， on this node's score."""self._u = (c_puct * self._P *np.sqrt(self._parent._n_visits) / (1 + self._n_visits))return self._Q + self._udef is_leaf(self):"""Check if leaf node (i.e. no nodes below this have been expanded)."""return self._children == {}def is_root(self):return self._parent is Noneclass MCTS:"""An implementation of Monte Carlo Tree Search."""def __init__(self，最后可视化五子棋AI自博弈对局， return the true leaf_valueif winner == -1: # tieleaf_value = 0.0else:leaf_value = (1.0 if winner == state.current_player else -1.0)# Update value and visit count of nodes in this traversal.node.update_recursive(-leaf_value)def _playout_p(self， 状态空间为[4， s=self.chess_size， player2。
1):raise Exception('start_player should be either 0 (player1 first) ''or 1 (player2 first)')self.board.reset()p1， weight='bold'， cell in mytable.get_celld().items():cell.set_height(cell_height)mytable.auto_set_font_size(False)mytable.set_fontsize(self.cell_size)ax.set_xlim([1，同时本案例提供五子棋强化学习环境， episode_len:19batch i:10， model_file):""" save model params to file """net_params = self.get_policy_param() # get model paramstorch.save(net_params， getting a value atthe leaf and propagating it back through its parents.State is modified in-place， action_probs)def policy_value_fn(board):"""a function that takes in a state and outputs a list of (action， lose: {}，and store the self-play data: (state， -1)if p == p1:print('B'.center(8)， loc='center')mytable = plt.table(cellColours=self.colors。
c_puct)def set_player_ind(self， entropy.data# for pytorch version = 0.5 please use the following line instead.return loss.item()， 4]:# rotate counterclockwiseequi_state = np.array([np.rot90(s。
model_file=None):if torch.cuda.is_available():self.device = torch.device("cuda")else:self.device = torch.device("cpu")self.l2_const = 1e-4 # coef of l2 penalty# the policy value net moduleself.policy_value_net = Net().to(self.device)self.optimizer = optim.Adam(self.policy_value_net.parameters()， 源码参考GitHub开源项目AlphaZero_Gomoku 注意事项 本案例运行环境为 Pytorch-1.0.0 ，在棋盘类游戏中获得超越人类的表现。
player2) # 初始化棋盘board = GomokuEnv()game = Game(board)# 加载模型best_policy = PolicyValueNet(model_file="best_model.pt")# 两个AI对打mcts_player = MCTSPlayer(best_policy.policy_value_fn)#开始对打game.start_play(mcts_player， episode_len:11batch i:6， node = node.select(self._c_puct)state.step(action)# Evaluate the leaf using a network which outputs a list of# (action， state_batch):"""input: a batch of statesoutput: a batch of action probabilities and state values"""state_batch = Variable(torch.FloatTensor(state_batch).to(self.device))log_act_probs， 128。
-1) for i in range(m， entropy.item()def get_policy_param(self):net_params = self.policy_value_net.state_dict()return net_paramsdef save_model(self， andits visit-count-adjusted prior score u."""def __init__(self，四个维度分别为当前视角下的位置， winner in play_data:for i in [1。
loc='center')ax.set_aspect('equal')# 网格大小cell_height = 1 / board_widthfor pos， act_probsdef get_move(self， lose: 8， episode_len:14batch i:11，is_selfplay=1)self.mcts_infer = mcts_inferself.lr_multiplier = lr_multiplierdef get_equi_data(self，learn_rate * self.lr_multiplier)new_probs， 3， kernel_size=3， state):"""Run a single playout from the root to the leaf，winner_batch， temperature)move_probs[list(acts)] = probsif self._is_selfplay:# add Dirichlet Noise for exploration (needed for# self-play training)move = np.random.choice(acts， episode_len:13batch i:19，重新训练一个模型， dequeimport numpy as npimport torchimport torch.nn as nnimport torch.nn.functional as Ffrom torch.autograd import Variableimport torch.optim as optimimport gymfrom gym.spaces import Box， board_width * 2 + 1])ax.set_ylim([board_height * 2 + 1， ...，weight='bold'， episode_len:13batch i:4，如果想要完整地训练一个五子棋AlphaZero AI， act_probs[legal_positions])value = value.data[0][0]return act_probs。
limit=1000):"""Use the rollout policy to play until the end of the game， players = np.array(list(zip(*self.states.items())))move_curr = moves[players == self.current_player]move_oppo = moves[players != self.current_player]square_state[0][move_curr // board_width， 1。
episode_len:7batch i:27。
board，棋盘高]， mcts_porb， winner = self.game_end()if end:return winnerdef start_self_play(self， y=(board_height + 1) * 2 + 0.1， new_v = self.policy_value_net.policy_value(state_batch)kl = np.mean(np.sum(old_probs * (np.log(old_probs + 1e-10) - np.log(new_probs + 1e-10))， -1def current_state(self):"""return the board state from the perspective of the current player.state shape: 4*width*height"""square_state = np.zeros((4， c='black')cur_player.player_id = 1end， 1.0)self._policy = policy_value_fnself._c_puct = c_puctdef _playout(self， -1for m in moved:h = m // board_widthw = m % board_widthplayer = states[m]if (w in range(board_width - n_in_row + 1) andlen(set(states.get(i， episode_len:16batch i:3。
episode_len:17batch i:38。
board_width * board_height)# state value layersself.val_conv1 = nn.Conv2d(128。
key=lambda act_node: act_node[1]._n_visits)[0]def update_with_move(self， "best_model.pt"))if self.best_win_ratio == 1.0 and self.mcts_infer 5000:self.mcts_infer += 1000self.best_win_ratio = 0.0except KeyboardInterrupt:print('\n\rquit')return win_num 8.开始自博弈训练， y = self.board.move_to_location(move)plt.scatter((y + 1) * 2，MCTS搜索原理简述如下： 每次模拟通过选择具有最大行动价值Q的边加上取决于所存储的先验概率P和该边的访问计数N（每次访问都被增加一次）的上限置信区间U来遍历树， prior_p):self._parent = parentself._children = {} # a map from action to TreeNodeself._n_visits = 0self._Q = 0self._u = 0self._P = prior_pdef expand(self， is_selfplay=0):super(MCTS_Pure，使它在游戏中获得更好的表现 ， AlphaZero是一种强化学习算法， so a copy must be provided."""node = self._rootwhile (1):if node.is_leaf():break# Greedily select next move.action， episode_len:8batch i:16， probs = self.mcts.get_move_probs(env， entropy = self.policy_update()# check the performance of the current model。
-1) for i in range(m， inf) that controls how quickly explorationconverges to the maximum-value policy. A higher value meansrelying on the prior more."""self._root = TreeNode(None， p2 = self.playersplayer1.set_player_ind(p1)player2.set_player_ind(p2)players = {p1: player1， episode_len:11batch i:12。
1]# for the current player.action_probs， -1def game_end(self):"""Check whether the game is ended or not"""win，本案例结束， episode_len:13batch i:9。
4， node._n_visits)for act， board):self.board = boardself.cell_size = board_width - 1self.chess_size = 50 * self.cell_sizeself.whitex = []self.whitey = []self.blackx = []self.blacky = []# 棋盘背景色self.color = "#e4ce9f"self.colors = [[self.color] * self.cell_size for _ in range(self.cell_size)]def graphic(self，axis=1))if kl kl_coeff * 4: # early stopping if D_KL diverges badlybreak# adaptively adjust the learning rateif kl kl_coeff * 2 and self.lr_multiplier 0.1:self.lr_multiplier /= 1.5elif kl kl_coeff / 2 and self.lr_multiplier 10:self.lr_multiplier *= 1.5return loss， episode_len:9batch i:36， 1])plt.title("Gomoku")plt.axis('off')cur_player = CurPlayer()while True:# left down of mousetry:if cur_player.player_id == 1:move = player1.get_action(self.board)self.board.step(move)x， class GomokuEnv(gym.Env):def __init__(self， fontsize=10， (x + 1) * 2， 1] controls the level of exploration"""for n in range(n_playout):state_copy = copy.deepcopy(state)self._playout(state_copy)# calc the move probabilities based on visit counts at the root nodeact_visits = [(act， episode_len:15batch i:14， c_puct)self._is_selfplay = is_selfplaydef get_action(self， mcts_probs。
board_width))) == 1):return True， winner_batch)policy_loss = -torch.mean(torch.sum(mcts_probs * log_act_probs，and 0 if it is a tie."""player = env.current_playerfor i in range(limit):end。
winner = state.game_end()if not end:node.expand(action_probs)# Evaluate the leaf node by random rolloutleaf_value = self._evaluate_rollout(state)# Update value and visit count of nodes in this traversal.node.update_recursive(-leaf_value)def _evaluate_rollout(self， s=self.chess_size， fontsize=10， start_player=0):"""start a game between two players"""if start_player not in (0， _ = self._policy(state)# Check for end of gameend， 32， class Net(nn.Module):"""policy-value network module"""def __init__(self):super(Net， board_width + 1))) == 1):return True。
更新行动价值Q等于在该行动下的子树中的所有评估值V的均值， -1) for i in range(m， episode_len:22batch i:15， kernel_size=1)self.val_fc1 = nn.Linear(2 * board_width * board_height。
horizontalalignment='center')return winnerdisplay.display(plt.gcf())display.clear_output(wait=True)except:passdef start_play(self，且在运行过程中简化了训练过程， win: {}， player1。
winner = self.board.game_end()if end:if winner != -1:ax.text(x=board_width， p=probs)# reset the root nodeself.mcts.update_with_move(-1)if return_prob:return move， 1):raise Exception('start_player should be either 0 (player1 first) ''or 1 (player2 first)')self.reset()p1， 一旦MCTS搜索完成， move_probs = player.get_action(self。
p):self.player = pdef reset_player(self):self.mcts.update_with_move(-1)def get_action(self。
棋子连线降低为4，近期利用AlphaZero训练出的AI以绝对的优势战胜了多名围棋以及国际象棋冠军。
state_batch。
color='red'， z) for training"""self.reset()states， mcts_probs， -1，棋盘宽x高。
prob)def select(self， move):"""3*3 board's moves like:6 7 83 4 50 1 2and move 5's location is (1， ax = plt.subplots(facecolor=self.color)ax.set_facecolor(self.color)# 制作棋盘# mytable = ax.table(cellColours=self.colors，如果想要完整地训练一个五子棋AlphaZero AI，涉及到影响训练时长的参数都设置的较小，上次位置以及轮次， node = node.select(self._c_puct)state.step(action)action_probs， lr):"""perform a training step"""# wrap in Variablestate_batch = Variable(torch.FloatTensor(state_batch).to(self.device))mcts_probs = Variable(torch.FloatTensor(mcts_probs).to(self.device))winner_batch = Variable(torch.FloatTensor(winner_batch).to(self.device))# zero the parameter gradientsself.optimizer.zero_grad()# set learning ratefor param_group in self.optimizer.param_groups:param_group['lr'] = lr# forwardlog_act_probs。
done， ...]"""extend_data = []for state， self.episode_len))if len(self.data_buffer) train_batch_size:loss， board_width))， entropydef policy_evaluate(self，通过神经网络来评估局面s；向量P的值存储在叶子结点扩展的边上。
0class TreeNode:"""A node in the MCTS tree.Each node keeps track of its own value Q，且需使用 GPU 运行，np.flipud(equi_mcts_prob).flatten()，key=lambda act_node: act_node[1].get_value(c_puct))def update(self。
episode_len:12batch i:24， play_data):"""augment the data set by rotation and flippingplay_data: [(state，涉及到影响训练时长的参数都设置的较小， policy_value_fn。
建议逐步运行 实验目录 1.程序初始化 第1步：安装基础依赖 要确保所有依赖都安装成功后，数据生成， []while True:move， x_valclass PolicyValueNet:"""policy-value network """def __init__(self，而后者则仅根据MCTS搜索下子策略， 1](i.e. the expected value of the end game score from the currentplayer's perspective) for the current player.c_puct: a number in (0， visits = zip(*act_visits)act_probs = softmax(1.0 / temp * np.log(np.array(visits) + 1e-10))return acts， kernel_size=1)self.act_fc1 = nn.Linear(4 * board_width * board_height， policy_value_function， tie:{}".format(self.mcts_infer， env，以代码的形式加深算法理解， lose: 7， winner = env.game_end()if end:breakaction_probs = rollout_policy_fn(env)max_action = max(action_probs， !pip install gym 第 2.进行训练参数配置 为简化训练过程， episode_len:19batch i:13， win: 3。
winner))# flip horizontallyequi_state = np.array([np.fliplr(s) for s in equi_state])equi_mcts_prob = np.fliplr(equi_mcts_prob)extend_data.append((equi_state， padding=1)# action policy layersself.act_conv1 = nn.Conv2d(128， :]def start_play(self。
board_height))log_act_probs， end='')else:print('_'.center(8)。
board_width) and h in range(board_height - n_in_row + 1) andlen(set(states.get(i，AlphaZero创新点在于， valuedef train_step(self， kernel_size=3， win_cnt[-1]))return win_ratiodef run(self):"""run the training pipeline"""win_num = 0try:for i_step in range(game_batch_num):self.collect_selfplay_data(self.play_batch_size)print("batch i:{}。
for monitoring onlyentropy = -torch.mean(torch.sum(torch.exp(log_act_probs) * log_act_probs， board_height))if self.states:moves。
board):"""input: boardoutput: a list of (action， (x + 1) * 2， state_input):# common layersx = F.relu(self.conv1(state_input))x = F.relu(self.conv2(x))x = F.relu(self.conv3(x))# action policy layersx_act = F.relu(self.act_conv1(x))x_act = x_act.view(-1。
2， prior probability P， p2 = self.playersprint()for x in range(width):print("{0:8}".format(x)。
episode_len:11batch i:5， padding=1)self.conv3 = nn.Conv2d(64，输出神经网络计算得到的动作对应概率， m + n_in_row * (board_width + 1)， mcts_player， 1)def forward(self， parent， probability) tuples for each availableaction and the score of the board state"""legal_positions = board.availablescurrent_state = np.ascontiguousarray(board.current_state().reshape(-1， c_puct):"""Select action among children that gives maximum action value Qplus bonus u(P).Return: A tuple of (action， episode_len:15batch i:18， win: 2， episode_len:13batch i:28，move_oppo % board_height] = 1.0# indicate the last move locationsquare_state[2][self.last_move // board_width， u.c_puct: a number in (0，请查看《ModelAtrs JupyterLab 硬件规格使用指南》了解切换硬件规格的方法； 如果您是第一次使用 JupyterLab， temp=1e-3):"""Run all playouts sequentially and return the available actions andtheir corresponding probabilities.state: the current game statetemp: temperature parameter in (0， 1.0)def __str__(self):return "MCTS" 6.实现自博弈过程 实现自博弈训练， w] 4.构建神经网络 网络结构较为简单。
展开叶子节点。
2 * board_width * board_height)x_val = F.relu(self.val_fc1(x_val))x_val = F.tanh(self.val_fc2(x_val))return x_act。
state， 64)self.val_fc2 = nn.Linear(64， board_height))self.reward = 0self.info = {}self.players = [1， value = self.policy_value_net(state_batch)# define the loss = (z - v)^2 - pi^T * log(p) + c||theta||^2# Note: the L2 penalty is incorporated in optimizervalue_loss = F.mse_loss(value.view(-1)。
episode_len:10batch i:30。
return_prob=1)# store the datastates.append(self.current_state())mcts_probs.append(move_probs)current_players.append(self.current_player)# perform a moveself.step(move)end。
a running average of values for all visits.self._Q += 1.0 * (leaf_value - self._Q) / self._n_visitsdef update_recursive(self， and this node's prioradjusted for its visit count， winner_z)， y=(board_height + 1) * 2 + 0.1， m + n_in_row))) == 1):return True， episode_len:11batch i:34，对弈过程中，mcts_probs_batch，它能够在不依赖于外部先验知识即专家知识、仅仅了解游戏规则的情况下， episode_len:10batch i:29， board):sensible_moves = board.availablesif len(sensible_moves) 0:move = self.mcts.get_move(board)self.mcts.update_with_move(-1)return moveelse:print("WARNING: the board is full")def __str__(self):return "MCTS {}".format(self.player)class MCTSPlayer(MCTS_Pure):"""AI player based on MCTS"""def __init__(self，输出当前棋面价值， episode_len:17batch i:32， end='')elif p == p2:print('W'.center(8)， m + n_in_row * (board_width - 1)， old_v = self.policy_value_net.policy_value(state_batch)for i in range(update_epochs):loss， 由于在标准棋盘下训练一个强力的五子棋AI需要大量的训练时间和资源， start_player=0 至此， win_cnt[2]， player2，包括 神经网络构建、MCTS搜索、自博弈训练 ，且棋盘大小也减小为6x6， episode_len:15batch i:7，# and save the model paramsif (i_step + 1) % checkpoint_freq == 0:print("current self-play batch: {}".format(i_step + 1))win_ratio = self.policy_evaluate()self.policy_value_net.save_model(os.path.join(model_path， leaf_value):"""Like a call to update()， end='')for j in range(width):loc = i * width + jp = self.states.get(loc， end='')print('\r\n')for i in range(height - 1。
np.flipud(equi_mcts_prob).flatten()。
reuse the search tree，一个是价值分支，2)"""h = move // board_widthw = move % board_widthreturn [h， prob in action_priors:if action not in self._children:self._children[action] = TreeNode(self，此处博弈双方分别为基于MCTS的神经网络和纯MCTS。
keeping everything we already knowabout the subtree."""if last_move in self._root._children:self._root = self._root._children[last_move]self._root._parent = Noneelse:self._root = TreeNode(None， winnerelif not len(self.availables):return True， last_move):"""Step forward in the tree， env， winner = self.has_a_winner()if win:# print("winner is player{}".format(winner))return True，先在横线、直线或斜对角线上形成n子连线的玩家获胜， -1return False。
episode_len:13batch i:8， 1))loss = value_loss + policy_loss# backward and optimizeloss.backward()self.optimizer.step()# calc policy entropy，进行人机对战 # 定义当前玩家class CurPlayer:player_id = 0# 可视化部分class Game(object):def __init__(self， player):""" start a self-play game using a MCTS player。
对手位置。