第十六章：五子棋AI编程
我们在第七章学习了编写五子棋游戏，在第九章学习了蒙特卡罗方法，本章将会基于蒙特卡罗方法来构造一个能自动下棋的AI引擎，能和人类玩家对弈。本章介绍的具体称为蒙特卡罗树方法，简写为MCTS，它会利用树类的数据结构，并结合蒙特卡罗模拟来计算。
一、MCTS的整体思路
当人类下棋的时候，我们实际上是在做选择题，也就是给定当前的棋盘局面，我们需要从N个可选择的行动中，选择一个可能最有利的位置落子。如果预先有神仙告诉我们，这N个位置下对应的N个可能的胜率，我们自然就好办些，直接选择胜率最高的那个位置落子即可，就像是开了外挂一样。但这个外挂怎么得到呢？
想像一下，如果我们可以拿到一份完美的棋谱，里面记载了从古到今所有名家高手的所有下棋数据，那我们可以来做一个简单的统计。给定某个局面，汇总计算下所有棋谱中下一步的落子位置，就能得到N个可能的落子位置，以及对应的次数。棋谱中出现次数最多的落子位置可能是比较好的位置，我们就可以直接落子在这个位置上。
为什么棋谱上出现多的落子位置更好呢？就像是我们徒步进入一个山区，因为不熟悉而迷失了方向。此时我们面前有两个岔路，一条路被人踩踏的次数更多一些，路面更宽一些，说明选择这条路的人多，另一条路窄一些，林木更茂盛些，说明选择这条路的人少一些。我们自然会选择路宽一些的方向，因为这条路走的人多，可能更有希望走出去。这种决策思路就是我们上面走棋时的思路。这种方法有个前提条件，就是需要历史信息来进行汇总统计。这个历史信息是从棋谱而来，那这个棋谱的信息又从何而来呢？
退一步想，如果我们得不到这份完美的棋谱，我们可不可以自己生成一份棋谱呢？如果我们有两个会下棋的机器人，虽然它们不是名家高手，水平一般的很。但机器人可以对弈很多盘，三个臭皮匠赛过诸葛亮。所以一种可能的思路是，当面临某个局面我们不知道如何落子应对时，我们让机器人基于此棋局去反复多轮对弈，保存对弈棋谱和最终胜负结局，再根据这份机器人棋谱来得到统计数据，进而来帮助我们实际的落子决策。这里还存在一个疑问，机器人只知道下棋的规则，机器人如何去选择落子呢？随机落子是一个思路。虽然随机走棋有点瞎搞的意思，不过暴力出奇迹。
整个故事的逻辑可以归纳如下，让两个机器人之间基于随机落子反复模拟对弈，得到大量棋谱数据，基于棋谱数据进行统计汇总，得到某个局面下所有可能的落子位置、次数和胜负情况。在真正对弈的时候，基于这个统计数据来进行决策。如果是以之前的山区徒步例子来类比的话，我们面临很多岔路，我们也不知道哪些路更宽，哪些路更窄。我们就派出很多的机器人去探路，让他们返回信息。这就是我们在第九章最后一小节玩井字棋的编程思路。
不过，玩井字棋的编程思路有个缺点。我们每次模拟的时候，它们的信息并不是共享的。也就是说，第一轮模拟后得到的落子位置价值信息，并没在第二轮模拟时用上。各轮模拟都是从头开始的。如果能把这些信息累积起来，计算效果将会更上一层楼。就类似于我们每碰到一个岔路，都需要重新派出机器人去探路，获取新的信息。但是有可能这个岔路在过去已经被探测过了。这些被抛弃的历史信息是有些浪费的。实际上，前一次的历史探路信息，我们也可以利用起来。所以我们需要在每个岔路口，都要让探路机器人能保存一些信息。从计算机编程上考虑，这就需要一个树状的数据结构。每个岔路口就是树上的一个节点，代替了某个局面状态信息，而这个节点上会带有信息属性。那探路机器人需要保留什么样的信息呢？需要机器人留下两个路牌插在岔路口，一个是Q，一个是N，Q意思是顺利走出山区的概率，数字N表示有多少机器人走过这条路。
这里还有一个问题，探路机器人要怎么探路？有两种机器人探路的方式，一种是当机器人来到岔路口，发现路牌上什么信息都没有，也就是这个方向上还没有任何机器人来探索过，那么它就采取随机的方式来选择方向。另一种是当机器人来到岔路口，发现路牌上已经有信息了，说明这个方向上曾经有机器人来探索过，那它应该如何选择？此时需要综合考虑探索和利用。
和强化学习中的探索目的一样，探索是为了挖掘更多信息，利用是为了结果更好。就好像我们去餐厅吃饭，餐厅A我们去过十几次了，感觉不错，可以打70分，餐厅B我们只去过一次，那次感觉一般，只有60分，但是因为去餐厅B的次数少，也许厨师发挥不好呢，我们还是可能选择去餐厅B，给它更多的机会。所以我们会有一个公式来综合探索和利用，这个称为UCT。当机器人来到路牌上有信息的路口时，会根据UCT公式来计算，选择UCT值最大的那个方向前进。
那我们总结一下，棋局中的不同状态发展就像一个充满岔路的山区，计算机中用树类数据结构来表示。进入山区后，我们首先派出机器人来探路，机器人有两种探路方式，来获取各选择方向的信息。当人类拿到信息后，根据多数机器人走的方向来决策。
换到下棋的例子，每一个节点就是一个棋局状态，真实玩家有不同的落子选择，当选择其中一个落子后，就进入一个新的棋局状态，也称为子节点。玩家选择什么位置落子，依然是根据机器人模拟走棋得到的信息，看哪个位置是更多机器人选择的落子位置。那么机器人如何模拟下棋落子，也是有两种方式，一种方式是当某个节点没有任何信息时，则进入随机对弈的走子方式，当节点存在信息时，则基于信息和UCT来对弈走子。
二、MCTS代码实现
首先需要定义一个树节点类TreeNode，它用于保存每一个棋局状态节点信息，其内部包括了几个要素，节点的上层父节点是谁，子节点有哪些，这个节点被访问了多少次n_visits，也就是有多少个机器人从这个方向走过，以及重要的Q，它最后成功的概率，以及u值，这个u值是根据n_visits计算出来的。get_value函数的返回值就是UCT公式，它平衡考虑了探索和利用。c_puct是一个超参数，用来平衡二者。
class TreeNode:
def __init__(self, parent):
self._parent = parent
self._children = {}
self._n_visits = 0
self._Q = 0
self._u = 0
def get_value(self, c_puct):
self._u = (c_puct * np.sqrt(self._parent._n_visits) / (1 + self._n_visits))
return self._Q + self._u
expand函数是用于扩展子节点的函数，当棋局从第一步开始时，我们只有一个空棋盘状态，其子节点就若干个可落子位置的节点。函数中基于for循环，将可落子的位置放入_children字典中，构造成子节点。
def expand(self, actions):
for action in actions:
if action not in self._children:
self._children[action] = TreeNode(self)
当机器人来到岔路口时，需要进行探路的选择，如果节点被访问过，也就是存在路牌信息时，根据get_value函数计算出来的值选择最大的子节点。select函数就是通过计算节点中的UCT值，选择最大值对应的子节点。
def select(self, c_puct):
return max(self._children.items(),
key=lambda act_node: act_node[1].get_value(c_puct))
update函数是用于更新本节点的路牌信息，当机器人访问过这个节点后，其n_visits会增加，机器人最终的探路结果，也就是leaf_value值，也会送回这个节点进行累计。update_recursive函数则是用于递归更新各层父节点的路牌信息。这里之所以将leaf_value转换成负值，是因为子节点和父节点之间玩家交换了，在某个玩家视角来看，能带来胜利的局面，是对手玩家不愿意看到的。
def update(self, leaf_value):
self._n_visits += 1
self._Q += 1.0*(leaf_value - self._Q) / self._n_visits
def update_recursive(self, leaf_value):
if self._parent:
self._parent.update_recursive(-leaf_value)
self.update(leaf_value)
基础的节点类完成后，我们会基于节点类来构造MCTS类。初始化函数确定了两个参数，一个是c_puct系数，另一个是机器人的模拟探路的次数n_playout，同时定义了根节点。
class MCTS:
def __init__(self, c_puct=5, n_playout=400):
self._root = TreeNode(parent=None)
self._c_puct = c_puct
self._n_playout = n_playout
playout函数的功能是负责给定当前局面下进行一轮模拟下棋，它以根节点为起始节点，只要它还有子节点，就会从子节点中去找到最好的子节点，也就是找到UCT值最大的方向去选择行动，然后在棋盘上去反馈这个行动。当到达叶节点，也就是没有子节点的状态时，检查游戏是否结束，如果没结束，就扩展叶节点，得到对应的子节点，因为叶节点是没有路牌信息的，所以用随机走子的函数去下棋，得到叶节点的信息，最后用更新函数更新路牌信息。
def _playout(self, state):
node = self._root
while(1):
if node.is_leaf():
break
action, node = node.select(self._c_puct)
state.do_move(action)
end, winner = state.game_end()
if not end:
node.expand(state.availables)
leaf_value = self._evaluate_rollout(state)
node.update_recursive(-leaf_value)
函数evaluate_rollout是负责模拟双方均为随机走子，直至下完一局棋的功能。核心部分是根据rollout_policy_fn函数生成的概率来选择概率最大的一步，输入到do_move函数中进行落子。得到最终棋局结果。
def _evaluate_rollout(self, state, limit=5000):
player = state.current_player
for i in range(limit):
end, winner = state.game_end()
if end:
break
action_probs = self.rollout_policy_fn(state)
max_action = max(action_probs, key=itemgetter(1))[0]
state.do_move(max_action)
else:
print("WARNING: rollout reached move limit")
if winner == -1: # tie
return 0
else:
return 1 if winner == player else -1
静态函数rollout_policy_fn是负责一步的随机概率生成，它从当前状态state中获取所有可落子位置，然后返回这些位置的概率值，它们都是随机生成了，所以上面选择最大值，也是随机选择的。
@staticmethod
def rollout_policy_fn(state):
action_probs = np.random.rand(len(state.availables))
return zip(state.availables, action_probs)
最后两个函数负责真实下棋的决策。get_move函数会负责运行多遍搜索，也就是将playout函数运行多次，基于当前状态来计算哪个子节点的访问次数最大，选择最多人走的方向。实际走子后，选择最好的子节点就成为当前节点，所以利用update_with_move函数，来将根节点更新为其子节点。
def get_move(self, state):
for n in trange(self._n_playout):
state_copy = copy.deepcopy(state)
self._playout(state_copy)
return max(self._root._children.items(),
key=lambda act_node: act_node[1]._n_visits)[0]
def update_with_move(self, last_move):
if last_move in self._root._children:
self._root = self._root._children[last_move]
self._root._parent = None
else:
self._root = TreeNode(None)
最后，我们来构造MCTSPlayer类，初始化函数中对MCTS类进行了实例化。reset_player用于树结构的信息重置。get_action函数负责最重要的对弈功能，它封装了对弈的基本逻辑，当人类玩家走子后，它从树结构数据中找到人类玩家的行动节点，进行信息更新，以此为根节点。再调用get_move函数，计算得到最优走子位置，再将此位置作为新的根节点。
class MCTSPlayer:
def __init__(self, c_puct=5, n_playout=400):
self.mcts = MCTS(c_puct, n_playout)
def reset_player(self):
self.mcts.update_with_move(-1)
def get_action(self, board,is_selfplay=False,print_probs_value=0):
sensible_moves = board.availables
if board.last_move != -1:
self.mcts.update_with_move(last_move=board.last_move)
if len(sensible_moves) > 0:
move = self.mcts.get_move(board)
self.mcts.update_with_move(move)
else:
print("WARNING: the board is full")
return move, None
完整的代码可以参考mcts_pure.py文件。
三、五子棋游戏程序改造
在第七章的五子棋代码中，我们构造了Board类和Game类，通过Game类中的play_human函数来和用户交互。因为假设是两个人类玩家在下棋，因此我们只需要考虑人类玩家的输入即可。在本章中，我们会加入AI玩家来和人类玩家对弈。AI玩家将基于前文描述的MCTS算法来进行落子决策。五子棋程序中本有的代码不需要修改，我们只需要增加一个和AI对弈的函数就可以了。这个函数的功能概述如下：
- 初始化基于MCTS的AI引擎对象
- 判断当前是人类玩家落子还是AI玩家落子
- 如果是人类玩家落子则需要负责鼠标输入，如果是AI玩家落子则需要将当前棋局状态信息输入，得到AI计算的落子位置
- 将玩家的落子位置进行棋盘渲染，并进行棋局状态判断，以确定是否结束棋局。
我们将整体逻辑包装在一个play_AI函数中。首先会重置AI玩家状态和棋盘状态。之后进入游戏主循环。根据不同玩家的编号，来显示提示信息，并在棋盘的消息区域显示这些信息。
def play_AI(self, AI, start_player=1):
AI.reset_player()
self.board.reset_board(start_player)
while True:
if not self.game_end:
print('current_player', self.board.current_player)
if self.board.current_player == 1:
text = "请玩家{x}落子，人类到你了".format(x=self.board.current_player)
else:
text = "请玩家{x}落子，AI思考中".format(x=self.board.current_player)
self.message_area.draw(self.surface,text,self.TextSize)
如果判断当前玩家轮换到AI了，而且游戏还未结束，则调用AI的get_action函数，让它根据当前棋局状态来输出落子位置。如果轮到人类玩家，则会接受人类玩家输入，根据其鼠标点击位置，进行相应的操作。这部分和第七章是类似的。
if self.board.current_player == 2 and not self.game_end:
move, _ = AI.get_action(self.board)
else:
user_input = self.get_input()
if user_input.action == 'quit':
break
if user_input.action == 'RestartGame':
self.game_end = False
self.board.reset_board(start_player)
self.restart_game()
continue
if user_input.action == 'SwitchPlayer':
self.game_end = False
start_player = start_player % 2 + 1
self.board.reset_board(start_player)
self.restart_game()
continue
if user_input.action == 'move' and not self.game_end:
move = user_input.value
当处理完游戏参与双方的输入后，需要根据它们的输入信息来处理和显示落子。首先是用render_step来处理渲染落子，并在棋盘对象中处理落子。使用game_end函数来判断是否对弈结束。如果游戏结束，则显示相应的信息。
if not self.game_end:
self.render_step(move)
self.board.do_move(move)
self.game_end, winner = self.board.game_end()
if self.game_end:
if winner != -1:
print("Game end. Winner is player", winner)
text = "玩家{x}胜利".format(x=winner)
self.message_area.draw(self.surface,text,self.TextSize)
else:
text = "二位旗鼓相当！"
self.message_area.draw(self.surface,text,self.TextSize)
在实际运行时，只需要将MCTSPlayer类进行实例化，再运行play_AI函数即可。
if __name__ == '__main__':
board_size = 9
n = 5
start_player = 1
game = Game(width=board_size, height=board_size, n_in_row=n)
mcts_player = MCTSPlayer(5,10000)
game.play_AI(mcts_player,start_player)
完整的代码可以参见gomoku.py文件。然后我们来运行游戏，在终端输入如下命令
本章小结
本章实现在蒙特卡罗树搜索算法，它的代码实现会比较难以理解，但基本思路和第九章的例子是类似的。只是为了让每次模拟的结果能在后续重复使用，我们构造了一个树的数据结构。大家可以和这个算法进行多轮对弈，看看这个算法的下棋水平如何。你会发现，当棋盘比较大的时候，搜索空间会比较大，算法难以准确评估每个落子位置的价值。所以棋盘较大的时候，需要更多轮的搜索模拟计算。我们默认的值是设置为一万，你可以尝试用更大或更小的数值来对弈看看。