代码注释解读说明，运行于notebook

首先我们安装依赖，避免报错

1. 更新并安装系统依赖，忽略符号链接警告

!apt update -o Dir::Etc::sourcelist="/etc/apt/sources.list.d/*.list" -o Dir::Etc::sourceparts="-" -y
!apt install -y libgl1-mesa-glx libosmesa6 libglew2.2 python3-opengl xvfb -o Dpkg::Options::="--force-overwrite"

2. 安装现代版依赖，彻底避免 setup.py 报错

!pip install gymnasium[box2d] pygame pyvirtualdisplay tqdm numpy torch

3.导入我们需要的库

from pyvirtualdisplay import Display
#这个库，常用来刷新强化学习，实时更新智能体玩游戏的画面
virtual_display = Display(visible=0, size=(1400, 900))
virtual_display.start()

%matplotlib inline
import matplotlib.pyplot as plt

from IPython import display

import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
from torch.distributions import Categorical
#表示概率分布
from tqdm.notebook import tqdm
 

4.固定随机种子，保证实验可复现

seed = 543 # Do not change this固定随机种子，保证实验可复现
def fix(env, seed):
  env.seed(seed)        #给环境本身设置种子
  env.action_space.seed(seed) #给动作空间设置种子
  torch.manual_seed(seed)   #给pytorch CPU算子设置种子
  torch.cuda.manual_seed(seed) #给pytorch 单GPU算子设置种子
  torch.cuda.manual_seed_all(seed)#Pytorch 所有GPU算子设置种子
  np.random.seed(seed)      #给Numpy 随机数生成器设置种子
  random.seed(seed)       #random 模块设置种子
  # torch.set_deterministic(True)
  torch.backends.cudnn.benchmark = False  

#关闭 cudnn 自动优化：避免不同硬件上的算法选择不同
  torch.backends.cudnn.deterministic = True

#强制cudnn 使用确定性算法：强制cuDNN使用确定性算法，保证卷积等操作结果完全可复现
 

%%capture
import gymnasium as gym   #强化学习的标准游乐场，提供了各种经典环境
import random
env = gym.make('LunarLander-v2')                    #创建月球着陆器环境
fix(env, seed)                                                   #调用fix函数，锁死随机数                

观测空间，这是一个八维向量连续空间，按照顺序，（0）火箭水平坐标范围，（1）垂直坐标范围，（2）水平速度，（3）垂直速度，（4）火箭倾斜角度，（5），倾斜角速度，（6）左腿是否接触地面，（7）右腿是否接触地面

print(env.observation_space)    

动作空间，是一个离散空间，四个操作，0不操作，1启动左侧引擎，2启动主引擎，3启动右引擎

print(env.action_space)  

重置环境，获得初始状态，每局游戏开始时，必须调用的函数，返回8维浮点数向量，代表火箭初始状态，是对观测空间的初始化

initial_state = env.reset()   

random_action = env.action_space.sample()   

#动作空间随机采样个动作
print(random_action)#输出0到3的动作

#observation：执行动作后新的八维状态向量; reward：这一步获得的即时奖励; done：bool表示这一局游戏是否结束; info：调试用的额外信息

observation, reward, done, info = env.step(random_action)

---

接下来我们看看样例：

env.reset()

img = plt.imshow(env.render(mode='rgb_array'))  

#将当前环境渲染成一个RGB图像数组

done = False
while not done:
    action = env.action_space.sample()    #四个动作里随机选一个，随机智能体的决策逻辑
    observation, reward, done, _ = env.step(action) 

    img.set_data(env.render(mode='rgb_array'))  #更新img对象的像素数据
    display.display(plt.gcf())                  

    #plt.gcf（）可获得当前plt图像，display将图像显示在notebook中
    display.clear_output(wait=True)       #清除上一步的输出然后等待新的图像显示

接下来我们开始定义模型：

class PolicyGradientNetwork(nn.Module):

    def __init__(self):
        super().__init__()
        self.fc1 = nn.Linear(8, 16)
        self.fc2 = nn.Linear(16, 16)
        self.fc3 = nn.Linear(16, 4)

    def forward(self, state):
        hid = torch.tanh(self.fc1(state))
        hid = torch.tanh(self.fc1(hid))
        return F.softmax(self.fc3(hid), dim=-1)

class PolicyGradientAgent():
    
    def __init__(self, network):
        self.network = network
        self.optimizer = optim.SGD(self.network.parameters(), lr=0.001)
        
    def forward(self, state):
        return self.network(state)

    def sample(self, state):
        action_prob = self.network(torch.FloatTensor(state))  

        #将状态输出网络中，得到四个动作的决策概率
        action_dist = Categorical(action_prob)         

        #把概率向量包装成了一个分布，方便按概率采样
        action = action_dist.sample()             

        #根据概率分布随机选动作，概率越高越可能被选中
        log_prob = action_dist.log_prob(action)         

#已经抽好的动作，原来的的概率取对数。这个对数概率值，目的是为了参与损失计算，对数概率的数值稳定性更强
        return action.item(), log_prob

    def learn(self, log_probs, rewards):
        loss = (-log_probs * rewards).sum()  

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

network = PolicyGradientNetwork()
agent = PolicyGradientAgent(network)

开始训练，大体上与深度学习训练步骤相同

agent.network.train() 
EPISODE_PER_BATCH = 5  

# 每收集五局游戏的数据，更新一次智能体
NUM_BATCH = 500        

# 总共更新500次，500x5=2500局游戏

avg_total_rewards, avg_final_rewards = [], []

#记录每一批（5）局的平均奖励，total是单批总奖励平均值，单批最后一步奖励平均值（降落或坠毁）

prg_bar = tqdm(range(NUM_BATCH))  

#进度条显示
for batch in prg_bar:

    log_probs, rewards = [], []       #存储每一步的对数概率与奖励
    total_rewards, final_rewards = [], []   #总奖励和最后的奖励

    # collect trajectory
    for episode in range(EPISODE_PER_BATCH):
        
        state = env.reset()
        total_reward, total_step = 0, 0
        seq_rewards = []
        while True:

            action, log_prob = agent.sample(state) 
            next_state, reward, done, _ = env.step(action)

            log_probs.append(log_prob) 
            # seq_rewards.append(reward)
            state = next_state
            total_reward += reward
            total_step += 1
            rewards.append(reward) # change here
            if done:
                final_rewards.append(reward)
                total_rewards.append(total_reward)
                
                break

    print(f"rewards looks like ", np.shape(rewards))  

    avg_total_reward = sum(total_rewards) / len(total_rewards)
    avg_final_reward = sum(final_rewards) / len(final_rewards)

    avg_total_rewards.append(avg_total_reward)

    avg_final_rewards.append(avg_final_reward)

    prg_bar.set_description(f"Total: {avg_total_reward: 4.1f}, Final: {avg_final_reward: 4.1f}")

    rewards = (rewards - np.mean(rewards)) / (np.std(rewards) + 1e-9)  

   # 奖励归一化，均值为0，方差唯一，能让训练更稳定

    agent.learn(torch.stack(log_probs), torch.from_numpy(rewards))      

    #将对数概率张量堆叠成一个二维张量<-->[tensor(...),tensor(...)...]-->tensor([.........])

    print("logs prob looks like ", torch.stack(log_probs).size())

    print("torch.from_numpy(rewards) looks like ", torch.from_numpy(rewards).size())

看一下总奖励

plt.plot(avg_total_rewards)
plt.title("Total Rewards")
plt.show()

看一下最后一次奖励，意味着着陆的身位

plt.plot(avg_final_rewards)
plt.title("Final Rewards")
plt.show()

最后我们开始测试

fix(env, seed)
agent.network.eval()  
NUM_OF_TEST = 5 
test_total_reward = []
action_list = []
for i in range(NUM_OF_TEST):
  actions = []
  state = env.reset()

  img = plt.imshow(env.render(mode='rgb_array'))

  total_reward = 0

  done = False
  while not done:
      action, _ = agent.sample(state)
      actions.append(action)
      state, reward, done, _ = env.step(action)

      total_reward += reward

      img.set_data(env.render(mode='rgb_array'))
      display.display(plt.gcf())
      display.clear_output(wait=True)
      
  print(total_reward)
  test_total_reward.append(total_reward)

  action_list.append(actions)