【强化学习】DDPG : 强化学习之倒立摆

import gymimport paddlefrom itertools import countimport numpy as npfrom collections import dequeimport randomimport timefrom visualdl import LogWriter

class Critic(paddle.nn.Layer):
    def __init__(self):
        super(Critic,self).__init__()       
        # 状态空间为3
        self.fc1=paddle.nn.Linear(3,256) 
      
        # 连接了动作空间，所以＋1
        self.fc2=paddle.nn.Linear(256+1,128)        # 
        self.fc3=paddle.nn.Linear(128,1)

        self.relu=paddle.nn.ReLU()    
    def forward(self,x,a):

        x=self.relu(self.fc1(x))

        x=paddle.concat((x,a),axis=1)

        x=self.relu(self.fc2(x))

        x=self.fc3(x)        return x

class Actor(paddle.nn.Layer):
    def __init__(self,is_tranin=True):
        super(Actor,self).__init__()
        
        self.fc1=paddle.nn.Linear(3,256)
        self.fc2=paddle.nn.Linear(256,128)
        self.fc3=paddle.nn.Linear(128,1)

        self.relu=paddle.nn.ReLU() # max(0,x)
        self.tanh=paddle.nn.Tanh() # 双切正切曲线 (e^(x)-e^(-x))/(e^(x)+e^(-x))
        self.noisy=paddle.distribution.Normal(0,0.2)

        self.is_tranin=is_tranin    
    def forward(self,x):
        x=self.relu(self.fc1(x))
        x=self.relu(self.fc2(x))
        x=self.tanh(self.fc3(x))        return x    
    def select_action(self,epsilon,state):

        state=paddle.to_tensor(state,dtype='float32').unsqueeze(0)        # 创建一个上下文来禁用动态图梯度计算。在此模式下，每次计算的结果都将具有stop_gradient=True。
        with paddle.no_grad():            # squeeze():会删除输入Tensor的Shape中尺寸为1的维度
            # noisy抽样结果是1维的tensor
            action=self.forward(state).squeeze()+self.is_tranin*epsilon*self.noisy.sample([1]).squeeze(0)        
        # paddle.clip：将输入的所有元素进行剪裁，使得输出元素限制在[min, max]内
        # 动作空间是[-2,2]
        return 2*paddle.clip(action,-1,1).numpy()

class Memory(object):
   
    def __init__(self,memory_size):
        self.memory_size=memory_size
        self.buffer=deque(maxlen=self.memory_size)    

    # 增加经验，因为经验数组是存放在deque中的，deque是双端队列，
    # 我们的deque指定了大小，当deque满了之后再add元素，则会自动把队首的元素出队
    def add(self,experience):
        self.buffer.append(experience)    

    def size(self):
        return len(self.buffer)    # continuous=True则表示连续取batch_size个经验
    def sample(self , batch_szie , continuous = True):

        # 选取的经验数目是否超过缓冲区内经验的数目
        if batch_szie>len(self.buffer):
            batch_szie=len(self.buffer)        
        # 是否连续取经验
        if continuous:            # random.randint(a，b) 返回[a，b]之间的任意整数
            rand=random.randint(0,len(self.buffer)-batch_szie)            return [self.buffer[i] for i in range(rand,rand+batch_szie)]        else:            # numpy.random.choice(a, size=None, replace=True, p=None)
            # a 如果是数组则在数组中采样；a如果是整数，则从[0,a-1]这个序列中随机采样
            # size 如果是整数则表示采样的数量
            # replace为True可以重复采样；为false不会重复
            # p 是一个数组，表示a中每个元素采样的概率；为None则表示等概率采样
            indexes=np.random.choice(np.arange(len(self.buffer)),size=batch_szie,replace=False)            return [self.buffer[i] for i in indexes]    

    def clear(self):
        self.buffer.clear()

env=gym.make('Pendulum-v0')

actor=Actor()
critic=Critic()

critic_optim=paddle.optimizer.Adam(parameters=critic.parameters(),learning_rate=3e-5)
actor_optim=paddle.optimizer.Adam(parameters=actor.parameters(),learning_rate=1e-5)

explore=50000epsilon=1 # 控制动作的抽样gamma=0.99tau=0.001memory_replay=Memory(50000)
begin_train=Falsebatch_szie=32learn_steps=0epochs=100writer=LogWriter('./logs')

s_time=time.time()for epoch in range(0,epochs):

    start_time=time.time()

    state=env.reset()

    episode_reward=0 # 记录一个epoch的奖励综合

    # 一个epoch进行2000的时间步
    for time_step in range(2000):        # 步骤1，策略网络根据初始状态做预测
        action=actor.select_action(epsilon,state)

        next_state,reward,done,_=env.step([action])

        episode_reward+=reward        # 这一步在做什么？ reward [-1,1]
        reward=(reward+8.2)/8.2

        memory_replay.add((state,next_state,action,reward))        # 当经验回放数组中的存放一定量的经验才开始利用
        if memory_replay.size()>1280:
            learn_steps+=1

            if not begin_train:                print('train begin!')
                begin_train=True
            
            # false : 不连续抽样
            experiences=memory_replay.sample(batch_szie,False)

            batch_state,batch_next_state,batch_action,batch_reward=zip(*experiences)

            batch_state=paddle.to_tensor(batch_state,dtype='float32')
            batch_next_state=paddle.to_tensor(batch_next_state,dtype='float32')
            batch_action=paddle.to_tensor(batch_action,dtype='float32').unsqueeze(1)
            batch_reward=paddle.to_tensor(batch_reward,dtype='float32').unsqueeze(1)            
            with paddle.no_grad():                
                # 策略网络做预测
                at,at1=actor(batch_state),actor(batch_next_state)                # 价值网络做预测
                qt,qt1=critic(batch_state,batch_action),critic(batch_next_state,at1)                # 计算TD目标
                Q_traget=batch_reward+gamma*qt1            
            # TD误差函数
            # 该OP用于计算预测值和目标值的均方差误差。(预测值，目标值)
            # reduction='mean'，默认值，计算均值
            critic_loss=paddle.nn.functional.mse_loss(qt,Q_traget)            # 更新价值网络
            critic_optim.clear_grad()
            critic_loss.backward()
            critic_optim.step()

            writer.add_scalar('critic loss',critic_loss.numpy(),learn_steps)            # 更新策略网络
            critic.eval()
            actor_loss=-critic(batch_state,at)
            actor_loss=actor_loss.mean()

            actor_optim.clear_grad()
            actor_loss.backward()
            actor_optim.step()

            critic.train()

            writer.add_scalar('actor loss',actor_loss.numpy(),learn_steps)        # epsilon 逐渐减小
        if epsilon>0:
            epsilon-=1/explore
        state=next_state
     
    writer.add_scalar('episode reward',episode_reward/2000,epoch)    
    if epoch%10==0:        print("Epoch:{}, episode reward is {:.3f}, use time is {:.3f}s ".format(epoch,episode_reward/2000,time.time()-start_time))    

print("************ all time is {:.3f} h:".format((time.time()-s_time)/3600))

actor=Actor()
critic=Critic()
critic_optim=paddle.optimizer.Adam(parameters=critic.parameters(),learning_rate=3e-5)
actor_optim=paddle.optimizer.Adam(parameters=actor.parameters(),learning_rate=1e-5)# 目标网络，缓解自举造成的高估问题actor_target=Actor()
critic_target=Critic()

explore=50000epsilon=1 # 控制动作的抽样gamma=0.99tau=0.001memory_replay=Memory(50000)
begin_train=Falsebatch_szie=32learn_steps=0epochs=100twriter=LogWriter('./newlogs')

def soft_update(target,source,tau):
    # zip() 函数用于将可迭代的对象作为参数，将对象中对应的元素打包成一个个元组，然后返回由这些元组组成的列表。
    for target_param,param in zip(target.parameters(),source.parameters()):
        target_param.set_value(target_param*(1.0-tau)+param*tau)

s_time=time.time()
k=0for epoch in range(0,epochs):

    start_time=time.time()

    state=env.reset()

    episode_reward=0 # 记录一个epoch的奖励综合

    # 一个epoch进行2000的时间步
    for time_step in range(2000):        # 步骤1，策略网络根据初始状态做预测
        action=actor.select_action(epsilon,state)

        next_state,reward,done,_=env.step([action])

        episode_reward+=reward        # reward缩放到 [-1,1]
        reward=(reward+8.2)/8.2

        memory_replay.add((state,next_state,action,reward))        # 当经验回放数组中的存放一定量的经验才开始利用
        if memory_replay.size()>1280:
            learn_steps+=1

            if not begin_train:                print('train begin!')
                begin_train=True
            
            # false : 不连续抽样
            # 从经验数组中随机抽取四元组（s_t,a_t,r_t,s_t+1）
            experiences=memory_replay.sample(batch_szie,False)

            batch_state,batch_next_state,batch_action,batch_reward=zip(*experiences)

            batch_state=paddle.to_tensor(batch_state,dtype='float32')
            batch_next_state=paddle.to_tensor(batch_next_state,dtype='float32')
            batch_action=paddle.to_tensor(batch_action,dtype='float32').unsqueeze(1)
            batch_reward=paddle.to_tensor(batch_reward,dtype='float32').unsqueeze(1)            # 目标策略网络做预测，输入是 s_t+1 预测出下一个动作a_t+1
            # 目标价值网络做预测，输入是s_t+1和a_t+1，输出是q_tt
            # 计算TD目标，y= r_t+gamma*q_tt
            with paddle.no_grad():                # Q_next traget是TD目标
                Q_next=critic_target(batch_next_state,actor_target(batch_next_state))
                Q_traget=batch_reward+gamma*Q_next            
            # 价值网络做预测，输入是s_t,a_t,输出是q_t
            # 计算TD误差，error=q_t-y
            # 误差函数
            # 该OP用于计算预测值和目标值的均方差误差。(预测值，目标值)
            # reduction='mean'，默认值，计算均值
            critic_loss=paddle.nn.functional.mse_loss(critic(batch_state,batch_action),Q_traget)            
            # 更新价值网络
            critic_optim.clear_grad()
            critic_loss.backward()
            critic_optim.step()

            twriter.add_scalar('critic loss',critic_loss.numpy(),learn_steps)            if k==5:                #更新策略网络
                # 使用Critic网络给定值的平均值来评价Actor网络采取的行动。最大化该值。
                # 更新Actor()网络，对于一个给定状态，它产生的动作尽量让Critic网络给出高的评分
                critic.eval()
                actor_loss=-critic(batch_state,actor(batch_state))
                actor_loss=actor_loss.mean()

                actor_optim.clear_grad()
                actor_loss.backward()
                actor_optim.step()

                critic.train()

                twriter.add_scalar('actor loss',actor_loss.numpy(),learn_steps)                # 更新目标网络
                soft_update(actor_target,actor,tau)
                soft_update(critic_target,critic,tau)

                k=0
            else:
                k+=1
        
        # epsilon 逐渐减小
        if epsilon>0:
            epsilon-=1/explore
        state=next_state
     
    twriter.add_scalar('episode reward',episode_reward/2000,epoch)    
    if epoch%10==0:        print("Epoch:{}, episode reward is {:.3f}, use time is {:.3f}s ".format(epoch,episode_reward/2000,time.time()-start_time))    

print("************ all time is {:.3f} h:".format((time.time()-s_time)/3600))