利用强化学习训练“井字棋”下棋策略的详细介绍

文章目录

1. 引言
2. 基本要素
- 2.1 棋盘状态
- 2.2 智能体（棋手）
- 2.3 用户交互接口
3. 训练过程
- 3.1 模拟对弈过程
- 3.2 策略的训练及检验
- 3.3 用户实践

本篇文章的强化学习例子的完整代码来自刘建平老师的Github仓库。

1. 引言

井字棋游戏（Tic-Tac-Toe）是强化学习入门常用的一个简单案例，它的规则是，在一个 $3\times 3$ 的九宫格里，两个人轮流下棋，直到有个人的棋子能连成长度为 $3$ 的线（一行一列或者一对角线）则赢得比赛，如果九宫格填满后也没人胜出，则和棋。

这里通过强化学习训练一个下棋机器人，由于棋盘规模小，对局状态有限且能完全遍历，因此由人来和 AI 棋手对弈是不可能赢得了训练好之后的机器人，最多和棋。

2. 基本要素

该井字棋的强化学习过程：先生成两个 AI 棋手进行对弈，用于训练的对弈棋手的探索率 $\epsilon$ 不为 $0$ ，当训练好对弈策略之后，把探索率设置为 $0$ ，测试对弈的 AI 棋手基于训练好的策略赢棋的概率。最后创建一个 HumanPlayer 来代表用户玩家与 AI 棋手进行对局，该“棋手”的执行动作由玩家手动输入。

2.1 棋盘状态

首先先介绍下 State 棋盘状态类型，每个状态（对象）表示的是棋盘的一种布局。具体到棋盘上的每一格，共有 $3$ 种状态，分别是空格（取值 $0$ ），棋手双方的棋子（取值 $1$ 或者 $- 1$ ），一共有 $9$ 个格子，则棋盘状态空间大小为 $3^9=19683$ ，但由于是对弈双方一人下一步，且每次只能挑空白的格子下，因此实际的状态空间会更小，可以通过递归的方式生成所有的状态空间，并存放在 all_states 字典当中，键是下面会提到的状态哈希值，字典值存放的是 State 对象，以及该状态是否结束的布尔值。

棋盘的状态设置了 $4$ 个成员属性，首先是棋盘布局 data，由每个格子的状态值组成一个 $3\times 3$ 的数组；winner 记录该棋盘状态下是否有赢家；end 记录该棋盘状态是否结束（分出赢家或者和棋）；hash_val 则是一个唯一表征该棋盘状态的哈希值，目的是为了方便通过该值进行存储和检索。

class State:
    def __init__(self):
        self.data = np.zeros((BOARD_ROWS, BOARD_COLS))
        self.winner = None
        self.hash_val = None
        self.end = None

棋盘状态类还包含如下的成员方法，每个方法的逻辑如下：

（1）hash()：计算哈希值

该方法基于棋盘布局，计算出一个能唯一表征棋盘的值，用于后续能够作为哈希表的键，具体逻辑如下：

$((i_1 \times 3 + i_2)\times 3 + i_4....)\times 3 +i_9$

该哈希值的计算函数既考虑了每个格子的取值大小，也考虑每个值在棋盘当中的位置，能够保证不论是值的差异还是值位置差异，都会改变哈希值。

def hash(self):
    if self.hash_val is None:
        self.hash_val = 0
        for i in self.data.reshape(BOARD_ROWS * BOARD_COLS):
            if i == -1:
                i = 2
            self.hash_val = self.hash_val * 3 + i
    return int(self.hash_val)

（2）iS_end() ：判断是否结束

当前棋盘状态结束的条件只有两种，一种是某一方赢棋，将三个相同值的棋子连成线（按行、按列、按对角线），另一种是没人赢棋，且所有的空格都被占满，此时视为和棋。除了这两种情况，其他情况的棋局都没有结束。

# check whether a player has won the game, or it's a tie
def is_end(self):
    if self.end is not None:
        return self.end
    results = []
    # check row
    for i in range(0, BOARD_ROWS):
        results.append(np.sum(self.data[i, :]))
    # check columns
    for i in range(0, BOARD_COLS):
        results.append(np.sum(self.data[:, i]))

    # check diagonals
    results.append(0)
    for i in range(0, BOARD_ROWS):
        results[-1] += self.data[i, i]
    results.append(0)
    for i in range(0, BOARD_ROWS):
        results[-1] += self.data[i, BOARD_ROWS - 1 - i]

    for result in results:
        if result == 3:
            self.winner = 1
            self.end = True
            return self.end
        if result == -3:
            self.winner = -1
            self.end = True
            return self.end

    sum = np.sum(np.abs(self.data))
    if sum == BOARD_ROWS * BOARD_COLS:
        self.winner = 0
        self.end = True
        return self.end

    self.end = False
    return self.end

（3）next_state()：棋盘状态转化

在双方棋手每执行一个动作之后，棋盘模型的状态变化是确定，即落子的位置会确定地变为某个状态（ $0, 1, - 1$ ），因此棋盘状态的转移概率均为 $1$ ，不存在某个状态在某个动作下会以一定概率转换到不同的几种状态，这种不存在环境随机因素的问题相对简单。

def next_state(self, i, j, symbol):
    new_state = State()
    new_state.data = np.copy(self.data)
    new_state.data[i, j] = symbol
    return new_state

（4）print()：可视化输出棋盘状态

为了方便用户在下棋过程中可以直观地看到棋盘的状态，以还原最直接的下棋体验，可以通过该方法将棋盘状态输出成如下的棋盘格式，具体格式可以自定义修改。

def print(self):
        for i in range(0, BOARD_ROWS):
            print('-------------')
            out = '| '
            for j in range(0, BOARD_COLS):
                if self.data[i, j] == 1:
                    token = '*'
                if self.data[i, j] == 0:
                    token = '0'
                if self.data[i, j] == -1:
                    token = 'x'
                out += token + ' | '
            print(out)
        print('-------------')

在这里插入图片描述

2.2 智能体（棋手）

其次介绍模拟下棋的智能体 Player 类，每个智能体对象在对弈的时候，会根据对已知的状态价值的估计，来决定下一步的动作，由于棋盘状态的奖励非常确定，智能体只对下一步动作之后生成的状态价值进行判断，但这可能会错过当前步的状态价值并不是最高，但总的动作路线更优的情况。因此在训练阶段，为每个棋手增加了探索率 $\epsilon=0.1$ ，即在每次决定下一步动作时，有一定概率不按照状态价值进行判断，而随机选择一个动作。

智能体类 Player 有 $6$ 个成员属性，分别为：estimations 表示智能体对所有棋盘状态的价值估计；step_size 是更新价值估计的学习率；epsilon 是棋手随机落子的概率；states 是棋手在下棋过程中，棋盘状态的每一次变化；greedy 是棋盘每一次变化时，该棋手是否是选择了当前看来最优的落子位置，是的话值为 $1$ ，反之值为 $- 1$ ，意味着棋手在当前步是随机落子的；symbol 是下棋双方的编号，值为 $1$ 的是先手落子，值为 $- 1$ 的是后手落子。

class Player:
    def __init__(self, step_size=0.1, epsilon=0.1):
        self.estimations = dict()
        self.step_size = step_size
        self.epsilon = epsilon
        self.states = []
        self.greedy = []
        self.symbol = None

智能体（棋手）类还包含如下的成员方法，每个方法的逻辑如下：

（1）reset() 和 set_symbol()：重置智能体的相关属性

由于创建的两个智能体要重复地进行对弈，因此在每次进行对弈之前都需要对智能体中存放历史对弈情况的属性进行重置，包括 states、greedy、symbol，和智能体对各个棋盘状态的价值估计 estimations。

def reset(self):
    self.states = []
    self.greedy = []

def set_symbol(self, symbol):
    self.symbol = symbol
    for hash_val in all_states.keys():
        (state, is_end) = all_states[hash_val]
        if is_end: 
            if state.winner == self.symbol:
                self.estimations[hash_val] = 1.0
            elif state.winner == 0:
                self.estimations[hash_val] = 0.5
            else:
                self.estimations[hash_val] = 0
        else:
            self.estimations[hash_val] = 0.5

（2）set_state() 存储棋盘状态

每当棋盘状态发生改变之后，都会将棋盘状态更新到棋手双方的 states 和 greedy 属性当中，以便在对弈结束后能通过 backup() 函数对状态价值估计进行更新。

def set_state(self, state):
    self.states.append(state)
    self.greedy.append(True)

（3）act() 基于棋盘状态的下棋策略

前面说到，每次棋盘发生改变时，都会将棋盘状态存储到 states，当轮到某一棋手下棋，则从 states 中取出最后的棋盘状态，并找到该棋盘状态上的所有的空的格子，以一定概率（探索率 $\epsilon$ ）随机选择落子位置，以 $1-\epsilon$ 概率选择落子后棋盘状态价值最大的位置。

def act(self):
    state = self.states[-1] 
    next_states = []
    next_positions = []
    for i in range(BOARD_ROWS):
        for j in range(BOARD_COLS):
            if state.data[i, j] == 0:
                next_positions.append([i, j])
                next_states.append(state.next_state(i, j, self.symbol).hash())

    if np.random.rand() < self.epsilon:
        action = next_positions[np.random.randint(len(next_positions))]
        action.append(self.symbol)
        self.greedy[-1] = False
        return action

    values = []
    for hash, pos in zip(next_states, next_positions):
        values.append((self.estimations[hash], pos))
    np.random.shuffle(values)
    values.sort(key=lambda x: x[0], reverse=True)
    action = values[0][1]
    action.append(self.symbol)
    return action

（4）backup()更新状态价值估计

按照最开始的状态价值估计，每次挑选落子后棋盘状态最大的位置（贪心），或者是用随机的方式落子（随机）。如果是贪心的方法，且状态价值提升 $0.5\rightarrow1$ ，则加强该状态的价值估计，因为碰到这个状态则只差一步就能取胜；反之，如果价值状态下降 $0.5\rightarrow 0$ ，则会降低该状态的价值估计，只差一步就输棋。具体更新过程如下：

$q(s_t)=q(s_t)+\alpha(q(s_{t+1})-q(s))$

def backup(self):
    self.states = [state.hash() for state in self.states]
    for i in reversed(range(len(self.states) - 1)):
        state = self.states[i]
        td_error = self.greedy[i] * (self.estimations[self.states[i + 1]] - self.estimations[state])
        self.estimations[state] += self.step_size * td_error

这里的状态价值函数更新只考虑当前动作的现有价值，以及下一个状态的估计价值，忽略了再往后的棋盘状态的价值，主要是由于三格井字棋的简易性和特殊性，能在一步内决定赢或者输，此时的奖励衰减因子 $\gamma=0$ 。而学习率能加快棋手策略的收敛，多次训练后面对相同棋盘状态总能采用相同的动作。

（5）save_policy() 和 load_policy()：读写状态价值估计（策略）

当训练好智能体对各个状态的价值估计函数后，即获得了智能体的对局策略。为了保持策略的持久化，保证程序重启或退出时，已训练好的策略也不会丢失，避免从头开始训练，甚至于方便将策略分享出去，在各个不同的环境下进行测试验证，都离不开对训练好的策略进行本地化存储。具体的存储和加载操作如下：

def save_policy(self):
   with open('policy_%s.bin' % ('first' if self.symbol == 1 else 'second'), 'wb') as f:
        pickle.dump(self.estimations, f)

def load_policy(self):
    with open('policy_%s.bin' % ('first' if self.symbol == 1 else 'second'), 'rb') as f:
        self.estimations = pickle.load(f)

2.3 用户交互接口

强化学习代码会创建两个智能体进行对弈训练，当训练结束后，需要由用户作为其中的一方棋手，为了保持对弈程序的正常使用，需要模拟一个用户智能体的类 HumanPlayer 来进行对弈，只不过该智能体的一些属性和方法由用户的操作替代，例如，该对象没有存储对各个状态的价值估计，因为用户本身就有自己的策略。具体与 Player 类相似，如下：

class HumanPlayer:
    def __init__(self, **kwargs):
        self.symbol = None
        self.keys = ['q', 'w', 'e', 'a', 's', 'd', 'z', 'x', 'c']
        self.state = None
        return

    def reset(self):
        return

    def set_state(self, state):
        self.state = state

    def set_symbol(self, symbol):
        self.symbol = symbol
        return

    def backup(self, _):
        return

    def act(self):
        self.state.print()
        key = input("Input your position:")
        data = self.keys.index(key)
        i = data // int(BOARD_COLS)
        j = data % BOARD_COLS
        return (i, j, self.symbol)

注意这里，会通过用户输入的关键字符，来映射用户所要落子的位置，具体也可以通过行列值来输入想要落子的位置，只要能唯一指代棋盘上个各个位置即可。

在这里插入图片描述

3. 训练过程

这里强化训练的关键是，先让两个棋手对弈，然后根据它们的对弈结果，来回过头来调整它们的下棋策略，循环反复，直到达到了最大的训练代数。因此先来看是如何让两个智能体模拟对弈的过程的。

3.1 模拟对弈过程

模拟对弈其实可以视为是在实战，从实战中获取训练数据，通过不断地调整来优化棋手的下棋策略，以保证大概率能获得最大的奖励（赢棋或和棋），对于九宫格井字棋规模较小的游戏而言，训练好的 AI 棋手能保证不输棋。

具体的模拟对弈过程 play() 的逻辑如下：

首先初始化一个 Judger 对象，传入两个 Player 对象，根据传入的顺序给两个棋手分别安排下棋的顺序，例如，先手的 symbol 等于 $1$ ，后手的则为 $- 1$ ，并通过 set_symbol() 初始化每个棋手对棋盘状态的价值估计；
调用 Judger.play() 执行对弈：
1. 重置两个棋手的历史对弈记录 states 和 greedy；
2. 初始化当前棋盘状态为空棋盘；
3. 通过 alternate() 生成器来轮流取出其中一方棋手来下棋，每个棋手根据自身的状态价值估计来决定下哪里，落子后更新棋盘状态并将该棋盘状态录入两个棋手的状态历史 states，循环往复，直到对弈结束，返回对局的赢家（返回结果为 $1, - 1, 0$ ，和棋用 $0$ 表示）。

class Judger:
    def __init__(self, player1, player2):
        self.p1 = player1
        self.p2 = player2
        self.current_player = None
        self.p1_symbol = 1
        self.p2_symbol = -1
        self.p1.set_symbol(self.p1_symbol)
        self.p2.set_symbol(self.p2_symbol)
        self.current_state = State()    

    def reset(self):
        self.p1.reset()
        self.p2.reset()

    def alternate(self):
        while True:
            yield self.p1
            yield self.p2

    def play(self, print=False):
        alternator = self.alternate()
        self.reset()
        current_state = State()
        self.p1.set_state(current_state)
        self.p2.set_state(current_state)
        while True:
            player = next(alternator)
            if print:
                current_state.print()
            [i, j, symbol] = player.act()
            next_state_hash = current_state.next_state(i, j, symbol).hash()
            current_state, is_end = all_states[next_state_hash]
            self.p1.set_state(current_state)
            self.p2.set_state(current_state)
            if is_end:
                if print:
                    current_state.print()
                return current_state.winner

3.2 策略的训练及检验

前面提到，训练过程首先会创建两个 AI 棋手，然后将这两个对象传入 Judger 进行初始化，此时这两个 Player 会围绕着训练的整个过程，且棋手的状态价值评估函数 estimations 会一直更新下去。

这里的训练过程，一共训练 $epochs=10^5$ ，并记录下训练过程中双方棋手的赢局次数，每训练一次，调用一次 judger.play()，会重置一次棋手的历史下棋记录，但不会重置棋手的状态价值评估函数，只会根据对弈结束后，回顾每次棋盘状态变化后的价值，来更新对棋盘状态的价值评估。当循环训练结束后，保存两个棋手的对弈策略。

def train(epochs):
    player1 = Player(epsilon=0.01)
    player2 = Player(epsilon=0.01)
    
    judger = Judger(player1, player2)
    player1_win = 0.0
    player2_win = 0.0
    for i in range(1, epochs + 1):
        winner = judger.play(print=False)
        if winner == 1:
            player1_win += 1
        if winner == -1:
            player2_win += 1
        print('Epoch %d, player 1 win %.02f, player 2 win %.02f' % (i, player1_win / i, player2_win / i))
        player1.backup()
        player2.backup()
        judger.reset()
    player1.save_policy()
    player2.save_policy()

注意：根据实际训练过程中打印的结果，可以看到对弈的棋手都有一定的概率赢棋，概率大约在 $0.01\sim0.03$ ，这是由于训练过程中设置了对弈双方的探索率 $\epsilon=0.01$ 导致的，即有一定概率因随机落子而输棋。因此，为了检验训练后的对弈策略效果，额外安排了 $\epsilon=0$ 的对弈检验。流程与训练过程基本一致，只是在初始化 player 对象时，将已经训练好的对弈策略加载进来，且不再对策略进行调整，具体过程如下。

def compete(turns):
    player1 = Player(epsilon=0)
    player2 = Player(epsilon=0)
    judger = Judger(player1, player2)
    player1.load_policy()
    player2.load_policy()
    player1_win = 0.0
    player2_win = 0.0
    for i in range(0, turns):
        winner = judger.play()
        if winner == 1:
            player1_win += 1
        if winner == -1:
            player2_win += 1
        judger.reset()
    print('%d turns, player 1 win %.02f, player 2 win %.02f' % (turns, player1_win / turns, player2_win / turns))

结果如下，经过 $1000$ 次的检验发现，对弈的双方彼此都不能赢对方，这说明对弈策略是成功的。

1000 turns, player 1 win 0.00, player 2 win 0.00

3.3 用户实践

最后是用户实践，即让人与训练好下棋策略进行 PK，如下代码，创建了 HumanPlayer 的对象代表人，从初始话 judger 可以看出，用户是作为先手下棋的。而每次轮到用户下棋，则会先输出当前的棋局状态，再询问用户的落子位置，循环反复，直到对局结束后，重新开启新的对局。

def play():
    while True:
        player1 = HumanPlayer()
        player2 = Player(epsilon=0)
        judger = Judger(player1, player2)
        player2.load_policy()
        winner = judger.play()
        if winner == player2.symbol:
            print("You lose!")
        elif winner == player1.symbol:
            print("You win!")
        else:
            print("It is a tie!")

该强化学习案例的主程序入口如下：

if __name__ == '__main__':
    train(int(1e5))
    compete(int(1e3))
    play()

以上就是用强化学习训练九宫格井字棋的简单案例的全部内容，后续会继续分享相关有趣的案例，并结合主流的一些强化学习框架进行介绍，欢迎交流~