强化学习算法

置顶深圳市进化图灵智能科技有限公司

已于 2025-03-12 23:25:13 修改

阅读量362

点赞数

分类专栏： AI 文章标签：人工智能

于 2023-08-10 16:02:19 首次发布

本文链接：https://blog.csdn.net/WMX843230304WMX/article/details/132210434

版权

AI 专栏收录该内容

60 篇文章

订阅专栏

本文围绕强化学习算法展开，介绍了游戏模型、策略轨迹概率。阐述了奖励机制，包括累积奖励、折扣奖励，目标是最大化期望奖励。还讲解了梯度更新的推导，可借助TensorFlow求解最大值并更新。最后介绍了REINFORCE算法及相关代码版本信息。

摘要生成于 C知道，由 DeepSeek-R1 满血版支持，前往体验 >

强化学习算法

加群（AI学习交流群）：
请添加图片描述
游戏模型如下：
在这里插入图片描述

策略网络输入状态s，输出动作a的概率分布如下： $\pi(a|s)$

多次训练轨迹如下

r表示回报
横轴为T, 1个回合的步骤数
纵轴为N, 回合数，1行代表1条轨迹，符合概率分布P

$\begin{bmatrix} s_{11} a_{11} r_{11} …… s_{1t} a_{1t} r_{1t} …… s_{1T} a_{1T} r_{1T}\\ …… …… ……\\ s_{n1} a_{n1} r_{n1} …… s_{nt} a_{nt} r_{nt}…… s_{nT} a_{nT} r_{nT} \\ …… …… ……\\ s_{N1} a_{N1} r_{N1} …… s_{Nt} a_{Nt} r_{Nt} …… s_{NT} a_{NT} r_{NT}\\ \end{bmatrix}$

策略轨迹 $\tau = s_{1} a_{1} , s_{2} a_{2}，……，s_{T} a_{T}$

发生的概率
$P(\tau) = P(s_{1} a_{1} , s_{2} a_{2}，……，s_{T} a_{T})$
$P(s_{1})\pi(a_{1}|s_{1})P(s_{2}|s_{1},a_{1})\pi(a_{2}|s_{2})P(s_{3}|s_{1},a_{1},s_{2},a_{2})……$
$P(s_{1})\prod_{t=1}^{T-1}\pi(a_{t}|s_{t})P(s_{t+1}|s_{1},a_{1},......,s_{t},a_{t})$

根据马尔科夫性(Markov Property)，简化为：
$P(\tau) = P(s_{1})\prod_{t=1}^{T-1}\pi(a_{t}|s_{t})P(s_{t+1}|s_{t},a_{t})$

奖励

在每次智能体与环境的交互过程中，均会得到一个滞后的奖励
$r_{t} = r(s_{t},a_{t})$

一次交互轨迹𝜏的累积奖励：
$R_{\tau } = \sum_{t=1}^{T-1} r_{t}$
其中𝑇为轨迹的步数。如果只考虑从轨迹的中间状态𝑠𝑡开始𝑠𝑡, 𝑠𝑡+1, . . . , 𝑠𝑇的累积奖励:
$R(s_{t}) = \sum_{k=1}^{T-t-1} r_{t+k}$

为了权衡近期奖励与长期奖励的重要性，更多地使用随着时间衰减的折扣奖励:

$R_{\tau } = \sum_{t=1}^{T-1} \gamma ^{t-1} r_{t}$
其中𝛾 ∈ [0,1]叫做折扣率.
我们希望找到一个策略𝜋(𝑎|𝑠)模型，使得在策略𝜋(𝑎|𝑠)控制下的智能体与环境交互产
生的轨迹𝜏的总奖励𝑅(𝜏)越高越好。由于环境状态转移和策略都具有随机性，同样的策略模
型作用于同初始状态的同一环境，也可能产生截然不同的轨迹序列𝜏。因此，强化学习的目
标是最大化期望奖励(Expected Return)：
$J(\pi (\theta )) = E_{\tau \sim P(\tau) } \left [ R(\tau) \right ] = E_{\tau \sim P(\tau) }\left [ \sum_{t=1}^{T-1} \gamma ^{t-1} r_{t} \right ]$
把期望写成求和平均形式：
$J(\pi (\theta )) =\frac{1}{N}\sum_{n=1}^{N} \left ( P(\tau ^{n}) R\left ( \tau ^{n} \right ) \right ) ，其中 \tau ^{n} 表示第n条轨迹$

训练的目标是寻找一组参数𝜃代表的策略网络𝜋𝜃，使得𝐽(𝜋𝜃)最大：
$\theta ^{*} = \underset{\theta }{argmax} E_{\tau \sim P(\tau) } \left [ R(\tau) \right ]$
其中𝑝(𝜏)代表了轨迹𝜏的分布，它由状态转移概率𝑝(𝑠′|𝑠, 𝑎)和策略𝜋(𝑎|𝑠)共同决定。策略𝜋
的好坏可以通过𝐽(𝜋𝜃)衡量，期望奖励越大，策略越优良；反之则策略越差。

梯度更新

$\frac{\partial J}{\partial \theta } = \frac{1}{N}\sum_{n=1}^{N} \frac{\partial}{\partial \theta } \left ( \left ( P(s_{1})\prod_{t=1}^{T-1}\pi(a_{t}|s_{t})P(s_{t+1}|s_{t},a_{t}) \right ) R(n) \ \right )$

考虑
$\frac{d\log_{}{f(x)} }{dx} = \frac{1}{f(x)} \frac{df(x)}{dx}$
将log ∏∙转换为∑ log(∙)形式

$\frac{\partial J}{\partial \theta } = \frac{1}{N}\sum_{n=1}^{N} R(n) \frac{\partial}{\partial\theta }\left ( \log_{}{p(s_{1}) } +\sum_{t=1}^{T-1} \log_{}{\pi _{\theta } (a_{t}| s_{t} )} + \log_{}{p(s_{t+1}|s_{t},a_{t})} \right )$
第n条轨迹分布𝑝(n)与𝜃 无关，奖励R(n)与𝜃 无关，看成常数：
$\frac{\partial J}{\partial \theta } = \frac{1}{N}\sum_{n=1}^{N}R(n) \left ( \sum_{t=1}^{T-1} \frac{\partial}{\partial\theta }\log_{}{\pi _{\theta } (a_{t}| s_{t} )} \right )$
写成期望形式：

$\frac{\partial J}{\partial \theta } = \ E_{\tau \sim P(\tau) }\left [ \left ( \sum_{t=1}^{T-1} \frac{\partial}{\partial\theta }\log_{}{\pi _{\theta } (a_{t}| s_{t} )} \right ) R(\tau)\right ]$

有了上述的表达式后，我们就可以通过 TensorFlow 的自动微分工具方便地求解出𝐽(𝜋𝜃)的最大值
最后利用梯度上升算法更新即可

强化学习算法(REINFORCE 算法)

start : 随机初始化参数𝜽

for ：

    根据策略𝝅𝜽(𝒂𝒕|𝒔𝒕)与环境交互， 生成多条轨迹{𝝉(𝒏)}
    计算𝑹(𝝉(𝒏))
    计算𝝏𝑱(𝜽)
    更新网络参数𝜽′ ← 𝜽 + 𝜼 ∙ 𝝏𝑱(𝜽)

end : 训练回合数达到要求

out : 策略网络𝝅𝜽(𝒂𝒕|𝒔𝒕)

REINFORCE.py

其中 gym version = 0.26.2 https://github.com/openai/gym/releases/tag/0.26.2

import 	gym,os
import  numpy as np
import  matplotlib
from 	matplotlib import pyplot as plt
# Default parameters for plots
matplotlib.rcParams['font.size'] = 18
matplotlib.rcParams['figure.titlesize'] = 18
matplotlib.rcParams['figure.figsize'] = [9, 7]
matplotlib.rcParams['font.family'] = ['KaiTi']
matplotlib.rcParams['axes.unicode_minus']=False 

import 	tensorflow as tf
from    tensorflow import keras
from    tensorflow.keras import layers,optimizers,losses
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'
assert tf.__version__.startswith('2.')

from    PIL import Image

SEED_NUM = 2222
env = gym.make('CartPole-v1')  # 创建游戏环境
# env.seed(SEED_NUM)
tf.random.set_seed(SEED_NUM)
np.random.seed(SEED_NUM)


learning_rate = 0.0002
gamma         = 0.98

class Policy(keras.Model):
    # 策略网络，生成动作的概率分布
    def __init__(self):
        super(Policy, self).__init__()
        self.data = [] # 存储轨迹
        # 输入为长度为4的向量，输出为左、右2个动作
        self.fc1 = layers.Dense(128, kernel_initializer='he_normal')
        self.fc2 = layers.Dense(2, kernel_initializer='he_normal')
        # 网络优化器
        self.optimizer = optimizers.Adam(lr=learning_rate)

    def call(self, inputs, training=None):
        # 状态输入s的shape为向量：[4] s,a,s_new,r
        x = tf.nn.relu(self.fc1(inputs))
        x = tf.nn.softmax(self.fc2(x), axis=1)
        return x

    def put_data(self, item):
        # 记录r,log_P(a|s)
        self.data.append(item)

    def train_net(self, tape):
        # 计算梯度并更新策略网络参数。tape为梯度记录器
        R = 0 # 终结状态的初始回报为0
        for r, log_prob in self.data[::-1]:#逆序取
            R = r + gamma * R # 计算每个时间戳上的回报
            # 每个时间戳都计算一次梯度
            # grad_R=-log_P*R*grad_theta
            loss = -log_prob * R 
            with tape.stop_recording():
                # 优化策略网络
                grads = tape.gradient(loss, self.trainable_variables)
                # print(grads)
                self.optimizer.apply_gradients(zip(grads, self.trainable_variables))
        self.data = [] # 清空轨迹

def main():
    pi = Policy() # 创建策略网络
    pi(tf.random.normal((4,4))) # 随机初始化 s,a,s_new,r
    pi.summary()
    score = 0.0 # 计分
    print_interval = 20 # 打印间隔
    returns = []

    for n_epi in range(400):
        s,info = env.reset(seed=SEED_NUM) # 回到游戏初始状态，返回s0
        with tf.GradientTape(persistent=True) as tape:
            for t in range(501): # CartPole-v1 forced to terminates at 500 step.
                # 送入状态向量，获取策略
                s = tf.constant(s,dtype=tf.float32)
                # s: [4] => [1,4]
                s = tf.expand_dims(s, axis=0)
                prob = pi(s) # 动作分布:[1,2]
                # 从动作类别分布中采样1个动作, shape: [1] , 返回prob中每行1个索引组成的张量，[0] 表示取值
                a = tf.random.categorical(tf.math.log(prob), 1)[0]
                a = int(a) # Tensor转数字
                s_prime, r, done, truncated, info = env.step(a)  
                # 记录动作a和动作产生的奖励r
                # prob shape:[1,2] 
                # prob[0][a] 动作a的概率, 其中a就是动作对应的索引
                pi.put_data((r, tf.math.log(prob[0][a])))
                s = s_prime # 刷新状态
                score += r # 累积奖励

                if n_epi >1000:
                    env.render()
                    # im = Image.fromarray(s)
                    # im.save("res/%d.jpg" % info['frames'][0])

                if done:  # 当前episode终止
                    break
            # episode终止后，训练一次网络
            pi.train_net(tape)
        del tape

        if n_epi%print_interval==0 and n_epi!=0:
            returns.append(score/print_interval)
            print(f"# of episode :{n_epi}, avg score : {score/print_interval}")
            score = 0.0
    env.close() # 关闭环境

    plt.plot(np.arange(len(returns))*print_interval, returns)
    plt.plot(np.arange(len(returns))*print_interval, returns, 's')
    plt.xlabel('回合数')
    plt.ylabel('总回报')
    plt.savefig('reinforce-tf-cartpole.svg')

if __name__ == '__main__':
    print(tf.config.list_physical_devices('GPU'))
    main()