添加 DDPG、DPAC、Off-PAC 算法实现及连续动作空间训练代码

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'])

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'])

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()