Double DQN性能远逊于Vanilla DQN的技术问题求助

环境与任务概述 #

• 100×100网格环境（每格宽10米），每回合随机生成多个静止用户（ues）
• 用户资源需求随智能体距离变化：距离越近需求越少，用户获得全部所需资源则标记为已满足
• 智能体目标：找到能满足最多用户的位置（用户位置未知）
• 状态空间：智能体当前位置、已满足用户数、用户资源请求列表
• 动作空间：9种动作（前后移动、侧移、停留等）
• 训练配置：每回合初始位置随机；Epsilon在550回合内从1衰减至0.1

奖励函数规则 #

• 移动后满足用户数增加，或未增加但未出现过更优位置：奖励+2
• 移动后满足用户数不变且存在过更优位置：惩罚-0.001
• 移动后满足用户数减少：惩罚-0.002

核心问题 #

采用tau=1e-3的目标网络软更新策略，但Double DQN性能远差于Vanilla DQN，需排查原因（奖励函数或实现问题）。

代码实现问题分析 #

从提供的DDQN代码来看，存在多个关键问题可能导致性能不佳：

1. 输出层激活函数错误 #

DQN的输出是动作的Q值（连续值），需使用线性激活（无激活函数），但代码中用了softmax激活：

output_layer = Dense(output_nodes, activation="softmax")(dense_layer_3)

softmax会将输出归一化为概率分布，破坏Q值的绝对大小，导致DDQN中"用主网络选动作、目标网络算Q值"的逻辑完全失效。

2. 损失函数不匹配 #

代码使用categorical_crossentropy（分类任务损失），但DQN是回归任务（预测Q值），应使用**均方误差(MSE)**或Huber损失：

dnn.compile(loss="categorical_crossentropy", optimizer=opt, metrics=['accuracy'])

分类损失会导致模型训练目标偏离Q值预测，无法正确学习动作价值。

3. 训练效率与稳定性问题 #

代码对minibatch中的样本逐个调用fit训练，而非批量处理：

for index, (current_state, action, reward, new_current_state, done) in enumerate(minibatch):
    # ...单样本处理逻辑
    self.model_uav_pos.fit([pos, reqs, number_satisfaction], target_f, ...)

单样本训练会导致梯度更新噪声大、训练速度极慢，模型难以收敛。

4. 状态特征未归一化 #

位置（0-99）、已满足用户数（0-总用户数）等特征尺度差异大，但代码未做归一化处理：

# create_pos_dnn中直接使用原始输入，无归一化

特征尺度不一致会导致模型对大尺度特征过度敏感，训练不稳定。

5. Epsilon衰减逻辑不符预期 #

代码中设置self.epsilon_decay = 0.8，但描述要求550回合内从1衰减至0.1。指数衰减的正确decay值应为：
$$\text{decay} = (0.1/1)^{1/550} \approx 0.998$$
当前0.8的衰减速度过快，会过早结束探索，导致智能体无法充分探索环境。

可能的改进方向 #

1. 修正网络结构：将输出层改为线性激活，损失函数替换为mse或huber_loss
2. 批量训练：将minibatch样本批量处理，一次性输入模型训练
3. 特征归一化：对位置、已满足用户数等特征进行归一化（如缩放到0-1区间）
4. 调整Epsilon衰减：计算符合550回合要求的衰减系数，保证探索-利用平衡
5. 对齐Vanilla DQN配置：确保DDQN与Vanilla DQN的 replay memory大小、学习率、批量大小等参数完全一致，排除参数差异影响

附用户提供的DDQN代码 #

DISCOUNT = 0.9 #0.99
REPLAY_MEMORY_SIZE = 10_000  
MIN_REPLAY_MEMORY_SIZE = 10_000  # Minimum number of steps in a memory to start training
MINIBATCH_SIZE = 32 # How many steps (samples) to use for training
class DDQNAgent(object):
    def __init__(self):
        #self.gamma = 0.95 
        self.epsilon = 1.0
        self.epsilon_decay = 0.8
        self.epsilon_min = 0.1
        self.learning_rate = 10e-4 #0.0005 #0.25 #1e-4
        self.tau = 1e-3
        
        self.plot_loss_acc = PlotLearning()
                
        # Main models
        self.model_uav_pos = self._build_pos_model()

        # Target networks
        self.target_model_uav_pos = self._build_pos_model()
        # Copy weights
        self.target_model_uav_pos.set_weights(self.model_uav_pos.get_weights())

        # An array with last n steps for training
        self.replay_memory_pos_nn = deque(maxlen=REPLAY_MEMORY_SIZE)

        tboard_log_dir_pos = os.path.join("logs", MODEL_NAME_POS_DDQN)
        self.tensorboard_pos = ModifiedTensorBoard(MODEL_NAME_POS_DDQN, log_dir=tboard_log_dir_pos)
        
    def _build_pos_model(self): # compile the DNN
        # create the DNN model
        dnn = self.create_pos_dnn()
        
        opt = Adam(learning_rate=self.learning_rate) #, decay=self.epsilon_decay)
        dnn.compile(loss="categorical_crossentropy", optimizer=opt, metrics=['accuracy'])
        dnn.call = tf.function(dnn.call, jit_compile=True)
        
        return dnn
    
    ''' Don't forget to normalize the inputs '''
    def create_pos_dnn(self): 
        # initialize the input shape (The shape of an array is the number of elements in each dimension)
        pos_input_shape = (2,)
        requests_input_shape = (len(env.ues),)
        number_of_satisfied_ues_input_shape = (1,)
        # How many possible outputs we can have
        output_nodes = n_possible_movements
        
        # Initialize the inputs
        uav_current_position = Input(shape=pos_input_shape, name='pos')
        ues_requests = Input(shape=requests_input_shape, name='requests')
        number_of_satisfied_ues = Input(shape=number_of_satisfied_ues_input_shape, name='number_of_satisfied_ues')
        
        # Put them in a list
        list_inputs = [uav_current_position, ues_requests, number_of_satisfied_ues]
        
        # Merge all input features into a single large vector
        x = layers.concatenate(list_inputs)
        
        # Add a 1st Hidden (Dense) Layer
        dense_layer_1 = Dense(512, activation="relu")(x)
        
        # Add a 2nd Hidden (Dense) Layer
        dense_layer_2 = Dense(512, activation="relu")(dense_layer_1)
        
        # Add a 3rd Hidden (Dense) Layer
        dense_layer_3 = Dense(256, activation="relu")(dense_layer_2)
        
        # Output layer
        output_layer = Dense(output_nodes, activation="softmax")(dense_layer_3)

        model = Model(inputs=list_inputs, outputs=output_layer)
                        
        # return the DNN
        return model
    
    def remember_pos_nn(self, state, action, reward, next_state, done):
        self.replay_memory_pos_nn.append((state, action, reward, next_state, done)) # list of previous experiences, enabling re-training later
        
    def act_upon_choosing_a_new_position(self, state): # state is a tuple (uav_position, requests_array)
        if np.random.rand() <= self.epsilon: # if acting randomly, take random action
            return random.randrange(n_possible_movements)
        pos =  np.array([state[0]])
        reqs =  np.array([state[1]])
        number_satisfaction = np.array([state[2]])
        act_values = self.model_uav_pos([pos, reqs, number_satisfaction]) # if not acting randomly, predict reward value based on current state
        return np.argmax(act_values[0]) #env.possible_positions[np.argmax(act_values[0])] # pick the action that will give the highest reward
        
    def train_pos_nn(self):
        print("In Training..")

        # Start training only if certain number of samples is already saved
        if len(self.replay_memory_pos_nn) < MIN_REPLAY_MEMORY_SIZE:
            print("Exiting Training: Replay Memory Not Full Enough...")
            return

        # Get a minibatch of random samples from memory replay table
        list_memory = list(self.replay_memory_pos_nn)
        random.shuffle(list_memory)
        minibatch = random.sample(list_memory, MINIBATCH_SIZE)

        start_time = time.time()
        # Enumerate our batches
        for index, (current_state, action, reward, new_current_state, done) in enumerate(minibatch):
            print('...Starting Training...')
            target = 0
            pos =  np.array([current_state[0]])
            reqs =  np.array([current_state[1]])
            number_satisfaction = np.array([current_state[2]])
            pos_next = np.array([new_current_state[0]])
            reqs_next = np.array([new_current_state[1]])
            number_satisfaction_next = np.array([new_current_state[2]])

            # If not a terminal state, get new q from future states, otherwise set it to 0
            # almost like with Q Learning, but we use just part of equation here
            if not done:
                max_action = np.argmax(self.model_uav_pos([pos_next, reqs_next, number_satisfaction_next])[0])
                target = reward + DISCOUNT * self.target_model_uav_pos([pos_next, reqs_next, number_satisfaction_next])[0][max_action]
            else:
                target = reward

            # Update Q value for a given state
            target_f = self.model_uav_pos([pos, reqs, number_satisfaction])
            target_f = np.array(target_f)
            target_f[0][action] = target

            self.model_uav_pos.fit([pos, reqs, number_satisfaction], \
                                   target_f, \
                                   verbose=2, \
                                   shuffle=False, \
                                   callbacks=None, \
                                   epochs=1 \
                                  )  
        end_time = time.time()
        print("Time", end_time - start_time)
        # Update target network counter every episode
        self.target_train() 

    def target_train(self):
        weights = self.model_uav_pos.get_weights()
        target_weights = self.target_model_uav_pos.get_weights()
        for i in range(len(target_weights)):
            target_weights[i] = weights[i] * self.tau + target_weights[i] * (1 - self.tau)
        self.target_model_uav_pos.set_weights(target_weights)

内容的提问来源于stack exchange，提问作者Ness