Actor-Critic：强化学习之摆车
深度强化学习入门；经典案例详细分析。 这个我的强化学习入门之作，参考官方样例，添加详细注释，易于理解。欢迎fork；欢迎评论；欢迎共同探讨，共同进步。
·
1.配置环境，导入需要的包
import gym
import os
import sys
from itertools import count
import paddle
from paddle.distribution import Categorical
print('python version',sys.version)
print('paddle version',paddle.__version__)
python version 3.7.4 (default, Aug 13 2019, 20:35:49)
[GCC 7.3.0]
paddle version 2.2.0
2.搭建Actor-Critci 网络
2.1 获取当前运行设备（CPU or GPU）
device=paddle.get_device()
print(device)
cpu
2.2 加载摆车环境
env=gym.make('CartPole-v0')
print(env)
<TimeLimit<CartPoleEnv<CartPole-v0>>>
2.3 获取状态空间大小与动作空间大小
关于 observation_space 与 action_space 代表的到底是什么内容，可以参考CSDN博文：https://wuxian.blog.csdn.net/article/details/89576003
目前我的理解：
- observation_space 代表的是观测到的状态，也就是【摆车在轨道上的位置，杆子与竖直方向的夹角，小车速度，角度变化率】
- action_space 代表的是动作空间的大小，【左移，右移】
state_=env.observation_space
print(state_)
state_size=state_.shape[0]
print(state_size)
action_=env.action_space
print(action_)
action_size=action_.n
print(action_size)
Box(4,)
4
Discrete(2)
2
2.4 定义学习率
lr=0.001
2.5 定义Actor网络（实际定义的网络并没有严格按照图示，多了一个抽样）
-
截图来自 王树森 老师的书籍《深度强化学习》，Github直达链接：https://github.com/wangshusen/DRL
-
**如果s是张量，那么使用卷积网络捕获特征向量；如果s是向量，那么使用全连接网络。**这句话摘抄于上述书籍中，个人认为其所述张量应为：大于1维的张量；注意区分标量，向量，矩阵，张量，个人理解如下：
- 标量是仅包含一个数字，也是0维的张量
- 向量是数字组成的数组，也是1维的张量
- 矩阵是向量组成的数组，也是2维的张量
- 张量是深度学习的主要数据载体。标量，向量，矩阵是张量的一些特殊情况的常用名。
在（初中）高中数学中我们也学过向量，比如计算投影，这时向量是有方向的，但是在深度学习中，我理解的方向与此是有点区别的，即不用考虑方向。
class Actor(paddle.nn.Layer):
def __init__(self,state_size,action_size):
super(Actor,self).__init__()
self.state_size=state_size
self.action_size=action_size
self.l1=paddle.nn.Linear(self.state_size,128)
self.l2=paddle.nn.Linear(128,256)
self.l3=paddle.nn.Linear(256,self.action_size)
def forward(self,state):
out=paddle.nn.functional.relu(self.l1(state))
out=paddle.nn.functional.relu(self.l2(out))
out=self.l3(out)
# 根据动作的概率（加和为1），生成一个类别分布.paddle框架目前提供了三种分类函数【类别分类，正态分布，均匀分布】
distribution=Categorical(paddle.nn.functional.softmax(out,axis=-1))
return distribution
2.6 定义Critic网络（实际定义网络并没有返回每个动作的价值，而是仅返回一个动作的价值）
class Critic(paddle.nn.Layer):
def __init__(self,state_size,action_size):
super(Critic,self).__init__()
self.state_size=state_size
self.action_size=action_size
self.l1=paddle.nn.Linear(self.state_size,128)
self.l2=paddle.nn.Linear(128,256)
self.l3=paddle.nn.Linear(256,1)
def forward(self,state):
# paddle.nn.ReLU() 与paddle.nn.functional.relu()的区别是：前者是面向对象，是class,在类的fordward中调用了后者;后者是面向过程，是def。
out=paddle.nn.functional.relu(self.l1(state)) # relu(x)=max(0,x)
out=paddle.nn.functional.relu(self.l2(out))
value=self.l3(out)
return value
3. 训练模型
3.1 定义模型存储路径
actor_path='model/actor.pdparams'
critic_path='model/critic.pdparams'
3.2 TD目标的计算
看可以看王树森的书，但是感觉不太一样，理论与实践存在一些差别，但是总体思想是一致的。
def compute_returns(next_value,rewards,masks,gamma=0.99):
# 相当于n+1时刻的价值
R=next_value
returns=[]
# masks=[1,1,1,1,1,...,1,1,1,0]
# rewards=[r0,r1,r2,r3,r4,...,rn-3,rn-2,rn-1,rn]
# 倒序
for step in reversed(range(len(rewards))):
# TD目标的计算？？
R=rewards[step]+gamma*R*masks[step]
returns.insert(0,R)
# returns=[t0,t1,t2,t3,t4,...,tn-3,tn-2,tn-1,tn]
return returns
3.3 训练过程（一）
带有很多输出，方便了解每一行代码在做什么事情。 3.3与3.4内容是一样的，只不过3.3用来理解每一步的输出；3.4用来训练，保证输出的简洁。
def trainIters1(actor,critic,n_iters):
# 定义两个网络的优化器
optA=paddle.optimizer.Adam(lr,parameters=actor.parameters())
optC=paddle.optimizer.Adam(lr,parameters=critic.parameters())
for iter in range(n_iters):
state=env.reset() # 环境初始化 state形如[ 0.04700963 -0.0149178 0.01601383 0.03912796]
print(state)
log_probs=[] # 对数概率密度函数
rewards=[] # 奖励列表
values=[] # 价值，critic网络的输出
masks=[] # done or not done
entropy=[] # 交叉熵
env.reset() # 这里为什么又要重置环境？？
# python的迭代器 count(初值=0，步长=1)
for i in count():
# env.render()
print('*************************** /n/n/n')
print('i = ',i)
# state转为paddle.tensor,不指定place,根据环境自行判断
state=paddle.to_tensor(state,dtype='float32')
print(state)
dist,value=actor(state),critic(state)
print('actor distribution:',dist) # 这是一个分类分布（class）
print('critic value:',value)
# 从分布中进行采样（具体如何采样尚且不知，根据概率进行抽样？？）
action=dist.sample([1])
print('action is:',action,type(action))
# 环境执行一个时间步长，得到下一个状态，奖励，done，以及info(用不到,这里返回是空的字典)
# env.step()的输入格式是什么？： <class 'numpy.ndarray'>
# 将tensor拷贝到CPU上，不懂意图是什么,如果decive是在GPU上，则会有作用。
print('action.cpu()=',action.cpu())
# 删除axis=0上尺度为1的维度
print('action.cpu().squeeze(0)=',action.cpu().squeeze())
# 获取数值
print('action.cpu().squeeze(0).numpy()=',action.cpu().squeeze().numpy(),type(action.cpu().squeeze().numpy()))
next_state,reward,done,info=env.step(action.cpu().squeeze(0).numpy())
print('next_state=',next_state)
print('reward=',reward)
print('done=',done)
print('info=',info,type(info))
# 计算当前动作的对数概率密度函数
log_prob=dist.log_prob(action)
print('log_prob = ',log_prob)
# 加入到对数概率密度函数列表中
log_probs.append(log_prob)
# 将价值（critic）网络的输出加入到价值列表中
values.append(value)
# 将当前奖励加入到奖励列表中
rewards.append(paddle.to_tensor([reward],dtype='float32'))
# 将当前的done标志加入到masks列表中,False=0,
masks.append(paddle.to_tensor([1-done],dtype='float32'))
print('1-done = ',1-done)
state=next_state
if done:
if iter%10==0:
# score 就是小车坚持了多少个时间步
print("Iteration:{},score:{}".format(iter,i))
break
break #如果想要查看这个for循环每一步的输入，打开这个break
# end for count() 小车运行一次，也就是进行一句游戏
print()
next_state=paddle.to_tensor(next_state,dtype='float32')
# 让critic网络根据next_state预测next_value
next_value=critic(next_state)
print('next_value = ',next_value)
returns=compute_returns(next_value,rewards,masks)
print('returns == ',returns)
print('len returns = ',len(returns))
log_probs=paddle.concat(log_probs)
# detach() 返回一个新的Tensor，从当前计算图分离。
# 作用是什么???
returns=paddle.concat(returns).detach()
values=paddle.concat(values)
print('concat log_probs: ',log_probs)
print('concat returns: ',returns)
print('concat values: ',values)
# 负的 TD 误差 ，是让values接近returns，TD误差是 values-returns
advantage=returns-values
# actor的loss计算
# critic的loss计算
actor_loss=-(log_probs*advantage.detach()).mean()
critic_loss=advantage.pow(2).mean()
print('actor_loss: ',actor_loss)
print('critic_loss: ',critic_loss)
# 更新参数，完成一句游戏更新一次
# 梯度反向传播，注意actor网络是做梯度上升；critic网络是做梯度下降
optA.clear_grad()
optC.clear_grad()
actor_loss.backward()
critic_loss.backward()
optA.step()
optC.step()
break # 看一个iter的输出，注意打开这个break
paddle.save(actor.state_dict(),actor_path)
paddle.save(critic.state_dict(),critic_path)
env.close
print("111111111111111-------overover****************************************")
3.4 训练过程（二）
将很多输出注释掉，输出界面的简洁。3.3与3.4内容是一样的，只不过3.3用来理解每一步的输出；3.4用来训练，保证输出的简洁。
def trainIters2(actor,critic,n_iters):
# 定义两个网络的优化器
optA=paddle.optimizer.Adam(lr,parameters=actor.parameters())
optC=paddle.optimizer.Adam(lr,parameters=critic.parameters())
for iter in range(n_iters):
state=env.reset() # 环境初始化 state形如[ 0.04700963 -0.0149178 0.01601383 0.03912796]
#print(state)
log_probs=[] # 对数概率密度函数
rewards=[] # 奖励列表
values=[] # 价值，critic网络的输出
masks=[] # done or not done
entropy=[] # 交叉熵
env.reset() # 这里为什么又要重置环境？？
# python的迭代器 count(初值=0，步长=1)
for i in count():
# env.render() # 在线运行好像不支持
#print('*************************** /n/n/n')
#print('i = ',i)
# state转为paddle.tensor,不指定place,根据环境自行判断
state=paddle.to_tensor(state,dtype='float32')
#print(state)
dist,value=actor(state),critic(state)
#print('actor distribution:',dist) # 这是一个分类分布（class）
#print('critic value:',value)
# 从分布中进行采样（具体如何采样尚且不知，根据概率进行抽样？？）
action=dist.sample([1])
#print('action is:',action,type(action))
# 环境执行一个时间步长，得到下一个状态，奖励，done，以及info(用不到,这里返回是空的字典)
# env.step()的输入格式是什么？： <class 'numpy.ndarray'>
# 将tensor拷贝到CPU上，不懂意图是什么,如果decive是在GPU上，则会有作用。
#print('action.cpu()=',action.cpu())
# 删除axis=0上尺度为1的维度
#print('action.cpu().squeeze(0)=',action.cpu().squeeze())
# 获取数值
#print('action.cpu().squeeze(0).numpy()=',action.cpu().squeeze().numpy(),type(action.cpu().squeeze().numpy()))
next_state,reward,done,info=env.step(action.cpu().squeeze(0).numpy())
# print('next_state=',next_state)
# print('reward=',reward)
# print('done=',done)
# print('info=',info,type(info))
# 计算当前动作的对数概率密度函数
log_prob=dist.log_prob(action)
#print('log_prob = ',log_prob)
# 加入到对数概率密度函数列表中
log_probs.append(log_prob)
# 将价值（critic）网络的输出加入到价值列表中
values.append(value)
# 将当前奖励加入到奖励列表中
rewards.append(paddle.to_tensor([reward],dtype='float32'))
# 将当前的done标志加入到masks列表中,False=0,
masks.append(paddle.to_tensor([1-done],dtype='float32'))
#print('1-done = ',1-done)
state=next_state
if done:
if iter%10==0:
# score 就是小车坚持了多少个时间步
print("Iteration:{},score:{}".format(iter,i))
break
# break #如果想要查看这个for循环每一步的输入，打开这个注释
# end for count() 小车运行一次，也就是进行一句游戏
#print()
next_state=paddle.to_tensor(next_state,dtype='float32')
# 让critic网络根据next_state预测next_value
next_value=critic(next_state)
#print('next_value = ',next_value)
returns=compute_returns(next_value,rewards,masks)
#print('returns == ',returns)
#print('len returns = ',len(returns))
log_probs=paddle.concat(log_probs)
# detach() 返回一个新的Tensor，从当前计算图分离。
# 作用是什么???
returns=paddle.concat(returns).detach()
values=paddle.concat(values)
# print('concat log_probs: ',log_probs)
# print('concat returns: ',returns)
# print('concat values: ',values)
# 负的 TD 误差 ，是让values接近returns，TD误差是 values-returns
advantage=returns-values
# actor的loss计算
# critic的loss计算
actor_loss=-(log_probs*advantage.detach()).mean()
critic_loss=advantage.pow(2).mean()
# 更新参数，完成一句游戏更新一次
# 梯度反向传播，注意actor网络是做梯度上升；critic网络是做梯度下降
optA.clear_grad()
optC.clear_grad()
actor_loss.backward()
critic_loss.backward()
optA.step()
optC.step()
# break
paddle.save(actor.state_dict(),actor_path)
paddle.save(critic.state_dict(),critic_path)
env.close
print("222222222222222-------overover****************************************")
3.5 主函数，从文件中加载网络或是从零开始。可自行修改。
def main():
actor=Actor(state_size,action_size)
critic=Critic(state_size,action_size)
if os.path.exists(actor_path):
amodel_state_dict=paddle.load(actor_path)
actor.set_state_dict(amodel_state_dict)
print("load actor model from file")
if os.path.exists(critic_path):
cmodel_state_dict=paddle.load(critic_path)
critic.set_state_dict(cmodel_state_dict)
print("load critic model from file")
# 在这里决定是用3.3 还是 用3.4
#trainIters1(actor,critic,n_iters=200)
trainIters2(actor,critic,n_iters=200)
main()
load actor model from file
load critic model from file
Iteration:0,score:17
Iteration:10,score:11
Iteration:20,score:12
Iteration:30,score:12
Iteration:40,score:16
Iteration:50,score:17
Iteration:60,score:13
Iteration:70,score:26
Iteration:80,score:47
Iteration:90,score:139
Iteration:100,score:199
Iteration:110,score:155
Iteration:120,score:199
Iteration:130,score:199
Iteration:140,score:199
Iteration:150,score:199
Iteration:160,score:199
Iteration:170,score:199
Iteration:180,score:199
Iteration:190,score:199
222222222222222-------overover****************************************
4.总结
本项目使用强化学习经典算法：Actor-Critic在经典实验环境CartPole-v0进行实验，并对代码及一些知识点进行了详细分析。
个人认为我在这个项目中的贡献除了提供详细的代码之外，还分享了王树森老师的书籍（开源的），可以供入门的朋友参考学习。
本人也是刚刚入门强化学习，还有很多不足之处，请大家批评指正。
学大模型，用大模型上飞桨星河社区！每天8点V100G算力免费领！免费领取ERNIE 4.0 100w Token >>>
更多推荐
3
0- 0
已为社区贡献1438条内容
所有评论(0)