Python详细了解强化学习算法并基于强化学习Qlearning让电脑玩flappybird游戏qq42686550的博客-

完整代码：https://github.com/Connor666/flappy_bird-RL
首先，如果你是为了追求一个非常高的强化学习效果，也就是flappy bird的分数，那么建议出门右拐选择DQN，但是如果你是为了通过一个项目，想详细了解Qleaning的原理，以及如何建立一个强化学习的模型，那么你来对地方了。

本文的代码可在我个人的gitHub上找到完整版，由于代码较长文件较多，我这边这讲解跟算法相关的关键代码。完整代码请看：https://github.com/Connor666/flappy_bird-RL
首先感谢，flappy bird游戏Python代码制作人sourabhv本游戏的代码是基于这个代码而修改的，以及我老师Dr Luke Dickens提供的一些建议。

1.什么是Q learning

强化学习是机器学习中的一个领域，强调如何基于环境而行动，以取得最大化的预期利益。其灵感来源于心理学中的行为主义理论，即有机体如何在环境给予的奖励或惩罚的刺激下，逐步形成对刺激的预期，产生能获得最大利益的习惯性行为.而为了实现强化学习，Qlearning是其中的一个主要算法。
作为一个强化学习问题，有几个要素时必须的：状态（state），奖励（reward）以及动作（action） 具体如何定义和制定这三个要素我们再后面会讲到。
对于Qlearning而言：

上图中的s,a,r分别为状态，动作与奖励。
每次采取一个动作时，会产生一个新的状态以及奖励，根据以上公式来更新Q表（Q-table），进而在下一次根据这个更新后的Q表来选取动作，在Qlearning中每次选取动作是根据一个叫贪心算法（greedy policy） 而选取行动的，贪心算法会在下一个部分提到。可以注意到，每次更新Q表的时候，是根据下一个状态选取了Q表最大的更新的，这一点与SARSA不同，也因此被叫做off-policy。
用代码表示如下：

def q_learning(         env, gamma, alpha, epsilon, num_episodes, max_steps=np.inf,         initial_Q=None):     num_states = env.num_states     num_actions = env.num_actions     if initial_Q is None:         Q = np.zeros((num_states, num_actions)) else:         Q = initial_Q     policy = get_epsilon_greedy_policy(epsilon, Q, env.absorbing) for _ in range(num_episodes): # initialise state         s = env.reset()         steps = 0         done = False while not env.is_terminal() and steps < max_steps: # choose the action             a = choose_from_policy(policy, s)             next_s, r = env.next(a) # get td_error (called delta on slides)             td_error = r + gamma * np.max(Q[next_s, :]) - Q[s, a] # update the Q function estimate             Q[s, a] += alpha * td_error             # update the policy (only need to do so for current state)             policy[s, :] = get_epsilon_greedy_policy(                 epsilon, Q[s, :].reshape((1, num_actions))) # set next state and action to current state and action             s = next_s             # increment the number of steps             steps += 1 # return the policy return get_greedy_policy(Q, absorbing=env.absorbing), Q 

2. Greedy policy

贪婪算法，举个直观的例子，加入开左边门你可以获得10元，开右边你会获得1元，你会怎么选？你肯定会选择开左边的门，因此greedy policy的逻辑就是每次选取的时候都会选取能获利最大的地方。

但是这样并不一定有利于机器学习，因为这样会让机器无论在任何情况下都去开左边的门，这样右边门的情况完全不了解，或许在右边门开了100此后收益增加到了50元呢。因此就有了ε-Soft Policies：
有ε概率，会在每一次采取动作的时候随机采取动作：
用代码写出来：

def get_epsilon_greedy_policy(epsilon, Q, absorbing=None):       num_actions = Q.shape[1]     greedy_policy = get_greedy_policy(Q, absorbing=absorbing)     policy = (1 - epsilon) * greedy_policy + epsilon * np.ones(Q.shape) / num_actions     return policy   def get_greedy_policy(Q, absorbing=None):      num_states, num_actions = Q.shape     dominant_actions = np.argmax(Q, axis=1)     policy = np.zeros((num_states, num_actions))     policy[np.arange(num_states), dominant_actions] = 1. if not absorbing is None: # np.nan means Not-a-number, so the rows for absorbing states are # not valid probability vectors         policy[absorbing, :] = np.nan     return policy 

好了，基本概念说完，我们要结合这个flappy bird游戏来看一看如何制定，状态，奖励与动作。

3. State action and reward

从易到难。
动作
首先我们思考，flappy bird有哪些动作，似乎很简单，只有跳与不跳。因此我们假设当action=1是跳，等于0不跳。
奖励
我们的目的是让鸟尽可能获得更多分，为了简化这个过程，我们可以理解为，鸟存活时间越长越好。因此我们设置每一次移动：
存活 获得一分
死亡 获得负一千分
为什么死亡要扣一千分，而存活只获得一分呢。因为每次获得一分时，是一个画面，换言之如果一个30帧的游戏，每存活一秒就获得了30分，那么如果存活10秒钟就是300分了，相比较死亡扣1000，在这种情况下比较符合实际游戏的。
状态
状态十分重要，我们希望使用较少的状态来表达出复杂的情况，因为对于一个Q表而言，其大小为numer of state * number of action:

如果你的状态太多，那么极有可能内存存不下，那么你就需要使用 functional approximation。当然对于一个flappy bird游戏而言，还不至于使用 functional approximation（当然你想用也是可以的）。
思考每次我们选择跳与不跳的因素是什么，是不是根据小鸟到下一个管子的距离？那么我们可以根据这个来指定我们的状态：

根据每次小鸟距离下一个下方管子的距离来作为state
那么已知下方管子最远x坐标为488（第一根管子）最左边坐标为0，y坐标最下方为322和最上方为48。
小鸟x坐标固定不变为57，最上方假设可达到-100，最下方为380.那么：

$Delta x1=488-57=431, Delta x2=0-57=-57$

Δ x1=488−57=431,Δ x2=0−57=−57

$Delta y1=48-380=-332, Delta y2=322+100=422$

Δ y1=48−380=−332,Δ y2=322+100=422
即有（431+57）*（422+332）=367952个状态。
但是!! 我们知道每次小鸟或管子移动的时候，并不是每次坐标+1，而是+4或-4（游戏作者提供的参数），因此理论上来说这些状态数量是可以除以16的（x除以4，y除以4）。不过即使现在，总共不到四十万个状态，属于可以接受的状态因此我们保持不变。

4. 代码表示

原游戏作者提供的代码是在一个基于pygame的一个方程，作为使用强化学习的对象，我们有必要把这个代码改成面向对象格式，并基于一些强化学习的要素：

lookup方法，是为了建立一个state的list和字典，这样当输入一个statex 和statey时可以返回一个对应的index。

 def lookup(self):         self.state_lookup = {}         state_names = []         x=int(self.SCREENWIDTH * 0.2) for x in range(0-x,int((self.SCREENWIDTH + (200)) - x)+1): for y in range(-322,440+1):                 state_names.append((x, y)) for i, (statex, statey) in enumerate(state_names):             self.state_lookup[(statex, statey)] = i 

Next方程，类似强化学习的transaction function（在此不多提），其目的是基于一个新的动作时候，产生一个与之对应的奖励和新的状态。新的奖励，就是上文提到的，新的状态就是根据look up返回的index

 def next(self,a):         crashInfo=self.mainGame(a) if crashInfo==None:             reward=1 else:             reward=-1000          state=self.state_lookup[(self.deltax,self.deltay)] #print(state,reward) return state,reward 

剩余的就是一些其他必要因素了：如终止程序（就是鸟挂了），一些重置方程（重新开始一局）等。

 def is_terminal(self): if self.crashTest[0]==True: return True else: return False 

我来将一下这个rest方程，原作者，是当按下跳跃键时，开始游戏，因此我们每次都默认游戏开始，那么设置初始动作为1（跳），并且开始第一个默认动作也为跳。

 def reset(self):         action = 1         movementInfo = self.showWelcomeAnimation(action)         self.init(movementInfo)         self.mainGame(1) return self.state_lookup[(self.deltax,self.deltay)] 

5. 其他需要注意的地方

为了加快训练速度，不像人一样一次次点，我这里加大了FPS的数值至1200，也就是每秒钟有1200个画面，是普通游戏的40倍，如果你想速度慢一点请设置为30正常游戏速度。PIPEGAPSIZE也就是上下管子的间距，原游戏为100，但同样为了加快速度，这里设置为150（难度调低了一点）。建议不要太高，不然你会发现无法收敛，因为在原作者的游戏里面，管道与下一个管道间距较小，不太利于机器去学习，所以我降低了一些难度。
然后音效部分被我注释掉了，你也可以打开（如果你对你电脑以及声卡有信心了话）反正我是遇到过因此而死机。

 def __init__(self):         self.FPS = 1200 # you can increase this value and have a quicker training         self.SCREENWIDTH  = 288         self.SCREENHEIGHT = 512         self.PIPEGAPSIZE  = 150 # gap between upper and lower part of pipe         self.BASEY        = self.SCREENHEIGHT * 0.79 # image, sound and hitmask  dicts         self.IMAGES, self.SOUNDS, self.HITMASKS = {}, {}, {} 

6. 结果展示

第一幅图为前4000次的训练效果，我们可以看见前1000次基本都是没有分的，直到后面开始增加，等达到第4000次的时候，差不多每次可以取得6分了

第二幅图为后1000次的效果（总共训练了5000次），可以看出非常明显的增长，第5000次的时候已经可以平均获得1200+分了，远超出人的能力。当然这个分数还有可能再继续增加，但是训练时间也会更长，不过可以似乎看出已经开始得分增加减慢了，有可能会停止增加。
不过无论怎样这已经是一个比较令人满意的训练结果了。

当然你也可以直接使用，我已经训练好的模型：
将第二行注释掉，开启第一行，就是从头训练模式。

    Q = 0*np.ones((sim.num_states,sim.num_actions))     Q = np.load('save.npy') 

def evaluate_learning(         sim,series_size=50, num_series=20, gamma=0.99, alpha=0.7,         epsilon=0.0001): '''     :param sim: The queuing object     :param series_size: the size of series     :param num_series: the number of series     :return: a result policy     ''' # initialise Q values #Q = 0*np.ones((sim.num_states,sim.num_actions))     Q = np.load('save.npy') print(Q.shape)     figrew, axrew = plt.subplots()     total_reward_seq = [0]     total_episodes = 0 for series in range(num_series): print("series = %r/%d" % (series,num_series) ) # Print the current stage         rewardlist=[] for episode in range(series_size):             policy, Q = q_learning(                 sim, gamma=gamma, alpha=alpha, epsilon=epsilon,                 num_episodes=1, initial_Q=Q)             rewardlist.append(                 sim.score)             total_episodes += 1         Q2=Q         total_reward_seq.append(np.mean(np.array(rewardlist)))          np.save('save.npy',Q2)     total_reward_seq = np.array(total_reward_seq)     axrew.plot(         np.arange(0, total_episodes+1, series_size),         total_reward_seq)     axrew.set_xlabel("episodes")     axrew.set_ylabel("average score") return policy 

完整代码：https://github.com/Connor666/flappy_bird-RL