运行dqn.test无执行无报错，请求排查解决方案

问题描述 #

运行scores = dqn.test(env, nb_episodes=100, visualize=False)时，代码长时间无输出、无报错，陷入卡顿状态。该代码用于测试DQN智能体在自定义EnergyOptimizationEnv环境中100个episode的性能，未启用可视化，但执行耗时远超预期。

已尝试的解决措施 #

• 确认所有依赖库已安装并更新至最新版本
• 验证DQN智能体与环境的配置兼容性
• 核对dqn.test函数的参数使用符合官方文档要求
• 测试GreedyQPolicy、BoltzmannQPolicy等不同策略对执行时间的影响
• 排查Google Colab平台的资源限制问题
• 将测试episode数降至10，问题依然存在
• 实现独立的Q值预测模型（代码如下）：

from rl.policy import GreedyQPolicy
from keras.models import Model
from keras.layers import Input, Reshape

scores = []
num_episodes = 100

policy = GreedyQPolicy()  # 创建GreedyQPolicy策略

# 构建独立的Q值预测模型
input_shape = env.observation_space.shape
inputs = Input(shape=input_shape)
reshape = Reshape((1,) + input_shape)(inputs)
q_values = dqn.model(reshape)
q_model = Model(inputs=inputs, outputs=q_values)

for episode in range(num_episodes):
    state = env.reset()
    done = False
    episode_reward = 0
    
    while not done:
        q_values = q_model.predict(np.expand_dims(state, axis=0))
        action = policy.select_action(q_values[0])
        next_state, reward, done, _ = env.step(action)
        episode_reward += reward
        state = next_state
    
    scores.append(episode_reward)

average_score = np.mean(scores)
print(average_score)

问题根源分析 #

查看自定义EnergyOptimizationEnv环境的实现，发现done条件仅在总能耗超过阈值energy_threshold=10000时触发。而测试过程中未设置最大步数限制：

• 若智能体持续选择action=0（什么都不做），总能耗不会增加，永远无法触发done条件，导致单个episode进入无限循环，最终整个测试过程卡顿。
• 训练时使用了nb_max_episode_steps=50限制单episode步数，但测试时未添加该参数。

解决方案 #

方案1：给dqn.test添加最大步数限制 #

在测试时指定与训练一致的单episode最大步数，避免无限循环：

scores = dqn.test(env, nb_episodes=100, visualize=False, nb_max_episode_steps=50)
print(np.mean(scores.history['episode_reward']))

方案2：修改环境的done条件 #

在环境中添加步数限制，从根源避免无限循环：
修改EnergyOptimizationEnv类，添加步数计数器：

class EnergyOptimizationEnv(gym.Env):
    def __init__(self):
        # 原有初始化代码...
        self.max_steps = 50  # 添加最大步数限制
        self.current_step = 0
    
    def step(self, action):
        self.current_step += 1  # 步数递增
        # 原有step代码...
        
        # 修改done条件：超过阈值或达到最大步数
        done = self.check_threshold() or (self.current_step >= self.max_steps)
        return self.state, reward, done, info
    
    def reset(self):
        # 原有reset代码...
        self.current_step = 0  # 重置步数计数器
        return self.state

高效测试的替代方案 #

优化自定义测试循环，添加步数限制并增加进度反馈，方便实时观察测试状态：

import numpy as np
from rl.policy import GreedyQPolicy

scores = []
num_episodes = 100
max_steps_per_episode = 50  # 与训练时的步数限制一致

policy = GreedyQPolicy()

for episode in range(num_episodes):
    state = env.reset()
    done = False
    episode_reward = 0
    step_count = 0
    
    while not done and step_count < max_steps_per_episode:
        # 使用dqn内置模型直接预测，无需重新构建
        q_values = dqn.model.predict(np.expand_dims(state, axis=0), verbose=0)
        action = policy.select_action(q_values[0])
        next_state, reward, done, _ = env.step(action)
        episode_reward += reward
        state = next_state
        step_count += 1
    
    scores.append(episode_reward)
    # 每10个episode打印一次进度
    if (episode + 1) % 10 == 0:
        print(f"Episode {episode+1}/{num_episodes}, 当前episode奖励: {episode_reward}")

average_score = np.mean(scores)
print(f"100个episode平均奖励: {average_score}")

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