新增模型自动查找功能

src/Unmanned_vehicle_AD_DQN/Hyperparameters.py

@@ -74,4 +74,21 @@
 # 成功阈值
 
 LEARNING_RATE = 0.00005
-# 优化器的学习率 - 降低以获得更稳定训练
+# 优化器的学习率 - 降低以获得更稳定训练
+
+# PER (优先经验回放) 参数
+PER_ALPHA = 0.6
+# 优先级程度 (0 = 均匀采样, 1 = 完全优先级)
+
+PER_BETA_START = 0.4
+# 重要性采样权重起始值
+
+PER_BETA_FRAMES = 100000
+# beta线性增长的帧数
+
+# Dueling DQN 参数
+USE_DUELING = True
+# 是否使用Dueling DQN架构
+
+USE_PER = True
+# 是否使用优先经验回放

src/Unmanned_vehicle_AD_DQN/Model.py

@@ -9,9 +9,8 @@
 import math
 import matplotlib.pyplot as plt
 from collections import deque
-from tensorflow.keras.applications.xception import Xception
 from tensorflow.keras.layers import Dense, GlobalAveragePooling2D, Input, Concatenate, Conv2D, AveragePooling2D, Activation, \
-    Flatten, Dropout, BatchNormalization, MaxPooling2D, Multiply
+    Flatten, Dropout, BatchNormalization, MaxPooling2D, Multiply, Add, Lambda, Subtract
 from tensorflow.keras.optimizers import Adam
 from tensorflow.keras.models import Sequential, Model
 from tensorflow.keras.callbacks import TensorBoard
@@ -54,16 +53,90 @@ def update_stats(self, **stats):
                 self.writer.flush()
 
 
-# DQN智能体类
+# 优先经验回放缓冲区
+class PrioritizedReplayBuffer:
+    def __init__(self, max_size=REPLAY_MEMORY_SIZE, alpha=0.6, beta_start=0.4, beta_frames=100000):
+        self.max_size = max_size
+        self.alpha = alpha  # 优先级程度 (0 = 均匀采样, 1 = 完全优先级)
+        self.beta_start = beta_start  # 重要性采样权重起始值
+        self.beta_frames = beta_frames  # beta线性增长的帧数
+        self.frame = 1
+
+        # 使用循环缓冲区
+        self.buffer = deque(maxlen=max_size)
+        self.priorities = deque(maxlen=max_size)
+
+    def __len__(self):
+        return len(self.buffer)
+
+    def beta(self):
+        """线性递增的beta值，用于重要性采样权重"""
+        return min(1.0, self.beta_start + self.frame * (1.0 - self.beta_start) / self.beta_frames)
+
+    def add(self, experience, error=None):
+        """添加经验到缓冲区"""
+        if error is None:
+            priority = max(self.priorities) if self.priorities else 1.0
+        else:
+            priority = (abs(error) + 1e-5) ** self.alpha
+
+        self.buffer.append(experience)
+        self.priorities.append(priority)
+
+    def sample(self, batch_size):
+        """从缓冲区中采样一批经验"""
+        if len(self.buffer) == 0:
+            return [], [], [], []
+
+        # 计算采样概率
+        priorities = np.array(self.priorities, dtype=np.float32)
+        probs = priorities ** self.alpha
+        probs /= probs.sum()
+
+        # 采样索引
+        indices = np.random.choice(len(self.buffer), batch_size, p=probs)
+
+        # 获取样本
+        samples = [self.buffer[i] for i in indices]
+
+        # 计算重要性采样权重
+        total = len(self.buffer)
+        weights = (total * probs[indices]) ** (-self.beta())
+        weights /= weights.max()  # 归一化
+
+        # 更新帧计数器
+        self.frame += 1
+
+        return indices, samples, weights
+
+    def update_priorities(self, indices, errors):
+        """更新采样经验的优先级"""
+        for idx, error in zip(indices, errors):
+            if 0 <= idx < len(self.priorities):
+                self.priorities[idx] = (abs(error) + 1e-5) ** self.alpha
+
+
+# DQN智能体类 - 升级版
 class DQNAgent:
-    def __init__(self):
+    def __init__(self, use_dueling=True, use_per=True):
         # 创建主网络和目标网络
-        self.model = self.create_model()
-        self.target_model = self.create_model()
+        self.use_dueling = use_dueling
+        self.use_per = use_per
+
+        if use_dueling:
+            self.model = self.create_dueling_model()
+            self.target_model = self.create_dueling_model()
+        else:
+            self.model = self.create_model()
+            self.target_model = self.create_model()
+
         self.target_model.set_weights(self.model.get_weights())
 
-        # 经验回放缓冲区
-        self.replay_memory = deque(maxlen=REPLAY_MEMORY_SIZE)
+        # 经验回放缓冲区 - 使用PER或标准缓冲区
+        if use_per:
+            self.replay_buffer = PrioritizedReplayBuffer(max_size=REPLAY_MEMORY_SIZE)
+        else:
+            self.replay_memory = deque(maxlen=REPLAY_MEMORY_SIZE)
 
         # 自定义TensorBoard
         self.tensorboard = ModifiedTensorBoard(log_dir=f"logs/{MODEL_NAME}-{int(time.time())}")
@@ -76,8 +149,8 @@ def __init__(self):
         self.training_initialized = False
 
     def create_model(self):
-        """创建深度Q网络模型 - 优化网络结构"""
-        # 使用函数式API以支持注意力机制
+        """创建标准深度Q网络模型"""
+        # 使用函数式API
         inputs = Input(shape=(IM_HEIGHT, IM_WIDTH, 3))
 
         # 第一卷积块
@@ -105,7 +178,7 @@ def create_model(self):
         # 展平层
         x = Flatten()(x)
 
-        # 全连接层 - 增加层深度
+        # 全连接层
         x = Dense(512, activation='relu')(x)
         x = Dropout(0.3)(x)
         x = Dense(256, activation='relu')(x)
@@ -122,65 +195,147 @@ def create_model(self):
         model = Model(inputs=inputs, outputs=outputs)
 
         # 编译模型
-        model.compile(loss="huber", optimizer=Adam(lr=LEARNING_RATE), metrics=["mae"])
+        model.compile(loss="huber", optimizer=Adam(learning_rate=LEARNING_RATE), metrics=["mae"])
+        return model
+
+    def create_dueling_model(self):
+        """创建Dueling DQN模型架构"""
+        inputs = Input(shape=(IM_HEIGHT, IM_WIDTH, 3))
+
+        # 共享的特征提取层
+        # 第一卷积块
+        x = Conv2D(32, (5, 5), strides=(2, 2), padding='same')(inputs)
+        x = Activation('relu')(x)
+        x = BatchNormalization()(x)
+        x = MaxPooling2D(pool_size=(2, 2))(x)
+
+        # 第二卷积块
+        x = Conv2D(64, (3, 3), padding='same')(x)
+        x = Activation('relu')(x)
+        x = BatchNormalization()(x)
+        x = MaxPooling2D(pool_size=(2, 2))(x)
+
+        # 第三卷积块
+        x = Conv2D(128, (3, 3), padding='same')(x)
+        x = Activation('relu')(x)
+        x = BatchNormalization()(x)
+        x = MaxPooling2D(pool_size=(2, 2))(x)
+
+        # 空间注意力机制
+        attention = Conv2D(1, (1, 1), padding='same', activation='sigmoid')(x)
+        x = Multiply()([x, attention])
+
+        # 展平层
+        x = Flatten()(x)
+
+        # 共享的全连接层
+        shared = Dense(512, activation='relu')(x)
+        shared = Dropout(0.3)(shared)
+        shared = Dense(256, activation='relu')(shared)
+
+        # 价值流 (V(s))
+        value_stream = Dense(128, activation='relu')(shared)
+        value_stream = Dropout(0.2)(value_stream)
+        value = Dense(1, activation='linear', name='value')(value_stream)
+
+        # 优势流 (A(s,a))
+        advantage_stream = Dense(128, activation='relu')(shared)
+        advantage_stream = Dropout(0.2)(advantage_stream)
+        advantage = Dense(5, activation='linear', name='advantage')(advantage_stream)
+
+        # 合并: Q(s,a) = V(s) + (A(s,a) - mean(A(s,a)))
+        # 减去均值使得优势函数的均值为0
+        mean_advantage = Lambda(lambda a: tf.reduce_mean(a, axis=1, keepdims=True))(advantage)
+        advantage_centered = Subtract()([advantage, mean_advantage])
+        q_values = Add()([value, advantage_centered])
+
+        # 创建模型
+        model = Model(inputs=inputs, outputs=q_values)
+
+        # 编译模型
+        model.compile(loss="huber", optimizer=Adam(learning_rate=LEARNING_RATE), metrics=["mae"])
+
         return model
 
     def update_replay_memory(self, transition):
         """更新经验回放缓冲区"""
         # transition = (当前状态, 动作, 奖励, 新状态, 完成标志)
-        self.replay_memory.append(transition)
+        if self.use_per:
+            # PER: 初始添加时使用最大优先级
+            self.replay_buffer.add(transition, error=1.0)  # 初始误差设为1.0
+        else:
+            self.replay_memory.append(transition)
 
     def minibatch_chooser(self):
         """改进的经验采样策略"""
-        if len(self.replay_memory) < MIN_REPLAY_MEMORY_SIZE:
-            return random.sample(self.replay_memory, min(len(self.replay_memory), MINIBATCH_SIZE))
+        if self.use_per:
+            # PER采样
+            if len(self.replay_buffer) < MIN_REPLAY_MEMORY_SIZE:
+                return [], [], [], []
+
+            indices, samples, weights = self.replay_buffer.sample(MINIBATCH_SIZE)
+            return indices, samples, weights
+        else:
+            # 标准采样
+            if len(self.replay_memory) < MIN_REPLAY_MEMORY_SIZE:
+                return random.sample(self.replay_memory, min(len(self.replay_memory), MINIBATCH_SIZE))
+
+            # 分类经验样本
+            positive_samples = []    # 高奖励经验
+            negative_samples = []    # 负奖励/碰撞经验
+            neutral_samples = []     # 中性奖励经验
 
-        # 分类经验样本
-        positive_samples = []    # 高奖励经验
-        negative_samples = []    # 负奖励/碰撞经验
-        neutral_samples = []     # 中性奖励经验
-
-        for sample in self.replay_memory:
-            _, _, reward, _, done = sample
+            for sample in self.replay_memory:
+                _, _, reward, _, done = sample
+
+                if done and reward < -5:  # 碰撞或严重错误
+                    negative_samples.append(sample)
+                elif reward > 1:  # 积极经验
+                    positive_samples.append(sample)
+                else:  # 中性经验
+                    neutral_samples.append(sample)
+
+            # 平衡采样
+            batch = []
+
+            # 采样负经验 (20%)
+            num_negative = min(len(negative_samples), MINIBATCH_SIZE // 5)
+            batch.extend(random.sample(negative_samples, num_negative))
 
-            if done and reward < -5:  # 碰撞或严重错误
-                negative_samples.append(sample)
-            elif reward > 1:  # 积极经验
-                positive_samples.append(sample)
-            else:  # 中性经验
-                neutral_samples.append(sample)
-
-        # 平衡采样
-        batch = []
-
-        # 采样负经验 (20%)
-        num_negative = min(len(negative_samples), MINIBATCH_SIZE // 5)
-        batch.extend(random.sample(negative_samples, num_negative))
-
-        # 采样正经验 (30%)
-        num_positive = min(len(positive_samples), MINIBATCH_SIZE // 3)
-        batch.extend(random.sample(positive_samples, num_positive))
-
-        # 用中性经验补全批次
-        remaining = MINIBATCH_SIZE - len(batch)
-        if remaining > 0:
-            batch.extend(random.sample(neutral_samples, min(remaining, len(neutral_samples))))
-
-        # 如果还不够，从整个记忆库随机采样
-        if len(batch) < MINIBATCH_SIZE:
-            additional = MINIBATCH_SIZE - len(batch)
-            batch.extend(random.sample(self.replay_memory, additional))
+            # 采样正经验 (30%)
+            num_positive = min(len(positive_samples), MINIBATCH_SIZE // 3)
+            batch.extend(random.sample(positive_samples, num_positive))
 
-        random.shuffle(batch)  # 打乱批次
-        return batch
+            # 用中性经验补全批次
+            remaining = MINIBATCH_SIZE - len(batch)
+            if remaining > 0:
+                batch.extend(random.sample(neutral_samples, min(remaining, len(neutral_samples))))
+
+            # 如果还不够，从整个记忆库随机采样
+            if len(batch) < MINIBATCH_SIZE:
+                additional = MINIBATCH_SIZE - len(batch)
+                batch.extend(random.sample(self.replay_memory, additional))
+
+            random.shuffle(batch)  # 打乱批次
+            return batch
 
     def train(self):
         """训练DQN网络"""
-        if len(self.replay_memory) < MIN_REPLAY_MEMORY_SIZE:
-            return
-
-        # 选择小批量经验
-        minibatch = self.minibatch_chooser()
+        if self.use_per:
+            if len(self.replay_buffer) < MIN_REPLAY_MEMORY_SIZE:
+                return
+
+            # PER: 采样并获取权重
+            indices, minibatch, weights = self.replay_buffer.sample(MINIBATCH_SIZE)
+            if len(minibatch) == 0:
+                return
+        else:
+            if len(self.replay_memory) < MIN_REPLAY_MEMORY_SIZE:
+                return
+
+            # 标准采样
+            minibatch = self.minibatch_chooser()
+            weights = np.ones(len(minibatch))  # 标准训练权重为1
 
         # 准备训练数据
         current_states = np.array([transition[0] for transition in minibatch]) / 255
@@ -191,6 +346,7 @@ def train(self):
 
         x = []  # 输入状态
         y = []  # 目标Q值
+        errors = []  # TD误差（用于PER）
 
         # 计算目标Q值
         for index, (current_state, action, reward, new_state, done) in enumerate(minibatch):
@@ -201,21 +357,33 @@ def train(self):
             else:
                 new_q = reward  # 终止状态
 
-            current_qs = current_qs_list[index]
+            current_qs = current_qs_list[index].copy()
+            old_q = current_qs[action]  # 用于计算TD误差
             current_qs[action] = new_q  # 更新对应动作的Q值
+
+            # 计算TD误差
+            td_error = abs(new_q - old_q)
+            errors.append(td_error)
 
             x.append(current_state)
             y.append(current_qs)
 
+        # PER: 更新优先级
+        if self.use_per and len(errors) > 0:
+            self.replay_buffer.update_priorities(indices, errors)
+
         # 记录日志判断
         log_this_step = False
         if self.tensorboard.step > self.last_logged_episode:
             log_this_step = True
             self.last_logged_episode = self.tensorboard.step
 
-        # 训练模型
-        self.model.fit(np.array(x) / 255, np.array(y), batch_size=TRAINING_BATCH_SIZE, verbose=0, shuffle=False,
-                       callbacks=[self.tensorboard] if log_this_step else None)
+        # 训练模型（带样本权重）
+        self.model.fit(np.array(x) / 255, np.array(y), 
+                      batch_size=TRAINING_BATCH_SIZE, 
+                      sample_weight=weights if self.use_per else None,
+                      verbose=0, shuffle=False,
+                      callbacks=[self.tensorboard] if log_this_step else None)
 
         # 更新目标网络
         if log_this_step: