• [PyTorch][chapter 61][强化学习-免模型学习1]


    前言:

          在现实的学习任务中,环境 E=<X,A,P,R>

        其中的转移概率P,奖赏函数R 是未知的,或者状态X也是未知的

        称为免模型学习(model-free learning)

    目录:

         1:  蒙特卡洛强化学习

         2:同策略-蒙特卡洛强化学习

         3: 异策略- 蒙特卡洛强化学习


    一  蒙特卡洛强化学习

       

       在免模型学习的情况下,策略迭代算法会遇到两个问题:

      1:  是策略无法评估
                因为无法做全概率展开。此时 只能通过在环境中执行相应的动作观察得到的奖赏和转移的状态、
           
           解决方案:一种直接的策略评估代替方法就是“采样”,然后求平均累积奖赏,作为期望累积奖赏的近似,这称为“蒙特卡罗强化学习”。

      2:  策略迭代算法估计的是 状态值函数(state value function) V,而最终的策略是通过 状态        动作值函数(state-action value function) Q 来获得。
            模型已知时,有很简单的从 V 到 Q 的转换方法,而模型未知 则会出现困难。
           
           解决方案:所以我们将估计对象从 V 转为 Q,即:估计每一对 “状态-动作”的值函数。

         模型未知的情况下,我们从起始状态出发,使用某种策略进行采样,执行该策略T步,

    并获得轨迹 < x_0,a_0,r_1,a_1,r_2,...x_{T-1},a_{T-1},r_T,x_T>,

     然后 对轨迹中出现的每一对 状态-动作,记录其后的奖赏之和,作为 状态-动作 对的一次

    累积奖赏采样值. 多次采样得到多条轨迹后,将每个状态-动作对的累积奖赏采样值进行平均。即得到 状态-动作值函数的估计.


    二  同策略蒙特卡洛强化学习

       target   Policy : 要优化的策略,也称为原始策略

       behavior policy: 用来产生轨迹的策略

     在on-policy 方案里面,两个策略相同。

        要获得好的值函数估计,就需要不同的采样轨迹。 

       我们将确定性的策略\pi 称为原始策略

      原始策略上使用 \epsilon-贪心法的策略记为\pi^{\epsilon }(x)=

       以概率1-\epsilon 选择策略1:           策略1 :\pi(x)

       以概率\epsilon    选择策略2:             策略2:均匀概率选取动作,

    对于最大化值函数的原始策略

    \pi= argmax_{a}Q(x,a)

    其中\epsilon-贪心策略\pi^{\epsilon }中:

    当前最优动作被选中的概率   1-\epsilon +\frac{\epsilon }{|A|}

     每个非最优动作选中的概率 \frac{\epsilon }{|A|},多次采样后将产生不同的采样轨迹。

    因此对于最大值函数的原始策略\pi^{'},同样有

    算法中,每采样一条轨迹,就根据该轨迹涉及的所有"状态-动作"对值函数进行更新

    同策略蒙特卡罗强化学习算法最终产生的是E-贪心策略。然而,引入E-贪心策略是为了便于策略评估,而不是最终使用


    三  同策略蒙特卡洛算法 Python

      整个算法分为三部分:

    迭代

        1: 使用behavior policy 生成轨迹

        2: 根据轨迹的样本,进行策略评估

        3: 根据策略评估的结果Q(累积奖赏),优化target policy(选择状态-动作 最大值)

    1. import numpy as ap
    2. # -*- coding: utf-8 -*-
    3. """
    4. Created on Fri Nov 3 09:37:32 2023
    5. @author: chengxf2
    6. """
    7. # -*- coding: utf-8 -*-
    8. """
    9. Created on Thu Nov 2 19:38:39 2023
    10. @author: cxf
    11. """
    12. import random
    13. from enum import Enum
    14. class State(Enum):
    15. #状态空间#
    16. shortWater =1 #缺水
    17. health = 2 #健康
    18. overflow = 3 #溢水
    19. apoptosis = 4 #凋亡
    20. class Action(Enum):
    21. #动作空间A#
    22. water = 1 #浇水
    23. noWater = 2 #不浇水
    24. class Env():
    25. def getReward(self, nextState):
    26. #针对转移到新的环境奖赏
    27. r = -100
    28. if nextState is State.shortWater:
    29. r =-1
    30. elif nextState is State.health:
    31. r = 1
    32. elif nextState is State.overflow:
    33. r= -1
    34. else:
    35. r = -100
    36. return r
    37. def action(self, state, action):
    38. if state is State.shortWater:
    39. #print("\n state--- ",state, "\t action---- ",action)
    40. if action is Action.water :
    41. S =[State.shortWater, State.health]
    42. proba =[0.4, 0.6]
    43. else:
    44. S =[State.shortWater, State.apoptosis]
    45. proba =[0.5, 0.5]
    46. elif state is State.health:
    47. #健康
    48. if action is Action.water :
    49. S =[State.health, State.overflow]
    50. proba =[0.6, 0.4]
    51. else:
    52. S =[State.shortWater, State.health]
    53. proba =[0.6, 0.4]
    54. elif state is State.overflow:
    55. #溢水
    56. if action is Action.water :
    57. S =[State.overflow, State.apoptosis]
    58. proba =[0.6, 0.4]
    59. else:
    60. S =[State.health, State.overflow]
    61. proba =[0.6, 0.4]
    62. else:
    63. #凋亡
    64. S =[State.apoptosis]
    65. proba =[1.0]
    66. #print("\n S",S, "\t prob ",proba)
    67. nextState = random.choices(S, proba)[0]
    68. reward= self.getReward(nextState)
    69. return nextState,reward
    70. def __init__(self):
    71. self.name = "环境空间"
    72. class Agent():
    73. def initPolicy(self):
    74. self.Q ={} #累积奖赏
    75. self.count ={} #[state,action] 出现的次数
    76. for state in self.S:
    77. for action in self.A:
    78. self. Q[state, action] = 0
    79. self.count[state,action]= 0
    80. def behavior(self,state):
    81. #使用行为策略 behavior-policy
    82. r = random.random()
    83. if r<self.epsi:
    84. #使用随机策略
    85. A= self.A
    86. randProb =[0.5,0.5]
    87. action = random.choices(A, randProb)[0]
    88. else:#使用target policy >1-epsi
    89. action = self.policy[state]
    90. return action
    91. def generate(self,T):
    92. #使用行为策略(e贪心策略),产生轨迹
    93. #轨迹[s0,a0,r0],[s1,a1,r1]......[rT-1,aT-1,rT-1]
    94. env = Env()
    95. state = State.shortWater #从缺水开始
    96. trajectory ={}
    97. for t in range(T):
    98. action = self.behavior(state)
    99. nextState,reward = env.action(state, action)
    100. trajectory[t] =[state, action, reward]
    101. state = nextState
    102. return trajectory
    103. def evaluate(self,T ,trajectory):
    104. #策略评估
    105. denominator =0
    106. for t in range(0,T):
    107. denominator =T-t
    108. R_t = 0
    109. acc = 0
    110. for i in range(t,T):
    111. r = trajectory[i][-1]
    112. acc = acc+r
    113. R_t = acc/denominator #t时刻的累积奖赏
    114. state_t = trajectory[t][0]
    115. action_t = trajectory[t][1]
    116. count = self.count[state_t, action_t]
    117. self.Q[state_t, action_t] = (self.Q[state_t, action_t]*count+R_t)/(count+1)
    118. self.count[state_t, action_t] +=1
    119. def improve(self):
    120. #策略提升
    121. for state in self.S:
    122. maxReward = self.Q[state,Action.water]
    123. for action in self.A:
    124. reward = self.Q[state,action]
    125. if reward>=maxReward:
    126. maxReward = reward
    127. self.policy[state] = action
    128. #print("\n state ",state, "\t maxReward ",maxReward)
    129. def learn(self):
    130. self.initPolicy()
    131. #target policy 最终要优化的策略
    132. self.policy ={}
    133. T =15
    134. for state in self.S:
    135. self.policy[state] = Action.noWater
    136. for s in range(1,self.epsidode): #采样第S 条轨迹
    137. #使用e 贪心策略生成轨迹
    138. trajectory = self.generate(T)
    139. #策略评估
    140. self.evaluate(T, trajectory)
    141. print("\n 迭代次数 %d"%s ,"\t 缺水:",self.policy[State.shortWater].name,
    142. "\t 健康:",self.policy[State.health].name,
    143. "\t 溢水:",self.policy[State.overflow].name,
    144. "\t 凋亡:",self.policy[State.apoptosis].name)
    145. #策略improve
    146. self.improve()
    147. def __init__(self):
    148. self.S = [State.shortWater, State.health, State.overflow, State.apoptosis]
    149. self.A = [Action.water, Action.noWater]
    150. self.Q ={} #累积奖赏
    151. self.count ={}
    152. self.policy ={}
    153. self.epsidode =1000
    154. self.epsi = 0.2
    155. if __name__ == "__main__":
    156. agent = Agent()
    157. agent.learn()

  • 相关阅读:
    【Java面试】MongoDB
    【JavaWeb】Servlet属性设置
    EVE111
    原始资料的收集方法———定性资料的收集
    oracle在Windows正常使用需要启动哪些服务
    普中51-独立按键实验
    Spring MVC 的核心组件有哪些?
    java计算机毕业设计springboot+vue社区养老管理系统(源码+系统+mysql数据库+Lw文档)
    百日刷题计划 ———— DAY2
    基于JAVA医院预约挂号系统计算机毕业设计源码+系统+mysql数据库+lw文档+部署
  • 原文地址:https://blog.csdn.net/chengxf2/article/details/134183448