• 强化深度学习中使用Dyna-Q算法确定机器人问题中不同规划的学习和策略实战（超详细 附源码）

    需要源码请点赞关注收藏后评论区留下QQ并且私信~~~

    一、模型、学习、规划简介

    1：模型

    Agent可以通过模型来预测环境并做出反应，这里所说的模型通常指模拟模型，即在给定一个状态和动作时，通过模型可以对下一状态和奖赏做出预测

    模型通常可以分为分布模型和样本模型两种类型

    分布模型：该模型可以生成所有可能的结果及其对应的概率分布

    样本模型：该模型能够从所有可能的情况中产生一个确定的结果

    从功能上讲，模型是用于模拟环境和产生模拟经验的。与样本模型相比，分布模型包含更多的信息，只是现实任务中难以获得所有的状态转移概率

    2：学习

    学习过程是从环境产生的真实经验中进行学习，根据经验的使用方法，学习过程可以分为直接强化学习和模型学习两种类型

    直接强化学习：在真实环境中采集真实经验，根据真实经验直接更新值函数或策略，不受模型偏差的影响

    模型学习：在真实环境中采集真实经验，根据真实经验来构建和改进模拟模型，提供模拟模型精度，使其更接近真实环境

    3：规划

    规划过程是基于模拟环境或经验模型，从模拟经验中更新值函数，实现改进策略的目的，学习和规划的核心都是通过回溯操作来评估值函数，不同之处在于：在规划过程中，Agent并没有与真实环境交互

    规划通常可分为状态空间规划和方案空间规划，状态空间规划是在状态空间中寻找最优策略，值函数的计算都是基于状态的，通常该规划方法视为搜索方法，其基本思想如下

    1：所有规划算法都以计算值函数作为策略改进的中间关键步骤

    2：所有规划算法都可以通过基于模型产生的模拟经验来计算值函数

     二、Dyna-Q结构及其算法

    Dyna-Q架构包含了在线规划Agent所需要的主要功能，该架构讲学习和规划有机地结合在一起，是有模型和无模型方法的融合，其数据来源包括基于真实环境采样的真实经验以及基于模拟模型采样的模拟经验，通过直接强化学习或间接强化学习来更新值函数或者策略

    架构图如下

     三、Dyna-Q不同规划对学习步数的影响

    机器人环境搭建以及背景可点击如下链接了解

    机器人环境

    此处比较不同规划步数对实验效果的影响，当机器人离开边界或者撞到障碍物则得到-10的奖赏，到达充电桩获得+1的奖赏，其他情况奖赏均为0，不同需要不同情节数，可视化结果如下

     

     代码如下

    1.  
    2.  
    3. import gym
    4. from gym import spaces
    5. from gym.utils import seeding
    6. from random import random, choice
    7. import matplotlib.pyplot as plt
    8. from matplotlib.font_manager import FontProperties
    9.  
    10.  
    11. class Grid(object):
    12.     def __init__(self, x:int = None,
    13.                        y:int = None,
    14.                        type:int = 0,
    15.                        reward:float = 0.0):
    16.         self.x = x                          # 坐标x
    17.         self.y = y
    18.         self.type = type                    # 类别值（0：空；1：障碍或边界）
    19.         self.reward = reward                # 该格子的即时奖励
    20.         self.name = None                    # 该格子的名称
    21.         self._update_name()
    22.  
    23.     def _update_name(self):
    24.         self.name = "X{0}-Y{1}".format(self.x, self.y)
    25.  
    26.     def __str__(self):
    27.         return "name:{3}, x:{0}, y:{1}, type:{2}".format(self.x,
    28.                                                                     self.y,
    29.                                                                     self.type,
    30.                                                                     self.name
    31.                                                                     )
    32.  
    33. class GridMatrix(object):
    34.     def __init__(self, n_width:int,                 # 水平方向格子数
    35.                        n_height:int,                # 竖直方向格子数
    36.                        default_type:int = 0,        # 默认类型
    37.                        default_reward:float = 0.0,  # 默认即时奖励值
    38.                        ):
    39.         self.grids = None
    40.         self.n_height = n_height
    41.         self.n_width = n_width
    42.         self.len = n_width * n_height
    43.         self.default_reward = default_reward
    44.         self.default_type = default_type
    45.         self.reset()
    46.  
    47.     def reset(self):
    48.         self.grids = []
    49.         for x in range(self.n_height):
    50.             for y in range(self.n_width):
    51.                 self.grids.append(Grid(x,
    52.                                        y,
    53.                                        self.default_type,
    54.                                        self.default_reward))
    55.  
    56.     def get_grid(self, x, y=None):
    57.         '''获取一个格子信息
    58.         args:坐标信息，由x，y表示或仅有一个类型为tuple的x表示
    59.         return:grid object
    60.         '''
    61.         xx, yy = None, None
    62.         if isinstance(x, int):
    63.             xx, yy = x, y
    64.         elif isinstance(x, tuple):
    65.             xx, yy = x[0], x[1]
    66.         assert(xx >= 0 and yy >= 0 and xx < self.n_width and yy < self.n_height), "任意坐标值应在合理区间"
    67.         index = yy * self.n_width + xx
    68.         return self.grids[index]
    69.  
    70.     def set_reward(self, x, y, reward):
    71.         grid = self.get_grid(x, y)
    72.         if grid is not None:
    73.             grid.reward = reward
    74.         else:
    75.             raise("grid doesn't exist")
    76.  
    77.     def set_type(self, x, y, type):
    78.         grid = self.get_grid(x, y)
    79.         if grid is not None:
    80.             grid.type = type
    81.         else:
    82.             raise("grid doesn't exist")
    83.  
    84.     def get_reward(self, x, y):
    85.         grid = self.get_grid(x, y)
    86.         if grid is None:
    87.             return None
    88.         return grid.reward
    89.  
    90.  
    91.     def get_type(self, x, y):
    92.         grid = self.get_grid(x, y)
    93.         if grid is None:
    94.             return None
    95.         return grid.type
    96.  
    97. # 格子世界环境
    98. class GridWorldEnv(gym.Env):
    99.  
    100.     metadata = {
    101.         'render.modes': ['human', 'rgb_array'],
    102.         'video.frames_per_second': 30
    103.     }
    104.  
    105.     def __init__(self, n_width: int=5,
    106.                        n_height: int = 5,
    107.                        u_size=40,
    108.                        default_reward: float = 0.0,
    109.                        default_type=0):
    110.         self.u_size = u_size                        # 当前格子绘制尺寸
    111.         self.n_width = n_width                      # 格子世界宽度（以格子数计）
    112.         self.n_height = n_height                    # 高度
    113.         self.width = u_size * n_width               # 场景宽度 screen width
    114.         self.height = u_size * n_height             # 场景长度
    115.         self.default_reward = default_reward
    116.         self.default_type = default_type
    117.  
    118.         self.grids = GridMatrix(n_width=self.n_width,
    119.                                 n_height=self.n_height,
    120.                                 default_reward=self.default_reward,
    121.                                 default_type=self.default_type)
    122.         self.reward = 0                             # for rendering
    123.         self.action = None                          # for rendering
    124.  
    125.         # 0,1,2,3 represent up, down, left, right
    126.         self.action_space = spaces.Discrete(4)
    127.         # 观察空间由low和high决定
    128.         self.observation_space = spaces.Discrete(self.n_height * self.n_width)
    129.  
    130.         self.ends = [(0, 0)]  # 终止格子坐标，可以有多个
    131.         self.start = (0, 4)  # 起始格子坐标，只有一个
    132.         self.types = [(2, 2, 1)]
    133.         self.rewards = []
    134.         self.refresh_setting()
    135.         self.viewer = None  # 图形接口对象
    136.         self.seed()  # 产生一个随机子
    137.         self.reset()
    138.  
    139.     def seed(self, seed=None):
    140.         # 产生一个随机化时需要的种子，同时返回一个np_random对象，支持后续的随机化生成操作
    141.         self.np_random, seed = seeding.np_random(seed)
    142.         return [seed]
    143.  
    144.     def step(self, action):
    145.         assert self.action_space.contains(action), "%r (%s) invalid" % (action, type(action))
    146.         self.action = action                        # action for rendering
    147.         old_x, old_y = self._state_to_xy(self.state)
    148.         new_x, new_y = old_x, old_y
    149.  
    150.         if action == 0: new_y += 1  # up
    151.         elif action == 1: new_y -= 1  # down
    152.         elif action == 2: new_x -= 1  # left
    153.         elif action == 3: new_x += 1  # right
    154.  
    155.         # boundary effect
    156.         if new_x < 0: new_x = 0
    157.         if new_x >= self.n_width: new_x = self.n_width - 1
    158.         if new_y < 0: new_y = 0
    159.         if new_y >= self.n_height: new_y = self.n_height - 1
    160.  
    161.         # wall effect:
    162.         # 类型为1的格子为障碍格子，不可进入
    163.         if self.grids.get_type(new_x, new_y) == 1:
    164.             new_x, new_y = old_x, old_y
    165.  
    166.         self.reward = self.grids.get_reward(new_x, new_y)
    167.         done = self._is_end_state(new_x, new_y)
    168.         self.state = self._xy_to_state(new_x, new_y)
    169.         # 提供格子世界所有的信息在info内
    170.         info = {"x": new_x, "y": new_y, "grids": self.grids}
    171.         return self.state, self.reward, done, info
    172.  
    173.     # 将状态变为横纵坐标
    174.     def _state_to_xy(self, s):
    175.         x = s % self.n_width
    176.         y = int((s - x) / self.n_width)
    177.         return x, y
    178.  
    179.     def _xy_to_state(self, x, y=None):
    180.         if isinstance(x, int):
    181.             assert (isinstance(y, int)), "incomplete Position info"
    182.             return x + self.n_width * y
    183.         elif isinstance(x, tuple):
    184.             return x[0] + self.n_width * x[1]
    185.         return -1  # 未知状态
    186.  
    187.     def refresh_setting(self):
    188.         '''用户在使用该类创建格子世界后可能会修改格子世界某些格子类型或奖励值
    189.         的设置，修改设置后通过调用该方法使得设置生效。
    190.         '''
    191.         for x, y, r in self.rewards:
    192.             self.grids.set_reward(x, y, r)
    193.         for x, y, t in self.types:
    194.             self.grids.set_type(x, y, t)
    195.  
    196.     def reset(self):
    197.         self.state = self._xy_to_state(self.start)
    198.         return self.state
    199.  
    200.     # 判断是否是终止状态
    201.     def _is_end_state(self, x, y=None):
    202.         if y is not None:
    203.             xx, yy = x, y
    204.         elif isinstance(x, int):
    205.             xx, yy = self._state_to_xy(x)
    206.         else:
    207.             assert (isinstance(x, tuple)), "坐标数据不完整"
    208.             xx, yy = x[0], x[1]
    209.         for end in self.ends:
    210.             if xx == end[0] and yy == end[1]:
    211.                 return True
    212.         return False
    213.  
    214.     # 图形化界面
    215.     def render(self, mode='human', close=False):
    216.         if close:
    217.             if self.viewer is not None:
    218.                 self.viewer.close()
    219.                 self.viewer = None
    220.             return
    221.         zero = (0, 0)
    222.         u_size = self.u_size
    223.         m = 2                                       # 格子之间的间隙尺寸
    224.  
    225.         # 如果还没有设定屏幕对象，则初始化整个屏幕具备的元素。
    226.         if self.viewer is None:
    227.             from gym.envs.classic_control import rendering
    228.             self.viewer = rendering.Viewer(self.width, self.height)
    229.  
    230.             # 绘制格子
    231.             for x in range(self.n_width):
    232.                 for y in range(self.n_height):
    233.                     v = [(x * u_size + m, y * u_size + m),
    234.                          ((x + 1) * u_size - m, y * u_size + m),
    235.                          ((x + 1) * u_size - m, (y + 1) * u_size - m),
    236.                          (x * u_size + m, (y + 1) * u_size - m)]
    237.  
    238.                     rect = rendering.FilledPolygon(v)
    239.                     r = self.grids.get_reward(x, y) / 10
    240.                     if r < 0:
    241.                         rect.set_color(0.9 - r, 0.9 + r, 0.9 + r)
    242.                     elif r > 0:
    243.                         rect.set_color(0.3, 0.5 + r, 0.3)
    244.                     else:
    245.                         rect.set_color(0.9, 0.9, 0.9)
    246.                     self.viewer.add_geom(rect)
    247.                     # 绘制边框
    248.                     v_outline = [(x * u_size + m, y * u_size + m),
    249.                                  ((x + 1) * u_size - m, y * u_size + m),
    250.                                  ((x + 1) * u_size - m, (y + 1) * u_size - m),
    251.                                  (x * u_size + m, (y + 1) * u_size - m)]
    252.                     outline = rendering.make_polygon(v_outline, False)
    253.                     outline.set_linewidth(3)
    254.  
    255.                     if self._is_end_state(x, y):
    256.                         # 给终点方格添加金黄色边框
    257.                         outline.set_color(0.9, 0.9, 0)
    258.                         self.viewer.add_geom(outline)
    259.                     if self.start[0] == x and self.start[1] == y:
    260.                         outline.set_color(0.5, 0.5, 0.8)
    261.                         self.viewer.add_geom(outline)
    262.                     if self.grids.get_type(x, y) == 1:  # 障碍格子用深灰色表示
    263.                         rect.set_color(0.3, 0.3, 0.3)
    264.                     else:
    265.                         pass
    266.             # 绘制个体
    267.             self.agent = rendering.make_circle(u_size / 4, 30, True)
    268.             self.agent.set_color(1.0, 1.0, 0.0)
    269.             self.viewer.add_geom(self.agent)
    270.             self.agent_trans = rendering.Transform()
    271.             self.agent.add_attr(self.agent_trans)
    272.  
    273.             # 更新个体位置
    274.         x, y = self._state_to_xy(self.state)
    275.         self.agent_trans.set_translation((x + 0.5) * u_size, (y + 0.5) * u_size)
    276.  
    277.         return self.viewer.render(return_rgb_array= mode == 'rgb_array')
    278. # 环境参数设定
    279. class Agent():
    280.     def __init__(self, env):
    281.         self.episode = 1
    282.         self.Q = {}
    283.         self.actions = [0, 1, 2, 3]
    284.         self.position = env.start
    285.         self.model = {}
    286.     # 建立模型
    287.     def make_model(self, pos, act, reward, next_state):
    288.         self.model["{0},{1}".format(pos, act)] = "{0},{1}".format(reward, next_state)
    289.     # 模型规划
    290.     def q_planning(self,n):
    291.         for i in range(0,n):
    292.  
    293.             a = [i for i in self.model.keys()]
    294.             done = False
    295.             if a != []:
    296.                 str = choice(a)
    297.                 pos = str.split(",")[0]+","+str.split(",")[1]
    298.                 act = int(str.split(",")[2])
    299.                 reward = float(self.model[str].split(",")[0])
    300.                 next_state = self.model[str].split(",")[1]+","+self.model[str].split(",")[2]
    301.                 if next_state == "(0,0)" or next_state == "(4,3)":
    302.                     done = True
    303.                 self.updateQ(pos, act, next_state, reward, done)
    304.  
    305.     def chaxunQ(self, pos, act):
    306.         judge = False
    307.         for i in self.Q:
    308.  
    309.             if i == "{0},{1}".format(pos, act):
    310.                 judge = True
    311.                 break
    312.         if judge == True:
    313.             return True
    314.         else:
    315.             self.Q["{0},{1}".format(pos, act)] = float(format(random()/10000, '.3f'))
    316.             return
    317.     # 更新状态动作值Q函数
    318.     def updateQ(self, pos, action, next_pos, reward, done):
    319.         if done == False:
    320.             self.chaxunQ(pos, action)
    321.             old_q = self.Q["{0},{1}".format(pos, action)]
    322.             action1 = self.performmax(next_pos)
    323.             # self.chaxunQ(next_pos, action1)
    324.             new_q = self.Q["{0},{1}".format(next_pos, action1)]
    325.             old_q = old_q + 0.1 * (reward+0.9 * new_q - old_q)
    326.             self.Q["{0},{1}".format(pos, action)] = float(format(old_q, '.3f'))
    327.  
    328.         else:
    329.             self.chaxunQ(pos, action)
    330.             self.Q["{0},{1}".format(pos, action)] = float(format(reward, '.3f'))
    331.             # print(pos, action,reward)
    332.     # 动作选取策略
    333.     def perform(self, pos):
    334.         eplison = random()
    335.  
    336.         self.chaxunQ(pos, choice([0, 1, 2, 3]))
    337.         if eplison > 1/self.episode:
    338.             maxq = -1000
    339.             act = ""
    340.             for i in self.Q:
    341.                 list = i.split(",")
    342.                 state = list[0] + "," + list[1]
    343.                 if state == str(pos):
    344.                     if self.Q[i] > maxq:
    345.                         maxq = self.Q[i]
    346.                         act = list[2]
    347.             return int(act)
    348.         else:
    349.             return choice([0, 1, 2, 3])
    350.     # argmaxQ
    351.     def performmax(self, pos):
    352.         maxq = -1000
    353.         str1 = ""
    354.         self.chaxunQ(pos,choice([0,1,2,3]))
    355.         for i in self.Q:
    356.             list = i.split(",")
    357.             state = list[0]+","+list[1]
    358.             if state == str(pos):
    359.                 if self.Q[i] > maxq:
    360.                     maxq = self.Q[i]
    361.                     str1 = list[2]
    362.         return int(str1)
    363.  
    364. def run(n):
    365.     agent = Agent(env)
    366.     total_j = 0
    367.     total_r = 0
    368.     a = []
    369.     b = []
    370.     env.refresh_setting()
    371.     for i in range(0, 300):
    372.         done = False
    373.         env.reset()
    374.         r = 0
    375.         j = 0
    376.         while done == False:
    377.             state = env._state_to_xy(env.state)
    378.             action = agent.perform(state)
    379.             next_state, reward, done, info = env.step(action)
    380.             next_state = env._state_to_xy(next_state)
    381.             # 更新Q值
    382.             agent.updateQ(state, action, next_state, reward,done)
    383.             # 更新模型
    384.             agent.make_model(state, action, reward, next_state)
    385.             r += reward
    386.             # 模型规划
    387.             agent.q_planning(n)
    388.             j = j+1
    389.         tpend(total_r)
    390.         agent.episode += 1
    391.         # print(agent.Q)
    392.         total_j += j
    393.         if i != 0:
    394.             b.append(j)
    395.         print("回合={0},步数={1},奖赏={2}".format(i, j, '%.3f' % r))
    396.     # return (np.array(a)/np.array(total_j)).tolist()
    397.     # return a
    398.     return b
    399. if __name__ == "__main__":
    400.     n_width = 5
    401.     n_height = 5
    402.     default_reward = 0
    403.     env = GridWorldEnv(n_width, n_height, default_reward=default_reward)
    404.     env.types = [(2, 2, 1)]
    405.     env.rewards = [(0, 0, 1), (4, 3, 1)]  # 奖赏值设定
    406.     env.start = (0, 4)
    407.     env.ends = [(0, 0), (4, 3)]
    408.     env.refresh_setting()
    409.  
    410.     x = range(1, 300)
    411.     ln1, = plt.plot(x, run(0), label=u"n=0")
    412.     ln2, = plt.plot(x, run(5), label=u"n=5")
    413.     ln3, = plt.plot(x, run(10), label=u"n=10")
    414.     ln4, = plt.plot(x, run(30), label=u"n=30")
    415.     font1 = FontProperties(fname=r"c:\windows\fonts\simsun.ttc", size=15)
    416. ontproperties=font1)
    417.     plt.ylabel(u'步数', fontproperties=font1)
    418.     plt.show()

    四、Dyan-Q算法对策略的影响

    同样Agent在不同的情节中采用不同的n获得的策略也不一样，当n=50的时候明显快，而且策略更加广泛

     代码如下

    1.  
    2.  
    3. from random import random, choice
    4. import gym
    5. from gym import spaces
    6. from gym.utils import seeding
    7.  
    8. class Grid(object):
    9.     def __init__(self, x:int = None,
    10.                        y:int = None,
    11.                        type:int = 0,
    12.                        reward:float = 0.0):
    13.         self.x = x                          # 坐标x
    14.         self.y = y
    15.         self.type = type                    # 类别值（0：空；1：障碍或边界）
    16.         self.reward = reward                # 该格子的即时奖励
    17.         self.name = None                    # 该格子的名称
    18.         self._update_name()
    19.  
    20.     def _update_name(self):
    21.         self.name = "X{0}-Y{1}".format(self.x, self.y)
    22.  
    23.     def __str__(self):
    24.         return "name:{3}, x:{0}, y:{1}, type:{2}".format(self.x,
    25.                                                                     self.y,
    26.                                                                     self.type,
    27.                                                                     self.name
    28.                                                                     )
    29.  
    30. class GridMatrix(object):
    31.     def __init__(self, n_width:int,                 # 水平方向格子数
    32.                        n_height:int,                # 竖直方向格子数
    33.                        default_type:int = 0,        # 默认类型
    34.                        default_reward:float = 0.0,  # 默认即时奖励值
    35.                        ):
    36.         self.grids = None
    37.         self.n_height = n_height
    38.         self.n_width = n_width
    39.         self.len = n_width * n_height
    40.         self.default_reward = default_reward
    41.         self.default_type = default_type
    42.         self.reset()
    43.  
    44.     def reset(self):
    45.         self.grids = []
    46.         for x in range(self.n_height):
    47.             for y in range(self.n_width):
    48.                 self.grids.append(Grid(x,
    49.                                        y,
    50.                                        self.default_type,
    51.                                        self.default_reward))
    52.  
    53.     def get_grid(self, x, y=None):
    54.         '''获取一个格子信息
    55.         args:坐标信息，由x，y表示或仅有一个类型为tuple的x表示
    56.         return:grid object
    57.         '''
    58.         xx, yy = None, None
    59.         if isinstance(x, int):
    60.             xx, yy = x, y
    61.         elif isinstance(x, tuple):
    62.             xx, yy = x[0], x[1]
    63.         assert(xx >= 0 and yy >= 0 and xx < self.n_width and yy < self.n_height), "任意坐标值应在合理区间"
    64.         index = yy * self.n_width + xx
    65.         return self.grids[index]
    66.  
    67.     def set_reward(self, x, y, reward):
    68.         grid = self.get_grid(x, y)
    69.         if grid is not None:
    70.             grid.reward = reward
    71.         else:
    72.             raise("grid doesn't exist")
    73.  
    74.     def set_type(self, x, y, type):
    75.         grid = self.get_grid(x, y)
    76.         if grid is not None:
    77.             grid.type = type
    78.         else:
    79.             raise("grid doesn't exist")
    80.  
    81.     def get_reward(self, x, y):
    82.         grid = self.get_grid(x, y)
    83.         if grid is None:
    84.             return None
    85.         return grid.reward
    86.  
    87.  
    88.     def get_type(self, x, y):
    89.         grid = self.get_grid(x, y)
    90.         if grid is None:
    91.             return None
    92.         return grid.type
    93.  
    94.  
    95. class GridWorldEnv(gym.Env):
    96.  
    97.     metadata = {
    98.         'render.modes': ['human', 'rgb_array'],
    99.         'video.frames_per_second': 30
    100.     }
    101.  
    102.     def __init__(self, n_width: int=5,
    103.                        n_height: int = 5,
    104.                        u_size=40,
    105.                        default_reward: float = 0.0,
    106.                        default_type=0):
    107.         self.u_size = u_size                        # 当前格子绘制尺寸
    108.         self.n_width = n_width                      # 格子世界宽度（以格子数计）
    109.         self.n_height = n_height                    # 高度
    110.         self.width = u_size * n_width               # 场景宽度 screen width
    111.         self.height = u_size * n_height             # 场景长度
    112.         self.default_reward = default_reward
    113.         self.default_type = default_type
    114.  
    115.         self.grids = GridMatrix(n_width=self.n_width,
    116.                                 n_height=self.n_height,
    117.                                 default_reward=self.default_reward,
    118.                                 default_type=self.default_type)
    119.         self.reward = 0                             # for rendering
    120.         self.action = None                          # for rendering
    121.  
    122.         # 0,1,2,3 represent up, down, left, right
    123.         self.action_space = spaces.Discrete(4)
    124.         # 观察空间由low和high决定
    125.         self.observation_space = spaces.Discrete(self.n_height * self.n_width)
    126.  
    127.         self.ends = [(0, 0)]  # 终止格子坐标，可以有多个
    128.         self.start = (0, 4)  # 起始格子坐标，只有一个
    129.         self.types = [(2, 2, 1)]
    130.         self.rewards = []
    131.         self.refresh_setting()
    132.         self.viewer = None  # 图形接口对象
    133.         self.seed()  # 产生一个随机子
    134.         self.reset()
    135.  
    136.     def seed(self, seed=None):
    137.         # 产生一个随机化时需要的种子，同时返回一个np_random对象，支持后续的随机化生成操作
    138.         self.np_random, seed = seeding.np_random(seed)
    139.         return [seed]
    140.  
    141.     def step(self, action):
    142.         assert self.action_space.contains(action), "%r (%s) invalid" % (action, type(action))
    143.         self.action = action                        # action for rendering
    144.         old_x, old_y = self._state_to_xy(self.state)
    145.         new_x, new_y = old_x, old_y
    146.  
    147.         if action == 0: new_y += 1  # up
    148.         elif action == 1: new_y -= 1  # down
    149.         elif action == 2: new_x -= 1  # left
    150.         elif action == 3: new_x += 1  # right
    151.  
    152.         # boundary effect
    153.         if new_x < 0: new_x = 0
    154.         if new_x >= self.n_width: new_x = self.n_width - 1
    155.         if new_y < 0: new_y = 0
    156.         if new_y >= self.n_height: new_y = self.n_height - 1
    157.  
    158.         # wall effect:
    159.         # 类型为1的格子为障碍格子，不可进入
    160.         if self.grids.get_type(new_x, new_y) == 1:
    161.             new_x, new_y = old_x, old_y
    162.  
    163.         self.reward = self.grids.get_reward(new_x, new_y)
    164.         done = self._is_end_state(new_x, new_y)
    165.         self.state = self._xy_to_state(new_x, new_y)
    166.         # 提供格子世界所有的信息在info内
    167.         info = {"x": new_x, "y": new_y, "grids": self.grids}
    168.         return self.state, self.reward, done, info
    169.  
    170.     # 将状态变为横纵坐标
    171.     def _state_to_xy(self, s):
    172.         x = s % self.n_width
    173.         y = int((s - x) / self.n_width)
    174.         return x, y
    175.  
    176.     def _xy_to_state(self, x, y=None):
    177.         if isinstance(x, int):
    178.             assert (isinstance(y, int)), "incomplete Position info"
    179.             return x + self.n_width * y
    180.         elif isinstance(x, tuple):
    181.             return x[0] + self.n_width * x[1]
    182.         return -1  # 未知状态
    183.  
    184.     def refresh_setting(self):
    185.         '''用户在使用该类创建格子世界后可能会修改格子世界某些格子类型或奖励值
    186.         的设置，修改设置后通过调用该方法使得设置生效。
    187.         '''
    188.         for x, y, r in self.rewards:
    189.             self.grids.set_reward(x, y, r)
    190.         for x, y, t in self.types:
    191.             self.grids.set_type(x, y, t)
    192.  
    193.     def reset(self):
    194.         self.state = self._xy_to_state(self.start)
    195.         return self.state
    196.  
    197.     # 判断是否是终止状态
    198.     def _is_end_state(self, x, y=None):
    199.         if y is not None:
    200.             xx, yy = x, y
    201.         elif isinstance(x, int):
    202.             xx, yy = self._state_to_xy(x)
    203.         else:
    204.             assert (isinstance(x, tuple)), "坐标数据不完整"
    205.             xx, yy = x[0], x[1]
    206.         for end in self.ends:
    207.             if xx == end[0] and yy == end[1]:
    208.                 return True
    209.         return False
    210.  
    211.     # 图形化界面
    212.     def render(self, mode='human', close=False):
    213.         if close:
    214.             if self.viewer is not None:
    215.                 self.viewer.close()
    216.                 self.viewer = None
    217.             return
    218.         zero = (0, 0)
    219.         u_size = self.u_size
    220.         m = 2                                       # 格子之间的间隙尺寸
    221.  
    222.         # 如果还没有设定屏幕对象，则初始化整个屏幕具备的元素。
    223.         if self.viewer is None:
    224.             from gym.envs.classic_control import rendering
    225.             self.viewer = rendering.Viewer(self.width, self.height)
    226.  
    227.             # 绘制格子
    228.             for x in range(self.n_width):
    229.                 for y in range(self.n_height):
    230.                     v = [(x * u_size + m, y * u_size + m),
    231.                          ((x + 1) * u_size - m, y * u_size + m),
    232.                          ((x + 1) * u_size - m, (y + 1) * u_size - m),
    233.                          (x * u_size + m, (y + 1) * u_size - m)]
    234.  
    235.                     rect = rendering.FilledPolygon(v)
    236.                     r = self.grids.get_reward(x, y) / 10
    237.                     if r < 0:
    238.                         rect.set_color(0.9 - r, 0.9 + r, 0.9 + r)
    239.                     elif r > 0:
    240.                         rect.set_color(0.3, 0.5 + r, 0.3)
    241.                     else:
    242.                         rect.set_color(0.9, 0.9, 0.9)
    243.                     self.viewer.add_geom(rect)
    244.                     # 绘制边框
    245.                     v_outline = [(x * u_size + m, y * u_size + m),
    246.                                  ((x + 1) * u_size - m, y * u_size + m),
    247.                                  ((x + 1) * u_size - m, (y + 1) * u_size - m),
    248.                                  (x * u_size + m, (y + 1) * u_size - m)]
    249.                     outline = rendering.make_polygon(v_outline, False)
    250.                     outline.set_linewidth(3)
    251.  
    252.                     if self._is_end_state(x, y):
    253.                         # 给终点方格添加金黄色边框
    254.                         outline.set_color(0.9, 0.9, 0)
    255.                         self.viewer.add_geom(outline)
    256.                     if self.start[0] == x and self.start[1] == y:
    257.                         outline.set_color(0.5, 0.5, 0.8)
    258.                         self.viewer.add_geom(outline)
    259.                     if self.grids.get_type(x, y) == 1:  # 障碍格子用深灰色表示
    260.                         rect.set_color(0.3, 0.3, 0.3)
    261.                     else:
    262.                         pass
    263.             # 绘制个体
    264.             self.agent = rendering.make_circle(u_size / 4, 30, True)
    265.             self.agent.set_color(1.0, 1.0, 0.0)
    266.             self.viewer.add_geom(self.agent)
    267.             self.agent_trans = rendering.Transform()
    268.             self.agent.add_attr(self.agent_trans)
    269.  
    270.             # 更新个体位置
    271.         x, y = self._state_to_xy(self.state)
    272.         self.agent_trans.set_translation((x + 0.5) * u_size, (y + 0.5) * u_size)
    273.  
    274.         return self.viewer.render(return_rgb_array= mode == 'rgb_array')
    275.  
    276. # 环境参数设定
    277. class Agent():
    278.     def __init__(self, env):
    279.         self.episode = 1
    280.         self.Q = {}
    281.         self.actions = [0, 1, 2, 3]
    282.         self.position = env.start
    283.         self.model = {}
    284.     def make_model(self, pos, act, reward, next_state):
    285.         self.model["{0},{1}".format(pos, act)] = "{0},{1}".format(reward, next_state)
    286.     def q_planning(self,n):
    287.         for i in range(0, n):
    288.             a = [i for i in self.model.keys()]
    289.             done=False
    290.             if a != []:
    291.                 str = choice(a)
    292.                 pos = str.split(",")[0]+","+str.split(",")[1]
    293.                 act = int(str.split(",")[2])
    294.                 reward = float(self.model[str].split(",")[0])
    295.                 next_state = self.model[str].split(",")[1]+","+self.model[str].split(",")[2]
    296.                 if next_state == "(8,5)" or next_state == "(1,6)":
    297.                     done = True
    298.                 self.updateQ(pos,act,next_state,reward,done)
    299.  
    300.     def chaxunQ(self, pos, act):
    301.         judge = False
    302.         for i in self.Q:
    303.  
    304.             if i == "{0},{1}".format(pos, act):
    305.                 judge = True
    306.                 break
    307.         if judge == True:
    308.             return True
    309.         else:
    310.             self.Q["{0},{1}".format(pos, act)] = float(format(random()/10000, '.3f'))
    311.             return
    312.     # 更新状态动作值Q函数
    313.     def updateQ(self, pos,action,next_pos,reward,done):
    314.         if done == False:
    315.             self.chaxunQ(pos, action)
    316.             old_q = self.Q["{0},{1}".format(pos, action)]
    317.             action1 = self.performmax(next_pos)
    318.             new_q = self.Q["{0},{1}".format(next_pos, action1)]
    319.             old_q = old_q + 0.1 * (reward+0.9 * new_q - old_q)
    320.             self.Q["{0},{1}".format(pos, action)] = float(format(old_q, '.3f'))
    321.  
    322.         else:
    323.             self.chaxunQ(pos, action)
    324.             self.Q["{0},{1}".format(pos, action)] = float(format(reward, '.3f'))
    325.     # 动作选取策略
    326.     def perform(self, pos):
    327.         eplison = random()
    328.  
    329.         self.chaxunQ(pos, choice([0, 1, 2, 3]))
    330.         if eplison > 1/self.episode:
    331.             maxq = -1000
    332.             act = ""
    333.             for i in self.Q:
    334.                 list = i.split(",")
    335.                 state = list[0] + "," + list[1]
    336.                 if state == str(pos):
    337.                     if self.Q[i] > maxq:
    338.                         maxq = self.Q[i]
    339.                         act = list[2]
    340.             return int(act)
    341.         else:
    342.             return choice([0, 1, 2, 3])
    343.     # argmaxQ
    344.     def performmax(self, pos):
    345.         maxq = -1000
    346.         str1 = ""
    347.         self.chaxunQ(pos,choice([0,1,2,3]))
    348.         for i in self.Q:
    349.             list = i.split(",")
    350.             state = list[0]+","+list[1]
    351.             if state == str(pos):
    352.                 if self.Q[i] > maxq:
    353.                     maxq = self.Q[i]
    354.                     str1 = list[2]
    355.         return int(str1)
    356.  
    357. def run(n):
    358.     agent = Agent(env)
    359.     total_j = 0
    360.     total_r = 0
    361.     a = []
    362.     b = []
    363.     env.refresh_setting()
    364.     for i in range(0, 300):
    365.         done = False
    366.         env.reset()
    367.         r = 0
    368.         j = 0
    369.         while done == False and j < 50:
    370.             j = j + 1
    371.             state = env._state_to_xy(env.state)
    372.             action = agent.perform(state)
    373.             next_state, reward, done, info = env.step(action)
    374.             next_state = env._state_to_xy(next_state)
    375.             # 更新模型
    376.             agent.make_model(state, action, reward, next_state)
    377.             # 更新Q值
    378.             agent.updateQ(state, action, next_state, reward, done)
    379.             r += reward
    380.             # 模型规划
    381.             agent.q_planning(n)
    382.  
    383.         if i >= 2:
    384.             total_r += r
    385.             a.append(total_r)
    386.         agent.episode += 1
    387.         action_Q = {}
    388.         for i in sorted(list(agent.Q.keys()), key=lambda x: [x[1], x[4], x[5], x[-1]]):
    389.             action_Q[i] = agent.Q[i]
    390.         print("n={0}:Q值{1}".format(n, action_Q))
    391.         # P_A:状态采取策略
    392.         P_A = {}
    393.         Q_keys = list(action_Q.keys())
    394.         for i in range(0, len(Q_keys)-4, 4):
    395.             temp_action_list = []
    396.             max_a_value = max(action_Q[Q_keys[j]] for j in range(i, i+4))
    397.             # temp_num:标记四个动作最大值
    398.             temp_num = [0, 0, 0, 0]
    399.             # PA:四个动作概率
    400.             PA = [0, 0, 0, 0]
    401.             for k in range(i, i+4):
    402.                 if action_Q[Q_keys[k]] == max_a_value:
    403.                     temp_action_list.append(Q_keys[k])
    404.                     temp_num[k-i] = 1
    405.             valid_action_p = round(1/len(temp_action_list), 2)
    406.             for m in range(4):
    407.                 if temp_num[m] == 1:
    408.                     PA[m] = valid_action_p
    409.             P_A[Q_keys[i][0:-2]] = PA
    410.         print("Q_A: ", P_A)
    411.         total_j += j
    412.         b.append(j)
    413.     return a
    414. if __name__ == "__main__":
    415.     n_width = 11
    416.     n_height = 7
    417.     default_reward = -0.1
    418.     env = GridWorldEnv(n_width, n_height, default_reward=default_reward)
    419.     env.types = [(1, 2, 1), (2, 2, 1), (3, 2, 1), (4, 2, 1), (5, 2, 1), (6, 2, 1), (7, 2, 1), (8, 2, 1), (9, 2, 1),
    420.                  (10, 2, 1)]
    421.     env.rewards = [(8, 5, 1), (1, 6, 1)]  # 奖赏值设定
    422.     env.start = (4, 0)  # 机器人出发点坐标
    423.     env.ends = [(8, 5), (1, 6)]  # 吸入状态坐标
    424.     env.refresh_setting()
    425.  
    426.     for i in range(0, 2):
    427.         print("episode=", i+1)
    428.         run(0)
    429.     for i in range(0, 2):
    430.         print("episode=", i+1)
    431.         run(50)

    创作不易 觉得有帮助请点赞关注收藏~~~

  • 相关阅读:
    Visual studio代码提示（IntelliSense）的语言（包括汉化等）修改
    WGCNA分析教程五 | [更新版]
    学习笔记2
    全网最全超详细.htaccess语法讲解
    细谈VR全景：数字营销时代的宠儿
    Coupang走什么物流？Coupang火箭颜色什么意思？——站斧浏览器
    迷宫问题详解（数据结构实验）
    Google gtest事件机制
    通过async方式在浏览器中调用web worker
    elasticsearch搜索IK分词器实现单个字搜索
  • 原文地址：https://blog.csdn.net/jiebaoshayebuhui/article/details/128018045