一些有用的书中句子摘抄。
初探强化学习
强化学习用智能体（agent）这个概念来表示做决策的机器。相比于有监督学习中的“模型”，强化学习中的“智能体”强调机器不但可以感知周围的环境信息，还可以通过做决策来直接改变这个环境，而不只是给出一些预测信号。
智能体有3种关键要素，即感知、决策和奖励。
感知。智能体在某种程度上感知环境的状态，从而知道自己所处的现状。例如，下围棋的智能体感知当前的棋盘情况；无人车感知周围道路的车辆、行人和红绿灯等情况。
智能体根据当前的状态计算出达到目标需要采取的动作的过程叫作决策。例如，针对当前的棋盘决定下一颗落子的位置；针对当前的路况，无人车计算出方向盘的角度和刹车、油门的力度。策略是智能体最终体现出的智能形式，是不同智能体之间的核心区别。
奖励。环境根据状态和智能体采取的动作，产生一个标量信号作为奖励反馈。这个标量信号衡量智能体这一轮动作的好坏。例如，围棋博弈是否胜利；无人车是否安全、平稳且快速地行驶。最大化累积奖励期望是智能体提升策略的目标，也是衡量智能体策略好坏的关键指标。
从以上分析可以看出，面向决策任务的强化学习和面向预测任务的有监督学习在形式上是有不少区别的。首先，决策任务往往涉及多轮交互，即序贯决策；而预测任务总是单轮的独立任务。如果决策也是单轮的，那么它可以转化为“判别最优动作”的预测任务。其次，因为决策任务是多轮的，智能体就需要在每轮做决策时考虑未来环境相应的改变，所以当前轮带来最大奖励反馈的动作，在长期来看并不一定是最优的。
强化学习的环境是动态的
说一个环境是动态的，意思就是它会随着某些因素的变化而不断演变（比如一座城市的交通），这在数学和物理中往往用随机过程来刻画。对于一个随机过程，其最关键的要素就是状态以及状态转移的条件概率分布。
环境的下一刻状态的概率分布将由当前状态和智能体的动作来共同决定，即：
换而言之，在动态随机过程中学习和在一个固定的数据分布下学习是非常不同的。
奖励信号
智能体和环境每次进行交互时，环境会产生相应的奖励信号，其往往由实数标量来表示。这个奖励信号一般是诠释当前状态或动作的好坏的及时反馈信号，好比在玩游戏的过程中某一个操作获得的分数值。整个交互过程的每一轮获得的奖励信号可以进行累加，形成整体回报。
在强化学习中，我们关注回报的期望，并将其定义为价值（value）
价值的计算需要对交互过程中每一轮智能体采取动作的概率分布和环境相应的状态转移的概率分布做积分运算。
数据分布
在强化学习中，数据是在智能体与环境交互的过程中得到的。如果智能体不采取某个决策动作，那么该动作对应的数据就永远无法被观测到，所以当前智能体的训练数据来自之前智能体的决策结果。因此，智能体的策略不同，与环境交互所产生的数据分布就不同。（而监督学习的数据分布是永远不变的）
强化学习中有一个关于数据分布的概念，叫作占用度量（occupancy measure）：对于一个定义在状态空间 S 和动作空间 A 上的MDP，给定一个策略 π，其占用度量 ρ_π 是一个定义在 S × A 上的概率分布（或非归一化的度量）。它衡量了在执行策略 π 时，智能体“花费”在各个状态-动作对上的“时间”或“权重”。
通俗来说：“占用度量”可理解为智能体在环境中“踩出的地盘”或“留下的足迹”。它就是一张热力图，它记录了你在无数次走迷宫的过程中，在每个位置、以哪种动作停留的频率。
占用度量有一个很重要的性质：给定两个策略及其与一个动态环境交互得到的两个占用度量，那么当且仅当这两个占用度量相同时，这两个策略相同。
所以：
强化学习的策略在训练中会不断更新，其对应的数据分布（即占用度量）也会相应地改变。因此，强化学习的一大难点就在于，智能体看到的数据分布是随着智能体的学习而不断发生改变的。
由于奖励建立在状态动作对之上，一个策略对应的价值其实就是一个占用度量下对应的奖励的期望，因此寻找最优策略对应着寻找最优占用度量。
关于强化学习与监督学习：二者优化的途径是不同的，有监督学习直接通过优化模型对于数据特征的输出来优化目标，即修改目标函数而数据分布不变；强化学习则通过改变策略来调整智能体和环境交互数据的分布，进而优化目标，即修改数据分布而目标函数不变。
一般的有监督学习关注寻找一个模型，使其在给定数据分布下得到的损失函数的期望最小；
强化学习关注寻找一个智能体策略，使其在与动态环境交互的过程中产生最优的数据分布，即最大化该分布下一个给定奖励函数的期望。
多臂老虎机（multi-armed bandit，MAB）
MAB是简化版的强化学习问题，它不存在状态信息，只有动作和奖励，算是最简单的“和环境交互中的学习”的一种形式。
问题：有一个拥有K根拉杆的老虎机，拉动每一根拉杆都对应一个关于奖励的概率分布R 。我们每次拉动其中一根拉杆，就可以从该拉杆对应的奖励概率分布中获得一个奖励r。我们在各根拉杆的奖励概率分布未知的情况下，从头开始尝试，目标是在操作T次拉杆后获得尽可能高的累积奖励
定义最优期望为Q*，懊悔被定义为拉动当前拉杆的动作 a 与最优拉杆的期望奖励差，即 R(a)=Q*-Q(a)。问题的目标为最大化累积奖励，等价于最小化累积懊悔。
经典增量式更新
关于探索与利用：一个比较常用的思路是在开始时做比较多的探索，在对每根拉杆都有比较准确的估计后，再进行利用。目前已有一些比较经典的算法来解决这个问题，例如epsilon-贪婪算法、上置信界算法和汤普森采样算法等，我们接下来将分别介绍这几种算法。
epsilon-贪婪算法在完全贪婪算法的基础上添加了噪声，每次以概率epsilon选择以往经验中期望奖励估值最大的那根拉杆（利用），以概率1-epsilon随机选择一根拉杆（探索）：
import numpy as np
import matplotlib.pyplot as plt
from utils import plot_results
class BernoulliBandit:
def __init__(self,K):
'''如果只传入一个参数，它返回的是一个标量（单个浮点数），而不是数组。这里应该传入两个参数，或者使用(size=K)来生成K个随机数。'''
self.probs = np.random.uniform(size=K) # 随机生成K个0～1的数,作为拉动每根拉杆的获奖
self.best_idx = np.argmax(self.probs) # 获奖概率最大的拉杆
self.best_prob = self.probs[self.best_idx]
self.K = K
class Solver:
"""多臂老虎机算法的基本框架，将根据策略选择动作、根据动作获取奖励和更新期望奖励估值放在 run_one_step() 函数中"""
def __init__(self,bandit):
self.bandit = bandit
self.counts = np.zeros(bandit.K) # 记录每根拉杆被拉动的次数
self.regret = 0.0 # 累计遗憾值
self.regrets = [] # 记录每一步的遗憾值
self.actions = [] # 记录每一步选择的动作
def step(self,k):
"""拉动编号为 k 的拉杆，返回奖励"""
reward = np.random.binomial(1,self.bandit.probs[k]) # 伯努利分布（n=1代表丢一次）
return reward
def run_one_step(self):
raise NotImplementedError
def update_regret(self,idx):
instant_regret = self.bandit.best_prob - self.bandit.probs[idx] # 伯努利分布的期望就是它的概率
self.regret += instant_regret
self.regrets.append(self.regret) # 存储的是累计遗憾
def run(self,n_steps):
for _ in range(n_steps):
idx = self.run_one_step() # 当前选的杠杆编号
self.counts[idx] += 1
self.actions.append(idx)
self.update_regret(idx)
class EpsilonGreedy(Solver):
"""ε-贪婪算法"""
def __init__(self,bandit,epsilon=0.01,init_prob=1.0):
super(EpsilonGreedy,self).__init__(bandit)
self.epsilon = epsilon
# 初始化拉动所有拉杆的期望奖励估值
self.estimates = np.array([init_prob]*self.bandit.K)
def run_one_step(self):
if np.random.random() < self.epsilon:
k = np.random.randint(0,self.bandit.K) # 探索，返回随机选的杠杆编号
else:
k = np.argmax(self.estimates) # 利用
r = self.step(k)
self.counts[k] += 1
self.estimates[k] += (r - self.estimates[k]) / self.counts[k] # 更新期望奖励估值，对应Q_k的更新公式
return k
if __name__ == "__main__":
np.random.seed(1)
K = 10
bandit = BernoulliBandit(K) # 初始化一个老虎机，有K个臂
T = 5000
epsilongreedy_solver = EpsilonGreedy(bandit,epsilon=0.01)
epsilongreedy_solver.run(T)
print("Epsilon-Greedy 总遗憾值: ",epsilongreedy_solver.regret)
plot_results([epsilongreedy_solver],['Epsilon-Greedy'])
一个细节：self.estimates = np.array([init_prob]*self.bandit.K) 这里init_prob是1，原因如下：
首先，解释一下“期望奖励估值”：
在多臂赌博机问题中，每个臂都有一个真实的期望奖励（对于伯努利分布，就是成功概率p）。但是智能体并不知道这个真实值，它需要在每次尝试后更新对每个臂的奖励估计。这个估计值就是我们这里所说的“期望奖励估值”，也就是智能体认为每个臂可能带来的平均奖励。（所以这里不是必须初始化为p）
ε-贪婪算法中一种常用的技巧，称为“乐观初始值”。其思想是，将初始估计值设置得比可能的最大奖励还要高（比如1.0，而实际奖励最大为1）。这样，每个臂在初始时都被高估，因此智能体会在早期尝试每个臂，从而在早期鼓励探索。
结果：
贪婪算法的累积懊悔几乎是线性增长的（无论epsilon取值）。因为一旦做出了随机拉杆的探索，那么产生的懊悔值是固定的。
上面那句话的理解：
假设有K=10个臂，最佳臂的概率是p*，其他臂的平均概率是p_avg，那么：
探索时选中最佳臂的懊悔 = 0
探索时选中非最佳臂的平均懊悔 = p* - p_avg
探索一步的期望懊悔 = (1/K)×0 + ((K-1)/K)×(p* - p_avg) = 常数C
所以改进是epsilon值随时间衰减（让epsilon乘时间分之一）：
def run_one_step(self):
self.total_count += 1
if np.random.random() < 1 / self.total_count:
...上置信界算法
与贪婪算法不同，UCB不依赖于随机探索，而是基于不确定性来指导探索。
一根拉杆的不确定性越大，它就越具有探索的价值，因为探索之后我们可能发现它的期望奖励很大。我们在此引入不确定性度量U(a)，它会随着一个动作被尝试次数的增加而减小。
Q_t(a)：动作a的真实期望奖励（未知，是我们希望估计的目标）
class UCB(Solver):
"""设置概率p为t分之一"""
def __init__(self,bandit,coef,init_prob=1.0):
super(UCB,self).__init__(bandit)
self.coef = coef
self.estimates = np.array([init_prob]*self.bandit.K)
self.total_counts = 0 # 记录总的拉杆次数
def run_one_step(self):
self.total_counts+=1
# ucb就是上置信界=期望估计+不确定性
ucb=self.estimates+self.coef*np.sqrt(np.log(self.total_counts)/(2*self.counts+1))
k=np.argmax(ucb) # 选择上置信界最大的拉杆
r=self.step(k)
self.counts[k] += 1
self.estimates[k] += (r - self.estimates[k]) / self.counts[k]
return k汤普森采样算法
“汤普森采样算法使用采样的方式，即根据当前每个动作 的奖励概率分布进行一轮采样，得到一组各根拉杆的奖励样本，再选择样本中奖励最大的动作。可以看出，汤普森采样是一种计算所有拉杆的最高奖励概率的蒙特卡洛采样方法。”
解释：汤普森采样采用贝叶斯观点，为每个动作的概率分布维持一个后验分布，然后从这个分布中采样来决定动作（先验分布：在没有观察数据之前，基于经验等对参数或概率做的初始判断；后验分布：在观察到数据之后，结合先验信息，对未知参数的更新）。
核心思想：
为每个臂的奖励概率维护一个概率分布
在每个时间步，从每个臂的分布中采样一个值
选择采样值最大的臂
根据实际观察到的奖励更新该臂的分布
Beta分布是定义在区间[0,1]上的连续概率分布，常用于描述概率的概率分布。它由两个正参数α和β控制，Beta(α, β) 是伯努利分布的共轭先验
α：观察到的成功次数 + 1；β：观察到的失败次数 + 1
通过选择不同的α, β，可以表示各种先验信念：
完全无知：Beta(1,1) 均匀分布
乐观先验：α大，β小，认为成功概率高
悲观先验：α小，β大，认为成功概率低
有信息先验：α, β都大，表示我们有很强的先验信念
回到老虎机，算法的流程是：先初始化，完全无知。对每个臂，采样一个值，选择采样最大的值（大模型的回答：不是选择"看起来最好"的，而是选择"这次抽样中运气最好"的，这就是汤普森采样的精髓：通过概率性的"想象"来指导探索，用现实来修正想象，最终找到真正最好的选择。）；选完臂之后拉动臂，得到奖励r（0或1），然后据此更新后验分布，然后重复：
if r == 1:
successes_a += 1
α_a += 1
else:
failures_a += 1
β_a += 1
class ThompsonSampling(Solver):
"""汤普森采样算法"""
def __init__(self,bandit):
super(ThompsonSampling, self).__init__(bandit)
self.successes = np.zeros(bandit.K)
self.failures = np.zeros(bandit.K)
def run_one_step(self):
sampled_theta = [np.random.beta(self.successes[k]+1,self.failures[k]+1) for k in range(self.bandit.K)]
k = np.argmax(sampled_theta) # 选择采样值最大的拉杆
r = self.step(k)
if r == 1:
self.successes[k] += 1
else:
self.failures[k] += 1
return k上面代码中的solver，其实代表的是不同的策略。多臂老虎机的每次交互的结果和以往的动作无关，所以可看作无状态的强化学习。有状态的强化学习对应马尔科夫决策。
马尔科夫决策过程（Markov decision process，MDP）
“如果要用强化学习去解决一个实际问题，第一步要做的事情就是把这个实际问题抽象为一个马尔可夫决策过程，也就是明确马尔可夫决策过程的各个组成要素。”
随机过程：有点像大学形式与自动机学的。在随机过程中，随机现象在某时刻的取值是一个向量随机变量，所有可能的状态组成状态集合S。当且仅当某时刻的状态只取决于上一时刻的状态时，一个随机过程被称为具有马尔可夫性质：
虽然t+1时刻的状态只与t时刻有关，但t时刻状态本身也传达了t-1时刻状态的，所以马尔可夫性可以大大简化运算，因为只要当前状态可知，所有的历史信息都不再需要了（暗含）
定义马尔科夫矩阵：
P(s_i|s_j)代表从状态s_i转移到s_j的概率，从某个状态出发，到达其他状态的概率和必须为 1，即状态转移矩阵的每一行的和为 1。
给定一个马尔可夫过程，我们就可以从某个状态出发，根据它的状态转移矩阵生成一个状态序列（episode），这个步骤也被叫做采样，比如从s1出发，可生成s1->s2->s4的序列……
马尔科夫奖励过程(MRP)
在马尔可夫过程的基础上加入奖励函数 和折扣因子，就可以得到马尔可夫奖励过程。引入折扣因子的理由为远期利益具有一定不确定性，有时我们更希望能够尽快获得一些奖励，所以我们需要对远期利益打一些折扣。
R_t表示在t时刻获得的奖励，从t时刻状态S_t开始，到终止态获得的奖励的衰减之和称为汇报G_t：
价值函数
在马尔可夫奖励过程中，一个状态的期望回报（即从这个状态出发的未来累积奖励的期望）被称为这个状态的价值，价值函数的输入为某个状态，输出为这个状态的价值。有：
由于计算复杂度是n^3，实际往往使用动态规划，蒙特卡洛方法和时序差分
def compute(P,rewards,gama,n):
'''n使状态数'''
rewards=np.array(rewards).reshape(-1,1) # 转换成列向量
I=np.eye(n)
value=np.dot(np.linalg.inv(I-gama*P),rewards)
return value马尔科夫决策过程(MDP)
马尔可夫过程和马尔可夫奖励过程都是自发改变的随机过程，在马尔可夫奖励过程（MRP）的基础上加入智能体的动作，就得到了马尔可夫决策过程
一艘小船在大海中随着水流自由飘荡的过程就是一个马尔可夫奖励过程，它如果凭借运气漂到了一个目的地，就能获得比较大的奖励；如果有个水手在控制着这条船往哪个方向前进，就可以主动选择前往目的地获得比较大的奖励（马尔科夫决策过程）
马尔科夫决策过程由五元组定义：
策略：通常用π表示。π(a|s)=P(A_t=a|S_t=s)是一个函数，表示在输入状态s的情况下采取动作a的概率
状态价值与动作价值
状态价值函数：定义为从状态s出发遵循策略π能获得的期望回报：
在 MDP 中，由于动作的存在，我们额外定义一个动作价值函数，表示在 MDP 遵循策略π时，对当前状态s执行动作a得到的期望回报：
显然有：状态的价值等于在该状态下基于策略采取所有动作的概率与相应的价值相乘再求和的结果：
使用策略π时，状态s下采取动作a的价值等于即时奖励加上经过衰减后的所有可能的下一个状态的状态转移概率与相应的价值的乘积：
贝尔曼期望方程
结合上面的式子，可推导下面的（价值函数和贝尔曼方程是强化学习非常重要的组成部分，之后的一些强化学习算法都是据此推导出来的）：
蒙特卡洛方法
经典统计模拟方法，比如估圆的面积，圆的面积与正方形面积之比就等于圆中点的个数与正方形中点的个数之比。由大数定律，当N趋于无穷时，有真实=期望。
一个状态的价值是它的期望回报，那么一个很直观的想法就是用策略在 MDP 上采样很多条序列，计算从这个状态出发的回报再求其期望就可以了
占用度量
可以定义策略的占用度量：它表示动作状态对(s,a)被访问到的概率。二者之间存在如下关系：
v那个函数是一个策略的状态分布，π那个函数表示执行a的概率，显然结果是二者相乘
1-γ是用来使得概率加和为 1 的归一化因子
贝尔曼最优方程与最优策略
强化学习的目标通常是找到一个策略，使得智能体从初始状态出发能获得最多的期望回报。
最优策略都有相同的状态价值函数，我们称之为最优状态价值函数
最优状态价值函数和最优动作价值函数之间显然有关系（和之前的一样）：
贝尔曼最优方程：
动态规划算法——策略迭代
动态规划的基本思想是将待求解问题分解成若干个子问题，先求解子问题，然后从这些子问题的解得到目标问题的解。动态规划会保存已解决的子问题的答案，在求解目标问题的过程中，需要这些子问题答案时就可以直接利用
基于动态规划的强化学习算法主要有两种：一是策略迭代（policy iteration），二是价值迭代（value iteration）。其中，策略迭代由两部分组成：策略评估（policy evaluation）和策略提升（policy improvement）。具体来说，策略迭代中的策略评估使用贝尔曼期望方程来得到一个策略的状态价值函数，这是一个动态规划的过程；而价值迭代直接使用贝尔曼最优方程来进行动态规划，得到最终的最优状态价值。
对于策略评估，再次强调贝尔曼期望方程：
V函数为状态价值函数，定义为从状态s出发遵循策略π能获得的期望汇报，π(a|s)是策略π在状态s下采取动作a的概率，p是状态转移函数，表示在状态s执行动作a之后到达s'的概率
方程本身就可以用动态规划化简：把计算下一个可能状态的价值当成一个子问题，把计算当前状态的价值看作当前问题，即：
当k趋于无穷时，序列V^k会收敛到V^π （实际可以看差值，如果两轮的V差距太小可以用早停直接终止)
对于策略提升：
使用策略评估计算得到当前策略的状态价值函数之后，我们可以据此来改进该策略。其本质上是贪心：
解释如下：
需要结合前面的定义：动作价值函数，表示在 MDP 遵循策略π时，对当前状态s执行动作a得到的期望回报；显然有：状态的价值等于在该状态下基于策略采取所有动作的概率与相应的价值相乘再求和的结果 （即状态价值是0.5概率*拿一块+0.5概率*拿不到；这里把策略更新，动作确定为1的概率拿一块）
所以由此产生策略迭代算法：
对当前的策略进行策略评估，得到其状态价值函数，然后根据该状态价值函数进行策略提升以得到一个更好的新策略，接着继续评估新策略、提升策略……直至最后收敛到最优策略
动态规划算法——价值迭代
我们是否必须要完全等到策略评估完成后再进行策略提升呢？试想一下，可能出现这样的情况：虽然状态价值函数还没有收敛，但是不论接下来怎么更新状态价值，策略提升得到的都是同一个策略。
利用贝尔曼最优方程，将其写为迭代形式：
时序差分算法
对于大部分强化学习现实场景，其马尔可夫决策过程的状态转移概率是无法写出来的，也就无法直接进行动态规划。在这种情况下，智能体只能和环境进行交互，通过采样到的数据来学习，这类学习方法统称为无模型的强化学习
不同于动态规划算法，无模型的强化学习算法不需要事先知道环境的奖励函数和状态转移函数，而是直接使用和环境交互的过程中采样到的数据来学习
在线策略学习和离线策略学习：在线策略学习要求使用在当前策略下采样得到的样本进行学习，一旦策略被更新，当前的样本就被放弃了，就好像在水龙头下用自来水洗手；而离线策略学习使用经验回放池将之前采样得到的样本收集起来再次利用，就好像使用脸盆接水后洗手。因此，离线策略学习往往能够更好地利用历史数据，并具有更小的样本复杂度，这使其被更广泛地应用。
时序差分方法：（类比算法中用于处理数组或序列的区间更新问题的差分法，反过来的是前缀和）时序差分算法用当前获得的奖励加上下一个状态的价值估计来作为在当前状态会获得的回报，α表示对价值估计更新的步长，即：
强调：时序差分算法的核心思想是用对未来动作选择的价值估计来更新对当前动作选择的价值估计
时序差分代表——Sarsa算法
策略评估可以通过时序差分算法实现，那么在不知道奖励函数和状态转移函数的情况下该怎么进行策略提升呢？答案是时可以直接用时序差分算法来估计动作价值函数Q：
然后用贪婪算法来选取在某个状态下动作价值最大的那个动作
如果在策略提升中一直根据贪婪算法得到一个确定性策略，可能会导致某些状态动作对永远没有在序列中出现，以至于无法对其动作价值进行估计，进而无法保证策略提升后的策略比之前的好，更实际的策略是使用epsilon-贪婪算法。
即：用贪婪算法根据动作价值选取动作来和环境交互，再根据得到的数据用时序差分算法更新动作价值估计。
class Sarsa:
""" Sarsa算法 """
def __init__(self, ncol, nrow, epsilon, alpha, gamma, n_action=4):
self.Q_table = np.zeros([nrow * ncol, n_action]) # 初始化Q(s,a)表格
self.n_action = n_action # 动作个数
self.alpha = alpha # 学习率
self.gamma = gamma # 折扣因子
self.epsilon = epsilon # epsilon-贪婪策略中的参数
def take_action(self, state): # 选取下一步的操作,具体实现为epsilon-贪婪
if np.random.random() < self.epsilon:
action = np.random.randint(self.n_action)
else:
action = np.argmax(self.Q_table[state])
return action
def best_action(self, state): # 用于打印策略
Q_max = np.max(self.Q_table[state])
a = [0 for _ in range(self.n_action)]
for i in range(self.n_action): # 若两个动作的价值一样,都会记录下来
if self.Q_table[state, i] == Q_max:
a[i] = 1
return a
def update(self, s0, a0, r, s1, a1):
td_error = r + self.gamma * self.Q_table[s1, a1] - self.Q_table[s0, a0]
self.Q_table[s0, a0] += self.alpha * td_error #td表示差分多步时序差分：“蒙特卡洛方法利用当前状态之后每一步的奖励而不使用任何价值估计，时序差分算法只利用一步奖励和下一个状态的价值估计。那它们之间的区别是什么呢？总的来说，蒙特卡洛方法是无偏（unbiased）的，但是具有比较大的方差，因为每一步的状态转移都有不确定性，而每一步状态采取的动作所得到的不一样的奖励最终都会加起来，这会极大影响最终的价值估计；时序差分算法具有非常小的方差，因为只关注了一步状态转移，用到了一步的奖励，但是它是有偏的，因为用到了下一个状态的价值估计而不是其真实的价值。那有没有什么方法可以结合二者的优势呢？答案是多步时序差分。多步时序差分的意思是使用n步的奖励，然后使用之后状态的价值估计。”
时序差分代表——Q-Learning
与前面最大的区别为时序差分更新方式（q-learning直接用max）：
算法流程和sarsa是一模一样的
Q-learning 算法中，我们以矩阵的方式建立了一张存储每个状态下所有动作Q值的表格。表格中的每一个动作价值Q(s,a)表示在状态s下选择动作a然后继续遵循某一策略预期能够得到的期望回报。
可以从价值迭代和贝尔曼最优方程的角度理解Q-Learning：
def best_action(self, state):
Q_max = np.max(self.Q_table[state])
a = [0 for _ in range(self.n_action)]
for i in range(self.n_action):
if self.Q_table[state, i] == Q_max:
a[i] = 1
return a
def update(self, s0, a0, r, s1):
td_error = r + self.gamma * self.Q_table[s1].max(
) - self.Q_table[s0, a0]
self.Q_table[s0, a0] += self.alpha * td_error强调：Sarsa 是在线策略（on-policy）算法，称 Q-learning 是离线策略（off-policy）算法
为什么sarsa是在线？看前面它的更新，它必须当前策略采样得到的数据（更新中用到的Q(s',a')的a'是当前策略在s'下的动作），而Q-Learning更新用的是max
我们称采样数据的策略为行为策略（behavior policy），称用这些数据来更新的策略为目标策略（target policy）。在线策略（on-policy）算法表示行为策略和目标策略是同一个策略；而离线策略（off-policy）算法表示行为策略和目标策略不是同一个策略。
对于 Q-learning，它的更新公式使用的是四元组(s,a,r,s’)来更新当前状态动作对的价值，数据中的s和a是给定的条件，r和s'皆由环境采样得到，该四元组并不需要一定是当前策略采样得到的数据，也可以来自行为策略，因此它是离线策略算法。
Dyna-Q算法
在强化学习中，“模型”通常指与智能体交互的环境模型，即对环境的状态转移概率和奖励函数进行建模。根据是否具有环境模型，强化学习算法分为两种：基于模型的强化学习和无模型的强化学习。在基于模型的强化学习中，模型可以是事先知道的，也可以是根据智能体与环境交互采样到的数据学习得到的，然后用这个模型帮助策略提升或者价值估计。
“基于模型的强化学习算法由于具有一个环境模型，智能体可以额外和环境模型进行交互，对真实环境中样本的需求量往往就会减少，因此通常会比无模型的强化学习算法具有更低的样本复杂度。但是，环境模型可能并不准确，不能完全代替真实环境，因此基于模型的强化学习算法收敛后其策略的期望回报可能不如无模型的强化学习算法。”
Dyna-Q 使用一种叫做 Q-planning 的方法来基于模型生成一些模拟数据，然后用模拟数据和真实数据一起改进策略。Q-planning 每次选取一个曾经访问过的状态s，采取一个曾经在该状态下执行过的动作a，通过模型得到转移后的状态s'以及奖励r，并根据这个模拟数据(s,a,r,s')，用 Q-learning 的更新方式来更新动作价值函数。（强调：在s下执行a后得到的s'和r是不确定的，这由环境的随机性决定。这被称为随机环境或概率性状态转移。）
Dyna-Q的算法框架：
import numpy as np
import matplotlib.pyplot as plt
from tqdm import tqdm
import random
import time
class Env:
def __init__(self,ncol,nrow):
self.ncol = ncol
self.nrow = nrow
self.x = 0
self.y = self.nrow - 1
def step(self,action):
# action: 0左，1右，2上，3下
change = [[0, -1], [0, 1], [-1, 0], [1, 0]]
self.x = min(self.ncol - 1, max(0, self.x + change[action][0]))
self.y = min(self.nrow - 1, max(0, self.y + change[action][1]))
next_state = self.y * self.ncol + self.x
reward = -1
done = False
if self.y == self.nrow - 1 and self.x > 0: # 下一个位置在悬崖或者目标
done = True
if self.x != self.ncol - 1:
reward = -100
return next_state, reward, done
def reset(self):
self.x = 0
self.y = self.nrow - 1
return self.y*self.ncol + self.x
class DynaQ:
def __init__(self,ncol,nrow,epsilon,alpha,gamma,n_planning,n_action=4):
self.q_table = np.zeros((nrow*ncol,n_action)) #q表初始化，这里用位置的组合表示状态，n_action为动作个数
self.epsilon = epsilon
self.alpha = alpha
self.gamma = gamma
self.n_planning = n_planning
self.n_action = n_action
self.model = dict() # 记录环境模型，key为(s,a),value为(r,s')
def take_action(self,state):
if np.random.random() < self.epsilon:
action = np.random.randint(0,self.n_action)
else:
action = np.argmax(self.q_table[state]) # 对应Q表选择价值最大的动作
return action
def q_learning(self,s0,a0,r,s1):
self.q_table[s0][a0] += self.alpha * (r + self.gamma * np.max(self.q_table[s1]) - self.q_table[s0][a0])
def update(self,s0,a0,r,s1):
self.q_learning(s0,a0,r,s1)
self.model[(s0,a0)] = (r,s1) # 更新环境模型
for _ in range(self.n_planning):
(s,a),(r,s_next) = random.choice(list(self.model.items()))
self.q_learning(s,a,r,s_next)
def run(n_planning):
ncol = 12
nrow = 4
env = Env(ncol,nrow)
epsilon = 0.01
alpha = 0.1
gamma = 0.9
agent = DynaQ(ncol,nrow,epsilon,alpha,gamma,n_planning)
n_episodes = 300 # 智能体在环境中运行多少条序列
return_list = [] # 记录每条序列的总奖励
for i in range(10):
with tqdm(total=n_episodes/10,desc='n_planning=%d'%n_planning) as pbar:
for j in range(n_episodes//10):
state = env.reset()
total_reward = 0
done = False
while not done:
action = agent.take_action(state)
next_state,reward,done = env.step(action)
agent.update(state,action,reward,next_state)
state = next_state
total_reward += reward # 这里的回报计算没计入衰减因子
if state == env.ncol - 1: # 到达右下角
break
return_list.append(total_reward)
if (j+1)%10 == 0:
pbar.set_postfix({'episode':i*n_episodes//10 + j+1,'mean_reward':np.mean(return_list[-10:])})
pbar.update(1)
return return_list
if __name__ == '__main__':
np.random.seed(0)
random.seed(0)
n_planning_list = [0, 2, 20]
for n_planning in n_planning_list:
print('Q-planning步数为：%d' % n_planning)
time.sleep(0.5)
return_list = run(n_planning)
episodes_list = list(range(len(return_list)))
plt.plot(episodes_list,
return_list,
label=str(n_planning) + ' planning steps')
plt.legend()
plt.xlabel('Episodes')
plt.ylabel('Returns')
plt.title('Dyna-Q on {}'.format('Cliff Walking'))
plt.show()上面的实现也可以顺便学习Q-Learning，重点是take_action q_learning update三个函数，如何对应到公式的！
“从上述结果中我们可以很容易地看出，随着 Q-planning 步数的增多，Dyna-Q 算法的收敛速度也随之变快。当然，并不是在所有的环境中，都是 Q-planning 步数越大则算法收敛越快，这取决于环境是否是确定性的，以及环境模型的精度。在上述悬崖漫步环境中，状态的转移是完全确定性的，构建的环境模型的精度是最高的，所以可以通过增加 Q-planning 步数来直接降低算法的样本复杂度。”
强化学习进阶
DQN算法
当状态数量很多或者状态是连续的时候，就没办法用Q表存储，所以，需要使用函数拟合的方法来估计 Q。这种函数拟合的方法存在一定的精度损失，因此被称为近似方法。DQN 算法便可以用来解决连续状态下离散动作的问题。
这里举得例子是车杆问题。对于用神经网络表示Q，若动作是连续（无限）的，神经网络的输入是状态和动作，然后输出一个标量，表示在状态下采取动作能获得的价值。若动作是离散（有限）的，除了可以采取动作连续情况下的做法，我们还可以只将状态输入到神经网络中，使其同时输出每一个动作的值（相当于输出端变为多分类）。
对于损失函数，从原先Q 值收敛的角度来理解：
此即为深度 Q 网络（deep Q network，DQN）算法。由于 DQN 是离线策略算法，因此我们在收集数据的时候可以使用一个-贪婪策略来平衡探索与利用。DQN 中还有两个非常重要的模块——经验回放和目标网络，它们能够帮助 DQN 取得稳定、出色的性能。
经验回放
“在一般的有监督学习中，假设训练数据是独立同分布的，我们每次训练神经网络的时候从训练数据中随机采样一个或若干个数据来进行梯度下降，随着学习的不断进行，每一个训练数据会被使用多次。在原来的 Q-learning 算法中，每一个数据只会用来更新一次Q值。为了更好地将 Q-learning 和深度神经网络结合，DQN 算法采用了经验回放（experience replay）方法，具体做法为维护一个回放缓冲区，将每次从环境中采样得到的四元组数据（状态、动作、奖励、下一状态）存储到回放缓冲区中，训练 Q 网络的时候再从回放缓冲区中随机采样若干数据来进行训练。这么做可以起到以下两个作用。
（1）使样本满足独立假设。在 MDP 中交互采样得到的数据本身不满足独立假设，因为这一时刻的状态和上一时刻的状态有关。非独立同分布的数据对训练神经网络有很大的影响，会使神经网络拟合到最近训练的数据上。采用经验回放可以打破样本之间的相关性，让其满足独立假设。
（2）提高样本效率。每一个样本可以被使用多次，十分适合深度神经网络的梯度学习。”
目标网络
经典DQN算法流程：——深度强化学习领域的开山之作
import random
import gym
import numpy as np
import collections
from tqdm import tqdm
import torch
import torch.nn.functional as F
import matplotlib.pyplot as plt
import rl_utils
class ReplayBuffer:
"""经验回放池，主要是加入数据和采样数据函数"""
def __init__(self, capacity):
self.buffer = collections.deque(maxlen=capacity)
def add(self, state, action, reward, next_state, done):
self.buffer.append((state, action, reward, next_state, done)) # (s,a,r,s',done)
def sample(self, batch_size):
transitions = random.sample(self.buffer, batch_size) # 从buffer中采样batch_size个数据
state, action, reward, next_state, done = zip(*transitions)
return np.array(state), action, reward, np.array(next_state), done
def size(self):
return len(self.buffer)
class Qnet(torch.nn.Module):
def __init__(self, state_dim, hidden_dim, action_dim):
super(Qnet, self).__init__()
self.fc1 = torch.nn.Linear(state_dim, hidden_dim)
self.fc2 = torch.nn.Linear(hidden_dim, action_dim)
def forward(self, x):
x = F.relu(self.fc1(x))
return self.fc2(x)
class DQN:
def __init__(self, state_dim, hidden_dim, action_dim, learning_rate, gamma, epsilon, target_update, device):
self.action_dim = action_dim
self.gamma = gamma
self.epsilon = epsilon
self.device = device
self.q_net = Qnet(state_dim, hidden_dim, action_dim).to(device)
self.target_net = Qnet(state_dim, hidden_dim, action_dim).to(device)
self.optimizer = torch.optim.Adam(self.q_net.parameters(), lr=learning_rate)
self.target_update = target_update
self.count = 0 # 记录更新次数
def take_action(self, state):
# epsilon-greedy策略选择动作(经典q-learning) 随机选动作和根据Q值最大的选
if np.random.random() < self.epsilon:
action = np.random.randint(self.action_dim)
else:
state = torch.tensor([state], dtype=torch.float).to(self.device)
q_values = self.q_net(state)
action = torch.argmax(q_values).item()
return action
def update(self, transition_dict):
# 采样字典
states = torch.tensor(transition_dict['states'], dtype=torch.float).to(self.device)
actions = torch.tensor(transition_dict['actions']).view(-1, 1).to(self.device)
rewards = torch.tensor(transition_dict['rewards'], dtype=torch.float).view(-1, 1).to(self.device)
next_states = torch.tensor(transition_dict['next_states'], dtype=torch.float).to(self.device)
dones = torch.tensor(transition_dict['dones'], dtype=torch.float).view(-1, 1).to(self.device)
q_values = self.q_net(states).gather(1, actions) # 训练网络正常预测Q(s,a)
max_next_q_values = self.target_net(next_states).max(1)[0].view(-1, 1) # max_a' Q_target(s',a')
q_targets = rewards + self.gamma * max_next_q_values * (1 - dones) # 时序差分
dqn_loss = torch.mean(F.mse_loss(q_values, q_targets)) # 均方误差损失函数
self.optimizer.zero_grad() # 梯度置0
dqn_loss.backward()
self.optimizer.step()
if self.count % self.target_update == 0:
self.target_net.load_state_dict(self.q_net.state_dict()) # 目标网络是滞后更新！
self.count += 1
if __name__ == "__main__":
lr = 2e-3
num_episodes = 500
hidden_dim = 128
gamma = 0.98
epsilon = 0.01
target_update = 10
buffer_size = 10000
minimal_size = 500
batch_size = 64
device = torch.device("cuda") if torch.cuda.is_available() else torch.device("cpu")
env_name = 'CartPole-v1'
env = gym.make(env_name)
random.seed(0)
np.random.seed(0)
torch.manual_seed(0)
replay_buffer = ReplayBuffer(buffer_size)
state_dim = env.observation_space.shape[0]
action_dim = env.action_space.n
agent = DQN(state_dim, hidden_dim, action_dim, lr, gamma, epsilon,
target_update, device)
return_list = []
for i in range(10):
with tqdm(total=int(num_episodes / 10), desc='Iteration %d' % i) as pbar:
for i_episode in range(int(num_episodes / 10)):
episode_return = 0
# 新版gym，reset()返回的是(state, info)元组
state = env.reset()
if isinstance(state, tuple):
state = state[0] # 取状态部分
done = False
truncated = False
while not (done or truncated):
action = agent.take_action(state)
result = env.step(action)
if len(result) == 5:
next_state, reward, terminated, truncated, info = result
done = terminated
else:
next_state, reward, done, info = result
truncated = False
replay_buffer.add(state, action, reward, next_state, done or truncated)
state = next_state
episode_return += reward
if replay_buffer.size() > minimal_size:
b_s, b_a, b_r, b_ns, b_d = replay_buffer.sample(batch_size)
transition_dict = {
'states': b_s,
'actions': b_a,
'next_states': b_ns,
'rewards': b_r,
'dones': b_d
}
agent.update(transition_dict)
return_list.append(episode_return)
if (i_episode + 1) % 10 == 0:
pbar.set_postfix({
'episode': '%d' % (num_episodes / 10 * i + i_episode + 1),
'return': '%.3f' % np.mean(return_list[-10:])
})
pbar.update(1)
episodes_list = list(range(len(return_list)))
plt.plot(episodes_list, return_list)
plt.xlabel('Episodes')
plt.ylabel('Returns')
plt.title('DQN on CartPole-v1')
plt.show()
mv_return = rl_utils.moving_average(return_list, 9)
plt.plot(episodes_list, mv_return)
plt.xlabel('Episodes')
plt.ylabel('Returns')
plt.title('DQN on CartPole-v1')
plt.show()仔细阅读代码，尤其是几个类的定义，辅助理解！
DQN算法拓展
拓展是以图像为输入：“在一些视频游戏中，智能体并不能直接获取这些状态信息，而只能直接获取屏幕中的图像。要让智能体和人一样玩游戏，我们需要让智能体学会以图像作为状态时的决策。我们将卷积层加入其网络结构以提取图像特征，最终实现以图像为输入的强化学习。”
class ConvolutionalQnet(torch.nn.Module):
def __init__(self, action_dim, in_channels=4):
super(ConvolutionalQnet, self).__init__()
self.conv1 = torch.nn.Conv2d(in_channels, 32, kernel_size=8, stride=4)
self.conv2 = torch.nn.Conv2d(32, 64, kernel_size=4, stride=2)
self.conv3 = torch.nn.Conv2d(64, 64, kernel_size=3, stride=1)
self.fc4 = torch.nn.Linear(7 * 7 * 64, 512)
self.head = torch.nn.Linear(512, action_dim)
def forward(self, x):
x = x / 255
x = F.relu(self.conv1(x))
x = F.relu(self.conv2(x))
x = F.relu(self.conv3(x))
x = F.relu(self.fc4(x))
return self.head(x)Double DQN： 红线部分辅助理解，方框部分为重点精髓
详细解释精髓：
在传统DQN的更新中，我们通过目标网络计算下一状态的最大Q值：max_a’ Q_target(s’, a’)。这里隐藏的问题是：由于q-target网络（或原始的DQN中单一的Q网络）本身存在估计误差，这个“取最大值”的操作会将所有可能的误差中正向的偏差（高估）挑选出来，而忽略负向的偏差（低估）。Double DQN的核心思想是：“选出最好动作”和“评估这个动作的价值”是两件不同的事，应该用两套独立的、有差异的参数来完成，以防止误差的“自催化”循环。
为什么是训练网络选动作，目标网络算Q值，而不是反过来？
我们需要一个决策者来选出当前认为最好的动作。我们希望这个决策相对准确、反应快。训练网络在每个时间步都更新，能更快地学习到最新的、更优的策略，因此用训练网络来选择动作
当选出动作后，我们需要一个评估者来客观、稳定地给出这个动作的估值。目标网络的参数定期从训练网络同步，更新频率低，因此它的输出更稳定、方差更小、偏差也相对稳定。用它来计算这个选定动作的Q值
如果反过来，比如用目标网络选动作，由于目标参数滞后，它选出来的可能已经不是当前策略下的最优动作
update函数里二者的区别：
if self.dqn_type == 'DoubleDQN': # DQN与Double DQN的区别
max_action = self.q_net(next_states).max(1)[1].view(-1, 1)
max_next_q_values = self.target_q_net(next_states).gather(1, max_action)
else: # DQN的情况
max_next_q_values = self.target_q_net(next_states).max(1)[0].view(-1, 1)Dueling DQN：引入了优势函数——状态动作价值函数Q-状态价值函数V（重申，状态价值函数定义为从状态s出发遵循策略π能获得的期望回报，表示在 MDP 遵循策略π时，对当前状态s执行动作a得到的期望回报）
公式可能会引入不稳定性，因为将V加上个常数，所有A值减这个常数，Q是不变的，所以还会在后面减一个max A保证唯一性：
为什么要引入优势函数？
——“将状态价值函数和优势函数分别建模的好处在于：某些情境下智能体只会关注状态的价值，而并不关心不同动作导致的差异，此时将二者分开建模能够使智能体更好地处理与动作关联较小的状态。在图 8-3 所示的驾驶车辆游戏中，智能体注意力集中的部位被显示为橙色（另见彩插图 4），当智能体前面没有车时，车辆自身动作并没有太大差异，此时智能体更关注状态价值，而当智能体前面有车时（智能体需要超车），智能体开始关注不同动作优势值的差异。”
Dueling DQN 能更高效学习状态价值函数。每一次更新时，函数V都会被更新，这也会影响到其他动作的Q值。而传统的 DQN 只会更新某个动作的Q值，其他动作的Q值就不会更新。因此，Dueling DQN 能够更加频繁、准确地学习状态价值函数。
补一个网络结构图：
前面是共享的，后面几层分别输出V和A~！
对应的代码：
class VAnet(torch.nn.Module):
''' 只有一层隐藏层的A网络和V网络 '''
def __init__(self, state_dim, hidden_dim, action_dim):
super(VAnet, self).__init__()
self.fc1 = torch.nn.Linear(state_dim, hidden_dim) # 共享网络部分
self.fc_A = torch.nn.Linear(hidden_dim, action_dim)
self.fc_V = torch.nn.Linear(hidden_dim, 1)
def forward(self, x):
A = self.fc_A(F.relu(self.fc1(x)))
V = self.fc_V(F.relu(self.fc1(x)))
Q = V + A - A.mean(1).view(-1, 1) # Q值由V值和A值计算得到
return Q策略梯度算法
上面的方法都是基于价值的方法，还有一类经典的方法是基于策略的。基于值函数的方法主要是学习值函数，然后根据值函数导出一个策略，学习过程中并不存在一个显式的策略；而基于策略的方法则是直接显式地学习一个目标策略。策略梯度是基于策略的方法的基础。
将策略参数化，具体表现为神经网络模型，目标函数定义为从初始状态出发的状态价值的期望（这个期望应该越大越好）
对目标函数求导，下面有非常重要的推导：
左边表示目标函数关于策略参数θ的梯度（▽是梯度算子）
小v表示在策略π_θ下的折扣状态分布，表示在遵循策略 πθ时，访问状态 s的折扣频率
π_θ(a|s)是参数化策略，表示在状态 s下选择动作 a的概率，前面加个梯度算子表示参数微小变化时，选择动作 a的概率如何变化
这是策略梯度定理的核心结果，证明过程略，仅从直觉观感上来解释
比例符号 ∝表示与右侧成正比，因为省略了常数因子
直觉：对每个状态 s（按访问频率加权），考虑所有可能动作 a
用 Qπθ(s,a)衡量动作好坏，乘以策略对该动作的梯度 ∇θπθ(a∣s)
好动作（高Q值）应该增加其选择概率
接着，推导的第二步左边乘π_θ(a|s)，右边除π_θ(a|s)，这样目的是构造期望形式，为下一步做准备
然后：
把它带进去得到下面的：
我们先对所有状态 s求和，权重是 νπθ(s)（这个状态被访问到的频率）。
对每个状态 s，我们又对所有动作 a求和，权重是 πθ(a∣s)（在这个状态下选择这个动作的概率）。
求和的对象是 Qπθ(s,a)∇θlogπθ(a∣s)
而这正是那个求和对象的期望的定义：
据对 (s,a)的生成过程是：先按 νπθ(s)的分布得到状态 s，然后根据 πθ(a∣s)的分布选择动作 a。
我们对这个生成过程中产生的 Qπθ(s,a)∇θlogπθ(a∣s)这个量取平均值。
策略梯度算法为在线策略（on-policy）算法，即必须使用当前策略采样得到的数据来计算梯度。直观理解一下策略梯度这个公式，可以发现在每一个状态下，梯度的修改是让策略更多地去采样到带来较高Q值的动作，更少地去采样到带来较低Q值的动作
公式需要用到Q(s,a)，REINFORCE 算法便是采用了蒙特卡洛方法来估计它。
REINFORCE
它的梯度更新策略：
把原公式的Q(s,a)换成了中间的两个求和，算法流程：
强调：REINFORCE 算法是一个在线策略算法，之前收集到的轨迹数据不会被再次利用（它马上就要更新策略了）。
实现：首先需要定义神经网络将策略参数化——输入是某个状态，输出则是该状态下的动作概率分布，这里采用在离散动作空间上的softmax()函数来实现一个可学习的多项分布
class PolicyNet(torch.nn.Module):
def __init__(self, state_dim, hidden_dim, action_dim):
super(PolicyNet, self).__init__()
self.fc1 = torch.nn.Linear(state_dim, hidden_dim)
self.fc2 = torch.nn.Linear(hidden_dim, action_dim)
def forward(self, x):
x = F.relu(self.fc1(x))
return F.softmax(self.fc2(x), dim=1)class REINFORCE:
def __init__(self, state_dim, hidden_dim, action_dim, learning_rate, gamma, device):
self.policy_net = PolicyNet(state_dim, hidden_dim,action_dim).to(device)
self.optimizer = torch.optim.Adam(self.policy_net.parameters(),lr=learning_rate)
self.gamma = gamma # 折扣因子
self.device = device
def take_action(self, state): # 根据动作概率分布随机采样
state = torch.tensor([state], dtype=torch.float).to(self.device)
probs = self.policy_net(state)
action_dist = torch.distributions.Categorical(probs) # PyTorch提供的分类分布类，里面还可以计算shang
action = action_dist.sample()
return action.item()
def update(self, transition_dict):
reward_list = transition_dict['rewards']
state_list = transition_dict['states']
action_list = transition_dict['actions']
G = 0
self.optimizer.zero_grad()
for i in reversed(range(len(reward_list))): # 从最后一步算起， 因为有个折扣因子，最后的乘的最多
reward = reward_list[i]
state = torch.tensor([state_list[i]], dtype=torch.float).to(self.device)
action = torch.tensor([action_list[i]]).view(-1, 1).to(self.device)
log_prob = torch.log(self.policy_net(state).gather(1, action))
G = self.gamma * G + reward
loss = -log_prob * G # 每一步的损失函数 目标函数最大化，取个负数就可以用最小化了
loss.backward() # 反向传播计算梯度
self.optimizer.step() # 梯度下降Actor-Critic算法
基于值函数的方法只学习一个价值函数，而基于策略的方法只学习一个策略函数。那么，一个很自然的问题是，有没有什么方法既学习价值函数，又学习策略函数呢？答案就是 Actor-Critic
Actor-Critic 算法本质上是基于策略的算法，因为这一系列算法的目标都是优化一个带参数的策略，只是会额外学习价值函数，从而帮助策略函数更好地学习。
再次强调前面目标函数的梯度：
从感性角度来理解更一般的形式，梯度算子右边表示选某个动作的概率，坐标代表奖励（动作好坏的评价指标）
Actor-Critic 分为两个部分：Actor（策略网络）和 Critic（价值网络），策略网络与环境交互，并在critic价值函数的指导下学习更好的策略
将价值网络表示为V_w，参数为w，可以对单个数据定义价值函数的损失函数：
然后可以从梯度下降的角度来更新参数
算法流程：
定义网络：
class PolicyNet(torch.nn.Module):
"""也是一样的将策略参数化"""
def __init__(self, state_dim, hidden_dim, action_dim):
super(PolicyNet, self).__init__()
self.fc1 = torch.nn.Linear(state_dim, hidden_dim)
self.fc2 = torch.nn.Linear(hidden_dim, action_dim)
def forward(self, x):
x = F.relu(self.fc1(x))
return F.softmax(self.fc2(x), dim=1)
class ValueNet(torch.nn.Module):
def __init__(self, state_dim, hidden_dim):
super(ValueNet, self).__init__()
self.fc1 = torch.nn.Linear(state_dim, hidden_dim)
self.fc2 = torch.nn.Linear(hidden_dim, 1)
def forward(self, x):
x = F.relu(self.fc1(x))
return self.fc2(x)强调valuenet输入是某个状态，输出则是状态的价值。（所以最后一维度为1） policy是actor网络的 actor根据输入的状态，forward出下一动作的概率序列。损失函数中下一状态的价值由valuenet给出
class ActorCritic:
def __init__(self, state_dim, hidden_dim, action_dim, actor_lr, critic_lr, gamma, device):
self.actor = PolicyNet(state_dim, hidden_dim, action_dim).to(device)
self.critic = ValueNet(state_dim, hidden_dim).to(device)
self.actor_optimizer = torch.optim.Adam(self.actor.parameters(), lr=actor_lr)
self.critic_optimizer = torch.optim.Adam(self.critic.parameters(),lr=critic_lr)
self.gamma = gamma
self.device = device
def take_action(self, state):
state = torch.tensor([state], dtype=torch.float).to(self.device)
probs = self.actor(state)
action_dist = torch.distributions.Categorical(probs)
action = action_dist.sample()
return action.item()
def update(self, transition_dict):
states = torch.tensor(transition_dict['states'], dtype=torch.float).to(self.device)
actions = torch.tensor(transition_dict['actions']).view(-1, 1).to(self.device)
rewards = torch.tensor(transition_dict['rewards'], dtype=torch.float).view(-1, 1).to(self.device)
next_states = torch.tensor(transition_dict['next_states'], dtype=torch.float).to(self.device)
dones = torch.tensor(transition_dict['dones'], dtype=torch.float).view(-1, 1).to(self.device)
# 时序差分目标
td_target = rewards + self.gamma * self.critic(next_states) * (1 - dones)
td_delta = td_target - self.critic(states) # 时序差分误差
log_probs = torch.log(self.actor(states).gather(1, actions))
actor_loss = torch.mean(-log_probs * td_delta.detach())
# 均方误差损失函数
critic_loss = torch.mean(F.mse_loss(self.critic(states), td_target.detach()))
self.actor_optimizer.zero_grad()
self.critic_optimizer.zero_grad()
actor_loss.backward() # 计算策略网络的梯度
critic_loss.backward() # 计算价值网络的梯度
self.actor_optimizer.step() # 更新策略网络的参数
self.critic_optimizer.step() # 更新价值网络的参数关于代码的一些分析：actor的损失函数=−E[logπθ(a∣s)⋅δ]（跟之前一样学习策略 δ指时序差分误差 logπθ(a∣s)对应log_probs，即实际执行动作的对数概率）
PS：这里算actor_loss的时候，td_delta记得detach，它返回一个新张量，与原始张量共享数据，但没有梯度历史（换而言之创建了一个计算图断点，由于log_prob是actor算的，所以actor_loss会被绑到actor的优化器上，而原先的td_delta是critic算的；如果不加，当计算actor_loss.backward()时，梯度会传播到Critic网络）
critic的目标是最小化贝尔曼误差：
根据贝尔曼方程：V(s)=E[r+γV(s′)]
在理想情况下，如果Critic完全准确：V(s)=r+γV(s′)
但在学习过程中，Critic的预测 V(s)和真实的 r+γV(s′)有差距，这个差距就是TD误差。
合在一起，即critic预测价值，actor更新策略，再次呼应前文：Actor-Critic 算法本质上是基于策略的算法，因为这一系列算法的目标都是优化一个带参数的策略，只是会额外学习价值函数，从而帮助策略函数更好地学习。
价值模块 Critic 在策略模块 Actor 采样的数据中学习分辨什么是好的动作，什么不是好的动作，进而指导 Actor 进行策略更新。后续章节中的 TRPO、PPO、DDPG、SAC 等深度强化学习算法都是在 Actor-Critic 框架下进行发展的。
TRPO算法
基于策略的方法：参数化智能体的策略，并设计衡量策略好坏的目标函数，通过梯度上升的方法来最大化这个目标函数，使得策略最优。策略梯度算法主要沿着 ▽J(θ) 方向迭代更新策略参数θ 。但是这种算法有一个明显的缺点：当策略网络是深度模型时， 沿着策略梯度更新参数，很有可能由于步长太长，策略突然显著变差。
感性理解这段话：以开车上山为例，算法告诉AI：“你刚才那一系列方向盘和油门的操作，整体上得到+x分。所以，把你所有的操作习惯，都朝着刚才那个方向再强化一下” 然后它大幅度调整大脑神经元的连接强度（参数θ进行一个大步长的更新 θ ← θ + α∇_θJ）
问题在于大步长，神经网络是一个高维的非线性空间，性能与参数从来不是平滑的斜坡
问题还在于所有，策略梯度把一整段旅程的“综合分”平均地反馈给了每一个操作，这会导致粗暴且不精确的更新。
TRPO（信任域策略优化）意识到上述教学方式是鲁莽的。它的核心是设立一个 “信任域” 。这个域不是地理上的，而是指在AI司机当前的“驾驶习惯”（旧策略π_θ_old）周围，划定一个它能力可及、我们确信不会出大错的“行为习惯变化范围”。
优势函数被拆成了两层期望，内层期望：在时间 t，状态 st服从新策略在第 t步的分布；外层期望：给定状态 st，动作 at服从新策略的条件分布
这个公式精确地告诉我们：新策略比旧策略好多少，取决于在新策略自己的状态分布下，其动作的平均优势有多大。
从直观感性的角度来理解：我们把"沿着轨迹的期望"拆解为"每个时间步的期望之和"。在每个时间步：
先看新策略能到达哪些状态（小v代表状态分布）
在新策略到达的状态下，看它会采取什么动作
评估这些动作相对于旧策略的优势
所以，我们要找的新策略应该满足优势函数大于0。但是直接求解该式是非常困难的，公式右边依赖于新策略的状态分布 ν^π_θ'。要知道这个分布，就必须先用新策略 π_θ'与环境交互，但 π_θ'恰恰是我们还没确定、正在优化的对象。这成了一个“先有鸡还是先有蛋”的死循环。。于是 TRPO 做了一步近似操作，对状态访问分布进行了相应处理。具体而言，忽略两个策略之间的状态访问分布变化，直接采用旧的策略π_θ的状态分布，定义如下替代优化目标：
当新旧策略非常接近时，状态访问分布变化很小，近似是合理的。其中，动作仍然用新策略采样得到，我们可以用重要性采样对动作分布进行处理：
重要性采样（Importance Sampling）是强化学习和蒙特卡洛方法中的核心技巧，它让我们能够用来自一个分布的样本来估计另一个分布下的期望值。比如：
从分布 p中采样困难或成本高；我们已经有了来自另一个分布 q的样本，我们可以这么干（离散版本更好理解）：
这样就可以基于旧策略已经采样出的数据来估计并优化新策略了，为了保证新旧策略足够接近，TRPO 使用了库尔贝克-莱布勒（Kullback-Leibler，KL）散度来衡量策略之间的距离，整体优化公式（θk替代了之前的θ，表示这是第k次迭代）：
直接求解上述问题比较困难，因为目标函数和约束都依赖于新策略。TRPO 通过在 θk处进行泰勒展开来近似（g代表目标函数在 θk处的梯度向量）：
目标函数用一阶近似：
约束条件用二阶近似：
所以得到近似的整体优化公式：
用拉格朗日数乘法求解，结合KKT公式：
共轭梯度法求解：看上面的解析解，直接计算 H−1g需要存储和求逆黑塞矩阵 H，而 H的维度等于策略参数的个数（成千上万），计算逆矩阵的时空复杂度极高（O(n3)时间，O(n2)存储），在深度强化学习中几乎不可行。
先介绍线性代数中的共轭梯度法：共轭梯度法是一种用于求解大型对称正定线性方程组的高效迭代算法。比如求解：Ax=b（A是对称正定矩阵，对称且所有特征值大于0）
对于对称正定矩阵 A，两个向量 u和 v是 A-共轭 的，如果：uTAv=0
这可以看作是在由 A定义的内积空间中的正交性。
如果有一组 A-共轭的搜索方向 {p0,p1,…,pn−1}，那么沿着每个方向 pk执行精确线搜索，最多 n步就能找到最优解 x∗
标准共轭梯度流程——其实就是搜索二次函数的最小值点
线性搜索：
广义优势估计：
重点掌握TRPO算法的流程：采样，然后根据之前重要性采样等推导，用旧策略的轨迹可得到目标函数L(θ')，算其梯度，利用共轭梯度法算x并带到前面推出来的策略更新公式中
详细研读下面的实现代码，找到代码和公式的一一对应~
import gym
import torch
import torch.nn.functional as F
import numpy as np
import copy
import matplotlib.pyplot as plt
import rl_utils
def compute_advantage(gamma, lmbda, td_delta):
"""计算GAE优势函数"""
td_delta = td_delta.detach().numpy()
advantage_list = []
advantage = 0.0
for delta in td_delta[::-1]:
advantage = gamma * lmbda * advantage + delta
advantage_list.append(advantage)
advantage_list.reverse()
return torch.tensor(advantage_list, dtype=torch.float)
class PolicyNet(torch.nn.Module):
def __init__(self, state_dim, hidden_dim, action_dim):
super(PolicyNet, self).__init__()
self.fc1 = torch.nn.Linear(state_dim, hidden_dim)
self.fc2 = torch.nn.Linear(hidden_dim, action_dim)
self.softmax = torch.nn.Softmax(dim=-1)
def forward(self, x):
x = F.relu(self.fc1(x))
x = self.fc2(x)
return self.softmax(x)
class ValueNet(torch.nn.Module):
"""价值网络是输出该状态的价值估计"""
def __init__(self, state_dim, hidden_dim):
super(ValueNet, self).__init__()
self.fc1 = torch.nn.Linear(state_dim, hidden_dim)
self.fc2 = torch.nn.Linear(hidden_dim, 1)
def forward(self, x):
x = F.relu(self.fc1(x))
return self.fc2(x)
class TRPO:
def __init__(self, hidden_dim, state_place, action_place, lmbda, kl_constraint, alpha, critic_lr, gamma, device):
state_dim = state_place.shape[0]
action_dim = action_place.n
self.actor = PolicyNet(state_dim, hidden_dim, action_dim).to(device)
self.critic = ValueNet(state_dim, hidden_dim).to(device)
# 强调policynet不需要优化器，价值的更新是通过那个公式来的
self.critic_optimizer = torch.optim.Adam(self.critic.parameters(), lr=critic_lr)
self.lmbda = lmbda # GAE中的λ参数
self.kl_constraint = kl_constraint # KL散度约束
self.alpha = alpha # 线性搜索参数
self.gamma = gamma # 折扣因子
self.device = device
def take_action(self, state):
state = torch.tensor([state], dtype=torch.float).to(self.device)
action_probs = self.actor(state)
action_dist = torch.distributions.Categorical(action_probs)
action = action_dist.sample()
return action.item()
def hessian_matrix_vector_product(self, states, old_action_probs, vector):
"""计算Hessian矩阵与向量的乘积"""
# 黑塞矩阵源自于两个策略之间KL距离的二阶导数
new_action_dists = torch.distributions.Categorical(self.actor(states))
kl = torch.mean(torch.distributions.kl.kl_divergence(old_action_probs, new_action_dists))
kl_grad = torch.autograd.grad(kl, self.actor.parameters(), create_graph=True)
kl_grad_vector = torch.cat([g.contiguous().view(-1) for g in kl_grad]) # 梯度展平为一维向量
# KL距离的梯度先和vector做点积
kl_grad_vector_product = torch.dot(kl_grad_vector, vector)
grad2 = torch.autograd.grad(kl_grad_vector_product, self.actor.parameters())
grad2_vector = torch.cat([g.contiguous().view(-1) for g in grad2])
return grad2_vector
def conjugate_gradient(self,grad,states,old_action_dists):
"""共轭梯度算法解方程"""
x = torch.zeros_like(grad)
r = grad.clone()
p = grad.clone()
r_dot_r = torch.dot(r, r)
for _ in range(10):
Ap = self.hessian_matrix_vector_product(states, old_action_dists, p)
alpha = r_dot_r / (torch.dot(p, Ap))
x += alpha * p
r -= alpha * Ap
new_r_dot_r = torch.dot(r, r)
if new_r_dot_r < 1e-10:
break
beta = new_r_dot_r / (r_dot_r + 1e-8)
p = r + beta * p
r_dot_r = new_r_dot_r
return x
def compute_surrogate_loss(self, states, actions, advantages, old_log_probs, actor):
"""计算策略的替代损失函数，也就是公式里面的目标函数"""
"""actor是策略网络，输入状态到策略网络，得到各动作的概率，1表示在第1维度（动作维度）上收集"""
new_log_probs = torch.log(actor(states).gather(1, actions))
"""重要性采样，A/B= exp(logA-logB)"""
ratio = torch.exp(new_log_probs - old_log_probs)
# 乘优势函数
surrogate_loss = torch.mean(ratio * advantages)
return surrogate_loss
def line_search(self, states, actions, advantages, old_log_probs, old_action_dists, max_vec):
"""线性搜索寻找合适的步长，max_vec是参数空间中的一个更新方向向量，由共轭梯度方法计算得到"""
old_para = torch.nn.utils.parameters_to_vector(self.actor.parameters())
old_obj = self.compute_surrogate_loss(states, actions, advantages, old_log_probs, self.actor)
for i in range(15):
step_frac = self.alpha ** i
new_para = old_para + step_frac * max_vec
new_actor = copy.deepcopy(self.actor)
torch.nn.utils.vector_to_parameters(new_para, new_actor.parameters())
new_action_dists = torch.distributions.Categorical(new_actor(states))
kl = torch.mean(torch.distributions.kl.kl_divergence(old_action_dists, new_action_dists))
new_obj = self.compute_surrogate_loss(states, actions, advantages, old_log_probs, new_actor)
if kl <= self.kl_constraint and new_obj >= old_obj:
return new_para
return old_para # 如果没有找到合适的步长，就返回旧参数
def policy_learn(self, states, actions, old_action_dists, old_log_probs, advantage):
# 更新策略网络
surrogate_obj = self.compute_surrogate_loss(states, actions, advantage, old_log_probs, self.actor)
grads = torch.autograd.grad(surrogate_obj, self.actor.parameters())
obj_grad = torch.cat([g.contiguous().view(-1) for g in grads]).detach()
# 共轭梯度法计算搜索方向 x= H^-1 g
descent_direction = self.conjugate_gradient(obj_grad, states, old_action_dists)
Hd = self.hessian_matrix_vector_product(states, old_action_dists, descent_direction)
# 这里max_coef是线性搜索更新公式中的sqrt(2*delta/(d^T H d))
max_coef = torch.sqrt(2 * self.kl_constraint / (torch.dot(descent_direction, Hd) + 1e-8))
new_para = self.line_search(states, actions, advantage, old_log_probs, old_action_dists, max_coef * descent_direction)
torch.nn.utils.vector_to_parameters(new_para, self.actor.parameters())
def update(self, transition_dict):
states = torch.tensor(transition_dict['states'], dtype=torch.float).to(self.device)
actions = torch.tensor(transition_dict['actions']).view(-1, 1).to(self.device)
rewards = torch.tensor(transition_dict['rewards'], dtype=torch.float).view(-1, 1).to(self.device)
next_states = torch.tensor(transition_dict['next_states'], dtype=torch.float).to(self.device)
dones = torch.tensor(transition_dict['dones'], dtype=torch.float).view(-1, 1).to(self.device)
td_target = rewards + self.gamma * self.critic(next_states) * (1 - dones)
# td误差是一个序列，每个元素对应一个时间步的TD误差
td_delta = td_target - self.critic(states)
# 计算GAE优势函数
advantage = compute_advantage(self.gamma, self.lmbda, td_delta.cpu()).to(self.device)
# 关键：旧策略分布/旧 log_probs 只作为“常量”参与 TRPO 的 KL 和重要性采样比值
# 必须与计算图彻底断开，否则在共轭梯度里多次 autograd.grad 会触发重复反向的报错。
with torch.no_grad():
old_action_probs = self.actor(states)
old_log_probs = torch.log(old_action_probs.gather(1, actions))
old_log_probs = old_log_probs.detach()
old_action_dists = torch.distributions.Categorical(old_action_probs.detach())
# critic网络的损失函数——参见时序差分的公式，TRPO主要是更新价值网络
critic_loss = torch.mean(F.mse_loss(self.critic(states), td_target.detach()))
self.critic_optimizer.zero_grad()
critic_loss.backward()
self.critic_optimizer.step()
self.policy_learn(states, actions, old_action_dists, old_log_probs, advantage)
if __name__ == "__main__":
num_episodes = 500
hidden_dim = 128
gamma = 0.98
lmbda = 0.95
critic_lr = 1e-2
kl_constraint = 0.0005
alpha = 0.5
device = torch.device("cuda") if torch.cuda.is_available() else torch.device(
"cpu")
print("Using device:", device)
env_name = 'CartPole-v1'
env = gym.make(env_name)
torch.manual_seed(0)
agent = TRPO(hidden_dim, env.observation_space, env.action_space, lmbda,
kl_constraint, alpha, critic_lr, gamma, device)
# 在线策略训练，包括数据采集和策略更新
return_list = rl_utils.train_on_policy_agent(env, agent, num_episodes)
episodes_list = list(range(len(return_list)))
plt.plot(episodes_list, return_list)
plt.xlabel('Episodes')
plt.ylabel('Returns')
plt.title('TRPO on {}'.format(env_name))
plt.show()
mv_return = rl_utils.moving_average(return_list, 9)
plt.plot(episodes_list, mv_return)
plt.xlabel('Episodes')
plt.ylabel('Returns')
plt.title('TRPO on {}'.format(env_name))
plt.show()PPO算法
TRPO计算过程非常复杂，每一步更新的运算量非常大。于是，TRPO 算法的改进版——PPO 算法在 2017 年被提出
如果我们想要尝试在一个新的环境中使用强化学习算法，那么 PPO 就属于可以首先尝试的算法。
TRPO的优化目标：
核心思想是通过最大化一个由旧策略数据加权（重要性采样）的优势函数期望来改进策略，同时施加KL散度约束以确保新策略的更新幅度不会太大。为了计算这个优化目标，TRPO 使用泰勒展开近似、共轭梯度、线性搜索等方法直接求解。PPO 的优化目标与 TRPO 相同，但 PPO 用了一些相对简单的方法来求解。具体来说，PPO 有两种形式，一是 PPO-惩罚，二是 PPO-截断
PPO-惩罚：用拉格朗日乘数法直接将 KL 散度的限制放进了目标函数中，这就变成了一个无约束的优化问题，在迭代的过程中不断更新 KL 散度前的系数。
β就像方向盘的“阻尼系数”，β越大，方向盘越“重”，你越难做出剧烈的转向（策略大幅改变）。算法并不手动设置一个固定的 β，而是设定一个期望的KL散度目标值 δ（一个超参数），然后让 β自动调整，以使实际的平均KL散度 d_k围绕 δ波动。 至于1.5和2，应该是作者经过深思熟虑选的叭
PPO-截断：在大多数时候都比惩罚表现的更好，也更主流
记住clip的作用是把x限制在[l,r]内
同样阅读代码，策略网络和价值网络和之前的TRPO是一样的（PPO就是在TRPO基础上改的）
第一点是不再像TRPO那样绕着算策略网络（TRPO没有这个的优化器，而是用一个policy learn函数专门更新参数，因为参数是通过公式算出来的，不是梯度下降出来的），PPO的actor也有优化器
take action和之前一样，也是按照动作概率序列选下一个动作
核心：
ratio = torch.exp(log_probs - old_log_probs)
surr1 = ratio * advantage
surr2 = torch.clamp(ratio, 1 - self.eps,1 + self.eps) * advantage # 截断
actor_loss = torch.mean(-torch.min(surr1, surr2)) # PPO损失函数
critic_loss = torch.mean(F.mse_loss(self.critic(states), td_target.detach()))
class PPO:
''' PPO算法,采用截断方式 '''
def __init__(self, state_dim, hidden_dim, action_dim, actor_lr, critic_lr, lmbda, epochs, eps, gamma, device):
self.actor = PolicyNet(state_dim, hidden_dim, action_dim).to(device)
self.critic = ValueNet(state_dim, hidden_dim).to(device)
self.actor_optimizer = torch.optim.Adam(self.actor.parameters(), lr=actor_lr)
self.critic_optimizer = torch.optim.Adam(self.critic.parameters(), lr=critic_lr)
self.gamma = gamma
self.lmbda = lmbda
self.epochs = epochs # 一条序列的数据用来训练轮数
self.eps = eps # PPO中截断范围的参数
self.device = device
def take_action(self, state):
state = torch.tensor([state], dtype=torch.float).to(self.device)
probs = self.actor(state)
action_dist = torch.distributions.Categorical(probs)
action = action_dist.sample()
return action.item()
def update(self, transition_dict):
states = torch.tensor(transition_dict['states'], dtype=torch.float).to(self.device)
actions = torch.tensor(transition_dict['actions']).view(-1, 1).to(self.device)
rewards = torch.tensor(transition_dict['rewards'], dtype=torch.float).view(-1, 1).to(self.device)
next_states = torch.tensor(transition_dict['next_states'], dtype=torch.float).to(self.device)
dones = torch.tensor(transition_dict['dones'], dtype=torch.float).view(-1, 1).to(self.device)
td_target = rewards + self.gamma * self.critic(next_states) * (1 - dones)
td_delta = td_target - self.critic(states)
advantage = rl_utils.compute_advantage(self.gamma, self.lmbda, td_delta.cpu()).to(self.device)
old_log_probs = torch.log(self.actor(states).gather(1, actions)).detach()
for _ in range(self.epochs):
log_probs = torch.log(self.actor(states).gather(1, actions))
"""重要性采样，A/B= exp(logA-logB)"""
ratio = torch.exp(log_probs - old_log_probs)
surr1 = ratio * advantage
surr2 = torch.clamp(ratio, 1 - self.eps,1 + self.eps) * advantage # 截断
actor_loss = torch.mean(-torch.min(surr1, surr2)) # PPO损失函数
critic_loss = torch.mean(F.mse_loss(self.critic(states), td_target.detach()))
self.actor_optimizer.zero_grad()
self.critic_optimizer.zero_grad()
actor_loss.backward()
critic_loss.backward()
self.actor_optimizer.step()
self.critic_optimizer.step()对于与连续动作交互的环境，小改一下，让策略网络输出连续动作高斯分布（Gaussian distribution）的均值和标准差。后续的连续动作则在该高斯分布中采样得到
DDPG算法
前面是在线策略算法，意味着他们样本效率比较低。用类似的思想来处理动作空间无限的环境并且使用的是离线策略算法——深度确定性策略梯度（deep deterministic policy gradient，DDPG），构造一个确定性策略，用梯度上升的方法来最大化Q值。DDPG 也属于一种 Actor-Critic 算法。我们之前学习的 REINFORCE、TRPO 和 PPO 学习随机性策略，而DDPG 则学习一个确定性策略。
在actor网络中，不再输出动作概率分布，而是某个具体的动作（确定性策略）Critic需要同时看到状态和Actor输出的动作，才能评估这个“状态-动作对”的价值。
注意下他有四个网络，其中 Actor 和 Critic 各用一个网络，此外它们都各自有一个目标网络。理由和DQN一样
在 DQN 中，每隔一段时间将Q网络直接复制给目标Q网络；而在 DDPG 中，目标网络的更新采取的是一种软更新的方式，即让目标Q网络缓慢更新，逐渐接近Q网络: τ是一个比较小的数，目标网络也用这种软更新
由于函数Q存在值过高估计Q的问题，DDPG 采用了 Double DQN 中的技术来更新Q网络。但是，由于 DDPG 采用的是确定性策略，它本身的探索仍然十分有限。所以其在行为策略上引入一个随机噪声来进行探索（和之前DQN的贪婪策略出发点一样）
补充：对于随机策略（之前的），他们的策略梯度：
对于确定策略，它的策略梯度：
它的算法流程：
代码：
class DDPG:
''' DDPG算法 '''
def __init__(self, state_dim, hidden_dim, action_dim, action_bound, sigma, actor_lr, critic_lr, tau, gamma, device):
self.actor = PolicyNet(state_dim, hidden_dim, action_dim, action_bound).to(device)
self.critic = QValueNet(state_dim, hidden_dim, action_dim).to(device)
self.target_actor = PolicyNet(state_dim, hidden_dim, action_dim, action_bound).to(device)
self.target_critic = QValueNet(state_dim, hidden_dim, action_dim).to(device)
# 初始化目标价值网络并设置和价值网络相同的参数
self.target_critic.load_state_dict(self.critic.state_dict())
# 初始化目标策略网络并设置和策略相同的参数
self.target_actor.load_state_dict(self.actor.state_dict())
self.actor_optimizer = torch.optim.Adam(self.actor.parameters(), lr=actor_lr)
self.critic_optimizer = torch.optim.Adam(self.critic.parameters(), lr=critic_lr)
self.gamma = gamma
self.sigma = sigma # 高斯噪声的标准差,均值直接设为0
self.tau = tau # 目标网络软更新参数
self.action_dim = action_dim
self.device = device
def take_action(self, state):
state = torch.tensor([state], dtype=torch.float).to(self.device)
action = self.actor(state).item()
# 给动作添加噪声，增加探索
action = action + self.sigma * np.random.randn(self.action_dim)
return action
def soft_update(self, net, target_net):
for param_target, param in zip(target_net.parameters(), net.parameters()):
param_target.data.copy_(param_target.data * (1.0 - self.tau) + param.data * self.tau)
def update(self, transition_dict):
states = torch.tensor(transition_dict['states'], dtype=torch.float).to(self.device)
actions = torch.tensor(transition_dict['actions'], dtype=torch.float).view(-1, 1).to(self.device)
rewards = torch.tensor(transition_dict['rewards'], dtype=torch.float).view(-1, 1).to(self.device)
next_states = torch.tensor(transition_dict['next_states'], dtype=torch.float).to(self.device)
dones = torch.tensor(transition_dict['dones'], dtype=torch.float).view(-1, 1).to(self.device)
"""这部分是经典，critic算下一状态的q值，然后loss是这一状态和下一状态，最小化旨在让他收敛"""
next_q_values = self.target_critic(next_states, self.target_actor(next_states))
q_targets = rewards + self.gamma * next_q_values * (1 - dones)
critic_loss = torch.mean(F.mse_loss(self.critic(states, actions), q_targets))
self.critic_optimizer.zero_grad()
critic_loss.backward()
self.critic_optimizer.step()
"""让actor在这一状态选的动作价值最大，即最小化这个负的"""
actor_loss = -torch.mean(self.critic(states, self.actor(states)))
self.actor_optimizer.zero_grad()
actor_loss.backward()
self.actor_optimizer.step()
self.soft_update(self.actor, self.target_actor) # 软更新策略网络
self.soft_update(self.critic, self.target_critic) # 软更新价值网络SAC算法
虽然 DDPG 是离线策略算法，但是它的训练非常不稳定，收敛性较差，对超参数比较敏感，也难以适应不同的复杂环境。一个更加稳定的算法：SAC被提出，它学习一个随机策略
熵：即H(X)=E[-logp(x)]，由此有最大熵强化学习
熵的引入增加了强化学习的探索程度，α越大，探索性就越强，有助于加速后续的策略学习，它主要解决：传统RL容易陷入局部最优，过度利用已知的好动作
传统的策略改进是确定性的，即：πnew(s)=argamaxQπ(s,a)，而soft策略迭代不仅追求高回报，还希望策略的随机性更大
这个改进的公式推导有点复杂：
由于归一化常数难以计算，所以又引入KL散度作为代理损失，之后就有更新公式了。
核心：两个目标网络Q，解决环境Q值过高的问题 连续动作空间要重参数化
考虑到设计中熵的正则项参数很重要，SAC设计了自动化熵参数：α在训练过程中改进，公式如下
它的算法框架（也是非常经典）：
给出代码实现中的一些key：
self.critic_1 = QValueNetContinuous(state_dim, hidden_dim, action_dim).to(device) # 第一个Q网络
self.critic_2 = QValueNetContinuous(state_dim, hidden_dim, action_dim).to(device) # 第二个Q网络
self.target_critic_1 = QValueNetContinuous(state_dim, hidden_dim, action_dim).to(device) # 第一个目标Q网络
self.target_critic_2 = QValueNetContinuous(state_dim, hidden_dim, action_dim).to(device) # 第二个目标Q网络
"""SAC有四个网络，注意 目标网络的引入参见DQN的想法 两个Q网络是为了减少高估Q值"""
def calc_target(self, rewards, next_states, dones): # 计算目标Q值
next_actions, log_prob = self.actor(next_states)
entropy = -log_prob
q1_value = self.target_critic_1(next_states, next_actions)
q2_value = self.target_critic_2(next_states, next_actions)
"""这里next_value计算考虑了熵"""
next_value = torch.min(q1_value, q2_value) + self.log_alpha.exp() * entropy
td_target = rewards + self.gamma * next_value * (1 - dones)
return td_target
"""更新两个Q网络——即critic网络，他们的loss目标是使预测出的动作价值Q和实际接近"""
td_target = self.calc_target(rewards, next_states, dones)
critic_1_loss = torch.mean(F.mse_loss(self.critic_1(states, actions), td_target.detach()))
critic_2_loss = torch.mean(F.mse_loss(self.critic_2(states, actions), td_target.detach()))
"""策略网络的更新参见前面L_π(θ)公式"""
new_actions, log_prob = self.actor(states)
entropy = -log_prob
q1_value = self.critic_1(states, new_actions)
q2_value = self.critic_2(states, new_actions)
actor_loss = torch.mean(-self.log_alpha.exp() * entropy - torch.min(q1_value, q2_value))
"""熵正则化参数α的更新，也和公式是一一对应的"""
# 使用alpha的log值,可以使训练结果比较稳定
self.log_alpha = torch.tensor(np.log(0.01), dtype=torch.float)
self.log_alpha.requires_grad = True # 可以对alpha求梯度
self.log_alpha_optimizer = torch.optim.Adam([self.log_alpha], lr=alpha_lr)
# 更新alpha值
alpha_loss = torch.mean((entropy - self.target_entropy).detach() * self.log_alpha.exp())
self.log_alpha_optimizer.zero_grad()
alpha_loss.backward()
self.log_alpha_optimizer.step()SAC也可以处理与离散动作交互的环境，修改：
策略网络的输出修改为在离散动作空间上的 softmax 分布；
价值网络直接接收状态和离散动作空间的分布作为输入。
策略网络输出一个离散的动作分布，所以在价值网络的学习过程中，不需要再对下一个动作进行采样，而是直接通过概率计算来得到下一个状态的价值。同理，在α的损失函数计算中，也不需要再对动作进行采样。
# 计算目标Q值,直接用策略网络的输出概率进行期望计算
def calc_target(self, rewards, next_states, dones):
next_probs = self.actor(next_states)
next_log_probs = torch.log(next_probs + 1e-8)
"""主要是entropy的计算有点区别"""
entropy = -torch.sum(next_probs * next_log_probs, dim=1, keepdim=True)
q1_value = self.target_critic_1(next_states)
q2_value = self.target_critic_2(next_states)
min_qvalue = torch.sum(next_probs * torch.min(q1_value, q2_value),
dim=1,
keepdim=True)
next_value = min_qvalue + self.log_alpha.exp() * entropy
td_target = rewards + self.gamma * next_value * (1 - dones)
return td_target
# 更新策略网络
probs = self.actor(states)
log_probs = torch.log(probs + 1e-8)
# 直接根据概率计算熵
entropy = -torch.sum(probs * log_probs, dim=1, keepdim=True) #
q1_value = self.critic_1(states)
q2_value = self.critic_2(states)
min_qvalue = torch.sum(probs * torch.min(q1_value, q2_value), dim=1, keepdim=True) # 直接根据概率计算期望
actor_loss = torch.mean(-self.log_alpha.exp() * entropy - min_qvalue)GRPO算法
学习教程来自于rethinkfun
DPO算法
ARPO算法
其他补充：大模型强化学习算法完全指南：从PPO到ARPO的7大算法选型与实战应用！-CSDN博客
