DQN pytorch 代码

DQN（Deep Q-Network）是一种基于深度学习的强化学习算法，用于解决离散动作空间的强化学习问题。它的核心思想是使用一个神经网络来估计每个状态下各个动作的价值，并根据这些价值来选择最优的动作。

在PyTorch中实现DQN算法的代码主要包括以下几个步骤：

1. 定义神经网络模型：使用PyTorch构建一个深度神经网络模型，通常是一个多层的全连接神经网络。该模型将输入状态作为输入，输出每个动作的Q值。

2. 定义经验回放缓存：为了减小样本之间的相关性，DQN使用经验回放缓存来存储智能体与环境交互的经验数据。可以使用PyTorch提供的`torch.utils.data.Dataset`来实现。

3. 定义损失函数和优化器：使用均方误差（MSE）作为损失函数，用于衡量预测的Q值与目标Q值之间的差异。同时选择一个优化器，如Adam或SGD，来更新神经网络的参数。

4. 定义训练过程：在每个时间步，从经验回放缓存中随机采样一批经验数据，然后计算当前状态下每个动作的Q值。根据贪婪策略或ε-greedy策略选择动作，并与环境进行交互，得到下一个状态、奖励和是否终止的信息。根据Bellman方程计算目标Q值，并更新神经网络的参数。

5. 进行训练：重复执行训练过程，直到达到预设的停止条件，如达到最大训练步数或达到目标性能。

下面是一个简单的DQN算法的PyTorch代码示例：

import torch
import torch.nn as nn
import torch.optim as optim
import random
from collections import deque

# 定义神经网络模型
class DQN(nn.Module):
    def __init__(self, input_dim, output_dim):
        super(DQN, self).__init__()
        self.fc1 = nn.Linear(input_dim, 64)
        self.fc2 = nn.Linear(64, 64)
        self.fc3 = nn.Linear(64, output_dim)

    def forward(self, x):
        x = torch.relu(self.fc1(x))
        x = torch.relu(self.fc2(x))
        x = self.fc3(x)
        return x

# 定义经验回放缓存
class ReplayBuffer:
    def __init__(self, capacity):
        self.buffer = deque(maxlen=capacity)

    def push(self, state, action, reward, next_state, done):
        experience = (state, action, reward, next_state, done)
        self.buffer.append(experience)

    def sample(self, batch_size):
        batch = random.sample(self.buffer, batch_size)
        state, action, reward, next_state, done = zip(*batch)
        return state, action, reward, next_state, done

# 定义训练过程
def train(model, target_model, replay_buffer, batch_size, gamma, optimizer, loss_fn):
    state, action, reward, next_state, done = replay_buffer.sample(batch_size)

    state = torch.tensor(state, dtype=torch.float32)
    action = torch.tensor(action, dtype=torch.long)
    reward = torch.tensor(reward, dtype=torch.float32)
    next_state = torch.tensor(next_state, dtype=torch.float32)
    done = torch.tensor(done, dtype=torch.float32)

    q_values = model(state).gather(1, action.unsqueeze(1)).squeeze(1)
    next_q_values = target_model(next_state).max(1)
    expected_q_values = reward + gamma * next_q_values * (1 - done)

    loss = loss_fn(q_values, expected_q_values.detach())

    optimizer.zero_grad()
    loss.backward()
    optimizer.step()

# 创建DQN模型和目标网络
input_dim = 4
output_dim = 2
model = DQN(input_dim, output_dim)
target_model = DQN(input_dim, output_dim)
target_model.load_state_dict(model.state_dict())
target_model.eval()

# 定义超参数和优化器
capacity = 10000
batch_size = 64
gamma = 0.99
lr = 0.001
optimizer = optim.Adam(model.parameters(), lr=lr)
loss_fn = nn.MSELoss()

# 创建经验回放缓存
replay_buffer = ReplayBuffer(capacity)

# 训练DQN模型
for episode in range(num_episodes):
    state = env.reset()
    done = False
    total_reward = 0

    while not done:
        action = select_action(state)
        next_state, reward, done, _ = env.step(action)
        replay_buffer.push(state, action, reward, next_state, done)
        state = next_state
        total_reward += reward

        if len(replay_buffer.buffer) > batch_size:
            train(model, target_model, replay_buffer, batch_size, gamma, optimizer, loss_fn)

    if episode % target_update == 0:
        target_model.load_state_dict(model.state_dict())

    print(f"Episode: {episode}, Total Reward: {total_reward}")

# 使用训练好的模型进行预测
state = env.reset()
done = False

while not done:
    action = model(torch.tensor(state, dtype=torch.float32)).argmax().item()
    next_state, reward, done, _ = env.step(action)
    state = next_state