Advertisement

深度强化学习 Deep Reinforcement Learning: An Introduction

阅读量:

作者:禅与计算机程序设计艺术

1.简介

深度强化学习(Deep reinforcement learning)属于机器学习领域中的新兴研究方向,在处理复杂环境下的控制与决策任务方面展现出显著潜力。其基本概念是利用多层次神经网络构建预测模型,并通过预测结果引导行为以实现长期记忆优化、策略改进以及避免陷入局部最优状态。作为一种新型强化学习算法——Actor-Critic(演员-评论家)方法——深度强化学习并非传统算法而是融合了经典强化学习理论与前沿研究成果并结合深度学习最新技术发展而成。本文旨在系统阐述深度强化学习的核心概念、典型算法及其实际应用案例

2.基本概念术语说明

(1)马尔可夫决策过程(Markov Decision Process, MDP)

在深度强化学习框架中, 智能体与环境之间的互动过程中会生成一系列的状态和奖励. 其中, 状态代表智能体所处的环境信息集合, 而奖励则是指智能体在特定状态下采取某一动作后获得的反馈. 整个交互过程可通过马尔可夫决策过程(MDP)进行建模. MDP由五个核心要素构成:

  • S: 环境的状态空间
    • A: 行动空间
    • T: 转移矩阵 P(s'|s,a),它表示智能体从状态 s 执行动作 a 后转移到状态 s' 的概率
    • R(s,a,s'): 奖励函数,在状态 s 执行动作 a 时获得的奖励
    • γ: 折现因子 γ 描述折现效应的作用范围及其优先级(取值范围为 [0,1])。当折现因子 γ 取值为 0 时,则不考虑长期影响;而当其取值为 1 时,则主要关注于最大化即时奖励而非未来收益。

MDP 具体阐述了一个智能体如何基于环境信息和执行动作获取奖励,并据此选择后续行动的方法。另一种表述则明确指出其定义了智能体与环境之间的互动过程的具体细节。

(2)状态值函数、状态价值函数、贝尔曼方程

在实际应用中,我们通常不会单纯依靠MDP模型进行建模,因为这种模型过于抽象,在精确描述智能体在不同状态下可能采取的行为模式和奖励机制方面存在一定局限性。因此,在深度强化学习领域中,我们需要通过构建状态值函数或状态价值函数来量化智能体在各个状态下所能获得的价值或奖励。这些函数的计算过程均依赖于贝尔曼方程的求解过程,并且其求解则主要基于动态规划等优化算法的支持。本节将系统阐述了状态值函数和状态价值函数的概念以及相关的贝尔曼方程,并在此基础上逐步展开讨论相关的求解方法和技术细节。

(2.1)状态值函数和状态价值函数

状态值函数(state value function)和状态价值函数(state-action value function)分别用于评估智能体在给定状态下可能获得的奖励情况,并指导智能体采取何种行动。

(2.1.1)状态值函数

价值函数(value function)V(s) 是用来表示处于特定状态下可获取到的期望奖励的一种度量工具。具体而言,则是:V^\pi(s)=\mathbb{E}_{\tau \sim p_\pi(\cdot|\tau_o)}[\sum_{t=0}^H R(s_{t}, a_{t})]其中 p_{\pi}(\cdot|\tau_o) 表示从初始状态 \tau_o 出发的状态序列 \{\dots, s_0, s_1, \dots, s_H\} 的发生概率。

(2.1.2)状态价值函数

状态-动作价值函数(state-action value function)Q(s, a) 定义为:在给定状态s并执行动作a后所能获得的预期累积奖励总量。具体来说:
Q^\pi(s,a)=\mathbb{E}_{\tau \sim p_\pi(\tau|\tau_o)}[\sum_{t=0}^H R(s_{t}, a_{t})]
其中\pi表示智能体所采用的策略(policy),通常也被称为A值函数(A value function)。

(2.1.3)Bellman方程

状态值函数与状态价值函数的计算均基于Bellman方程进行。该方程由两个公式构成,各自对应着状态值函数与状态价值函数。

(2.1.3.1)状态值函数的 Bellman方程

对于状态值函数而言,Bellman方程表示为:

V^\pi(s)=\underset{a}{\max} Q^\pi(s,a)+\gamma\sum_{s'}P(s'|s,a)[R(s',a)-V^\pi(s')]

其最佳期望奖励等于动作价值函数与折扣因子之积与转移概率、折扣因子及新旧奖励差值之积之和。在给定状态下,智能体所能达到的最大期望奖励即为上述方程所描述的关系。

(2.1.3.2)状态价值函数的 Bellman方程

该 Bellman 方程表明:在特定状态下 s 下,在执行动作 a 后所能积累的价值等于即时回报 R(s, a) 加上折扣因子 γ 乘以从当前状态 s 转移到新状态 s' 的概率 P(s'|s, a) 以及相应新状态下策略 π 下的价值 Vπ(s')。

(2.2)策略梯度

在强化学习的实际应用中,“依赖”于人类指导的智能体无法自主制定具体行动方案。“提出”的是一种优化方法,在此背景下,“为了使智能体根据其自身制定的‘优化’方案实现性能提升,“提出”的方法是基于策略梯度模型。“在执行‘优化’过程时”,每一次迭代都必须利用历史轨迹数据,“计算”每个状态下的状态价值函数。“随后依据这些计算出的价值函数数值进行相应的‘参数调整’操作”,直至达到最优状态。“而与之相对应的‘策略梯度算法’则无需依赖于历史轨迹数据”,而是直接通过评估当前状态下政策参数的变化情况来实现更快捷地收敛到最佳解决方案的目的

(3)Actor-Critic 方法

该方法属于一种将actor与critic相结合的策略梯度技术。它通过同步更新actor行为策略与critic价值函数来优化智能体的行动决策。在actor-critic框架下,actor依据当前策略分布产生动作样本;随后critic评估这些动作表现,并基于状态价值或状态-动作价值模型计算相应的价值评估。此方法通过actor-critic机制协调两者的更新过程,在一定程度上缓解了政策更新中的偏差积累问题。

(3.1)Actor

Actor 可以被视作一种决策函数,在特定策略下它能够输出动作信息。在 A2C 和 PPO 策略中,Actor 采用生成式模型(Generative model)来生成动作序列的具体内容。具体而言,在基于变分自动编码器(Variational Autoencoder, VAE)的方法中,它可以输出高质量的图像数据;而在基于多层感知机(MLP)的框架下,则能够生成高维的动作向量序列。值得注意的是,在 Actor-Critic 框架下,默认采用确定性策略网络(Deterministic Policy Network)进行操作决策,在这种情况下直接输出离散的动作值而不依赖于生成模型的支持。例如,在CNN架构中通过最大池化层提取特征信息后利用全连接层预测动作概率分布情况

(3.2)Critic

从本质上讲,Critic 赋予了评估智能体表现的能力,它是基于奖励和惩罚信号来进行判断的机制。在 A2C 和 PPO 方法中,Critic 部分依赖于 Value Function Approximation (VFA) 技术,即利用神经网络模型拟合出各状态下最优的状态值函数,并在此基础上最小化 TD 误差以更新神经网络参数。尽管Actor-Critic方法也采用了状态值函数进行评估,但其算法特性使得 Critic 部分能够比传统的 VFA 方法更快地收敛至最优解。具体而言,我们在 Critic 模块中引入 LSTM 或 GRU 这样的时序处理单元,能够更好地捕捉动态变化的状态信息,从而提升了模型的整体性能

(3.3)优势

相较于 PPO、A2C 等传统方法,Actor-Critic 方法有以下优势:

  • 可微性:Actor 与 Critic 采用任意深度学习架构,在无需预先设定状态与动作之间的关联关系的前提下实现灵活适应。
  • 稳定性:Actor 与 Critic 的参数能够训练出更为稳定的有效策略,并有效避免因旧策略影响而导致的性能退化。
  • 扩展性:该系统在复杂任务场景中展现出良好的收敛性和泛化能力。
  • 分布式计算:该架构支持利用分布式计算框架实现对更大规模状态空间和多智能体系统的有效处理。

3.核心算法原理和具体操作步骤以及数学公式讲解

(1)深度强化学习的特点

通过深度神经网络对环境及决策机制进行建模,在学习过程中得以掌握具有非线性的决策逻辑,并以更高效率和准确性执行相应的策略方案。
该系统具备分布式计算能力,在多智能体协同工作时展现出良好的性能特征。
该算法无需依赖以往的状态信息,在部分观测的状态空间下有效提升了基于策略梯度的学习效率。

(2)Actor-Critic 算法流程图

(3)Advantage Estimator

Actor-Critic 算法采用了优势估计量来更新 Actor 参数的过程如下:
即采用以下方法进行计算:
其核心公式为:
∇_{θ^μ}J(θ^μ,θ^q)=α \frac{1}{m}\left[\sum_{i=1}^{m}\left(\sum_{t'=t}^{T_i}\gamma^{t'-t}(r+\gamma V(s_{t'},θ^q))-\bar{A}(s_t,a_t)\right)∇_{θ^μ}\log\pi(a_t|s_t,θ^μ)\right]
其中 J(θ^μ,θ^q) 代表策略损失函数;θ^μ 是 Actor 网络的参数;θ^q 是 Critic 网络的参数;α 为学习率;m 是批量大小;T_i 是第 i 个轨迹的时间步数;r 是即时奖励;
V(s_{t'},θ^q) 则是 Critic 网络在状态 s_{t'} 处的输出值;\bar{A}(s_t,a_t) 则是基于价值函数估算的状态-动作 (s_t,a_t) 的优势值估计量。这种优势值估计量有助于帮助 Actor 更精确地评估其策略的有效性。

(4)注意力机制

attention mechanism 是一种用于在不同状态间分配注意力的方式;它通过 attention vector 来决定注意力分布;其主要目标是帮助智能体聚焦于比其他状态更为重要的信息;在 actor-critic 算法中应用多头注意力概念可以显著提高该算法的能力

4.具体代码实例和解释说明

下面展示一下 Actor-Critic 算法的一个具体实现。

复制代码
    import tensorflow as tf
    from tensorflow import keras
    from collections import deque
    
    
    class ActorCritic:
    def __init__(self, state_dim, action_dim):
        self.state_dim = state_dim
        self.action_dim = action_dim
    
        # Actor 模型
        inputs = layers.Input((state_dim,))
        hidden = layers.Dense(32, activation='relu')(inputs)
        mu = layers.Dense(action_dim)(hidden)
        stddev = layers.Dense(action_dim)(hidden)
        actor = Model(inputs, [mu, stddev])
        
        # Critic 模型
        critic = Sequential([
            InputLayer(input_shape=(None, state_dim)),
            Dense(32, activation='relu'),
            Dense(1),
            Lambda(lambda x: K.expand_dims(x))
        ])
    
        optimizer = Adam(lr=0.001)
        self.actor = actor
        self.critic = critic
    
    @tf.function
    def call(self, inputs):
        """
        Predict actions for given states and sample from the predicted distribution.
    
        Args:
          inputs (np.ndarray or Tensor): States to predict actions for.
    
        Returns:
          actions (Tensor): Predicted actions for given states.
        """
        means, stddevs = self.actor(inputs)
        distributions = tfp.distributions.Normal(means, stddevs)
        samples = distributions.sample()
        actions = tf.clip_by_value(samples, -1, 1)
        return actions
    
    @tf.function
    def train_step(self, data):
        """
        Perform one training step on a minibatch of transitions.
    
        Args:
          data (dict): Dictionary containing'states', 'actions','rewards',
                        'next_states', 'dones'.
    
        Returns:
          dict: Losses dictionary.
        """
        states = tf.convert_to_tensor(data['states'])
        actions = tf.convert_to_tensor(data['actions'], dtype=tf.float32)
        rewards = tf.convert_to_tensor(data['rewards'], dtype=tf.float32)
        next_states = tf.convert_to_tensor(data['next_states'])
        dones = tf.convert_to_tensor(data['dones'], dtype=tf.float32)
    
        with tf.GradientTape(persistent=True) as tape:
            # Compute advantages using target network
            v_values = self.critic(next_states)
            td_target = rewards + (1 - dones) * gamma * v_values
    
            # Compute advantage estimates using current network
            values = self.critic(states)
            td_errors = tf.math.subtract(td_target, values[:, 0])
            advantage_estimates = discounted_cumsum(td_errors, gamma*lam)
            
            # Compute policy loss
            pi_distribution = tfp.distributions.Normal(*self.actor(states))
            log_probs = pi_distribution.log_prob(actions)
            entropy = pi_distribution.entropy()
            pg_loss = tf.reduce_mean(-advantage_estimates * log_probs[:, :, None], axis=[0, 1])
            
            # Compute critic loss
            q_values = self.critic(states)
            v_values = q_values[:, 0][:, None]
            qf_loss = tf.reduce_mean(tf.square(td_target[:, :, None] - v_values))
            
        grads = tape.gradient(pg_loss, self.actor.trainable_variables)
        self.actor_optimizer.apply_gradients(zip(grads, self.actor.trainable_variables))
        
        grads = tape.gradient(qf_loss, self.critic.trainable_variables)
        self.critic_optimizer.apply_gradients(zip(grads, self.critic.trainable_variables))
        
        del tape
        
        return {'policy_loss': pg_loss, 'value_loss': qf_loss, 'entropy': entropy}
    
    def save_weights(self, filepath):
        self.actor.save_weights(filepath+'_actor')
        self.critic.save_weights(filepath+'_critic')
        
    def load_weights(self, filepath):
        self.actor.load_weights(filepath+'_actor')
        self.critic.load_weights(filepath+'_critic')
    
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读

全部评论 (0)

还没有任何评论哟~