Paddle强化学习 智能体玩俄罗斯方块

俄罗斯方块应该是80后90后很熟悉的小游戏，记忆中充满了各种小游戏中俄罗斯方块的旋律。近期想入门强化学习，突发奇想，试试智能体来盘俄罗斯方块。

b站视频链接

声明：本项目代码借鉴了https://github.com/uvipen/Tetris-deep-Q-learning-pytorch 项目，将pytorch代码迁移到了Paddle下，并进行训练。模型与训练架构自2013年DeepMind发布的《Playing Atari with Deep Reinforcement Learning》论文。

1、DQN(Deep Q Network)

Q-learning算法，本质上是一个决策选择，解决的是在当前状态S，应该采取动作A，达到最大收益R。

DQN是用深度神经网络与Q-learning算法相结合的产物。在《Playing Atari with Deep Reinforcement Learning》论文中，将RL与DL实现了融合，提出了存储记忆（Experience Replay）机制和Fixed-Q-Target，实现了一部分Atari游戏操控，甚至超过了人类水平。

2、代码实现与详解

本章内容主要对DQN部分进行分析

游戏环境部分不进行表述，感兴趣的同学可以直接看代码：tetris.py。

# 定义网络，只有三层，实验证明够用了
import paddle.nn as nn

class DeepQNetwork(nn.Layer):
    def __init__(self):
        super(DeepQNetwork, self).__init__()

        self.conv1 = nn.Sequential(nn.Linear(4, 64), nn.ReLU())
        self.conv2 = nn.Sequential(nn.Linear(64, 64), nn.ReLU())
        self.conv3 = nn.Sequential(nn.Linear(64, 1))

    def forward(self, x):
        x = self.conv1(x)
        x = self.conv2(x)
        x = self.conv3(x)

        return x

# 这一部分定义了引用库与输入的函数，变量暂时不做解释，将会在代码中详解
import argparse
import os
import shutil
from random import random, randint, sample

import numpy as np
import paddle
import paddle.nn as nn
from visualdl import LogWriter

from tetris import Tetris
from collections import deque


def get_args():
    parser = argparse.ArgumentParser(
        """Implementation of Deep Q Network to play Tetris""")
    parser.add_argument("--width", type=int, default=10, help="The common width for all images")
    parser.add_argument("--height", type=int, default=20, help="The common height for all images")
    parser.add_argument("--block_size", type=int, default=30, help="Size of a block")
    parser.add_argument("--batch_size", type=int, default=512, help="The number of images per batch")
    parser.add_argument("--lr", type=float, default=1e-3)
    parser.add_argument("--gamma", type=float, default=0.99)
    parser.add_argument("--initial_epsilon", type=float, default=1)
    parser.add_argument("--final_epsilon", type=float, default=1e-3)
    parser.add_argument("--num_decay_epochs", type=float, default=2000)
    parser.add_argument("--num_epochs", type=int, default=3000)
    parser.add_argument("--save_interval", type=int, default=10)
    parser.add_argument("--replay_memory_size", type=int, default=300,
                        help="Number of epoches between testing phases")
    parser.add_argument("--log_path", type=str, default="vdllog")
    parser.add_argument("--saved_path", type=str, default="trained_models")

    args = parser.parse_args()
    return args

def train(opt):

    # 设置DVL的log存储路径，可视化训练过程
    if os.path.isdir(opt.log_path):
        shutil.rmtree(opt.log_path)
    os.makedirs(opt.log_path)

    # 实例化环境,俄罗斯方块游戏的宽与高,方块大小
    env = Tetris(width=opt.width, height=opt.height, block_size=opt.block_size)

    # 实例化网络,这个网络是整个实验的核心,在训练中输出动作,指导游戏进行
    model = DeepQNetwork()

    # 设置学习率,优化器等
    lr = opt.lr
    optimizer = paddle.optimizer.Adam(parameters=model.parameters(), learning_rate=lr)
    criterion = nn.MSELoss()

    # 这里初始化游戏环境,相当于新建一局游戏
    state = env.reset()

    # 这个replay_memory是关键点,相当于记忆库,提供给网络进行学习
    # 记忆库的设计非常重要,要有一定的随机动作来走出局部最优解,也要有一些高得分的动作加速网络收敛
    # replay_memory_size是设置记忆库的大小,个人感觉大一点好
    replay_memory = deque(maxlen=opt.replay_memory_size)
    epoch = 0
    with LogWriter(logdir=opt.log_path) as writer:
        while epoch < opt.num_epochs:
            # 开始训练,env.get_next_states(),游戏就会进入下一步,在游戏中表现为在最上方给出一个方块
            next_steps = env.get_next_states()
            # Exploration or exploitation,这是文章的第二个关键点,控制replay_memory中随机动作与网络输出动作的比值
            # 一开始由于网络随机初始化不具备能力,随机动作较多,随着训练网络能力增强,随机动作减少,但一直保持一定的比例,来避免陷入局部最优无法跳出
            epsilon = opt.final_epsilon + (max(opt.num_decay_epochs - epoch, 0) * (
                    opt.initial_epsilon - opt.final_epsilon) / opt.num_decay_epochs)
            u = random()
            random_action = u <= epsilon


            # 打包动作与状态
            next_actions, next_states = zip(*next_steps.items())
            next_states = paddle.stack(next_states)

            
            # 执行随机动作,或执行网络输出的动作
            if random_action:
                index = randint(0, len(next_steps) - 1)
            else:
                model.eval()
                with paddle.no_grad():
                    predictions = model(next_states)[:, 0]
                index = paddle.argmax(predictions).item()

            
            # 执行动作,获取下一个状态,获取reward,获取游戏是否结束
            next_state = next_states[index, :]
            action = next_actions[index]
            reward, done = env.step(action, render=True)

            # 至此,获取了强化学习所需的所有要素[state, reward, next_state, done],放入replay_memory中,作为训练材料备用
            replay_memory.append([state, reward, next_state, done])
            if done:
                final_score = env.score
                final_tetrominoes = env.tetrominoes
                final_cleared_lines = env.cleared_lines
                state = env.reset()
            else:
                state = next_state
                continue
            # 这一句是判断replay_memory中的素材是否足够,等足够再开始训练,避免数据量太小无法挖掘出可用的知识
            if len(replay_memory) < opt.replay_memory_size / 10:
                print(len(replay_memory)/(opt.replay_memory_size / 10))
                continue
            # 开始训练
            model.train()
            epoch += 1
            # 采样,转张量
            batch = sample(replay_memory, min(len(replay_memory), opt.batch_size))
            state_batch, reward_batch, next_state_batch, done_batch = zip(*batch)
            state_batch = paddle.stack(tuple(state for state in state_batch))
            reward_batch = paddle.to_tensor(np.array(reward_batch, dtype=np.float32)[:, None])
            next_state_batch = paddle.stack(tuple(state for state in next_state_batch))

            # 状态state_batch进入网络,估算q_values 
            q_values = model(state_batch)

            # 预测下一个状态
            model.eval()
            with paddle.no_grad():
                next_prediction_batch = model(next_state_batch)
            model.train()

            # 理解了这一句,就理解了整个DQN算法,这一段在下一个Markdown里面详解
            y_batch = paddle.concat(
                tuple(reward if done else reward + opt.gamma * prediction for reward, done, prediction in
                      zip(reward_batch, done_batch, next_prediction_batch)))[:, None]

            # 计算损失,梯度回传,完成一步训练
            optimizer.clear_grad()
            loss = criterion(q_values, y_batch)
            loss.backward()
            optimizer.step()

            # 打印指标            
            print("Epoch: {}/{}, Action: {}, Score: {}, Tetrominoes {}, Cleared lines: {}".format(
                epoch,
                opt.num_epochs,
                action,
                final_score,
                final_tetrominoes,
                final_cleared_lines))

            # 写如log日志
            writer.add_scalar(tag="Train/Score", step=epoch-1, value=final_score)
            writer.add_scalar(tag="Train/Tetrominoes", step=epoch-1, value=final_tetrominoes)
            writer.add_scalar(tag="Train/Cleared lines", step=epoch-1, value=final_cleared_lines)

            if epoch > 0 and epoch % opt.save_interval == 0:
                paddle.save(model.state_dict(), "{}/tetris_{}".format(opt.saved_path, epoch))

    paddle.save(model.state_dict(), "{}/tetris".format(opt.saved_path))

3、算法思路的去公式化解释

为了避免同学们失去阅读兴趣，在这里不用任何公式，通过文字表示意图。

希望大家能愉快的意会，然后有需要的同学自己去查阅公式既可。

  • 关键点1：Q函数

通俗的来说，Q函数就是反复执行一个策略，得到最后的游戏得分

输入一个状态s1进入策略，策略输出一个动作a1，环境执行a1后，得到状态S2

（比如怪物有100点血量，这是S1；输入策略，策略说盘他，输出攻击动作，就是a1；游戏收到a1，玩家打了怪物一下，扣血1点，就进入了下一个状态上s2，在s2中，怪物还有99点血；过程中怪物掉血1点，就是得分）

迭代执行这一过程 s1执行a1可得s2，r1，s2执行a2，可得s3，r2，直到游戏结束

Q应该是所有r的和

（这里面还有一个衰减的参数，越往后的动作与当前状态相关性越小，所以贡献越小。这里不展开，因为这个参数不影响大家对算法的理解。有需要的同学请自行查阅。）

  • 关键点2：训练逻辑
  1. 随机初始化神经网络net
  2. 在状态s下执行行动a
  3. 得到奖励 r r r和新的状态s′
  4. 使用神经网络net计算r+Qs′
  5. 使用神经网络net计算Qs
  6. 以缩小r+Qs′与Qs为优化目标，计算L2损失
  7. 梯度回传优化
  8. 重复2。

在过程中，第四部和第五步是最巧妙的地方。我们的目标是令Qs = r+ Qs′

若等式成立，则代表着，Q可以完美预测s状态下执行a所得到的的r。（在这里只是逼近，并不是等于）所以Q可以知道在s状态下，哪一个动作有概率得到最高的得分。

所以在预测过程中，我们只需要执行最高得分概率的动作既可。

4、其他细节

由于AIStudio没有图形化界面，所以无法直接运行test.py，同理训练顾及也不行，会报错 cannot connect to X server

如果只是想玩玩感受下，建议下载压缩包Tetris-deep-Q-learning.zip（包含模型权重不到23k），解压后既可畅玩（需要paddle、cv2、PIL、matplotlib），运行test.py得到本项目开场的效果。

如果想自己训练，给出以下提示：

  1. 刚开始训练，会运行三万场游戏，填满记忆库，所以一开始有一段时间是不会训练模型的，大家只会看随机动作操作下到疯狂下落的方块。
  2. 记忆库填满后，一开始也不会有效果，原因是会有较大的概率执行随机动作，直到2000场之后，随机动作被压缩到千分之一概率，游戏画风开始趋近正常。
  3. 我训练最高的得分是十七万分，一局游戏玩了好久。

此文仅为搬运，原作链接：https://aistudio.baidu.com/aistudio/projectdetail/4337197

# paddle # python # 深度学习
Logo
百度飞桨AI Studio社区

学大模型，用大模型上飞桨星河社区！每天8点V100G算力免费领！免费领取ERNIE 4.0 100w Token >>>

更多推荐

cover

利用Amazon Bedrock畅玩Claude 3等多种领先模型，抢占AI高地(体验倒计时4小时)

avatar 百度飞桨星河社区
cover

LangChain的RAG实践

avatar 百度飞桨星河社区
cover

RAG综述汇总

avatar 百度飞桨星河社区