前言
在 REINFORCE 算法中,每次需要根据一个策略采集一条完整的轨迹,并计算这条轨迹上的回报。这种采样方式的方差比较大,学习效率也比较低。我们可以借鉴时序差分学习的思想,使用动态规划方法来提高采样效率,即从状态开始的总回报可以通过当前动作的即时奖励和下一个状态的值函数来近似估计。
为了解决 High Variance 和 High bias 之间的矛盾,可以把它们结合在一起,利用value based 和 policy based 两类方法各自的优势,还顺带把它们的短板都补上了,于是就有了集大成的 Actor-Critic 类方法。 Actor-Critic类算法是一种结合策略梯度和时序差分学习的强化学习方法,其中,Actor是指策略函数 ,即学习一个策略以得到尽可能高的回报。Critic是指价值函数,对当前策略的值函数进行估计,即评估Actor的好坏。借助于价值函数,Actor-Critic算法可以进行单步参数更新,不需要等到回合结束才进行更新。所以我们需要两个网络,一个负责生成策略的 Actor 和一个负责评价策略的 Critic。这就有点类似一个演员在表演,而同时一个评论家在随时纠正他的表现,而且两者都还在不断更新,这种互补式的训练方式会比单独的策略网络或者值函数网络更有效。
Actor-Critic是集合和ploicy gradient和Q-learning的方法。
回顾policy gradient和Q-learning方法
policy gradient:
可以看出policy gradient是基于回合更新的,因此我们需要需要很大的耐心和环境产生交互样本,进行回合训练,并且由于每个回合是不稳定的,因此我们需要的大量的样本。
Q-learning
Q-learning学习的是每个动作的价值,要求动作必须是离散的。
AC(Actor-critic)算法原理
Actor-Critic 算法是一个既基于策略也基于价值的方法。在上一节我们提到,在Reinfoce算法中可以用状态价值函数作为基准函数来降低梯度估计的方差。Actor-Critic 算法也沿用了相同的想法,同时学习行动者(Actor)函数(也就是智能体的策略函数)和批判者(Critic)函数(也就是状态价值函数)。此外,Actor-Critic算法还沿用了自举法(Bootstrapping)的思想来估计Q值函数。Reinfoce中的误差项被时间差分误差取代了,即 。
下面我们具体看一下AC算法的数学推导:
- 我们这里采用 L 步的时间差分误差,并通过最小化该误差的平方来学习批判者函数,即
- 式子中表示学习批判者函数的参数,是学习步长,并且
- 是智能体在 下 L 步之后到达的状态,所以
- 类似地,行动者函数决定每个状态 s 上所采取的动作或者动作空间上的一个概率分布。我们采用和初版策略梯度相似的方法来学习这个策略函数。
- 这里表示行动者函数的参数, 是学习步长,并且
注意到,我们这里分别用了和来表示策略函数和价值函数的参数。在实际应用中,当我们选择用神经网络来表示这两个函数的时候,经常会让两个网络共享一些底层的网络层作为共同的状态表征(State Representation)。此外,AC 算法中的 L 值经常设为 1, 也就是 TD(0) 误差。
- 值得注意的是,AC 算法也可以使用 Q 值函数作为其批判者。在这种情况下,优势函数可以用以下式子估计。
- 用来学习 Q 值函数这个批判者的损失函数为
AC算法的代码详解
以AC: CartPole-V0为例
Actor-Critic 算法通过 TD 方法计算基准,能在每次和环境交互后立刻更新策略,和 MC 非常不同。
在 Actor-Critic 算法中,我们建立了 2 个类:Actor 和 Critic,其结构如下所示。
class Actor(object):
def __init__(self, state_dim, action_num, lr=0.001): # 类初始化。创建模型、优化器及其
...
def learn(self, state, action, td_error): # 更新模型
...
def get_action(self, state, greedy=False): # 通过概率分布或者贪心方法选择动作
...
def save(self): # 存储训练模型
...
def load(self): # 载入训练模型
...
class Critic(object):
def __init__(self, state_dim, lr=0.01): # 类初始化。创建模型、优化器及其所需变量
...
def learn(self, state, reward, state_): # 更新模型
...
def save(self): # 存储训练模型
...
def load(self): # 载入训练模型
...
Actor 类的部分和策略梯度算法很像。唯一的区别是 learn() 函数使用了 TD 误差作为优势估计值进行更新,而不是使用折扣化奖励。
def learn(self, state, action, td_error):
with tf.GradientTape() as tape:
_logits = self.model(np.array([state]))
_exp_v = tl.rein.cross_entropy_reward_loss(logits=_logits, actions=[action],
rewards=td_error[0])
grad = tape.gradient(_exp_v, self.model.trainable_weights)
self.optimizer.apply_gradients(zip(grad, self.model.trainable_weights))
return _exp_v
和 PG 算法不同,AC 算法有一个带有价值网络的批判者,它能估计每个状态的价值。所以它初始化函数十分清晰,只需要创建网络和优化器即可。
class Critic(object):
def __init__(self, state_dim, lr=0.01):
input_layer = tl.layers.Input([1, state_dim], name=’state’)
layer = tl.layers.Dense(
n_units=30, act=tf.nn.relu6, W_init=tf.random_uniform_initializer(0, 0.01),
name=’hidden’
)(input_layer)
layer = tl.layers.Dense(n_units=1, act=None, name=’value’)(layer)
self.model = tl.models.Model(inputs=input_layer, outputs=layer, name="Critic")
self.model.train()
self.optimizer = tf.optimizers.Adam(lr)
在初始化函数之后,我们有了一个价值网络。下一步就是建立 learn() 函数。learn() 函数任务非常简单,通过公式 计算 TD 误差,之后将 TD 误差作为优势估计来计算损失。
def learn(self, state, reward, state_, done):
d = 0 if done else 1
v_ = self.model(np.array([state_]))
with tf.GradientTape() as tape:
v = self.model(np.array([state]))
td_error = reward + d * LAM * v_ - v
loss = tf.square(td_error)
grad = tape.gradient(loss, self.model.trainable_weights)
self.optimizer.apply_gradients(zip(grad, self.model.trainable_weights))
return td_error
存储和载入函数与往常一样。我们也可以将网络参数存储为.npz 格式的文件。
def save(self):
if not os.path.exists(os.path.join(’model’, ’ac’)):
os.makedirs(os.path.join(’model’, ’ac’))
tl.files.save_npz(self.model.trainable_weights, name=os.path.join(’model’, ’ac’,’model_critic.npz’))
def load(self):
tl.files.load_and_assign_npz(name=os.path.join(’model’, ’ac’,
’model_critic.npz’), network=self.model)
训练循环的代码和之前的代码非常相似。唯一的不同是更新的时机不同。使用 TD 误差的情况下,我们可以在每步进行更新。
if args.train:
all_episode_reward = []
for episode in range(TRAIN_EPISODES):
state = env.reset().astype(np.float32)
step = 0
episode_reward = 0
while True:
if RENDER: env.render()
action = actor.get_action(state)
state_new, reward, done, info = env.step(action)
state_new = state_new.astype(np.float32)
if done: reward = -20
episode_reward += reward
td_error = critic.learn(state, reward, state_new, done)
actor.learn(state, action, td_error)
state = state_new
step += 1
if done or step >= MAX_STEPS:
break