刚刚从大学毕业，近来闲来无事，开始了机器学习的旅程，深度学习是机器学习的重要一环，其可以使得机器自我尝试，并通过结果进行学习。
在机器学习的过程中，我自网上了解到大神morvanzhou，一个从土木工程转向了计算机的“聪明绝顶”的、英语特好的男人。
morvanzhou的python个人主页，请有兴趣的同学关注大神morvanzhou的python教程。
Policy Gradients的是一种基于策略的强化学习方法，基于策略的意思就是直接根据状态输出动作或者动作的概率。
我们使用神经网络输入当前的状态，网络就可以输出我们在这个状态下采取每个动作的概率。
与之前的DQN和Q-Learmnig这类基于价值的学习方法相比，二者的主要不同为：
1、基于价值的强化学习方法对应的最优策略通常是确定性策略，一般是从众多行为价值中选择一个最大价值的行为，仅有一定的概率会随机选择，而有些问题的最优策略却是随机策略，无法通过基于价值的学习方法求解。此时Policy Gradients可以起到很好的作用。
2、DQN之类的方法一般都是只处理离散动作，无法处理连续动作。
本文使用了OpenAI Gym中的CartPole-v0游戏来验证算法的实用性。
该环境只有两个离散动作，板子可以向左滑动或者向右滑动，保持杆子不倒即可。
state状态就是这个板子的位置和速度， pole的角度和角速度，4维的特征。
在Policy Gradients中，通过输入状态来得到每个动作的概率。
其建立的神经网络结构如下：
神经网络的输入量是self.tf_obs，指的是环境的状态。
神经网络的输出量是self.all_act_prob，指的是每个动作的概率。
神经网络共两层。
def _build_net(self):
# 输入量
# self.tf_obs指的是环境的观测量
# self.tf_acts指的是可执行的动作
# self.tf_vt指的是每个动作的奖惩情况
with tf.name_scope('inputs'):
self.tf_obs = tf.placeholder(tf.float32, [None, self.n_features], name="observations")
self.tf_acts = tf.placeholder(tf.int32, [None, ], name="actions_num")
self.tf_vt = tf.placeholder(tf.float32, [None, ], name="actions_value")
# 利用tf.layers.dense函数生成全连接层fc1
layer = tf.layers.dense(
inputs=self.tf_obs,
units=10,
activation=tf.nn.tanh, # tanh activation
kernel_initializer=tf.random_normal_initializer(mean=0, stddev=0.3),
bias_initializer=tf.constant_initializer(0.1),
name='fc1'
)
# 利用tf.layers.dense函数生成全连接层fc2
all_act = tf.layers.dense(
inputs=layer,
units=self.n_actions,
activation=None,
kernel_initializer=tf.random_normal_initializer(mean=0, stddev=0.3),
bias_initializer=tf.constant_initializer(0.1),
name='fc2'
)
self.all_act_prob = tf.nn.softmax(all_act, name='act_prob') # use softmax to convert to probability
with tf.name_scope('loss'):
# 可以通过最小化-(log_p * R)的方式实现最大化(log_p * R)
neg_log_prob = tf.nn.sparse_softmax_cross_entropy_with_logits(logits=all_act, labels=self.tf_acts)
loss = tf.reduce_mean(neg_log_prob * self.tf_vt) # reward guided loss
with tf.name_scope('train'):
self.train_op = tf.train.AdamOptimizer(self.lr).minimize(loss)
其根据神经网络的输出按照概率选择动作。
def choose_action(self, observation):
prob_weights = self.sess.run(self.all_act_prob, feed_dict={self.tf_obs: observation[np.newaxis, :]})
action = np.random.choice(range(prob_weights.shape[1]), p=prob_weights.ravel())
return action
神经网络的学习过程是这样的，通过记录每一轮开始直到done结束的所有的状态、动作及其得分。
通过所有的状态得到所有状态下每个动作的概率，利用：
neg_log_prob = tf.nn.sparse_softmax_cross_entropy_with_logits(logits=all_act, labels=self.tf_acts)
loss = tf.reduce_mean(neg_log_prob * self.tf_vt) # reward guided loss
计算每个动作的概率和实际动作的交叉熵，呈上实际得分，得到损失函数值。
在实际得分用于计算LOSS前，需要进行标准化。
def learn(self):
# 计算并标准化reward
discounted_ep_rs_norm = self._discount_and_norm_rewards()
# 训练
self.sess.run(self.train_op, feed_dict={
self.tf_obs: np.vstack(self.ep_obs), # shape=[None, n_obs]
self.tf_acts: np.array(self.ep_as), # shape=[None, ]
self.tf_vt: discounted_ep_rs_norm, # shape=[None, ]
})
# 初始化历史数据
self.ep_obs, self.ep_as, self.ep_rs = [], [], []
return discounted_ep_rs_norm
def _discount_and_norm_rewards(self):
# 标准化得分
# 生成与self.ep_rs数组shape相同的全0数组
discounted_ep_rs = np.zeros_like(self.ep_rs)
running_add = 0
for t in reversed(range(0, len(self.ep_rs))):
running_add = running_add * self.gamma + self.ep_rs[t]
discounted_ep_rs[t] = running_add
# 标准化
discounted_ep_rs -= np.mean(discounted_ep_rs)
# 计算标准差
discounted_ep_rs /= np.std(discounted_ep_rs)
return discounted_ep_rs
在每次学习完成后，需要对历史数据进行清零。
具体的实现代码分为两个部分，这是第一部分，主函数：
import gym
from RL_brain import PolicyGradient
import matplotlib.pyplot as plt
import math
RENDER = True # 是否显示图像
env = gym.make('CartPole-v0')
env.seed(1) # reproducible, general Policy gradient has high variance
env = env.unwrapped
print(env.action_space)
print(env.observation_space)
print(env.observation_space.high)
print(env.observation_space.low)
RL = PolicyGradient(
n_actions=env.action_space.n,
n_features=env.observation_space.shape[0],
learning_rate=0.02,
reward_decay=0.99,
# output_graph=True,
)
for i_episode in range(3000):
observation = env.reset()
while True:
if RENDER: env.render()
action = RL.choose_action(observation)
observation_, reward, done, info = env.step(action)
x, x_dot, theta, theta_dot = observation_
# r1代表车的 x水平位移 与 x最大边距 的距离差的得分
r1 = math.exp((env.x_threshold - abs(x))/env.x_threshold) - math.exp(1)/2
# r1代表棒子的 theta离垂直的角度 与 theta最大角度 的差的得分
r2 = math.exp((env.theta_threshold_radians - abs(theta))/env.theta_threshold_radians) - math.exp(1)/2
# 总 reward 是 r1 和 r2 的结合, 既考虑位置, 也考虑角度。
reward = r1 + r2
#在未done前进行数据存储
RL.store_transition(observation, action, reward)
if done:
ep_rs_sum = sum(RL.ep_rs)
if 'running_reward' not in globals():
running_reward = ep_rs_sum
else:
running_reward = running_reward * 0.99 + ep_rs_sum * 0.01
print("episode:", i_episode, " reward:", int(running_reward))
vt = RL.learn() # 学习, 输出 vt, 我们下节课讲这个 vt 的作用
break
observation = observation_
这是第二部分，RL_brain.py：
import numpy as np
import tensorflow as tf
np.random.seed(1)
tf.set_random_seed(1)
class PolicyGradient:
def __init__(
self,
n_actions,
n_features,
learning_rate=0.001,
reward_decay=0.95,
output_graph=False,
): # 初始化函数，需要输入action的数量，每个环境的features数量，学习率等
self.n_actions = n_actions
self.n_features = n_features
self.lr = learning_rate
# 衰减率
self.gamma = reward_decay
# ep_obs用于存储历史观测结果，ep_as用于存储历史动作，ep_rs用于存储历史奖惩
self.ep_obs, self.ep_as, self.ep_rs = [], [], []
self._build_net()
# 建立会话
self.sess = tf.Session()
if output_graph:
# tensorboard --logdir=logs
tf.summary.FileWriter("logs/", self.sess.graph)
self.sess.run(tf.global_variables_initializer())
def _build_net(self):
# 输入量
# self.tf_obs指的是环境的观测量
# self.tf_acts指的是可执行的动作
# self.tf_vt指的是每个动作的奖惩情况
with tf.name_scope('inputs'):
self.tf_obs = tf.placeholder(tf.float32, [None, self.n_features], name="observations")
self.tf_acts = tf.placeholder(tf.int32, [None, ], name="actions_num")
self.tf_vt = tf.placeholder(tf.float32, [None, ], name="actions_value")
# 利用tf.layers.dense函数生成全连接层fc1
layer = tf.layers.dense(
inputs=self.tf_obs,
units=10,
activation=tf.nn.tanh, # tanh activation
kernel_initializer=tf.random_normal_initializer(mean=0, stddev=0.3),
bias_initializer=tf.constant_initializer(0.1),
name='fc1'
)
# 利用tf.layers.dense函数生成全连接层fc2
all_act = tf.layers.dense(
inputs=layer,
units=self.n_actions,
activation=None,
kernel_initializer=tf.random_normal_initializer(mean=0, stddev=0.3),
bias_initializer=tf.constant_initializer(0.1),
name='fc2'
)
self.all_act_prob = tf.nn.softmax(all_act, name='act_prob') # use softmax to convert to probability
with tf.name_scope('loss'):
# 可以通过最小化-(log_p * R)的方式实现最大化(log_p * R)
neg_log_prob = tf.nn.sparse_softmax_cross_entropy_with_logits(logits=all_act, labels=self.tf_acts)
loss = tf.reduce_mean(neg_log_prob * self.tf_vt)
with tf.name_scope('train'):
self.train_op = tf.train.AdamOptimizer(self.lr).minimize(loss)
def choose_action(self, observation):
prob_weights = self.sess.run(self.all_act_prob, feed_dict={self.tf_obs: observation[np.newaxis, :]})
action = np.random.choice(range(prob_weights.shape[1]), p=prob_weights.ravel())
return action
def store_transition(self, s, a, r):
# 往历史记录中存储数据
# ep_obs用于存储历史观测结果，ep_as用于存储历史动作，ep_rs用于存储历史奖惩
self.ep_obs.append(s)
self.ep_as.append(a)
self.ep_rs.append(r)
def learn(self):
# 计算并标准化reward
discounted_ep_rs_norm = self._discount_and_norm_rewards()
# 训练
self.sess.run(self.train_op, feed_dict={
self.tf_obs: np.vstack(self.ep_obs), # shape=[None, n_obs]
self.tf_acts: np.array(self.ep_as), # shape=[None, ]
self.tf_vt: discounted_ep_rs_norm, # shape=[None, ]
})
# 初始化历史数据
self.ep_obs, self.ep_as, self.ep_rs = [], [], []
return discounted_ep_rs_norm
def _discount_and_norm_rewards(self):
# 标准化得分
# 生成与self.ep_rs数组shape相同的全0数组
discounted_ep_rs = np.zeros_like(self.ep_rs)
running_add = 0
for t in reversed(range(0, len(self.ep_rs))):
running_add = running_add * self.gamma + self.ep_rs[t]
discounted_ep_rs[t] = running_add
# 标准化
discounted_ep_rs -= np.mean(discounted_ep_rs)
# 计算标准差
discounted_ep_rs /= np.std(discounted_ep_rs)
return discounted_ep_rs
主函数和RL_brain部分我都进行了详细的备注。
由于代码并不是自己写的，所以就不上传github了，不过还是欢迎大家关注我和我的github。
https://github.com/bubbliiiing/
希望得到朋友们的喜欢。
有不懂的朋友可以评论询问噢。