【PyTorch深度强化学习】TD3算法（双延迟-确定策略梯度算法）的讲解及实战（超详细 附源码）

   从策略梯度方法已知，基于PG的强化学习存在过高估计问题，但由于DDPG评论家的目标值不是取最优动作值函数的，所以不存在最大化操作。此时，将Double DQN思想直接用于DDPG的评论家，构造如下目标函数：

   实际上，这样的处理效果并不好，这是因为在连续动作空间中，策略变化缓慢，行动者更新较为平缓，使得预测QQ值与目标QQ值相差不大，无法避免过高估计问题。

   考虑将Double Q-Learning思想应用于DDPG，采用两个独立的评论家Qw1Qw1、Qw2Qw2和两个独立的行动者μθ1μθ1、μθ2μθ2，以50%的概率利用Q1Q1产生动作，然后更新Q2Q2估计值，而另外50%的概率正好相反。构建更新所需的两个目标值分别为：    但由于样本均来自于同一经验池，不能保证样本数据完全独立，所以两个行动者的样本具有一定相关性，在一定的情况下，甚至会加剧高估问题。针对此种情形，秉持“宁可低估，也不要高估”的想法，对Double Q-Learning进行修改，构建基于Clipped Double Q-learning方法的目标值：    如式（bbb.13）所示，目标值只使用了一个行动者网络μθ1μθ1，取两个评论家网络Qw1Qw1和Qw2Qw2的最小值来作为值函数估计值。
   在更新评论家网络Qw1Qw1和Qw2Qw2时，均采用式（bbb.13）目标值y，共用如下损失函数：

   该算法相比于原算法的区别仅在于多了一个和原评论家Qw1Qw1同步更新的辅助评论家Qw2Qw2，在更新目标值y时取最小值。不过这一修改仍然会让人疑惑，Qw1Qw1和Qw2Qw2只有初始参数不同，后面的更新都一样，这样形成的两个类似的评论家能否有效消除TD误差带来的偏置估计。

 

 四、代码

部分源码如下

 
import numpy as np
import torch
import gym
import os
import copy
import numpy as np
import torch
import torch.nn as nn
import torch.nn.functional as F
 
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
 
class ReplayBuffer(object):
    def __init_
        self.ptr = 0
        self.size = 0
 
        self.state = np.zeros((max_size, state_dim))
        self.action = np.zeros((max_size, action_dim))
        self.next_state = np.zeros((max_size, state_dim))
        self.reward = np.zeros((max_size, 1))
        self.not_done = np.zeros((max_size, 1))
 
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
 
    def add(self, state, action, next_state, reward, done):
        self.state[self.ptr] = state
        self.action[self.ptr] = action
        self.next_state[self.ptr] = next_state
        self.reward[self.ptr] = reward
        self.not_done[self.ptr] = 1. - done
 
        self.ptr = (self.ptr + 1) % self.max_size
        self.size = min(self.size + 1, self.max_size)
 
    def sample(self, batch_size):
        ind = np.random.randint(0, self.size, size=batch_size)
 
        return (
            torch.FloatTensor(self.state[ind]).to(self.device),
            torch.FloatTensor(self.action[ind]).to(self.device),
            torch.FloatTensor(self.next_state[ind]).to(self.device),
            torch.FloatTensor(self.reward[ind]).to(self.device),
            torch.FloatTensor(self.not_done[ind]).to(self.device)
        )
 
class Actor(nn.Module):
        def __init__(self, state_dim, action_dim, max_action):
            super(Actor, self).__init__()
 
            self.l1 = nn.Linear(state_dim, 256)
            self.l2 = nn.Linear(256, 256)
            self.l3 = nn.Linear(256, action_dim)
 
            self.max_action = max_action
 
        def forward(self, state):
            a = F.relu(self.l1(state))
            a = F.relu(self.l2(a))
            return self.max_action * torch.tanh(self.l3(a))
 
class Critic(nn.Module):
        def __init__(self, state_dim, action_dim):
            super(Critic, self).__init__()
 
            # Q1 architecture
            self.l1 = nn.Linear(state_dim + action_dim, 256)
            self.l2 = nn.Linear(256, 256)
            self.l3 = nn.Linear(256, 1)
 
            # Q2 architecture
            self.l4 = nn.Linear(state_dim + action_dim, 256)
            self.l5 = nn.Linear(256, 256)
            self.l6 = nn.Linear(256, 1)
 
        def forward(self, state, action):
            sa = torch.cat([state, action], 1)
 
            q1 = F.relu(self.l1(sa))
            q1 = F.relu(self.l2(q1))
            q1 = self.l3(q1)
 
            q2 = F.relu(self.l4(sa))
            q2 = F.relu(self.l5(q2))
            q2 = self.l6(q2)
            return q1, q2
 
        def Q1(self, state, action):
            sa = torch.cat([state, action], 1)
 
            q1 = F.relu(self.l1(sa))
            q1 = F.relu(self.l2(q1))
            q1 = self.l3(q1)
            return q1
 
actor1=Actor(17,6,1.0)
for ch in actor1.children():
    print(ch)
print("*********************")
critic1=Critic(17,6)
for ch in critic1.children():
    print(ch)
 
class TD3(object):
    def __init__(
        self,
        state_dim,
        action_dim,
        max_action,
        discount=0.99,
        tau=0.005,
        policy_noise=0.2,
        noise_clip=0.5,
        policy_freq=2
    ):
 
        self.actor = Actor(state_dim, action_dim, max_action).to(device)
        self.actor_target = copy.deepcopy(self.actor)
        self.actor_optimizer = torch.optim.Adam(self.actor.parameters(), lr=3e-4)
 
        self.critic = Critic(state_dim, action_dim).to(device)
        self.critic_target = copy.deepcopy(self.critic)
        self.critic_optimizer = torch.optim.Adam(self.critic.parameters(), lr=3e-4)
 
        self.max_action = max_action
        self.discount = discount
        self.tau = tau
        self.policy_noise = policy_noise
        self.noise_clip = noise_clip
        self.policy_freq = policy_freq
 
        self.total_it = 0
 
 
    def select_action(self, state):
        state = torch.FloatTensor(state.reshape(1, -1)).to(device)
        return self.actor(state).cpu().data.numpy().flatten()
 
 
    def train(self, replay_buffer, batch_size=100):
        self.total_it += 1
 
        # Sample replay buffer
        state, action, next_state, reward, not_done = replay_buffer.sample(batch_size)
 
        with torch.no_grad():
            # Select action according to policy and add clipped noise
            noise = (
                torch.randn_like(action) * self.policy_noise
            ).clamp(-self.noise_clip, self.noise_clip)
 
            next_action = (
                self.actor_target(next_state) + noise
            ).clamp(-self.max_action, self.max_action)
 
            # Compute the target Q value
            target_Q1, target_Q2 = self.critic_target(next_state, next_action)
            target_Q = torch.min(target_Q1, target_Q2)
            target_Q = reward + not_done * self.discount * target_Q
 
        # Get current Q estimates
        current_Q1, current_Q2 = self.critic(state, action)
 
        # Compute critic loss
        critic_loss = F.mse_loss(current_Q1, target_Q) + F.mse_loss(current_Q2, target_Q)
 
        # Optimize the critic
        self.critic_optimizer.zero_grad()
        critic_loss.backward()
        self.critic_optimizer.step()
 
        # Delayed policy updates
        if self.total_it % self.policy_freq == 0:
 
            # Compute actor losse
            actor_loss = -self.critic.Q1(state, self.actor(state)).mean()
 
            # Optimize the actor
            self.actor_optimizer.zero_grad()
            actor_loss.backward()
            self.actor_optimizer.step()
 
            # Update the frozen target models
            for param, target_param in zip(self.critic.parameters(), self.critic_target.parameters()):
                target_param.data.copy_(self.tau * param.data + (1 - self.tau) * target_param.data)
 
            for param, target_param in zip(self.actor.parameters(), self.actor_target.parameters()):
                target_param.data.copy_(self.tau * param.data + (1 - self.tau) * target_param.data)
 
 
    def save(self, filename):
        torch.save(self.critic.state_dict(), filename + "_critic")
        torch.save(self.critic_optimizer.state_dict(), filename + "_critic_optimizer")
 
        torch.save(self.actor.state_dict(), filename + "_actor")
        torch.save(self.actor_optimizer.state_dict(), filename + "_actor_optimizer")
 
 
    def load(self, filename):
        self.critic.load_state_dict(torch.load(filename + "_critic"))
        self.critic_optimizer.load_state_dict(torch.load(filename + "_critic_optimizer"))
        self.critic_target = copy.deepcopy(self.critic)
 
        self.actor.load_state_dict(torch.load(filename + "_actor"))
        self.actor_optimizer.load_state_dict(torch.load(filename + "_actor_optimizer"))
        self.actor_target = copy.deepcopy(self.actor)
 
# Runs policy for X episodes and returns average reward
# A fixed seed is used for the eval environment
def eval_policy(policy, env_name, seed, eval_episodes=10):
    eval_env = gym.make(env_name)
    eval_env.seed(seed + 100)
 
    avg_reward = 0.
    for _ in range(eval_episodes):
        state, done = eval_env.reset(), False
        while not done:
            action = policy.select_action(np.array(state))
            state, reward, done, _ = eval_env.step(action)
            avg_reward += reward
 
    avg_reward /= eval_episodes
 
    print("---------------------------------------")
    print(f"Evaluation over {eval_episodes} episodes: {avg_reward:.3f}")
    print("---------------------------------------")
    return avg_reward
 
 
policy = "TD3"
env_name = "Walker2d-v4"  # OpenAI gym environment name
seed = 0  # Sets Gym, PyTorch and Numpy seeds
start_timesteps = 25e3  # Time steps initial random policy is used
eval_freq = 5e3  # How often (time steps) we evaluate
max_timesteps = 1e6  # Max time steps to run environment
expl_noise = 0.1  # Std of Gaussian exploration noise
batch_size = 256  # Batch size for both actor and critic
discount = 0.99  # Discount factor
tau = 0.005  # Target network update rate
policy_noise = 0.2  # Noise added to target policy during critic update
noise_clip = 0.5  # Range to clip target policy noise
policy_freq = 2  # Frequency of delayed policy updates
save_model = "store_true"  # Save model and optimizer parameters
load_model = ""  # Model load file name, "" doesn't load, "default" uses file_name
file_name = f"{policy}_{env_name}_{seed}"
print("---------------------------------------")
print(f"Policy: {policy}, Env: {env_name}, Seed: {seed}")
print("---------------------------------------")
if not os.path.exists("./results"):
    os.makedirs("./results")
if save_model and not os.path.exists("./models"):
    os.makedirs("./models")
env = gym.make(env_name)
# Set seeds
env.seed(seed)
torch.manual_seed(seed)
np.random.seed(seed)
state_dim = env.observation_space.shape[0]
action_dim = env.action_space.shape[0]
max_action = float(env.action_space.high[0])
kwargs = {
    "state_dim": state_dim,
    "action_dim": action_dim,
    "max_action": max_action,
    "discount": discount,
    "tau": tau,
    "policy_noise": policy_noise * max_action,
    "noise_clip": noise_clip * max_action,
    "policy_freq": policy_freq
}
policy = TD3(**kwargs)
if load_model != "":
    policy_file = file_name if load_model == "default" else load_model
    policy.load(f"./models/{policy_file}")
replay_buffer = ReplayBuffer(state_dim, action_dim)
# Evaluate untrained policy
evaluations = [eval_policy(policy, env_name, seed)]
state, done = env.reset(), False
episode_reward = 0
episode_timesteps = 0
episode_num = 0
for t in range(int(max_timesteps)):
    episode_timesteps += 1
    # Select action randomly or according to policy
    if t < start_timesteps:
        action = env.action_space.sample()
    else:
        action = (
                policy.select_action(np.array(state))
                + np.random.normal(0, max_action * expl_noise, size=action_dim)
        ).clip(-max_action, max_action)
l = float(done) if episode_timesteps < env._max_episode_steps else 0
    # Store data in replay buffer
    replay_buffer.add(state, action, next_state, reward, done_bool)
    state = next_state
    episode_reward += reward
    # Train agent after collecting sufficient data
    if t >= start_timesteps:
        policy.train(replay_buffer, batch_size)
    if done:
end(eval_policy(policy, env_name, seed))
        np.save(f"./results/{file_name}", evaluations)
    if save_model:
        policy.save(f"./models/{file_name}")
state_dim

创作不易 觉得有帮助请点赞关注收藏~~~