概述

Deep Q-Learning（深度 Q 學(xué)習(xí)）是一種強化學(xué)習(xí)算法，用于解決決策問題，其中代理（agent）通過學(xué)習(xí)在不同環(huán)境中采取行動來最大化累積獎勵。Lunar Lander 是一個經(jīng)典的強化學(xué)習(xí)問題，其中代理的任務(wù)是控制一個著陸艙在月球表面著陸，最小化著陸過程中的燃料消耗。

以下是使用 Deep Q-Learning 解決 Lunar Lander 問題的基本步驟：

1. 環(huán)境建模： 首先，需要對 Lunar Lander 環(huán)境進行建模。這包括定義狀態(tài)空間、動作空間、獎勵函數(shù)等。在 Lunar Lander 中，狀態(tài)可以包括著陸艙的位置、速度、角度等信息，動作可以是推力引擎的火力等。

2. 深度 Q 網(wǎng)絡(luò)： 創(chuàng)建一個深度神經(jīng)網(wǎng)絡(luò)，該網(wǎng)絡(luò)將輸入狀態(tài)，并輸出每個可能動作的 Q 值。Q 值表示在給定狀態(tài)下采取某個動作的累積獎勵的估計。網(wǎng)絡(luò)的目標(biāo)是通過學(xué)習(xí)調(diào)整 Q 值以最大化累積獎勵。

3. 經(jīng)驗回放： 使用經(jīng)驗回放機制來改善學(xué)習(xí)穩(wěn)定性。這涉及到存儲代理先前的經(jīng)驗，并從中隨機抽樣進行訓(xùn)練。這有助于解決樣本相關(guān)性問題，提高算法的收斂性。

4. ε-貪婪策略： 引入 ε-貪婪策略，以平衡探索和利用。在一部分情況下，代理將以高概率選擇當(dāng)前認(rèn)為最佳的動作（貪婪），而在其他情況下，它會以較高概率選擇一個隨機動作（探索）。

5. 目標(biāo) Q 值計算： 使用目標(biāo) Q 值來更新網(wǎng)絡(luò)參數(shù)。目標(biāo) Q 值通過將當(dāng)前狀態(tài)的即時獎勵與下一個狀態(tài)的最大 Q 值相結(jié)合得到。這有助于更有效地傳播獎勵信號。

6. 訓(xùn)練： 通過與環(huán)境的交互，不斷地更新深度 Q 網(wǎng)絡(luò)的參數(shù)。代理通過學(xué)習(xí)來優(yōu)化其行為，以最大化預(yù)期的累積獎勵。

7. 調(diào)優(yōu)： 對算法的超參數(shù)進行調(diào)優(yōu)，包括學(xué)習(xí)率、折扣因子、神經(jīng)網(wǎng)絡(luò)結(jié)構(gòu)等。這有助于提高算法的性能和穩(wěn)定性。

使用 Deep Q-Learning 解決 Lunar Lander 問題是一個復(fù)雜的任務(wù)，需要仔細(xì)調(diào)整和實驗。算法的性能可能受到許多因素的影響，包括網(wǎng)絡(luò)結(jié)構(gòu)的選擇、超參數(shù)的設(shè)置以及環(huán)境的建模等。

Lunar Lander環(huán)境

這個環(huán)境是 Box2D 環(huán)境的一部分，其中包含有關(guān)環(huán)境的一般信息。

描述 Description

該環(huán)境是一個經(jīng)典的火箭軌跡優(yōu)化問題。根據(jù)龐特里亞金最大值原理，最佳的方法是全速點火或關(guān)閉引擎。這就是為什么這個環(huán)境具有離散動作的原因：引擎開啟或關(guān)閉。

有兩個環(huán)境版本：離散或連續(xù)。著陸點始終位于坐標(biāo) (0,0)。坐標(biāo)是狀態(tài)向量中的前兩個數(shù)字。在著陸點外著陸是可能的。燃料是無限的，因此代理可以學(xué)會飛行，然后在第一次嘗試時著陸。

要查看啟發(fā)式著陸，請運行：

python gymnasium/envs/box2d/lunar_lander.py

動作空間 Action Space

有四個可用的離散動作：

0：什么都不做

1：點火左定向引擎

2：點火主引擎

3：點火右定向引擎

觀察空間 Observation Space

狀態(tài)是一個8維向量：著陸器在 x 和 y 上的坐標(biāo)，它在 x 和 y 上的線速度，它的角度，它的角速度，以及兩個表示每條腿是否與地面接觸的布爾值。

獎勵 Rewards

每一步都會獲得一個獎勵。一個 episode（回合）總獎勵是該 episode 中所有步驟的獎勵之和。

對于每一步，獎勵：

著陸器離著陸點越近/遠(yuǎn)，獎勵增加/減少。

著陸器移動越慢/快，獎勵增加/減少。

著陸器傾斜度越大，獎勵減少（角度非水平）。

每個與地面接觸的腿獎勵增加10分。

每幀側(cè)引擎點火，獎勵減少0.03分。

每幀主引擎點火，獎勵減少0.3分。

episode 因墜毀或安全著陸而額外獲得-100或+100分的獎勵。

如果一個 episode 得分至少為200分，則認(rèn)為它是一個解決方案。

起始狀態(tài) Starting State

著陸器位于視口的頂部中心，對其質(zhì)心施加隨機初始力。

回合終止條件 Episode Termination

如果滿足以下條件回合結(jié)束：

著陸器墜毀（著陸器本體與月球接觸）；

著陸器超出視口范圍（x 坐標(biāo)大于1）；

著陸器處于非喚醒狀態(tài)。根據(jù) Box2D 文檔，處于非喚醒狀態(tài)的身體是不移動且不與任何其他身體碰撞的身體：

當(dāng) Box2D 確定一個身體（或一組身體）已經(jīng)停止時，該身體進入了一種幾乎沒有 CPU 開銷的休眠狀態(tài)。如果一個身體醒著并與休眠中的身體發(fā)生碰撞，那么休眠中的身體會醒來。如果連接到它們的關(guān)節(jié)或接觸被銷毀，身體也會醒來。

參數(shù) Arguments

要使用連續(xù)環(huán)境，您需要指定 continuous=True 參數(shù)，如下所示：

import gymnasium as gym
env = gym.make(
    "LunarLander-v2",
    continuous: bool = False,
    gravity: float = -10.0,
    enable_wind: bool = False,
    wind_power: float = 15.0,
    turbulence_power: float = 1.5,
)

gravity 確定了重力常數(shù)，它被限制在 0 和 -12 之間。

如果傳遞 enable_wind=True，則著陸器將受到風(fēng)的影響。風(fēng)是使用函數(shù) tanh(sin(2 k (t+C)) + sin(pi k (t+C))) 生成的。k 設(shè)置為 0.01。C 在 -9999 到 9999 之間隨機抽樣。

wind_power 確定了施加在飛行器上的線性風(fēng)的最大幅度。wind_power 的推薦值在 0.0 到 20.0 之間。turbulence_power 確定了施加在飛行器上的旋轉(zhuǎn)風(fēng)的最大幅度。turbulence_power 的推薦值在 0.0 到 2.0 之間。

版本歷史 Version History

v2：計算能量消耗，并在 v0.24 版本中，添加了具有風(fēng)力和 turbulence_power 參數(shù)的湍流

v1：在狀態(tài)向量中添加了與地面接觸的腿；與地面接觸會獎勵 +10 分，如果失去接觸則減去 -10 分；獎勵重新調(diào)整為 200 分；更難的初始隨機推力。

v0：初始版本

備注 Notes

在環(huán)境實現(xiàn)中存在一些意外的錯誤。

著陸器身體上側(cè)推器的位置會隨著著陸器的方向而變化。這反過來導(dǎo)致對著陸器施加方向依賴性扭矩。

狀態(tài)的單位不一致。即：

角速度以每秒 0.4 弧度為單位。為了轉(zhuǎn)換為每秒弧度，需要將該值乘以 2.5 的因子。

對于 VIEWPORT_W、VIEWPORT_H、SCALE 和 FPS 的默認(rèn)值，比例因子相等：'x': 10 'y': 6.666 'vx': 5 'vy': 7.5 'angle': 1 'angular velocity': 2.5

在進行更正后，狀態(tài)的單位如下：'x':（單位） 'y':（單位） 'vx':（單位/秒） 'vy':（單位/秒） 'angle':（弧度） 'angular velocity':（弧度/秒）

示例代碼

################ 完整代碼
import gymnasium as gym
from gym.wrappers.monitoring.video_recorder import VideoRecorder
from IPython.display import HTML, display
import imageio
import base64
import io
import glob
import os
import random
import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
import torch.autograd as autograd
from torch.autograd import Variable
from collections import deque, namedtuple


######## 創(chuàng)建網(wǎng)絡(luò)架構(gòu)
class Network(nn.Module):
    def __init__(self, state_size, action_size, seed=42):
        super(Network, self).__init__()
        self.seed = torch.manual_seed(seed)
        self.fc1 = nn.Linear(state_size, 64)
        self.fc2 = nn.Linear(64, 64)
        self.fc3 = nn.Linear(64, action_size)


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


######## 設(shè)置環(huán)境：使用Gymnasium創(chuàng)建了LunarLander環(huán)境
state_shape = env.observation_space.shape
state_size = env.observation_space.shape[0]
number_actions = env.action_space.n
print('State shape: ', state_shape)
print('State size: ', state_size)
print('Number of actions: ', number_actions)
######## 初始化超參數(shù)：定義了學(xué)習(xí)率、批處理大小、折扣因子等超參數(shù)。
learning_rate = 5e-4
minibatch_size = 100
discount_factor = 0.99
replay_buffer_size = int(1e5)
interpolation_parameter = 1e-3


####### 實現(xiàn)經(jīng)驗回放：實現(xiàn)了經(jīng)驗回放（Experience Replay）的類 ReplayMemory，用于存儲和采樣Agent的經(jīng)驗
class ReplayMemory(object):
    def __init__(self, capacity):
        self.device = torch.device(
            "cuda:0" if torch.cuda.is_available() else "cpu")
        self.capacity = capacity
        self.memory = []


    def push(self, event):
        self.memory.append(event)
        if len(self.memory) > self.capacity:
            del self.memory[0]


    def sample(self, batch_size):
        experiences = random.sample(self.memory, k=batch_size)
        states = torch.from_numpy(np.vstack(
            [e[0] for e in experiences if e is not None])).float().to(self.device)
        actions = torch.from_numpy(
            np.vstack([e[1] for e in experiences if e is not None])).long().to(self.device)
        rewards = torch.from_numpy(np.vstack(
            [e[2] for e in experiences if e is not None])).float().to(self.device)
        next_states = torch.from_numpy(np.vstack(
            [e[3] for e in experiences if e is not None])).float().to(self.device)
        dones = torch.from_numpy(np.vstack([e[4] for e in experiences if e is not None]).astype(
            np.uint8)).float().to(self.device)
        return states, next_states, actions, rewards, dones


########## 實現(xiàn) DQN 代理：創(chuàng)建了一個Agent類，包含本地Q網(wǎng)絡(luò)和目標(biāo)Q網(wǎng)絡(luò)，包含了采取動作、學(xué)習(xí)、軟更新等方法。
class Agent():
    # 初始化函數(shù)，參數(shù)為狀態(tài)大小和動作大小
    def __init__(self, state_size, action_size):
        self.device = torch.device(
            "cuda:0" if torch.cuda.is_available() else "cpu")
        self.state_size = state_size
        self.action_size = action_size
        self.local_qnetwork = Network(state_size, action_size).to(self.device)
        self.target_qnetwork = Network(state_size, action_size).to(self.device)
        self.optimizer = optim.Adam(
            self.local_qnetwork.parameters(), lr=learning_rate)
        self.memory = ReplayMemory(replay_buffer_size)
        self.t_step = 0
    # 定義一個函數(shù)，用于存儲經(jīng)驗并決定何時從中學(xué)習(xí)
    def step(self, state, action, reward, next_state, done):
        self.memory.push((state, action, reward, next_state, done))
        self.t_step = (self.t_step + 1) % 4
        if self.t_step == 0:
            if len(self.memory.memory) > minibatch_size:
                experiences = self.memory.sample(100)
                self.learn(experiences, discount_factor)
    # 定義一個函數(shù)，根據(jù)給定的狀態(tài)和epsilon值選擇一個動作（epsilon貪婪動作選擇策略）0.表示浮點數(shù)
    def act(self, state, epsilon=0.):
        state = torch.from_numpy(state).float().unsqueeze(0).to(self.device)
        self.local_qnetwork.eval()
        with torch.no_grad():
            action_values = self.local_qnetwork(state)
        self.local_qnetwork.train()
        if random.random() > epsilon:
            return np.argmax(action_values.cpu().data.numpy())
        else:
            return random.choice(np.arange(self.action_size))
    # 定義一個函數(shù)，根據(jù)樣本經(jīng)驗更新代理的q值，參數(shù)為經(jīng)驗和折扣因子
    def learn(self, experiences, discount_factor):
        states, next_states, actions, rewards, dones = experiences
        next_q_targets = self.target_qnetwork(
            next_states).detach().max(1)[0].unsqueeze(1)
        q_targets = rewards + discount_factor * next_q_targets * (1 - dones)
        q_expected = self.local_qnetwork(states).gather(1, actions)
        loss = F.mse_loss(q_expected, q_targets)
        self.optimizer.zero_grad()
        loss.backward()
        self.optimizer.step()
        self.soft_update(self.local_qnetwork,
                         self.target_qnetwork, interpolation_parameter)
    # 定義一個函數(shù)，用于軟更新目標(biāo)網(wǎng)絡(luò)的參數(shù)，參數(shù)為本地模型，目標(biāo)模型和插值參數(shù)
    def soft_update(self, local_model, target_model, interpolation_parameter):
        for target_param, local_param in zip(target_model.parameters(), local_model.parameters()):
            target_param.data.copy_(interpolation_parameter * local_param.data + (
                1.0 - interpolation_parameter) * target_param.data)


####### 訓(xùn)練DQN代理
agent = Agent(state_size, number_actions)


number_episodes = 2000
maximum_number_timesteps_per_episode = 1000
epsilon_starting_value = 1.0
epsilon_ending_value = 0.01
epsilon_decay_value = 0.995
epsilon = epsilon_starting_value
scores_on_100_episodes = deque(maxlen=100)


for episode in range(1, number_episodes + 1):
    state, _ = env.reset()
    score = 0
    for t in range(maximum_number_timesteps_per_episode):
        action = agent.act(state, epsilon)
        next_state, reward, done, _, _ = env.step(action)
        agent.step(state, action, reward, next_state, done)
        state = next_state
        score += reward
        if done:
            break
    scores_on_100_episodes.append(score)
    epsilon = max(epsilon_ending_value, epsilon_decay_value * epsilon)
    print('\rEpisode {}\tAverage Score: {:.2f}'.format(
        episode, np.mean(scores_on_100_episodes)), end="")
    if episode % 100 == 0:
        print('\rEpisode {}\tAverage Score: {:.2f}'.format(
            episode, np.mean(scores_on_100_episodes)))
    if np.mean(scores_on_100_episodes) >= 200.0:
        print('\nEnvironment solved in {:d} episodes!\tAverage Score: {:.2f}'.format(
            episode - 100, np.mean(scores_on_100_episodes)))
        torch.save(agent.local_qnetwork.state_dict(), 'checkpoint.pth')
        break


####### 可視化結(jié)果
def show_video_of_model(agent, env_name):
    env = gym.make(env_name, render_mode='rgb_array')
    state, _ = env.reset()
    done = False
    frames = []
    while not done:
        frame = env.render()
        frames.append(frame)
        action = agent.act(state)
        state, reward, done, _, _ = env.step(action.item())
    env.close()
    imageio.mimsave('video.mp4', frames, fps=30)




show_video_of_model(agent, 'LunarLander-v2')




def show_video():
    mp4list = glob.glob('*.mp4')
    if len(mp4list) > 0:
        mp4 = mp4list[0]
        video = io.open(mp4, 'r+b').read()
        encoded = base64.b64encode(video)
        display(HTML(data='''<video alt="test" autoplay
                loop controls style="height: 400px;">
                <source src="data:video/mp4;base64,{0}" type="video/mp4" />
             </video>'''.format(encoded.decode('ascii'))))
    else:
        print("Could not find video")


show_video()

終端輸出：

State shape:  (8,)
State size:  8
Number of actions:  4
Episode 100     Average Score: -174.79
Episode 200     Average Score: -102.59
Episode 300     Average Score: -68.797
Episode 400     Average Score: -38.18
Episode 500     Average Score: 24.301
Episode 600     Average Score: 149.24
Episode 700     Average Score: 134.89
Episode 800     Average Score: 185.41
Episode 826     Average Score: 200.92
Environment solved in 726 episodes!     Average Score: 200.92
IMAGEIO FFMPEG_WRITER WARNING: input image is not divisible by macro_block_size=16, resizing from (600, 400) to (608, 400) to ensure video compatibility with most codecs and players. To prevent resizing, make your input image divisible by the macro_block_size or set the macro_block_size to 1 (risking incompatibility).
[swscaler @ 000001e276cb1f00] Warning: data is not aligned! This can lead to a speed loss
<IPython.core.display.HTML object>

總結(jié)：以上代碼是使用深度 Q 學(xué)習(xí)（DQN）算法訓(xùn)練一個智能體在月球著陸環(huán)境中控制火箭的示例。代碼分為以下幾個部分：

? 導(dǎo)入所需的庫和模塊，包括 gymnasium（一個開源的強化學(xué)習(xí)環(huán)境庫），torch（一個開源的深度學(xué)習(xí)框架），以及一些輔助的庫和模塊，如 imageio（用于處理圖像和視頻），base64（用于編碼和解碼數(shù)據(jù)），deque（用于實現(xiàn)雙端隊列），namedtuple（用于創(chuàng)建命名元組）等。

import io  # 導(dǎo)入io模塊，用于讀寫文件
import glob  # 導(dǎo)入glob模塊，用于查找文件
import gymnasium as gym  # 導(dǎo)入環(huán)境庫
import os  # os 是操作系統(tǒng)相關(guān)的庫，用來處理文件和目錄等。
import random  # random 是隨機數(shù)生成和操作的庫，用來實現(xiàn) epsilon 貪心策略等。
import numpy as np  # numpy 是科學(xué)計算的庫，用來處理多維數(shù)組和矩陣等。
import torch  # torch 是 PyTorch 框架的主要庫，用來實現(xiàn)張量和神經(jīng)網(wǎng)絡(luò)等。
import torch.nn as nn  # torch.nn 是 PyTorch 框架的神經(jīng)網(wǎng)絡(luò)模塊，用來定義神經(jīng)網(wǎng)絡(luò)的層和損失函數(shù)等。
import torch.optim as optim  # torch.optim 是 PyTorch 框架的優(yōu)化器模塊，用來定義優(yōu)化算法和更新參數(shù)等。
# torch.nn.functional 是 PyTorch 框架的函數(shù)式接口，用來實現(xiàn)激活函數(shù)和池化等。
import torch.nn.functional as F
# torch.autograd 是 PyTorch 框架的自動微分模塊，用來實現(xiàn)反向傳播和梯度計算等。
import torch.autograd as autograd
# torch.autograd.Variable 是 PyTorch 框架的變量類，用來封裝張量和梯度等。
from torch.autograd import Variable
# collections 是 Python 的內(nèi)置模塊，用來實現(xiàn)特殊的容器類型，如雙端隊列和命名元組等。
from?collections?import?deque,?namedtuple

? 定義 Network 類，繼承自 torch.nn.Module 類，用來構(gòu)建神經(jīng)網(wǎng)絡(luò)模型，包括三個全連接層和一個前向傳播函數(shù)，輸入是狀態(tài)的維度，輸出是動作的個數(shù)。

class Network(nn.Module):  # 繼承nn模塊中的Module類
    def __init__(self, state_size, action_size, seed=42):  # 初始化函數(shù)，參數(shù)為狀態(tài)大小（8），動作大小（4），隨機種子（42）
        super(Network, self).__init__()  # 調(diào)用父類的初始化函數(shù)
        self.seed = torch.manual_seed(seed)  # 設(shè)置隨機種子
        # 創(chuàng)建第一個全連接層，輸入為狀態(tài)大小，輸出為64個神經(jīng)元，這里64是為了適應(yīng)月球著陸的問題
        self.fc1 = nn.Linear(state_size, 64)
        self.fc2 = nn.Linear(64, 64)  # 創(chuàng)建第二個全連接層，輸入為64個神經(jīng)元，輸出為64個神經(jīng)元
        self.fc3 = nn.Linear(64, action_size)  # 創(chuàng)建第三個全連接層，輸入為64個神經(jīng)元，輸出為動作大小
        # 完成神經(jīng)網(wǎng)絡(luò)的構(gòu)建


    def forward(self, state):  # 前向傳播函數(shù)
        x = self.fc1(state)  # 從輸入層接收狀態(tài)
        x = F.relu(x)  # 使用激活函數(shù)
        x = self.fc2(x)  # 從第一個隱藏層接收輸出
        x = F.relu(x)  # 使用激活函數(shù)
        return self.fc3(x)  # 返回最后一層的輸出

? 創(chuàng)建月球著陸環(huán)境，使用 gym.make 函數(shù)，獲取環(huán)境的狀態(tài)空間和動作空間的屬性，如狀態(tài)的形狀，狀態(tài)的維度，動作的個數(shù)等。

# https://gymnasium.farama.org/environments/box2d/lunar_lander/
env = gym.make('LunarLander-v2')
# 導(dǎo)入月球著陸環(huán)境
state_shape = env.observation_space.shape  # 狀態(tài)的形狀，這里是8維向量
state_size = env.observation_space.shape[0]  # 狀態(tài)的大小，這里是8個元素，包括坐標(biāo)，速度等
number_actions = env.action_space.n  # 動作的數(shù)量，這里是4個
print('State shape: ', state_shape)
print('State size: ', state_size)
print('Number of actions: ', number_actions)

? 定義一些超參數(shù)，如學(xué)習(xí)率，批次大小，折扣因子，回放緩沖區(qū)的容量，軟更新的插值參數(shù)等。

learning_rate = 5e-4  # 學(xué)習(xí)率，這里是0.00005
minibatch_size = 100  # 批次大小，用于更新參數(shù)
discount_factor = 0.99  # 折扣因子，用于計算未來獎勵，越接近1越考慮未來，越接近0越考慮當(dāng)前
replay_buffer_size = int(1e5)  # 重放緩沖區(qū)的大小，用于存儲人工智能的經(jīng)驗，穩(wěn)定和改善學(xué)習(xí)，這里是10萬個經(jīng)驗
interpolation_parameter = 1e-3  # 插值參數(shù)，用于更新目標(biāo)網(wǎng)絡(luò)，這里是0.001
# 所有的參數(shù)都是通過實驗得到的最優(yōu)值

? 定義 ReplayMemory 類，用來實現(xiàn)經(jīng)驗回放機制，包括一個初始化函數(shù)，一個存儲經(jīng)驗的函數(shù)，一個采樣經(jīng)驗的函數(shù)，使用雙端隊列來存儲經(jīng)驗元組，使用隨機采樣的方法來獲取一批經(jīng)驗，將經(jīng)驗轉(zhuǎn)換為 PyTorch 張量并發(fā)送到 cpu 或 gpu 上。

# 定義一個重放緩沖區(qū)的類
class ReplayMemory(object):
    def __init__(self, capacity):  # 初始化函數(shù)，參數(shù)為緩沖區(qū)的容量
        # 判斷是否有cuda可用，如果有則使用cuda，否則使用cpu
        self.device = torch.device(
            "cuda:0" if torch.cuda.is_available() else "cpu")
        self.capacity = capacity  # 設(shè)置緩沖區(qū)的容量
        self.memory = []  # 創(chuàng)建一個列表，用于存儲經(jīng)驗，每個經(jīng)驗包括狀態(tài)，動作，獎勵，下一個狀態(tài)，是否結(jié)束等


    def push(self, event):  # 定義一個函數(shù)，用于向緩沖區(qū)中添加經(jīng)驗
        self.memory.append(event)  # 將經(jīng)驗添加到列表中
        if len(self.memory) > self.capacity:  # 如果列表的長度超過了容量
            del self.memory[0]  # 刪除最舊的經(jīng)驗


    def sample(self, batch_size):  # 定義一個函數(shù)，用于從緩沖區(qū)中隨機抽取一批經(jīng)驗
        experiences = random.sample(self.memory, k=batch_size)  # 從列表中隨機抽取k個經(jīng)驗
        # 從經(jīng)驗中提取每個元素，并將它們堆疊在一起
        # 使用列表推導(dǎo)式，從每個經(jīng)驗中提取狀態(tài)，使用np.vstack將它們垂直堆疊，然后轉(zhuǎn)換為pytorch張量，使用.float()將它們轉(zhuǎn)換為浮點數(shù)，使用.to(self.device)將它們發(fā)送到cpu或gpu
        states = torch.from_numpy(np.vstack(
            [e[0] for e in experiences if e is not None])).float().to(self.device)
        actions = torch.from_numpy(np.vstack([e[1] for e in experiences if e is not None])).long(
        ).to(self.device)  # 同理，從每個經(jīng)驗中提取動作，使用.long()將它們轉(zhuǎn)換為整數(shù)
        rewards = torch.from_numpy(np.vstack([e[2] for e in experiences if e is not None])).float(
        ).to(self.device)  # 同理，從每個經(jīng)驗中提取獎勵，使用.float()將它們轉(zhuǎn)換為浮點數(shù)
        next_states = torch.from_numpy(np.vstack([e[3] for e in experiences if e is not None])).float(
        ).to(self.device)  # 同理，從每個經(jīng)驗中提取下一個狀態(tài)，使用.float()將它們轉(zhuǎn)換為浮點數(shù)
        dones = torch.from_numpy(np.vstack([e[4] for e in experiences if e is not None]).astype(np.uint8)).float(
        ).to(self.device)  # 同理，從每個經(jīng)驗中提取是否結(jié)束的標(biāo)志，使用.astype(np.uint8)將它們轉(zhuǎn)換為無符號整數(shù)，使用.float()將它們轉(zhuǎn)換為浮點數(shù)
        return states, next_states, actions, rewards, dones  # 返回這些元素，注意順序要一致

? 定義 Agent 類，用來實現(xiàn)深度 Q 學(xué)習(xí)的智能體，包括一個初始化函數(shù)，一個執(zhí)行一步操作的函數(shù)，一個選擇動作的函數(shù)，一個學(xué)習(xí)的函數(shù)，一個軟更新的函數(shù)，使用兩個神經(jīng)網(wǎng)絡(luò)模型，一個是本地 Q 網(wǎng)絡(luò)，用來選擇和評估動作，一個是目標(biāo) Q 網(wǎng)絡(luò)，用來計算目標(biāo) Q 值，使用 Adam 優(yōu)化器來更新本地 Q 網(wǎng)絡(luò)的參數(shù)，使用 ReplayMemory 類的實例來存儲和采樣經(jīng)驗，使用時間步來控制何時學(xué)習(xí)，使用 epsilon 貪心策略來平衡探索和利用，使用均方誤差損失函數(shù)來計算預(yù)期 Q 值和目標(biāo) Q 值之間的差異，使用軟更新的方法來更新目標(biāo) Q 網(wǎng)絡(luò)的參數(shù)。

# 定義一個代理類
class Agent():
    def __init__(self, state_size, action_size):  # 初始化函數(shù)，參數(shù)為狀態(tài)大小和動作大小
        # 判斷是否有cuda可用，如果有則使用cuda，否則使用cpu
        self.device = torch.device(
            "cuda:0" if torch.cuda.is_available() else "cpu")
        self.state_size = state_size  # 創(chuàng)建對象變量，存儲狀態(tài)大小
        self.action_size = action_size  # 創(chuàng)建對象變量，存儲動作大小
        # Q學(xué)習(xí) --
        self.local_qnetwork = Network(state_size, action_size).to(
            self.device)  # 創(chuàng)建一個本地網(wǎng)絡(luò)，用于選擇動作，將其發(fā)送到設(shè)備上，隨機種子已經(jīng)在之前提供
        self.target_qnetwork = Network(state_size, action_size).to(
            self.device)  # 創(chuàng)建一個目標(biāo)網(wǎng)絡(luò)，用于計算目標(biāo)值，將其發(fā)送到設(shè)備上


        # 創(chuàng)建一個優(yōu)化器，用于更新本地網(wǎng)絡(luò)的參數(shù)，使用Adam算法，學(xué)習(xí)率為之前定義的值
        self.optimizer = optim.Adam(
            self.local_qnetwork.parameters(), lr=learning_rate)
        # 創(chuàng)建一個重放緩沖區(qū)，用于存儲人工智能的經(jīng)驗，容量為之前定義的值
        self.memory = ReplayMemory(replay_buffer_size)
        self.t_step = 0  # 時間步數(shù)


    def step(self, state, action, reward, next_state, done):  # 定義一個函數(shù)，用于存儲經(jīng)驗并決定何時從中學(xué)習(xí)
        self.memory.push((state, action, reward, next_state, done))  # 將經(jīng)驗推入緩沖區(qū)
        self.t_step = (self.t_step + 1) % 4  # 時間步數(shù)計數(shù)器（每4步學(xué)習(xí)一次）
        if self.t_step == 0:  # 如果時間步數(shù)為0
            # 如果緩沖區(qū)中的經(jīng)驗數(shù)量大于批次大小，self.memory是重放緩沖區(qū)的實例，memory屬性是__init__中定義的列表
            if len(self.memory.memory) > minibatch_size:
                experiences = self.memory.sample(100)  # 從緩沖區(qū)中隨機抽取100個經(jīng)驗
                self.learn(experiences, discount_factor)  # 從經(jīng)驗中學(xué)習(xí)，使用之前定義的折扣因子


    def act(self, state, epsilon=0.):  # 定義一個函數(shù)，根據(jù)給定的狀態(tài)和epsilon值選擇一個動作（epsilon貪婪動作選擇策略）0.表示浮點數(shù)
        # 將狀態(tài)轉(zhuǎn)換為pytorch張量，增加一個維度，表示批次的維度，0表示批次維度的索引，將其放在最前面，將張量發(fā)送到設(shè)備上
        state = torch.from_numpy(state).float().unsqueeze(0).to(self.device)
        self.local_qnetwork.eval()  # 將本地網(wǎng)絡(luò)設(shè)置為評估模式，不更新梯度，本地網(wǎng)絡(luò)是代理類的屬性，繼承自nn.Module類，有eval()方法
        with torch.no_grad():  # 不計算梯度，檢查是否處于推理模式而不是訓(xùn)練模式
            # 使用本地網(wǎng)絡(luò)對狀態(tài)進行前向傳播，得到每個動作的價值，這些價值將被epsilon貪婪策略選擇（這里我們得到的不是最終的值，而是對應(yīng)于狀態(tài)的q值）
            action_values = self.local_qnetwork(state)
        self.local_qnetwork.train()  # 將本地網(wǎng)絡(luò)設(shè)置為訓(xùn)練模式，更新梯度，本地網(wǎng)絡(luò)是代理類的屬性，繼承自nn.Module類，有train()方法
        # Eplison貪婪動作選擇策略 --（用于探索和利用的平衡）
        if random.random() > epsilon:  # 如果隨機數(shù)大于epsilon，random.random()是random庫的方法，返回一個0到1之間的隨機數(shù)
            # 返回價值最大的動作，np.argmax是numpy庫的方法，返回最大值的索引，action_values發(fā)送到cpu上，因為它是簡單的，data.numpy()將格式轉(zhuǎn)換為numpy的數(shù)據(jù)格式
            return np.argmax(action_values.cpu().data.numpy())
        else:  # 否則
            # 返回隨機的動作，random.choice是random庫的方法，從給定的列表中隨機選擇一個元素，np.arange是numpy庫的方法，返回一個從0到動作大小的數(shù)組
            return random.choice(np.arange(self.action_size))


    def learn(self, experiences, discount_factor):  # 定義一個函數(shù)，根據(jù)樣本經(jīng)驗更新代理的q值，參數(shù)為經(jīng)驗和折扣因子
        states, next_states, actions, rewards, dones = experiences  # 從經(jīng)驗中解包元素
        # 從目標(biāo)網(wǎng)絡(luò)中獲取下一個狀態(tài)的最大q值，self.target_qnetwork(next_states)返回每個動作的價值，.detach()將張量從計算圖中分離，我們不會在反向傳播中使用這些值，.max(1)沿著第一個維度取最大值，得到兩個張量（最大值和索引），因此我們添加.max[1][0]只獲取最大值，.unsqueeze(1)增加一個維度，表示批次的維度，但這次在第一個位置
        next_q_targets = self.target_qnetwork(
            next_states).detach().max(1)[0].unsqueeze(1) # 從目標(biāo)Q網(wǎng)絡(luò)的輸出中，選擇每個下一個狀態(tài)對應(yīng)的最大Q值，然后將其整理成一個列向量 next_q_targets。這個向量將用于計算Q-learning的目標(biāo)值，以便更新本地Q網(wǎng)絡(luò)
        q_targets = rewards + discount_factor * next_q_targets * \
            (1 - dones)  # 計算當(dāng)前狀態(tài)的目標(biāo)值，使用公式：獎勵加上折扣后的未來最大值，乘以1減去結(jié)束標(biāo)志
        q_expected = self.local_qnetwork(states).gather(
            1, actions)  # 獲取當(dāng)前狀態(tài)的預(yù)期值，.gather(1, actions)根據(jù)動作選擇對應(yīng)的價值
        loss = F.mse_loss(q_expected, q_targets)  # 計算預(yù)期值和目標(biāo)值之間的均方誤差
        self.optimizer.zero_grad()  # 清空優(yōu)化器的梯度，zero_grad()是Adam的方法
        loss.backward()  # 反向傳播誤差
        self.optimizer.step()  # 更新本地網(wǎng)絡(luò)的參數(shù)，step()是優(yōu)化器的方法
        # 使用軟更新的方法更新目標(biāo)網(wǎng)絡(luò)的參數(shù)，參數(shù)為本地網(wǎng)絡(luò)，目標(biāo)網(wǎng)絡(luò)和插值參數(shù)
        self.soft_update(self.local_qnetwork,
                         self.target_qnetwork, interpolation_parameter)


    # 定義一個函數(shù)，用于軟更新目標(biāo)網(wǎng)絡(luò)的參數(shù)，參數(shù)為本地模型，目標(biāo)模型和插值參數(shù)
    def soft_update(self, local_model, target_model, interpolation_parameter):
        # 遍歷目標(biāo)模型和本地模型的參數(shù)，zip()是一個函數(shù)，用于將參數(shù)打包成元組，parameters()是nn.Module的方法，返回模型的參數(shù)
        for target_param, local_param in zip(target_model.parameters(), local_model.parameters()):
            # 使用插值參數(shù)更新目標(biāo)模型的參數(shù)，將本地模型的參數(shù)乘以插值參數(shù)，再加上目標(biāo)模型的參數(shù)乘以1減去插值參數(shù)，.copy_()是一個方法，用于復(fù)制張量的數(shù)據(jù)
            target_param.data.copy_(interpolation_parameter * local_param.data + (
                1.0 - interpolation_parameter) * target_param.data)

? 創(chuàng)建 Agent 類的實例，傳入狀態(tài)維度和動作個數(shù)。

agent = Agent(state_size, number_actions)  # 創(chuàng)建一個代理或人工智能，參數(shù)為狀態(tài)大小和動作數(shù)量

? 定義訓(xùn)練過程，包括回合數(shù)，每個回合的最大時間步數(shù)，epsilon 值的初始值，終止值，衰減值，記錄最近 100 個回合的分?jǐn)?shù)，遍歷每個回合，重置環(huán)境，初始化分?jǐn)?shù)，遍歷每個時間步，選擇動作，執(zhí)行動作，存儲經(jīng)驗，更新狀態(tài)，累加獎勵，判斷是否結(jié)束，更新 epsilon 值，打印當(dāng)前回合和平均分?jǐn)?shù)，判斷是否達(dá)到目標(biāo)分?jǐn)?shù)，保存模型參數(shù)。

# ### 訓(xùn)練DQN代理
# %%
number_episodes = 2000  # 我們想要訓(xùn)練的次數(shù)
# 我們不想讓一個回合卡住，所以設(shè)置一個最大的時間步數(shù)（在月球上著陸的嘗試最多為1000個時間步）
maximum_number_timesteps_per_episode = 1000
epsilon_starting_value = 1.0  # epsilon的初始值
epsilon_ending_value = 0.01  # epsilon的結(jié)束值
epsilon_decay_value = 0.995  # epsilon的衰減率，按照(1*0.995 , 1*0.995*0.995,...)的方式遞減
epsilon = epsilon_starting_value  # epsilon的變量
scores_on_100_episodes = deque(maxlen=100)  # 最近100個回合的分?jǐn)?shù)（列表） 創(chuàng)建了一個雙端隊列，最大長度為100


for episode in range(1, number_episodes + 1):  # 循環(huán)直到2000次（上限固定）
    state, _ = env.reset()  # 重置環(huán)境（這里返回狀態(tài)和觀察值）
    score = 0  # 每個回合的累積分?jǐn)?shù)
    for t in range(maximum_number_timesteps_per_episode):  # 循環(huán)每個時間步
        action = agent.act(state, epsilon)  # 根據(jù)當(dāng)前狀態(tài)和epsilon貪婪策略選擇一個動作
        # 根據(jù)動作執(zhí)行環(huán)境的步驟，返回下一個狀態(tài)，獎勵，是否結(jié)束等值，_是丟棄不需要的值
        next_state, reward, done, _, _ = env.step(action)
        agent.step(state, action, reward, next_state, done)  # 調(diào)用代理的學(xué)習(xí)方法
        state = next_state  # 更新狀態(tài)
        score += reward  # 累加獎勵
        if done:  # 如果回合結(jié)束
            break  # 跳出循環(huán)
    scores_on_100_episodes.append(score)  # 將最近的回合分?jǐn)?shù)添加到列表中
    epsilon = max(epsilon_ending_value, epsilon_decay_value *
                  epsilon)  # 更新epsilon的值，使其按照衰減率遞減，但不低于結(jié)束值
    print('\rEpisode {}\tAverage Score: {:.2f}'.format(episode, np.mean(
        scores_on_100_episodes)), end="")  # \r覆蓋之前的輸出，\t制表符，.2f保留兩位小數(shù)，打印當(dāng)前的回合數(shù)和平均分?jǐn)?shù)，end表示不換行
    if episode % 100 == 0:  # 每100個回合
        print('\rEpisode {}\tAverage Score: {:.2f}'.format(
            episode, np.mean(scores_on_100_episodes)))  # \r覆蓋之前的輸出，打印當(dāng)前的回合數(shù)和平均分?jǐn)?shù)，換行
    if np.mean(scores_on_100_episodes) >= 200.0:  # 如果達(dá)到勝利的條件
        print('\nEnvironment solved in {:d} episodes!\tAverage Score: {:.2f}'.format(
            episode - 100, np.mean(scores_on_100_episodes)))  # \n換行，:d表示整數(shù)，打印環(huán)境在多少個回合內(nèi)解決，以及平均分?jǐn)?shù)
        torch.save(agent.local_qnetwork.state_dict(),
                   'checkpoint.pth')  # 將當(dāng)前Agent的本地Q網(wǎng)絡(luò)（agent.local_qnetwork）的參數(shù)保存到文件中
        break  # 跳出循環(huán)

? 定義展示模型表現(xiàn)的函數(shù)，使用 imageio 庫將每一幀的圖像保存為視頻文件，使用 HTML 標(biāo)簽和 base64 編碼在網(wǎng)頁中顯示視頻。

? 調(diào)用展示模型表現(xiàn)的函數(shù)，傳入智能體和環(huán)境的名稱。

from gym.wrappers.monitoring.video_recorder import VideoRecorder  # 導(dǎo)入gym模塊，用于錄制視頻
from IPython.display import HTML, display  # 導(dǎo)入IPython模塊，用于顯示HTML和視頻
import imageio  # 導(dǎo)入imageio模塊，用于處理圖像
import base64  # 導(dǎo)入base64模塊，用于編碼和解碼
import io  # 導(dǎo)入io模塊，用于讀寫文件
import glob  # 導(dǎo)入glob模塊，用于查找文件
import gymnasium as gym  # 導(dǎo)入環(huán)境庫
# %%




def show_video_of_model(agent, env_name):  # 定義一個函數(shù)，用于展示代理在環(huán)境中的表現(xiàn)
    env = gym.make(env_name, render_mode='rgb_array')  # 創(chuàng)建一個環(huán)境，渲染模式為rgb數(shù)組
    state, _ = env.reset()  # 重置環(huán)境，獲取初始狀態(tài)
    done = False  # 設(shè)置結(jié)束標(biāo)志為False
    frames = []  # 創(chuàng)建一個列表，用于存儲每一幀的圖像
    while not done:  # 循環(huán)直到結(jié)束
        frame = env.render()  # 渲染環(huán)境，獲取當(dāng)前幀的圖像
        frames.append(frame)  # 將圖像添加到列表中
        action = agent.act(state)  # 根據(jù)當(dāng)前狀態(tài)選擇一個動作
        state, reward, done, _, _ = env.step(
            action.item())  # 根據(jù)動作執(zhí)行環(huán)境的步驟，獲取下一個狀態(tài)，獎勵，是否結(jié)束等值
    env.close()  # 關(guān)閉環(huán)境
    # 使用imageio模塊，將列表中的圖像保存為視頻，幀率為30
    imageio.mimsave('video.mp4', frames, fps=30)




show_video_of_model(agent, 'LunarLander-v2')  # 調(diào)用函數(shù)，展示代理在月球著陸環(huán)境中的表現(xiàn)




def show_video():  # 定義一個函數(shù)，用于顯示視頻
    mp4list = glob.glob('*.mp4')  # 使用glob模塊，查找當(dāng)前目錄下的所有mp4文件
    if len(mp4list) > 0:  # 如果找到了
        mp4 = mp4list[0]  # 取第一個文件
        video = io.open(mp4, 'r+b').read()  # 使用io模塊，以二進制模式讀取文件
        encoded = base64.b64encode(video)  # 使用base64模塊，對文件進行編碼
        display(HTML(data='''<video alt="test" autoplay
                loop controls style="height: 400px;">
                <source src="data:video/mp4;base64,{0}" type="video/mp4" />
             </video>'''.format(encoded.decode('ascii'))))  # 使用IPython模塊，顯示HTML格式的視頻，使用base64編碼的數(shù)據(jù)作為源
    else:  # 如果沒有找到
        print("Could not find video")  # 打印提示信息
show_video()  # 調(diào)用函數(shù)，顯示視頻

筆記

參考網(wǎng)址：

https://gymnasium.farama.org/environments/box2d/lunar_lander/文章來源地址http://www.zghlxwxcb.cn/news/detail-823299.html