PyTorch 深度学习实战（11）：强化学习与深度 Q 网络（DQN）

news2025/3/17 23:44:32

在之前的文章中，我们介绍了神经网络、卷积神经网络（CNN）、循环神经网络（RNN）、Transformer 等多种深度学习模型，并应用于图像分类、文本分类、时间序列预测等任务。本文将介绍强化学习的基本概念，并使用 PyTorch 实现一个经典的深度 Q 网络（DQN）来解决强化学习中的经典问题——CartPole。

一、强化学习基础

强化学习（Reinforcement Learning, RL）是机器学习的一个重要分支，它通过智能体（Agent）与环境（Environment）的交互来学习策略，以最大化累积奖励。强化学习的核心思想是通过试错来学习，智能体在环境中采取行动，观察结果，并根据奖励信号调整策略。

1. 强化学习的基本要素

智能体（Agent）：学习并做出决策的主体。
环境（Environment）：智能体交互的外部世界。
状态（State）：环境在某一时刻的描述。
动作（Action）：智能体在某一状态下采取的行动。
奖励（Reward）：智能体采取动作后，环境返回的反馈信号。
策略（Policy）：智能体在给定状态下选择动作的规则。
价值函数（Value Function）：评估在某一状态下采取某一动作的长期回报。

2. Q-Learning 与深度 Q 网络（DQN）

Q-Learning 是一种经典的强化学习算法，它通过学习一个 Q 函数来评估在某一状态下采取某一动作的长期回报。Q 函数的更新公式为：

深度 Q 网络（DQN）将 Q-Learning 与深度学习结合，使用神经网络来近似 Q 函数。DQN 通过经验回放（Experience Replay）和目标网络（Target Network）来稳定训练过程。

二、CartPole 问题实战

CartPole 是强化学习中的经典问题，目标是控制一个小车（Cart）使其上的杆子（Pole）保持直立。我们将使用 PyTorch 实现一个 DQN 来解决这个问题。

1. 问题描述

CartPole 环境的状态空间包括小车的位置、速度、杆子的角度和角速度。动作空间包括向左或向右移动小车。智能体每保持杆子直立一步，就会获得 +1 的奖励，当杆子倾斜超过一定角度或小车移动超出范围时，游戏结束。

2. 实现步骤

安装并导入必要的库。
定义 DQN 模型。
定义经验回放缓冲区。
定义 DQN 训练过程。
测试模型并评估性能。

3. 代码实现

import gym
import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
import random
from collections import deque
import matplotlib.pyplot as plt

# 设置 Matplotlib 支持中文显示
plt.rcParams['font.sans-serif'] = ['SimHei']  # 设置字体为 SimHei（黑体）
plt.rcParams['axes.unicode_minus'] = False  # 解决负号显示问题

# 1. 安装并导入必要的库
env = gym.make('CartPole-v1')

# 2. 定义 DQN 模型
class DQN(nn.Module):
    def __init__(self, state_size, action_size):
        super(DQN, self).__init__()
        self.fc1 = nn.Linear(state_size, 64)
        self.fc2 = nn.Linear(64, 64)
        self.fc3 = nn.Linear(64, action_size)

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

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

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

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

    def __len__(self):
        return len(self.buffer)

# 4. 定义 DQN 训练过程
state_size = env.observation_space.shape[0]
action_size = env.action_space.n
model = DQN(state_size, action_size)
target_model = DQN(state_size, action_size)
target_model.load_state_dict(model.state_dict())
optimizer = optim.Adam(model.parameters(), lr=0.001)
buffer = ReplayBuffer(10000)

def train(batch_size, gamma=0.99):
    if len(buffer) < batch_size:
        return
    state, action, reward, next_state, done = buffer.sample(batch_size)
    state = torch.FloatTensor(state)
    next_state = torch.FloatTensor(next_state)
    action = torch.LongTensor(action)
    reward = torch.FloatTensor(reward)
    done = torch.FloatTensor(done)

    q_values = model(state)
    next_q_values = target_model(next_state)
    q_value = q_values.gather(1, action.unsqueeze(1)).squeeze(1)
    next_q_value = next_q_values.max(1)[0]
    expected_q_value = reward + gamma * next_q_value * (1 - done)

    loss = nn.MSELoss()(q_value, expected_q_value.detach())
    optimizer.zero_grad()
    loss.backward()
    optimizer.step()

# 5. 测试模型并评估性能
def test(env, model, episodes=10):
    total_reward = 0
    for _ in range(episodes):
        state = env.reset()
        done = False
        while not done:
            state = torch.FloatTensor(state).unsqueeze(0)
            action = model(state).max(1)[1].item()
            next_state, reward, done, _ = env.step(action)
            total_reward += reward
            state = next_state
    return total_reward / episodes

# 训练过程
episodes = 500
batch_size = 64
gamma = 0.99
epsilon = 1.0
epsilon_min = 0.01
epsilon_decay = 0.995
rewards = []

for episode in range(episodes):
    state = env.reset()
    done = False
    total_reward = 0

    while not done:
        if random.random() < epsilon:
            action = env.action_space.sample()
        else:
            state_tensor = torch.FloatTensor(state).unsqueeze(0)
            action = model(state_tensor).max(1)[1].item()

        next_state, reward, done, _ = env.step(action)
        buffer.push(state, action, reward, next_state, done)
        state = next_state
        total_reward += reward

        train(batch_size, gamma)

    epsilon = max(epsilon_min, epsilon * epsilon_decay)
    rewards.append(total_reward)

    if (episode + 1) % 50 == 0:
        avg_reward = test(env, model)
        print(f"Episode: {episode + 1}, Avg Reward: {avg_reward:.2f}")

# 6. 可视化训练结果
plt.plot(rewards)
plt.xlabel("Episode")
plt.ylabel("Total Reward")
plt.title("DQN 训练过程")
plt.show()