【强化学习小例子 DQN基于Pytorch】_pytorch 强化学习

前言

最近在学习强化学习，进行DQN算法实践过程中使用 gym 环境，训练模型完成一个小游戏。初始代码主要参考了：
DQN 模型解析，附Pytorch完整代码
然而，在运行过程中发现了一些错误，现在将我改后的代码分享出来，希望可以帮到大家。

---

一、主要代码

代码如下：

1.run_this

import gym
from RL_DQN import DQN, ReplayBuffer
import torch
from tqdm import tqdm
import matplotlib.pyplot as plt
import numpy as np
import random

# GPU运算
device = torch.device("cuda") if torch.cuda.is_available() \
        else torch.device("cpu")

# ------------------------------- #
# 全局变量
# ------------------------------- #


lr = 0.001  # 学习率
gamma = 0.95  # 折扣因子
epsilon = 0.1  # 贪心系数
target_update = 10  # 目标网络的参数的更新频率
batch_size = 64
capacity = 10000  # 经验池容量
min_size = 500  # 经验池超过200后再训练
n_hidden = 128  # 隐含层神经元个数
return_list = []  # 记录每个回合的回报

# 加载环境
env = gym.make("CartPole-v1", render_mode="human")
n_states = env.observation_space.shape[0]  # 4
n_actions = env.action_space.n  # 2

# 实例化经验池
replay_buffer = ReplayBuffer(capacity)
# 实例化DQN
agent = DQN(n_states=n_states,
            n_hidden=n_hidden,
            n_actions=n_actions,
            learning_rate=lr,
            gamma=gamma,
            epsilon=epsilon,
            target_update=target_update,
            device=device,
        )

# 训练模型
for i in range(100):  # 100回合
    # 打印训练进度，一共10回合
    with tqdm(total=10, desc='Iteration %d' % i) as pbar:
        for i_episode in range(10):
            # 每个回合开始前重置环境
            state = env.reset()[0]  # len=4
            # 记录每个回合的回报
            episode_return = 0
            done = False

            while True:
                # 获取当前状态下需要采取的动作
                action = agent.take_action(state)
                # 更新环境
                next_state, reward, done, _, _ = env.step(action)
                # 添加经验池
                replay_buffer.add(state, action, reward, next_state, done)
                # 更新当前状态
                state = next_state
                # 更新回合回报
                episode_return += reward

                # 当经验池超过一定数量后，训练网络
                if replay_buffer.size() > min_size:
                    # 从经验池中随机抽样作为训练集
                    s, a, r, ns, d = replay_buffer.sample(batch_size)
                    # 构造训练集
                    transition_dict = {
                        'states': s,
                        'actions': a,
                        'next_states': ns,
                        'rewards': r,
                        'dones': d,
                    }
                    # 网络更新
                    agent.update(transition_dict)
                # 找到目标就结束
                if done: break

            # 记录每个回合的回报
            return_list.append(episode_return)

            # 更新进度条信息
            pbar.set_postfix({
                'return': '%.3f' % return_list[-1]
            })
            pbar.update(1)

# 绘图
episodes_list = list(range(len(return_list)))
plt.plot(episodes_list, return_list)
plt.xlabel('Episodes')
plt.ylabel('Returns')
plt.title('DQN Returns')
plt.show()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102

2.RL_DQN

运行run_this时，不是从RL_DQN中引入了DQN, ReplayBuffer这俩Class嘛，所以以下代码必须命名为RL_DQN，或者改上一节的代码也行。

from torch import nn
from torch.nn import functional as F
import torch
import numpy as np
import random
import collections

# --------------------------------------- #
# 经验回放池
# --------------------------------------- #
class ReplayBuffer():
    def __init__(self, capacity):
        # 创建一个先进先出的队列，最大长度为capacity，保证经验池的样本量不变
        self.buffer = collections.deque(maxlen=capacity)

    # 将数据以元组形式添加进经验池
    def add(self, state, action, reward, next_state, done):
        self.buffer.append((state, action, reward, next_state, done))

    # 随机采样batch_size行数据
    def sample(self, batch_size):
        transitions = random.sample(self.buffer, batch_size)  # list, len=32
        # *transitions代表取出列表中的值，即32项
        state, action, reward, next_state, done = zip(*transitions)
        return np.array(state), action, reward, np.array(next_state), done

    # 目前队列长度
    def size(self):
        return len(self.buffer)

    # -------------------------------------- #
    # 构造深度学习网络，输入状态s，得到各个动作的reward
    # -------------------------------------- #


class Net(nn.Module):
    # 构造只有一个隐含层的网络
    def __init__(self, n_states, n_hidden, n_actions):
        super(Net, self).__init__()
        # [b,n_states]-->[b,n_hidden]
        self.fc1 = nn.Linear(n_states, n_hidden)
        # [b,n_hidden]-->[b,n_actions]
        self.fc2 = nn.Linear(n_hidden, n_actions)

    # 前传
    def forward(self, x):  # [b,n_states]
        # x = self.fc1(x)
        # x = self.fc2(x)
        # return x
        x = F.relu(self.fc1(x))  # 隐藏层使用ReLU激活函数
        return self.fc2(x)



# -------------------------------------- #
# 构造深度强化学习模型
# -------------------------------------- #

class DQN:
    # （1）初始化
    def __init__(self, n_states, n_hidden, n_actions,
                 learning_rate, gamma, epsilon,
                 target_update, device):
        # 属性分配
        self.n_states = n_states  # 状态的特征数
        self.n_hidden = n_hidden  # 隐含层个数
        self.n_actions = n_actions  # 动作数
        self.learning_rate = learning_rate  # 训练时的学习率
        self.gamma = gamma  # 折扣因子，对下一状态的回报的缩放
        self.epsilon = epsilon  # 贪婪策略，有1-epsilon的概率探索
        self.target_update = target_update  # 目标网络的参数的更新频率
        self.device = device  # 在GPU计算
        # 计数器，记录迭代次数
        self.count = 0

        # 构建2个神经网络，相同的结构，不同的参数
        # 实例化训练网络  [b,4]-->[b,2]  输出动作对应的奖励
        self.q_net = Net(self.n_states, self.n_hidden, self.n_actions).to(self.device)
        # 实例化目标网络
        self.target_q_net = Net(self.n_states, self.n_hidden, self.n_actions).to(self.device)

        # 优化器，更新训练网络的参数
        self.optimizer = torch.optim.Adam(self.q_net.parameters(), lr=self.learning_rate)

    # （3）网络训练
    def update(self, transition_dict):  # 传入经验池中的batch个样本
        # 获取当前时刻的状态 array_shape=[b,4]
        states = torch.tensor(transition_dict['states'], dtype=torch.float).to(self.device)
        # 获取当前时刻采取的动作 tuple_shape=[b]，维度扩充 [b,1]
        actions = torch.tensor(transition_dict['actions']).view(-1, 1).to(self.device)
        # 当前状态下采取动作后得到的奖励 tuple=[b]，维度扩充 [b,1]
        rewards = torch.tensor(transition_dict['rewards'], dtype=torch.float).view(-1, 1).to(self.device)
        # 下一时刻的状态 array_shape=[b,4]
        next_states = torch.tensor(transition_dict['next_states'], dtype=torch.float).to(self.device)
        # 是否到达目标 tuple_shape=[b]，维度变换[b,1]
        dones = torch.tensor(transition_dict['dones'], dtype=torch.float).view(-1, 1).to(self.device)

        # 输入当前状态，得到采取各运动得到的奖励 [b,4]==>[b,2]==>[b,1]
        # 根据actions索引在训练网络的输出的第1维度上获取对应索引的q值（state_value）
        q_values = self.q_net(states).gather(1, actions)  # [b,1]
        # 下一时刻的状态[b,4]-->目标网络输出下一时刻对应的动作q值[b,2]-->
        # 选出下个状态采取的动作中最大的q值[b]-->维度调整[b,1]
        max_next_q_values = self.target_q_net(next_states).max(1)[0].view(-1, 1)
        # 目标网络输出的当前状态的q(state_value)：即时奖励+折扣因子*下个时刻的最大回报
        q_targets = rewards + self.gamma * max_next_q_values * (1 - dones)

        # 目标网络和训练网络之间的均方误差损失
        dqn_loss = torch.mean(F.mse_loss(q_values, q_targets))
        # PyTorch中默认梯度会累积,这里需要显式将梯度置为0
        self.optimizer.zero_grad()
        # 反向传播参数更新
        dqn_loss.backward()
        # 对训练网络更新
        self.optimizer.step()

        # 在一段时间后更新目标网络的参数
        if self.count % self.target_update == 0:
            # 将目标网络的参数替换成训练网络的参数
            self.target_q_net.load_state_dict(
                self.q_net.state_dict())

        self.count += 1

    def take_action(self, state):  # epsilon-贪婪策略采取动作
        if np.random.random() < self.epsilon:  # 当随机数小于贪心系数，采取随机动作
            action = np.random.randint(self.n_actions)
        else:  # 否则采取Q网络输出最大值的动作，即动作价值最高的动作
            state = torch.tensor([state], dtype=torch.float).to(self.device)
            action = self.q_net(state).argmax().item()
        return action

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131

二、版本关系！！

我的各种包版本信息如下：
gym： 0.26.0
numpy： 1.26.3
pygame：2.5.2
torch： 1.12.1

其他包版本如下：

总结

• gym等包的版本信息一定要对应！！！对应不上会出现莫名其妙的Bug。
• 希望大家都能在强化学习领域从入门到秃头。
• 第一次发CSDN，求点赞。