解决DQN玩平衡车损失不收敛的问题
·
这几天一直在学习强化学习,试了比较经典的Pendulum-v1去做练手,但是跟着网上的教程一步一步做,发现我的回报不收敛。记录一下遇到的问题。(tensorflow)
首先我先过一遍我的代码。
导入包和初始化环境
import random
import gym
import numpy as np
from matplotlib import pyplot as plt
import tensorflow as tf
from collections import deque
from tensorflow.keras import layers
#跳过gpu,使用cpu
physical_devices = tf.config.list_physical_devices('GPU')
if physical_devices:
tf.config.set_visible_devices([], 'GPU')
初始化环境
env = gym.make("Pendulum-v1", render_mode="rgb_array")
env.reset()
神经网络的搭建
def build_model():
model = tf.keras.Sequential([
layers.Dense(512, input_dim=3, activation='relu'),
layers.BatchNormalization(),
layers.Dense(512, activation='relu'),
layers.BatchNormalization(),
layers.Dense(256, activation='relu'),
layers.BatchNormalization(),
layers.Dense(21, activation='linear')
])
model.compile(loss='mse', optimizer=tf.keras.optimizers.Adam(learning_rate=3e-4))
return model
创建神经网络
model = build_model()
构建一个和原网络一样的网络(目标网络)
target_model = build_model()
target_model.set_weights(model.get_weights())
epsilon = 1.0 epsilon_min = 0.01 epsilon_decay = 0.995 经验回放池可以适当调大一些 replay_buffer = deque(maxlen=50000) batch_size = 256 步长越小,更新越快 update_target_every = 10
动作策略,使用贪心策略并离散化
def get_action(state):
global epsilon
if np.random.rand() < epsilon:
return np.random.randint(0, 21), 0
state = np.reshape(state, [1, 3])
q_values = model.predict(state, verbose=0)
action_idx = np.argmax(q_values[0])
action_continuous = (action_idx / 20) * 4 - 2
return action_idx, action_continuous
定义添加经验元组的函数
def add_experience(state, action_idx, reward, next_state, done):
replay_buffer.append((state, action_idx, reward, next_state, done))
while len(replay_buffer) < 1000:
state = env.reset()[0]
for _ in range(200):
action_idx, action_cont = get_action(state)
next_state, reward, done, _, _ = env.step([action_cont])
reward = reward / 10
add_experience(state, action_idx, reward, next_state, done)
state = next_state if not done else env.reset()[0]
if done:
break
如果回放池小于1000就加记录
采样的函数
def sample_experiences():
samples = random.sample(replay_buffer, batch_size)
states = np.array([s[0] for s in samples])
actions = np.array([s[1] for s in samples])
rewards = np.array([s[2] for s in samples])
next_states = np.array([s[3] for s in samples])
dones = np.array([s[4] for s in samples])
return (
tf.convert_to_tensor(states, dtype=tf.float32),
tf.convert_to_tensor(actions, dtype=tf.int32),
tf.convert_to_tensor(rewards, dtype=tf.float32),
tf.convert_to_tensor(next_states, dtype=tf.float32),
tf.convert_to_tensor(dones, dtype=tf.bool)
)
最重要的一步,训练的函数(核心)~~
def train_step(states, actions, rewards, next_states, dones):
future_q_values = target_model.predict(next_states, verbose=0)#Q(S',a)
max_future_q = np.max(future_q_values, axis=1)#Q*(S',a)
target_q = rewards + (1 - dones.numpy()) * 0.99 * max_future_q# target = r+gama*Q(S',a)
with tf.GradientTape() as tape:
current_q = model(states)
action_masks = tf.one_hot(actions, 21)
current_q = tf.reduce_sum(current_q * action_masks, axis=1)
loss = tf.keras.losses.MSE(target_q, current_q)#LOSS
gradients = tape.gradient(loss, model.trainable_variables)#计算Loss 关于神经网络的参数的梯度
model.optimizer.apply_gradients(zip(gradients, model.trainable_variables))#更新参数
return loss.numpy()
#测试神经网络的函数,reward越高越好
def test_model(num_tests=5):
global epsilon
original_epsilon = epsilon
epsilon = 0
total_reward = 0
for _ in range(num_tests):
state = env.reset()[0]
done = False
episode_reward = 0
times_done = 0
while not done:
action_idx, action_cont = get_action(state)
state, reward, done, _, _ = env.step([action_cont])
episode_reward += reward
times_done += 1
if done or times_done >= 200:
break
total_reward += episode_reward
epsilon = original_epsilon
return total_reward / num_tests
开始训练
def train(total_epochs=200):
global epsilon
state = env.reset()[0]
training_loss = []
for epoch in range(total_epochs):
if len(replay_buffer) < batch_size:
state = env.reset()[0]
for _ in range(200):
action_idx, action_cont = get_action(state)
next_state, reward, done, _, _ = env.step([action_cont])
reward = reward / 10
add_experience(state, action_idx, reward, next_state, done)
state = next_state if not done else env.reset()[0]
if done:
break
continue
if epsilon > epsilon_min:
epsilon *= epsilon_decay
epoch_loss = []
for _ in range(200):
states, actions, rewards, next_states, dones = sample_experiences()
loss = train_step(states, actions, rewards, next_states, dones)
epoch_loss.append(loss)
action_idx, action_cont = get_action(state)
next_state, reward, done, _, _ = env.step([action_cont])
reward = reward / 10
add_experience(state, action_idx, reward, next_state, done)
state = next_state if not done else env.reset()[0]
if epoch % update_target_every == 0:
target_model.set_weights(model.get_weights())
if epoch % 20 == 0:
avg_reward = test_model()
print(
f"Epoch: {epoch}, Avg Reward: {avg_reward:.2f}, Loss: {np.mean(epoch_loss):.4f}, Epsilon: {epsilon:.3f}")
每20个epoch测试一下效果。每个epoch中执行200次 train_step。每10个epoch更新一次目标网络也就是2000步更新一次。
启动吧
train() test_model(num_tests=10) show()
遇到的问题
这个reward一直不收敛,效果一直很差
可能的原因。
问题:测试时未禁用探索,
解决:测试的时候禁用一下
问题:经验回放缓冲区过小。
解决:调大经验回放缓冲区
问题:学习率和训练步数不足。
解决:将学习率调整为 3e-4 或 5e-4,多试试。
问题:目标网络更新频率不当
解决:目标网络保持频率适中的更新大概在2000-4000步比较合理
问题:经验收集方式缺乏多样性
解决:在训练过程中持续收集经验,而不是在每个epoch开始时批量收集。
问题:Pendulum-v1的奖励范围较大(从-16.273到0),这可能导致Q值过大,训练不稳定。
解决:对奖励进行缩放,例如将奖励除以10,
在预填充经验池时也需要缩放奖励
只要奖励再200内就很不错了。
更多推荐

所有评论(0)