代码收藏家技术教程 2022-07-23

强化学习系列（二):Q learning算法简介及python实现Q learning求解TSP问题

一、什么是Q learning算法？

1.Q table

2.Q-learning算法伪代码

二、Q-Learning求解TSP的python实现

1）问题定义

2）创建TSP环境

3）定义 DeliveryQAgent类

4）定义每个episode下agent学习的过程

5) 定义训练的主函数

6）实验结果

1. 环境创建

2.实例化agent类

3.agent训练学习

一、什么是Q learning算法？

Q-learning算法非常适合新手入门理解强化学习，它是最容易编码和理解的。 Q-learning算法是一种model-free、off-policy/value_based的强化学习算法，即不考虑环境的特征，通过Q函数寻找最优的动作选择策略。Q值（action value function）计算的是当前状态下采取该动作的未来奖励期望，公式推导如下：

更多强化学习基本原理和概念见强化学习系列（一）：基本原理和概念

1.Q table

Q代表quality，即动作的质量。创建一个表格Q，将state-action-Q估计值存储进去，通过检索Q表，就能获取在当前state下选取各个action能够获得的未来奖励期望的估计值，Q-learning中最核心的就是不停更新Q表给出越来越好的近似。

2.Q-learning算法伪代码

步骤一：创建并初始化一个action-space*state space大小的Q表，一般初始化设置所有值为0；

步骤二：进入循环，直到达到迭代条件：

步骤三：检索Q表，在当前状态 s下根据Q的估计值和Policy选择一个action a；

步骤四：执行action a，检索Q表，转移到的状态 $s{}'$ 对应的Q最大值加上该动作得到的实时奖励reward是状态 s价值的真实值；

步骤五：根据贝尔曼方程更新Q表。

那么，开始时Q值都为0，我们该怎么选择下一个动作呢？这个时候就体现Policy的重要性了，常见做法是引入一个参数 $\varepsilon$ ，取值在0-1之间， $\varepsilon$ 体现了探索/利用（exploration/exploitation）的权衡。 $\varepsilon$ 越大，随机性/探索性越强，通常初始情况下 $\varepsilon$ 接近或等于1随机选择下一个动作进行大量的探索；随着agent的不断学习，对Q的估计越来越准确，我们将逐渐减小 $\varepsilon$ 的值，更多依赖利用当前的Q值。

二、Q-Learning求解TSP的python实现

1）问题定义

旅行商问题（ TSP）是一个典型的优化问题，目的是找到访问各个城市的最短路线。要使用RL的方法解决TSP问题，就需要把TSP问题转化为RL问题，定义RL的各要素：

agent：送货人；

environment：要交付的商品和要访问的城市节点位置；

state：当前送货员所在的城市节点；

action：在每个节点要做出的决策，下一步去哪一个节点；

reward：实时的奖励，两个节点之间的距离多长。

RL的目标goal是使得reward的求和最大，即访问路线的距离最短。

2）创建TSP环境

在python中创建一个简单的TSP环境非常简单，指定城市节点数量，随机生成城市节点坐标；并计算不同城市间的距离作为reward值。python具体实现代码如下，创建了一个DeliveryEnvironment类，默认的城市节点数是10个，随机选择一个节点作为出发点，定义了一个画图的函数展示TSP的Environment。

#导入需要的包
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import os
import time
from tqdm import tqdm_notebook
from scipy.spatial.distance import cdist
import imageio
from matplotlib.patches import Rectangle
from matplotlib.collections import PatchCollection

plt.style.use("seaborn-dark")

import sys
sys.path.append("../")
from rl.agents.q_agent import QAgent 

class DeliveryEnvironment(object): #初始化环境
    def __init__(self,n_stops = 10,max_box = 10,method = "distance",**kwargs):

        print(f"Initialized Delivery Environment with {n_stops} random stops")
        print(f"Target metric for optimization is {method}")

        # 参数初始化
        self.n_stops = n_stops
        self.action_space = self.n_stops
        self.observation_space = self.n_stops
        self.max_box = max_box
        self.stops = []
        self.method = method

        # 产生城市节点
        self._generate_constraints(**kwargs)
        self._generate_stops()
        self._generate_q_values()
        self.render()

        # 初始化环境
        self.reset()
    

    def _generate_stops(self):  #产生城市节点
            # Generate geographical coordinates
            xy = np.random.rand(self.n_stops,2)*self.max_box  #产生客户点坐标
            self.x = xy[:,0]
            self.y = xy[:,1]

    def _generate_q_values(self,box_size = 0.2):  #计算不同节点之间的距离充当reward

        # Generate actual Q Values corresponding to time elapsed between two points
        if self.method in ["distance"]:
            xy = np.column_stack([self.x,self.y])
            self.q_stops = cdist(xy,xy) #计算距离矩阵充当reward
        else:
            raise Exception("Method not recognized")
        
    #画图的函数
    def render(self,return_img = False):
        
        fig = plt.figure(figsize=(7,7))
        ax = fig.add_subplot(111)
        ax.set_title("Delivery Stops")

        # Show stops
        ax.scatter(self.x,self.y,c = "red",s = 50)

        # Show START
        if len(self.stops)>0:
            xy = self._get_xy(initial = True)  #生成的第一个点作为start点，文本位置在xy[1]-0.05
            xytext = xy[0]+0.1,xy[1]-0.05
            ax.annotate("START",xy=xy,xytext=xytext,weight = "bold")

        # Show itinerary
        if len(self.stops) > 1:
            ax.plot(self.x[self.stops],self.y[self.stops],c = "blue",linewidth=1,linestyle="--")
            
            # 路径结尾要回到出发点
            xy = self._get_xy(initial = False)
            xytext = xy[0]+0.1,xy[1]-0.05
            ax.annotate("END",xy=xy,xytext=xytext,weight = "bold")


        if hasattr(self,"box"):
            left,bottom = self.box[0],self.box[2]
            width = self.box[1] - self.box[0]
            height = self.box[3] - self.box[2]
            rect = Rectangle((left,bottom), width, height)
            collection = PatchCollection([rect],facecolor = "red",alpha = 0.2)
            ax.add_collection(collection)


        plt.xticks([])
        plt.yticks([])
        
        if return_img:
            # From https://ndres.me/post/matplotlib-animated-gifs-easily/
            fig.canvas.draw_idle()
            image = np.frombuffer(fig.canvas.tostring_rgb(), dtype='uint8')
            image  = image.reshape(fig.canvas.get_width_height()[::-1] + (3,))
            plt.close()
            return image
        else:
            plt.show()
#重置进入下一轮迭代
    def reset(self):

        # Stops placeholder
        self.stops = []

        # Random first stop
        first_stop = np.random.randint(self.n_stops) #随机生成第一个初始节点
        self.stops.append(first_stop)

        return first_stop

#根据reward选择下一个动作
    def step(self,destination):

        # Get current state  得到当前的state
        state = self._get_state()
        new_state = destination

        # Get reward for such a move  每个action得到reward
        reward = self._get_reward(state,new_state)

        # Append new_state to stops  进入下一个state
        self.stops.append(destination)
        done = len(self.stops) == self.n_stops

        return new_state,reward,done
    
#得到当前状态 即当前到达的节点位置
    def _get_state(self):
        return self.stops[-1]

#得到每个坐标的X和Y值
    def _get_xy(self,initial = False):
        state = self.stops[0] if initial else self._get_state()
        x = self.x[state]
        y = self.y[state]
        return x,y

#定义reward函数
    def _get_reward(self,state,new_state):  #
        base_reward = self.q_stops[state,new_state] #base_reward是两个节点之间的距离

        if self.method == "distance":
            return base_reward

    @staticmethod
    def _calculate_point(x1,x2,y1,y2,x = None,y = None):

        if y1 == y2:
            return y1
        elif x1 == x2:
            return x1
        else:
            a = (y2-y1)/(x2-x1)
            b = y2 - a * x2

            if x is None:
                x = (y-b)/a
                return x
            elif y is None:
                y = a*x+b
                return y
            else:
                raise Exception("Provide x or y")

3）定义 DeliveryQAgent类

决定选择下一个节点的Policy，随机产生一个0-1之间的值，如果比 $\varepsilon$ 大，选择Q值最大的action，否则随机选择一个未访问过的节点去访问。

class DeliveryQAgent(QAgent):

    def __init__(self,*args,**kwargs):
        super().__init__(*args,**kwargs)
        self.reset_memory()

    def act(self,s):

        # Get Q Vector  copyq表
        q = np.copy(self.Q[s,:])

        # Avoid already visited states  屏蔽已经走过的节点 这个地方应该可以加约束条件起到mask一些节点的作用
        q[self.states_memory] = -np.inf

        if np.random.rand() > self.epsilon:
            a = np.argmax(q)
        else:
            a = np.random.choice([x for x in range(self.actions_size) if x not in self.states_memory])

        return a
    

    def remember_state(self,s):
        self.states_memory.append(s)

    def reset_memory(self):
        self.states_memory = []

4）定义每个episode下agent学习的过程

每一次迭代需要将环境reset到初始状态，随机选择一个节点作为初始节点，然后根据Policy不断去选择下一个节点并更新Q值。

#每一个episode学习的函数
def run_episode(env,agent,verbose = 1):

    s = env.reset()
    agent.reset_memory()

    max_step = env.n_stops
    
    episode_reward = 0
    
    i = 0
    while i < max_step:   #节点个数

        # Remember the states 存储已经走过的点
        agent.remember_state(s)

        # Choose an action 选择一个action
        a = agent.act(s)
        
        # Take the action, and get the reward from environment 得到一个reward
        s_next,r,done = env.step(a)

        # Tweak the reward  加负号最小化问题变成最大化问题
        r = -1 * r
        
        if verbose: print(s_next,r,done)
        
        # Update our knowledge in the Q-table  更新reward在Q表中
        agent.train(s,a,r,s_next)  
        
        # Update the caches  累加reward
        episode_reward += r  
        s = s_next
        
        # If the episode is terminated
        i += 1
        if done:
            break
            
    return env,agent,episode_reward

在导入的QAgent.py里定义了更新Q值的函数如下，也就是贝尔曼公式，通常epsilon_decay小于1，也就是随着学习的不断进行， $\varepsilon$ 的值在不断减小，探索性降低。

 
def train(self,s,a,r,s_next):
        self.Q[s,a] = self.Q[s,a] + self.lr * (r + self.gamma*np.max(self.Q[s_next,a]) - self.Q[s,a])

        if self.epsilon > self.epsilon_min:
            self.epsilon *= self.epsilon_decay

5)定义训练的主函数

并输出训练过程的动图及迭代训练过程中reward值得变化趋势。

#模型训练的函数
def run_n_episodes(env,agent,name="training.gif",n_episodes=1000,render_each=10,fps=10):  #训练1000次，10次画图一次

    # Store the rewards  存储下reward和图形
    rewards = []  
    imgs = []

    # Experience replay
    for i in tqdm_notebook(range(n_episodes)):

        # Run the episode 迭代学习
        env,agent,episode_reward = run_episode(env,agent,verbose = 0)
        rewards.append(episode_reward)
        
        if i % render_each == 0:
            img = env.render(return_img = True)
            imgs.append(img)

    # Show rewards  画出reward的变化趋势
    plt.figure(figsize = (15,3))
    plt.title("Rewards over training")
    plt.plot(rewards)
    plt.show()

    # Save imgs as gif
    imageio.mimsave(name,imgs,fps = fps)  #输出动图,fps是帧率（每秒播放的帧数）

    return env,agent

6）实验结果

调用以上定义的类和函数，现简单实现一个规模为500个旅行商的TSP问题。

1. 环境创建：

指定规模为500，选定衡量reward的方法是distance；

env = DeliveryEnvironment(n_stops = 500,method = "distance")  #随机生成500个节点

输出环境如下：

还可以通过在各个走过的节点之间画线可视化路径，距离画出路径的前几个点：

for i in [0,1,2,3]:  #画出接下来几步要走的路径
    env.step(i)
env.render()

2.实例化agent类

agent = DeliveryQAgent(env.observation_space,env.action_space)  #env.observation_space和action_space都是节点数

3.agent训练学习

默认迭代学习1000次，可以根据需求更改；记录训练的时间。

start=time.time()
run_n_episodes(env,agent,"training_500_stops.gif")   #训练1000次reward的变化趋势  前400次基本上是在随机选择一个节点行走，后面就用到了之前行走的经验，Q表中没有的就设置reward，有的就更新
end=time.time()
print('运行时间',end-start)

模型输出的reward变化趋势及运行时间如下: