强化学习算法DQN实现

DQN的基本思想

Q学习：Q学习是一种基于值函数的强化学习方法，目的是通过学习状态-动作值函数Q(s, a)来指导智能体的动作选择。Q函数表示在状态s采取动作a后能够获得的期望总回报。
深度神经网络：使用深度神经网络来近似Q函数。输入是状态s，输出是每个动作的Q值。神经网络的参数通过与目标Q值的均方误差（MSE）损失函数进行反向传播来更新。
经验回放：经验回放机制用于解决样本之间的相关性问题。通过存储智能体的经验（状态，动作，奖励，下一个状态，是否终止）到回放池中，并从中随机抽取小批量样本进行训练，打破了样本之间的相关性，提高了样本利用效率。
目标网络：为了增强训练的稳定性，DQN引入了目标网络。目标网络的结构和参数与Q网络相同，但参数更新频率较低。目标Q值使用目标网络来计算，避免了训练过程中参数震荡的问题。

详细的训练过程

初始化：初始化Q网络和目标网络，设置超参数和经验回放池。
交互环境：在每一回合中，智能体根据当前策略与环境进行交互，选择动作并获得奖励，存储经验到回放池中。
经验采样：当回放池中的经验数量足够时，从中随机抽取一个小批量样本用于训练。
计算目标Q值：使用目标网络计算目标Q值，对于每个样本，目标Q值等于即时奖励加上下一状态的最大Q值乘以折扣因子。
更新Q网络：通过最小化预测Q值和目标Q值之间的均方误差来更新Q网络的参数。
更新目标网络：每隔一段时间，将Q网络的参数复制到目标网络中。
探索与利用：采用ε-greedy策略选择动作，即以ε的概率随机选择动作，以1-ε的概率选择当前Q网络认为最优的动作。随着训练的进行，ε逐渐减小，以增加利用率。
训练结束：在达到设定的回合数后，结束训练过程。

import gym
import numpy as np
import tensorflow as tf
from tensorflow.keras import layers# 环境设置
env = gym.make('CartPole-v1')# 超参数设置
gamma = 0.99  # 折扣因子
epsilon = 1.0  # 探索率
epsilon_min = 0.01  # 最小探索率
epsilon_decay = 0.995  # 探索率衰减
learning_rate = 0.001  # 学习率
batch_size = 64  # 批量大小
memory_size = 2000  # 经验回放池大小# 构建Q网络
def build_model(state_shape, action_size):model = tf.keras.Sequential()model.add(layers.Dense(24, input_shape=state_shape, activation='relu'))model.add(layers.Dense(24, activation='relu'))model.add(layers.Dense(action_size, activation='linear'))model.compile(loss='mse', optimizer=tf.keras.optimizers.Adam(lr=learning_rate))return model# 经验回放池
class ReplayMemory:def __init__(self, max_size):self.buffer = []self.max_size = max_sizedef add(self, experience):if len(self.buffer) >= self.max_size:self.buffer.pop(0)self.buffer.append(experience)def sample(self, batch_size):idx = np.random.choice(len(self.buffer), size=batch_size, replace=False)return [self.buffer[i] for i in idx]# 训练DQN
def train_dqn(episodes):state_size = env.observation_space.shape[0]action_size = env.action_space.nmodel = build_model((state_size,), action_size)target_model = build_model((state_size,), action_size)target_model.set_weights(model.get_weights())memory = ReplayMemory(memory_size)for episode in range(episodes):state = env.reset()state = np.reshape(state, [1, state_size])total_reward = 0while True:if np.random.rand() <= epsilon:action = np.random.choice(action_size)else:q_values = model.predict(state)action = np.argmax(q_values[0])next_state, reward, done, _ = env.step(action)next_state = np.reshape(next_state, [1, state_size])total_reward += rewardmemory.add((state, action, reward, next_state, done))state = next_stateif done:print(f"Episode: {episode + 1}, Reward: {total_reward}, Epsilon: {epsilon:.2f}")breakif len(memory.buffer) >= batch_size:experiences = memory.sample(batch_size)states, actions, rewards, next_states, dones = zip(*experiences)states = np.vstack(states)next_states = np.vstack(next_states)q_values = model.predict_on_batch(states)q_values_next = target_model.predict_on_batch(next_states)for i in range(batch_size):q_values[i][actions[i]] = rewards[i] + (1 - dones[i]) * gamma * np.amax(q_values_next[i])model.train_on_batch(states, q_values)if epsilon > epsilon_min:epsilon *= epsilon_decayif (episode + 1) % 10 == 0:target_model.set_weights(model.get_weights())train_dqn(500)
env.close()