强化学习实例：鸳鸯系统与动态规划求解

妖狐艹你老母 2022-11-17 13:36 209阅读 0赞

# 一、游戏介绍 #

鸳鸯系统游戏是指由两种鸳鸯在一个地图中，一只去寻找另外一只，图中有些障碍物是不能通行的，如果选择路径可以最快到达？

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjg1MjY1_size_16_color_FFFFFF_t_70][]

# 二、游戏系统搭建 #

具体可以看代码注释，主要思路是定义鸟的位置与状态，以及可以移动的方向，更新函数、碰撞检测等等，最后画图并渲染

#一些需要导入的头文件
    import pygame
    from load import *
    import random
    import numpy as np
    
    class YuanYangEnv:
        def __init__(self):
            # 状态空间0-99，共100个（10*10）
            self.states = []  # 0-99
            for i in range(0, 100):
                self.states.append(i)
            #四个移动方向
            self.actions = ['e', 'w', 'n', 's']
            self.gamma = 0.8
            # 值函数
            self.value = np.zeros((10, 10))  # 10*10的表格
    
            # 设置渲染属性
            self.viewer = None  # 一个渲染窗口
            # 帧速率是指程序每秒在屏幕山绘制图像的数目，我们可以用FPS来表示它。一般的计算机都能达到每秒60帧的速度。如果我们把帧速率讲得比较低，那么游戏也会看上去较为卡顿。
            self.FPSCLOCK = pygame.time.Clock()
            # 屏幕大小
            self.screen_size = (1200, 900)
            # 雄鸟当前位置
            self.bird_position = (0, 0)
            # 雄鸟在x方向每走一次像素为120
            self.limit_distance_x = 120
            # 雄鸟在y方向每走一次像素为90
            self.limit_distance_y = 90
            # 每个障碍物大小为120像素*90像素
            self.obstacle_size = [120, 90]
            # 一共有两个障碍物墙，每个障碍物墙由8个小障碍物组成
            self.obstacle1_x = []
            self.obstacle1_y = []
            self.obstacle2_x = []
            self.obstacle2_y = []
            for i in range(8):
                # 第一个障碍物
                self.obstacle1_x.append(360)
                if i <= 3:
                    self.obstacle1_y.append(90 * i)
                else:
                    self.obstacle1_y.append(90 * (i + 2))
                # 第二个障碍物
                self.obstacle2_x.append(720)
                if i <= 4:
                    self.obstacle2_y.append(90 * i)
                else:
                    self.obstacle2_y.append(90 * (i + 2))
            # 雄鸟初始位置
            self.bird_male_init_position = [0, 0]
            # 雄鸟当前位置
            self.bird_male_position = [0, 0]
            # 雌鸟初始位置
            self.bird_female_init_position = [1080, 0]
            # 声明路径
            self.path = []
    
        # 雄鸟碰撞检测子函数
        def collide(self, state_position):
            # 用标志flag,flag1,flag2分别表示是否与障碍物、障碍物墙1、障碍物墙2发生碰撞
            flag = 1
            flag1 = 1
            flag2 = 1
            # 检测雄鸟是否与第一个障碍物墙发生碰撞
            # 找到雄鸟与第一个障碍物所有障碍物x方向和y方向最近的障碍物的坐标差
            # 并判断最近的坐标差是否大于一个最小运动距离
            # 如果大于等于 就不会发生碰撞
            dx = []
            dy = []
            for i in range(8):
                dx1 = abs(self.obstacle1_x[i] - state_position[0])
                dx.append(dx1)
                dy1 = abs(self.obstacle1_y[i] - state_position[1])
                dy.append(dy1)
            mindx = min(dx)
            mindy = min(dy)
            if mindx >= self.limit_distance_x or mindy >= self.limit_distance_y:
                flag1 = 0  # 没碰
            # 是否与第二个障碍物墙碰撞
            second_dx = []
            second_dy = []
            for i in range(8):
                dx2 = abs(self.obstacle2_x[i] - state_position[0])
                second_dx.append(dx2)
                dy2 = abs(self.obstacle2_y[i] - state_position[1])
                second_dy.append(dy2)
            mindx = min(second_dx)
            mindy = min(second_dy)
            if mindx >= self.limit_distance_x or mindy >= self.limit_distance_y:
                flag2 = 0  # 没碰
            if flag1 == 0 and flag2 == 0:
                flag = 0  # 没碰
            # 是否超出边界，如果是，也认为发生碰撞
            if state_position[0] > 1080 or state_position[0] < 0 or state_position[1] > 810 or state_position[1] < 0:
                flag = 1  # 碰了
            # 返回碰撞标志位
            return flag
    
            # 雄鸟是否找到雌鸟子函数
    
        def find(self, state_position):
            # 设置标志位flag
            # 判断雄鸟当前位置和雌鸟位置坐标差，雄安与最小运动距离则为找到
            flag = 0
            if abs(state_position[0] - self.bird_female_init_position[0]) < self.limit_distance_x and abs(
                    state_position[1] - self.bird_female_init_position[1]) < self.limit_distance_y:
                flag = 1
            return flag
    
            # 状态转化为像素坐标子函数
    
        def state_to_position(self, state):
            i = int(state / 10)
            j = state % 10
            position = [0, 0]
            position[0] = 120 * j
            position[1] = 90 * i
            return position
    
            # 像素转化为状态坐标子函数
    
        def position_to_state(self, position):
            i = position[0] / 120
            j = position[1] / 90
            return int(i + 10 * j)
    
    
        def reset(self):
            #随机产生初始状态
            flag1=1
            flag2=1
            while flag1 or flag2 ==1:
                #随机产生初始状态，0~99，randoom.random() 产生一个0~1的随机数
                state=self.states[int(random.random()*len(self.states))]
                state_position = self.state_to_position(state)
                flag1 = self.collide(state_position)
                flag2 = self.find(state_position)
            return state
        def transform(self,state, action):
            #将当前状态转化为坐标
            current_position=self.state_to_position(state)
            next_position = [0,0]
            flag_collide=0
            flag_find=0
            #判断当前坐标是否与障碍物碰撞
            flag_collide=self.collide(current_position)
            #判断状态是否是终点
            flag_find=self.find(current_position)
            if flag_collide==1 or flag_find==1:
                return state, 0, True
            #状态转移
            if action=='e':
                next_position[0]=current_position[0]+120
                next_position[1]=current_position[1]
            if action=='s':
                next_position[0]=current_position[0]
                next_position[1]=current_position[1]+90
            if action=='w':
                next_position[0] = current_position[0] - 120
                next_position[1] = current_position[1]
            if action=='n':
                next_position[0] = current_position[0]
                next_position[1] = current_position[1] - 90
            #判断next_state是否与障碍物碰撞
            flag_collide = self.collide(next_position)
            #如果碰撞，那么回报为-1，并结束
            if flag_collide==1:
                return self.position_to_state(current_position),-1,True
            #判断是否终点
            flag_find = self.find(next_position)
            if flag_find==1:
                return self.position_to_state(next_position),1,True
            return self.position_to_state(next_position), 0, False
        def gameover(self):
            for event in pygame.event.get():
                if event.type == pygame.QUIT:
                    exit()
        def render(self):
            if self.viewer is None:
                pygame.init()
                #画一个窗口
                self.viewer=pygame.display.set_mode(self.screen_size,0,32)
                pygame.display.set_caption("yuanyang")
                #下载图片（这里我使用了自己查找的图像，导入的图像中，背景图为1200*900像素，其他三个为120*90像素）
                self.bird_male = pygame.image.load('公鸳鸯.jpeg')
                self.bird_female = pygame.image.load('母鸳鸯.jpeg')
                self.background = pygame.image.load('背景.png')
                self.obstacle = pygame.image.load('障碍物.jpeg')
                #self.viewer.blit(self.bird_male, self.bird_male_init_position)
                #在幕布上画图片
                self.viewer.blit(self.bird_female, self.bird_female_init_position)
                self.viewer.blit(self.background, (0, 0))
                self.font = pygame.font.SysFont('times', 15)
            self.viewer.blit(self.background,(0,0))
            #画直线
            for i in range(11):
                pygame.draw.lines(self.viewer, (255, 255, 255), True, ((120*i, 0), (120*i, 900)), 1)
                pygame.draw.lines(self.viewer, (255, 255, 255), True, ((0, 90* i), (1200, 90 * i)), 1)
            self.viewer.blit(self.bird_female, self.bird_female_init_position)
            #画障碍物
            for i in range(8):
                self.viewer.blit(self.obstacle, (self.obstacle1_x[i], self.obstacle1_y[i]))
                self.viewer.blit(self.obstacle, (self.obstacle2_x[i], self.obstacle2_y[i]))
            #画小鸟
            self.viewer.blit(self.bird_male,  self.bird_male_position)
            # 画值函数
            for i in range(10):
                for j in range(10):
                    surface = self.font.render(str(round(float(self.value[i, j]), 3)), True, (0, 0, 0))
                    self.viewer.blit(surface, (120 * i + 5, 90 * j + 70))
            # 画路径点
            for i in range(len(self.path)):
                rec_position = self.state_to_position(self.path[i])
                pygame.draw.rect(self.viewer, [255, 0, 0], [rec_position[0], rec_position[1], 120, 90], 3)
                surface = self.font.render(str(i), True, (255, 0, 0))
                self.viewer.blit(surface, (rec_position[0] + 5, rec_position[1] + 5))
            pygame.display.update()
            self.gameover()
            # time.sleep(0.1)
            self.FPSCLOCK.tick(30)
    
    
    if __name__=="__main__":
        yy=YuanYangEnv()
        yy.render()
        while True:
            for event in pygame.event.get():
                if event.type == pygame.QUIT:
                    exit()

# 三、游戏求解 #

使用动态规划求解，代码如下

# 便于生成随机数和时间延迟函数
    import random
    import time
    # 导入YuanYangEnv
    from YuanYangEnv import YuanYangEnv
    
    class DP_Policy_Iter:
        def __init__(self,yuanyang):
            # yuanyang为类的初始化参数,用于调用yuanyang系统
            # 将鸳鸯游戏系统的状态空间和动作空间赋给当前类的状态和动作
            self.states = yuanyang.states
            self.actions = yuanyang.actions
            # v 表示值函数
            self.v = [0.0 for i in range(len(self.states)+1)]
            # 声明一个数据结构为字典以便存储策略
            self.pi = dict()
            self.yuanyang = yuanyang
            self.gamma = yuanyang.gamma
            # 初始化策略
            for state in self.states:
                flag1 = 0
                flag2 = 0
                flag1 = yuanyang.collide(yuanyang.state_to_position(state))
                flag2 = yuanyang.find(yuanyang.state_to_position(state))
                if flag1 == 1 or flag2 == 1: continue
                # 利用随机函数初始化策略
                self.pi[state] = self.actions[int(random.random()*len(self.actions))]
    
        # 策略评估子函数
        # 策略评估包括两个训练，内层循环遍历状态空间中的每个状态，利用贝尔曼算子更新每个状态处的值函数；
        # 外层循环则迭代计算整个值函数。
        # 利用新旧值函数的累积和来控制策略评估是否收敛，如果累积和小于1e-6，那么说明收敛，则退出策略评估程序
        def policy_evaluate(self):
            # 策略评估计算值函数
            for i in range(100):
                delta = 0.0
                for state in self.states:
                    flag1 = 0
                    flag2 = 0
                    flag1 = yuanyang.collide(yuanyang.state_to_position(state))
                    flag2 = yuanyang.find(yuanyang.state_to_position(state))
                    if flag1 == 1 or flag2 ==1 : continue
                    action = self.pi[state]
                    s,r,t = yuanyang.transform(state,action)
                    # 更新值
                    new_v = r + self.gamma*self.v[s]
                    delta += abs(self.v[state]-new_v)
                    # 更新值替换原来的值函数
                    self.v[state] = new_v
                if delta < 1e-6:
                    print("策略评估迭代次数：",i)
                    break
    
    
        # 策略改善子函数
        # 有了值函数，就可以用贪婪策略对当前策略改善
        # 外循环是个状态遍历，对每个状态处的策略进行改善
        # 在每个状态处，利用当前的值函数找到对应的使之最大的动作
        def policy_improve(self):
            # 利用更新后值函数进行策略改善
            for state in self.states:
                flag1 = 0
                flag2 = 0
                flag1 = yuanyang.collide(yuanyang.state_to_position(state))
                flag2 = yuanyang.find(yuanyang.state_to_position(state))
                if flag1 == 1 or flag2 == 1: continue
                a1 = self.actions[0]
                s,r,t = yuanyang.transform(state,a1)
                v1 = r + self.gamma*self.v[s]
                # 找状态s时，采用哪种动作，值函数最大
                for action in self.actions:
                    s,r,t = yuanyang.transform(state,action)
                    if v1 < r + self.gamma*self.v[s]:
                        a1 = action
                        v1 = r + self.gamma*self.v[s]
                # 贪婪策略进行更新
                self.pi[state] = a1
    
    
        # 策略迭代子函数
        # 循环进行策略评估和策略改善，当策略不再发生变化时，结束
        def policy_iterate(self):
            for i in range(100):
                # 策略评估，变的是v
                self.policy_evaluate()
                # 策略改善
                pi_old = self.pi.copy()
                # 变的是pi
                self.policy_improve()
                if(self.pi == pi_old):
                    print("策略改善次数：",i)
                    break
    
    # 主函数测试
    # 首先，实例化一个鸳鸯游戏YuanYang;
    # 然后将之作为参数传入策略爹地啊类中实例化一个策略迭代类policy_value
    # 再利用该类调用策略迭代子函数policy_iterate()完成对策略的学习
    if __name__=="__main__":
        yuanyang = YuanYangEnv()
        policy_value = DP_Policy_Iter(yuanyang)
        policy_value.policy_iterate()
    
        # 对学到的策略进行测试，初始状态为0，当前的路径还不存在
        # 将策略迭代中学到的值函数给到游戏中，渲染出来
        flag = 1 # 标志位
        s = 0
        path = []
        # 将v值打印出来
        for state in range(100):
            i = int(state/10)
            j = state % 10
            yuanyang.value[j,i] = policy_value.v[state]
    
        step_num = 0
    
        # 下面是agent利用学到的策略pi与游戏环境进行交互
        # 并将交互结果渲染出来，在雄鸟移动过程中，我们将移动的状态和动作都打印出来
        # 将最优路径打印出来
        while flag:
            # 渲染路径点
            path.append(s)
            yuanyang.path = path
            a = policy_value.pi[s]
            print('%d->%s\t'%(s,a))
            yuanyang.bird_male_position = yuanyang.state_to_position(s)
            yuanyang.render()
            time.sleep(3)
            step_num+=1
            s_,r,t=yuanyang.transform(s,a)
            if t == True or step_num > 200: # t == True则结束transform
                flag = 0
            s = s_
        # 渲染最后的路径点
        yuanyang.bird_male_position = yuanyang.state_to_position(s)
        path.append(s)
        yuanyang.render()
        while True:
            yuanyang.render()

最后的结果为：

![20210409160324333.gif][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjg1MjY1_size_16_color_FFFFFF_t_70]: /images/20221022/30d00140d3c840a29842942b3e335d34.png
[20210409160324333.gif]: /images/20221022/205cb69690b44bbd8c78825abc55046c.png