dqn python代码

以下是DQN算法的Python实现（使用TensorFlow）：

import tensorflow as tf
import numpy as np
import gym

# 创建游戏环境
env = gym.make('CartPole-v0')

# 定义超参数
learning_rate = 0.01
discount_factor = 0.99
batch_size = 64
memory_size = 10000
epsilon = 1.0
epsilon_decay = 0.9995
min_epsilon = 0.01
n_episodes = 2000
n_steps_per_episode = 200
update_target_network_every = 100

# 创建经验回放存储器
memory = []

# 定义神经网络
class DQN(tf.keras.Model):
    def __init__(self, n_actions):
        super(DQN, self).__init__()
        self.dense1 = tf.keras.layers.Dense(32, activation='relu')
        self.dense2 = tf.keras.layers.Dense(32, activation='relu')
        self.dense3 = tf.keras.layers.Dense(n_actions)

    def call(self, inputs):
        x = self.dense1(inputs)
        x = self.dense2(x)
        x = self.dense3(x)
        return x

# 创建主网络和目标网络
n_actions = env.action_space.n
main_network = DQN(n_actions)
target_network = DQN(n_actions)
target_network.set_weights(main_network.get_weights())

# 定义损失函数和优化器
loss_fn = tf.keras.losses.MeanSquaredError()
optimizer = tf.keras.optimizers.Adam(learning_rate=learning_rate)

# 定义动作选择策略
def choose_action(state, epsilon):
    if np.random.rand() < epsilon:
        return env.action_space.sample()
    else:
        Q_values = main_network(tf.constant([state], dtype=tf.float32)).numpy()[0]
        return np.argmax(Q_values)

# 从经验回放存储器中随机采样一个批次进行训练
def train():
    batch = np.array(memory)[np.random.choice(len(memory), batch_size, replace=False)]
    states = np.array([b[0] for b in batch])
    actions = np.array([b[1] for b in batch])
    rewards = np.array([b[2] for b in batch])
    next_states = np.array([b[3] for b in batch])
    dones = np.array([b[4] for b in batch])
    Q_values_next = target_network(next_states).numpy()
    Q_values_target = rewards + discount_factor * np.max(Q_values_next, axis=1) * (1 - dones)
    Q_values = main_network(states).numpy()
    Q_values[np.arange(batch_size), actions] = Q_values_target
    with tf.GradientTape() as tape:
        loss = loss_fn(Q_values, main_network(states))
    grads = tape.gradient(loss, main_network.trainable_variables)
    optimizer.apply_gradients(zip(grads, main_network.trainable_variables))

# 训练主网络
total_reward_history = []
for episode in range(n_episodes):
    state = env.reset()
    total_reward = 0
    for step in range(n_steps_per_episode):
        epsilon = max(epsilon * epsilon_decay, min_epsilon)
        action = choose_action(state, epsilon)
        next_state, reward, done, _ = env.step(action)
        memory.append((state, action, reward, next_state, done))
        total_reward += reward
        state = next_state
        if len(memory) > memory_size:
            memory.pop(0)
        if episode % update_target_network_every == 0:
            target_network.set_weights(main_network.get_weights())
        if done:
            break
    if len(memory) > batch_size:
        train()
    total_reward_history.append(total_reward)
    print(f'Episode {episode} completed with total reward of {total_reward}.')