diff --git a/.gitignore b/.gitignore index c781834..ee4e2e8 100644 --- a/.gitignore +++ b/.gitignore @@ -51,3 +51,8 @@ Thumbs.db # *.jpg # *.jpeg # *.gif + +# Training results +training_results/ +*.png +*.mp4 diff --git a/RL_Algothrithms/agents/ddpg.py b/RL_Algothrithms/agents/ddpg.py new file mode 100644 index 0000000..a238c79 --- /dev/null +++ b/RL_Algothrithms/agents/ddpg.py @@ -0,0 +1,138 @@ +import torch +import torch.optim as optim +import torch.nn.functional as F +import numpy as np +import random +from networks_cont import ContActor, ContQCritic + +# --- 经验回放池 (Replay Buffer) 补丁 --- +class ReplayBuffer: + def __init__(self, capacity): + self.capacity = capacity + self.buffer = [] + self.position = 0 + + def add(self, state, action, reward, next_state, done): + if len(self.buffer) < self.capacity: + self.buffer.append(None) + self.buffer[self.position] = (state, action, reward, next_state, done) + self.position = (self.position + 1) % self.capacity + + def sample(self, batch_size): + batch = random.sample(self.buffer, batch_size) + # 解包数据,转换为 numpy 数组以提高效率 + state, action, reward, next_state, done = map(np.stack, zip(*batch)) + return state, action, reward, next_state, done + + def __len__(self): + return len(self.buffer) + +# --- DDPG 算法代理 --- +class DDPGAgent: + def __init__(self, state_dim, action_dim, max_action, device, + actor_lr=1e-4, critic_lr=1e-3, gamma=0.99, tau=0.005, buffer_size=100000): + self.device = device + self.gamma = gamma + self.tau = tau # 软更新系数 + self.max_action = max_action + + # --- 网络补丁:Online 与 Target 网络 --- + # 1. 创建 Online 网络 (负责被优化器更新) + self.actor = ContActor(state_dim, action_dim, max_action).to(self.device) + self.critic = ContQCritic(state_dim, action_dim).to(self.device) + + # 2. 创建影子 Target 网络 (不参与梯度下降) + self.actor_target = ContActor(state_dim, action_dim, max_action).to(self.device) + self.critic_target = ContQCritic(state_dim, action_dim).to(self.device) + + # 初始化影子网络的权重与主网络一致 + self.actor_target.load_state_dict(self.actor.state_dict()) + self.critic_target.load_state_dict(self.critic.state_dict()) + + # 优化器 + self.actor_optimizer = optim.Adam(self.actor.parameters(), lr=actor_lr) + self.critic_optimizer = optim.Adam(self.critic.parameters(), lr=critic_lr) + + # 经验回放池 + self.replay_buffer = ReplayBuffer(buffer_size) + + def select_action(self, state): + with torch.no_grad(): + state_tensor = torch.FloatTensor(state).unsqueeze(0).to(self.device) + # 动作由 Online Actor 输出 + action = self.actor(state_tensor).cpu().data.numpy().flatten() + + # 探索噪声:为了让行为策略具有探索性,加入高斯噪声 + noise = np.random.normal(0, 0.1 * self.max_action, size=action.shape) + action = np.clip(action + noise, -self.max_action, self.max_action) + return action + + def update(self, batch_size=64): + # 如果缓冲池数据不够,不进行更新 + if len(self.replay_buffer) < batch_size: + return + + # 1. 从回放池中随机抽取一批打乱的数据 (彻底打破时序相关性) + state, action, reward, next_state, done = self.replay_buffer.sample(batch_size) + + # 转换为 Tensor 维度 + state_t = torch.FloatTensor(state).to(self.device) + action_t = torch.FloatTensor(action).to(self.device) + reward_t = torch.FloatTensor(reward).unsqueeze(1).to(self.device) + next_state_t = torch.FloatTensor(next_state).to(self.device) + done_t = torch.FloatTensor(done).unsqueeze(1).to(self.device) + + # --- Critic 更新逻辑:计算稳定目标值 --- + # 1. 使用延迟反馈的影子网络 Target Actor 预测下一个状态的最理想动作 + next_mu_action = self.actor_target(next_state_t) + # 2. 使用影子网络 Target Critic 评估这个理想动作的 Q 值 (提供平滑参考) + next_q_value = self.critic_target(next_state_t, next_mu_action.detach()) + + # 计算稳定的 TD 目标 + td_target = reward_t + self.gamma * next_q_value * (1 - done_t) + + # 当前 Online Critic 的评估值 + current_q_value = self.critic(state_t, action_t) + + critic_loss = F.mse_loss(current_q_value, td_target.detach()) + + self.critic_optimizer.zero_grad() + critic_loss.backward() + self.critic_optimizer.step() + + # --- Actor 更新逻辑:只更新 Online Actor --- + mu_action = self.actor(state_t) + + # Actor 损失:让 Online Critic 给这个动作打分,越大越好 + actor_loss = -self.critic(state_t, mu_action).mean() + + self.actor_optimizer.zero_grad() + actor_loss.backward() + self.actor_optimizer.step() + + # --- 软更新 (Soft Update) 补丁 --- + # 影子网络缓慢向主网络靠近,解决目标乱动问题 + self._soft_update(self.actor_target, self.actor) + self._soft_update(self.critic_target, self.critic) + + def _soft_update(self, target_model, online_model): + """影子网络参数 = tau * 主网络参数 + (1 - tau) * 影子网络参数""" + for target_param, online_param in zip(target_model.parameters(), online_model.parameters()): + target_param.data.copy_( + target_param.data * (1.0 - self.tau) + online_param.data * self.tau + ) + + def save(self, path): + torch.save({ + 'actor': self.actor.state_dict(), + 'critic': self.critic.state_dict(), + 'actor_target': self.actor_target.state_dict(), + 'critic_target': self.critic_target.state_dict(), + }, path) + + def load(self, path): + checkpoint = torch.load(path, map_location=self.device) + self.actor.load_state_dict(checkpoint['actor']) + self.critic.load_state_dict(checkpoint['critic']) + self.actor_target.load_state_dict(checkpoint['actor_target']) + self.critic_target.load_state_dict(checkpoint['critic_target']) diff --git a/RL_Algothrithms/agents/dpac.py b/RL_Algothrithms/agents/dpac.py new file mode 100644 index 0000000..6d88b0c --- /dev/null +++ b/RL_Algothrithms/agents/dpac.py @@ -0,0 +1,75 @@ +import torch +import torch.optim as optim +import torch.nn.functional as F +import numpy as np +from networks_cont import ContActor, ContQCritic + +class DPACAgent: + def __init__(self, state_dim, action_dim, max_action, device, actor_lr=0.001, critic_lr=0.002, gamma=0.99): + self.device = device + self.gamma = gamma + self.max_action = max_action + + self.actor = ContActor(state_dim, action_dim, max_action).to(self.device) + self.critic = ContQCritic(state_dim, action_dim).to(self.device) + + self.actor_optimizer = optim.Adam(self.actor.parameters(), lr=actor_lr) + self.critic_optimizer = optim.Adam(self.critic.parameters(), lr=critic_lr) + + def select_action(self, state): + # 行为策略:在确定性目标策略的基础上加入高斯噪声,用于探索 + with torch.no_grad(): + state_tensor = torch.FloatTensor(state).unsqueeze(0).to(self.device) + action = self.actor(state_tensor).cpu().data.numpy().flatten() + + # 探索噪声 (行为策略 beta 与目标策略 mu 的区别就在这里) + noise = np.random.normal(0, 0.1 * self.max_action, size=action.shape) + action = np.clip(action + noise, -self.max_action, self.max_action) + return action + + def update(self, state, action, reward, next_state, next_action, done): + state_tensor = torch.FloatTensor(state).unsqueeze(0).to(self.device) + next_state_tensor = torch.FloatTensor(next_state).unsqueeze(0).to(self.device) + reward_tensor = torch.FloatTensor([reward]).unsqueeze(0).to(self.device) + # 注意:连续动作直接是浮点数,不需要转为索引 + action_tensor = torch.FloatTensor(action).unsqueeze(0).to(self.device) + + # --- Critic 更新 (Algorithm 10.4 核心逻辑) --- + # 1. 目标策略 mu 在下一个状态的理想输出 + next_mu_action = self.actor(next_state_tensor).detach() + # 2. 评估这个理想动作的 Q 值 + next_q_value = self.critic(next_state_tensor, next_mu_action).detach() + # 3. 计算 TD 目标 + td_target = reward_tensor + self.gamma * next_q_value * (1 - int(done)) + + # 4. 当前实际采取动作的 Q 值 + current_q_value = self.critic(state_tensor, action_tensor) + + critic_loss = F.mse_loss(current_q_value, td_target) + + self.critic_optimizer.zero_grad() + critic_loss.backward() + self.critic_optimizer.step() + + # --- Actor 更新 (链式法则) --- + # 1. 目标策略当前状态的输出 + mu_action = self.actor(state_tensor) + + # 2. 拿到这个动作去问 Critic:"给我打分"。 + # 为了让 Q 值最大化,我们加负号转化为梯度下降 + actor_loss = -self.critic(state_tensor, mu_action).mean() + + self.actor_optimizer.zero_grad() + actor_loss.backward() + self.actor_optimizer.step() + + def save(self, path): + torch.save({ + 'actor': self.actor.state_dict(), + 'critic': self.critic.state_dict(), + }, path) + + def load(self, path): + checkpoint = torch.load(path, map_location=self.device) + self.actor.load_state_dict(checkpoint['actor']) + self.critic.load_state_dict(checkpoint['critic']) diff --git a/RL_Algothrithms/agents/off_pac.py b/RL_Algothrithms/agents/off_pac.py new file mode 100644 index 0000000..4532996 --- /dev/null +++ b/RL_Algothrithms/agents/off_pac.py @@ -0,0 +1,81 @@ +import torch +import torch.optim as optim +import torch.nn.functional as F +import torch.distributions as distributions +from networks import Actor, VCritic + +class OffPACAgent: + def __init__(self, state_dim, action_dim, device, actor_lr=0.001, critic_lr=0.002, gamma=0.99, epsilon=0.1): + self.device = device + self.gamma = gamma + self.action_dim = action_dim + # epsilon 用于构建行为策略 beta 的探索率 + self.epsilon = epsilon + + # 沿用 A2C 的网络结构 + self.actor = Actor(state_dim, action_dim).to(self.device) + self.critic = VCritic(state_dim).to(self.device) + + self.actor_optimizer = optim.Adam(self.actor.parameters(), lr=actor_lr) + self.critic_optimizer = optim.Adam(self.critic.parameters(), lr=critic_lr) + + def select_action(self, state): + # 推理时禁用梯度图计算 + with torch.no_grad(): + state_tensor = torch.FloatTensor(state).unsqueeze(0).to(self.device) + # 获取目标策略 pi 的动作概率分布 + pi_probs = self.actor(state_tensor).squeeze(0) + + # 构建行为策略 beta:在 pi 的基础上加入 epsilon 的均匀随机噪声 + beta_probs = (1 - self.epsilon) * pi_probs + self.epsilon / self.action_dim + + # 根据行为策略 beta 采样动作,用于与环境交互 + action = distributions.Categorical(beta_probs).sample().item() + return action + + def update(self, state, action, reward, next_state, next_action, done): + state_tensor = torch.FloatTensor(state).unsqueeze(0).to(self.device) + next_state_tensor = torch.FloatTensor(next_state).unsqueeze(0).to(self.device) + reward_tensor = torch.FloatTensor([reward]).unsqueeze(0).to(self.device) + action_tensor = torch.tensor([action]).to(self.device) + + # --- 计算重要性采样权重 (Importance Weight) --- + # 重新获取当前最新目标策略 pi 下的动作概率 + pi_probs = self.actor(state_tensor).squeeze(0) + pi_prob_a = pi_probs[action] + + # 重建行为策略 beta 选出该动作的概率 (作为分母) + beta_prob_a = (1 - self.epsilon) * pi_prob_a.detach() + self.epsilon / self.action_dim + + # 计算重要性权重 rho = pi(a|s) / beta(a|s) + rho = (pi_prob_a / beta_prob_a).detach() + + # 【工程防崩技巧】:如果两个策略偏差过大,rho 会极大导致梯度爆炸。 + # 工业界通用的做法是对 rho 进行截断 (Clip),这也是 PPO 算法的前身思想。 + rho = torch.clamp(rho, 0.1, 10.0) + + # --- Critic 更新 (Off-Policy) --- + v_value = self.critic(state_tensor) + next_v_value = self.critic(next_state_tensor).detach() + + # 计算 TD 目标和优势函数 + td_target = reward_tensor + self.gamma * next_v_value * (1 - int(done)) + advantage = td_target - v_value + + # Critic 损失加入了重要性权重 rho + critic_loss = (rho * F.mse_loss(v_value, td_target, reduction='none')).mean() + + self.critic_optimizer.zero_grad() + critic_loss.backward() + self.critic_optimizer.step() + + # --- Actor 更新 (Off-Policy) --- + action_probs = self.actor(state_tensor) + dist = distributions.Categorical(action_probs) + log_prob = dist.log_prob(action_tensor) + + # Actor 梯度上升目标:rho * ln(pi) * Advantage + actor_loss = -(rho * log_prob * advantage.detach()).mean() + + self.actor_optimizer.zero_grad() + self.actor_optimizer.step() diff --git a/RL_Algothrithms/cont_main.py b/RL_Algothrithms/cont_main.py new file mode 100644 index 0000000..1ec0678 --- /dev/null +++ b/RL_Algothrithms/cont_main.py @@ -0,0 +1,167 @@ +import gymnasium as gym +import torch +import os +import pickle +from agents.dpac import DPACAgent +from agents.ddpg import DDPGAgent +from utils import plot_comparison + + +RESULTS_DIR = 'training_results' + + +# ---------- 路径工具 ---------- +def _path(agent_class, env_name, num_episodes, suffix): + os.makedirs(RESULTS_DIR, exist_ok=True) + return os.path.join(RESULTS_DIR, f'{agent_class.__name__}_{env_name}_{num_episodes}{suffix}') + + +def result_path(agent_class, env_name, num_episodes): + return _path(agent_class, env_name, num_episodes, '.pkl') + + +def model_path(agent_class, env_name, num_episodes): + return _path(agent_class, env_name, num_episodes, '_model.pt') + + +# ---------- 训练 & 保存 ---------- +def train(env_name, agent_class, device, num_episodes): + rp = result_path(agent_class, env_name, num_episodes) + mp = model_path(agent_class, env_name, num_episodes) + + if os.path.exists(rp) and os.path.exists(mp): + with open(rp, 'rb') as f: + rewards = pickle.load(f) + print(f"[{agent_class.__name__}] 已加载训练结果: {rp}") + return rewards + + print(f"[{agent_class.__name__}] 开始训练 ...") + if agent_class.__name__ == 'DDPGAgent': + rewards, agent = _train_ddpg(env_name, agent_class, device, num_episodes) + else: + rewards, agent = _train_dpac(env_name, agent_class, device, num_episodes) + + with open(rp, 'wb') as f: + pickle.dump(rewards, f) + agent.save(mp) + print(f"[{agent_class.__name__}] 训练完成 | 曲线: {rp} | 模型: {mp}") + return rewards + + +def _train_dpac(env_name, agent_class, device, num_episodes): + env = gym.make(env_name) + state_dim = env.observation_space.shape[0] # type: ignore + action_dim = env.action_space.shape[0] # type: ignore + max_action = float(env.action_space.high[0]) # type: ignore + + agent = agent_class(state_dim, action_dim, max_action, device) + rewards = [] + + for ep in range(num_episodes): + state, _ = env.reset() + episode_reward = 0 + while True: + action = agent.select_action(state) + ns, reward, terminated, truncated, _ = env.step(action) + agent.update(state, action, reward, ns, None, terminated or truncated) + state = ns + episode_reward += reward # type: ignore + if terminated or truncated: + break + rewards.append(episode_reward) + if (ep + 1) % 50 == 0: + avg = sum(rewards[-50:]) / 50 + print(f" 回合 {ep+1}/{num_episodes}, 近50回合均分: {avg:.2f}") + + env.close() + return rewards, agent + + +def _train_ddpg(env_name, agent_class, device, num_episodes, batch_size=64, start_steps=2000): + env = gym.make(env_name) + state_dim = env.observation_space.shape[0] # type: ignore + action_dim = env.action_space.shape[0] # type: ignore + max_action = float(env.action_space.high[0]) # type: ignore + + agent = agent_class(state_dim, action_dim, max_action, device) + rewards = [] + total_steps = 0 + + for ep in range(num_episodes): + state, _ = env.reset() + episode_reward = 0 + while True: + action = agent.select_action(state) + ns, reward, terminated, truncated, _ = env.step(action) + agent.replay_buffer.add(state, action, reward, ns, terminated or truncated) + total_steps += 1 + state = ns + episode_reward += reward # type: ignore + if total_steps > start_steps: + agent.update(batch_size) + if terminated or truncated: + break + rewards.append(episode_reward) + if (ep + 1) % 50 == 0: + avg = sum(rewards[-50:]) / 50 + print(f" 回合 {ep+1}/{num_episodes}, 近50回合均分: {avg:.2f}") + + env.close() + return rewards, agent + + +# ---------- 视频演示 ---------- +def demo(env_name, agent_class, device, num_demo=5, num_episodes=600): + mp = model_path(agent_class, env_name, num_episodes) + if not os.path.exists(mp): + print(f"[{agent_class.__name__}] 未找到模型权重 {mp},跳过演示。") + return + + video_dir = os.path.join(RESULTS_DIR, 'videos') + os.makedirs(video_dir, exist_ok=True) + + base_env = gym.make(env_name, render_mode='rgb_array') + env = gym.wrappers.RecordVideo(base_env, video_dir, name_prefix=f'{agent_class.__name__}_demo') + state_dim = env.env.observation_space.shape[0] # type: ignore + action_dim = env.env.action_space.shape[0] # type: ignore + max_action = float(env.env.action_space.high[0]) # type: ignore + + agent = agent_class(state_dim, action_dim, max_action, device) + agent.load(mp) + print(f"[{agent_class.__name__}] 加载模型: {mp},录制 {num_demo} 局演示 ...") + + for ep in range(num_demo): + state, _ = env.reset() + total_reward = 0.0 + while True: + action = agent.select_action(state) + state, reward, terminated, truncated, _ = env.step(action) + total_reward += float(reward) + if terminated or truncated: + break + print(f" 回合 {ep+1}/{num_demo}, 总奖励: {total_reward:.2f}") + + env.close() + base_env.close() + print(f"[{agent_class.__name__}] 视频已保存至: {video_dir}") + + +# ---------- 主程序 ---------- +if __name__ == "__main__": + device = torch.device("cuda" if torch.cuda.is_available() else "cpu") + print(f"计算设备: {device}\n") + + ENV_NAME = 'Pendulum-v1' + EPISODES = 600 + + dpac_rewards = train(ENV_NAME, DPACAgent, device, EPISODES) + ddpg_rewards = train(ENV_NAME, DDPGAgent, device, EPISODES) + + results = { + 'DPAC': dpac_rewards, + 'DDPG': ddpg_rewards, + } + plot_comparison(results, window=50, save_path='compare_dpac_vs_ddpg.png') + + demo(ENV_NAME, DPACAgent, device) + demo(ENV_NAME, DDPGAgent, device) diff --git a/RL_Algothrithms/disp_main.py b/RL_Algothrithms/disp_main.py new file mode 100644 index 0000000..55eea2b --- /dev/null +++ b/RL_Algothrithms/disp_main.py @@ -0,0 +1,95 @@ +import gymnasium as gym +import torch +import os +import pickle +from agents.qac import QACAgent +from agents.a2c import A2CAgent +from agents.off_pac import OffPACAgent +from utils import plot_comparison + +RESULTS_DIR = 'training_results' + +def get_result_path(agent_class, env_name, num_episodes): + os.makedirs(RESULTS_DIR, exist_ok=True) + return os.path.join(RESULTS_DIR, f'{agent_class.__name__}_{env_name}_{num_episodes}.pkl') + +def load_or_train(env_name, agent_class, device, num_episodes): + """检查是否有保存的结果,有则加载,没有则训练并保存。""" + result_path = get_result_path(agent_class, env_name, num_episodes) + if os.path.exists(result_path): + with open(result_path, 'rb') as f: + rewards = pickle.load(f) + print(f"[{agent_class.__name__}] 已找到训练结果文件,直接加载: {result_path}") + return rewards + + # 没有结果文件,开始训练 + print(f"[{agent_class.__name__}] 未找到结果文件,开始训练...") + rewards = _train_agent(env_name, agent_class, device, num_episodes) + + with open(result_path, 'wb') as f: + pickle.dump(rewards, f) + print(f"[{agent_class.__name__}] 训练完成,结果已保存至: {result_path}") + return rewards + +def _train_agent(env_name, agent_class, device, num_episodes): + """实际的训练循环逻辑""" + env = gym.make(env_name) + state_dim = env.observation_space.shape[0] # type: ignore + action_dim = int(env.action_space.n) # type: ignore + + agent = agent_class(state_dim, action_dim, device) + rewards_history = [] + + for episode in range(num_episodes): + state, _ = env.reset() + episode_reward = 0 + + action = agent.select_action(state) + + while True: + next_state, reward, terminated, truncated, _ = env.step(action) + done = terminated or truncated + + next_action = agent.select_action(next_state) + agent.update(state, action, reward, next_state, next_action, done) + + state = next_state + action = next_action + episode_reward += reward # type: ignore + + if done: + break + + rewards_history.append(episode_reward) + if (episode + 1) % 100 == 0: + print(f"[{agent_class.__name__}] 回合 {episode+1}/{num_episodes}, 近100回合均分: {sum(rewards_history[-100:])/100:.2f}") + + env.close() + return rewards_history + +if __name__ == "__main__": + # 检测 GPU + device = torch.device("cuda" if torch.cuda.is_available() else "cpu") + print(f"正在使用的计算设备: {device}") + + ENV_NAME = 'CartPole-v1' + EPISODES = 600 + + print("\n--- 开始训练 QAC ---") + qac_rewards = load_or_train(ENV_NAME, QACAgent, device, EPISODES) + + print("\n--- 开始训练 A2C ---") + a2c_rewards = load_or_train(ENV_NAME, A2CAgent, device, EPISODES) + + print("\n--- 开始训练 Off-Policy A2C ---") + off_pac_rewards = load_or_train(ENV_NAME, OffPACAgent, device, EPISODES) + + # 将其加入字典一起画图对比 + results = { + 'QAC': qac_rewards, + 'A2C': a2c_rewards, + 'Off-Policy A2C': off_pac_rewards + } + + + plot_comparison(results, window=50, save_path='qac_vs_a2c_gpu.png') diff --git a/RL_Algothrithms/main_cont.py b/RL_Algothrithms/main_cont.py new file mode 100644 index 0000000..e69de29 diff --git a/RL_Algothrithms/networks_cont.py b/RL_Algothrithms/networks_cont.py new file mode 100644 index 0000000..bacc2c6 --- /dev/null +++ b/RL_Algothrithms/networks_cont.py @@ -0,0 +1,32 @@ +import torch +import torch.nn as nn +import torch.nn.functional as F + +# 连续动作策略网络 (Deterministic Actor) +class ContActor(nn.Module): + def __init__(self, state_dim, action_dim, max_action): + super(ContActor, self).__init__() + self.fc1 = nn.Linear(state_dim, 128) + self.fc2 = nn.Linear(128, action_dim) + self.max_action = max_action + + def forward(self, state): + x = F.relu(self.fc1(state)) + # 使用 tanh 将输出限制在 [-1, 1],然后映射到物理实际的边界 (例如 [-2.0, 2.0]) + action = torch.tanh(self.fc2(x)) * self.max_action + return action + +# 连续动作 Q 值网络 (用于评价 (State, Action) 对) +class ContQCritic(nn.Module): + def __init__(self, state_dim, action_dim): + super(ContQCritic, self).__init__() + # 将状态和动作拼接后输入网络 + self.fc1 = nn.Linear(state_dim + action_dim, 128) + self.fc2 = nn.Linear(128, 1) + + def forward(self, state, action): + # 拼接维度 + xu = torch.cat([state, action], 1) + x = F.relu(self.fc1(xu)) + q_value = self.fc2(x) + return q_value