文章目錄
- 前言
- SAC處理連續動作空間問題 (Pendulum-v1)
-
- 核心代碼實現
-
- **工具函數與環境初始化**
- **ReplayBuffer、網絡結構與SAC算法**
- **訓練與結果**
- SAC處理離散動作空間問題 (CartPole-v1)
-
- 核心代碼實現
-
- **工具函數與環境初始化**
- **ReplayBuffer、網絡結構與SAC算法 (離散版)**
- **訓練與結果**
- 總結
前言
在深度強化學習(DRL)的探索之旅中,我們不斷尋求更高效、更穩定的算法來應對日益復雜的決策問題。傳統的在線策略算法(On-policy)如A2C、PPO等,雖然在很多場景下表現優異,但其采樣效率低下的問題也限制了它們在某些現實世界任務中的應用,尤其是在那些與環境交互成本高昂的場景中。
因此,離線策略(Off-policy)算法應運而生,它們能夠利用歷史數據(Replay Buffer)進行學習,極大地提高了數據利用率和學習效率。在眾多離線策略算法中,Soft Actor-Critic(SAC)算法以其出色的穩定性和卓越的性能脫穎而出。
正如上圖所述,與同為離線策略算法的DDPG相比,SAC在訓練穩定性和收斂性方面表現更佳,對超參數的敏感度也更低。 SAC的前身是Soft Q-learning,它們都屬于最大熵強化學習的范疇,即在最大化累積獎勵的同時,也最大化策略的熵,從而鼓勵智能體進行更充分的探索。 與Soft Q-learning不同,SAC引入了一個顯式的策略函數(Actor),從而優雅地解決了在連續動作空間中求解困難的問題。 SAC學習的是一個隨機策略,這使得它能夠探索多模態的最優策略,并在復雜的環境中表現出更強的魯棒性。
本篇博客將通過兩個PyTorch實現的SAC代碼示例,帶您深入理解SAC算法的精髓。我們將分別探討其在連續動作空間和離散動作空間中的具體實現,并通過代碼解析,讓您對策略網絡、價值網絡、經驗回放、軟更新以及核心的熵正則化等概念有更直觀的認識。
無論您是強化學習的初學者,還是希望深入了解SAC算法的實踐者,相信通過本文的代碼學習之旅,您都將有所收獲。
完整代碼:下載鏈接
SAC處理連續動作空間問題 (Pendulum-v1)
在連續控制任務中,SAC通過學習一個隨機策略,輸出動作的正態分布的均值和標準差,從而實現對連續動作的探索和決策。我們將以OpenAI Gym中的經典環境Pendulum-v1
為例,這是一個典型的連續控制問題,智能體的目標是利用有限的力矩將擺桿豎立起來。
核心代碼實現
以下是SAC算法在Pendulum-v1
環境下的完整PyTorch實現。代碼涵蓋了工具函數、環境初始化、核心網絡結構(ReplayBuffer、策略網絡、Q值網絡)、SAC算法主類以及訓練和可視化的全過程。
工具函數與環境初始化
首先,我們定義一個moving_average
函數用于平滑訓練過程中的獎勵曲線,以便更好地觀察訓練趨勢。然后,我們初始化Pendulum-v1
環境。
# utils"""
強化學習工具函數集
包含數據平滑處理功能
"""import torch
import numpy as np
def moving_average(data, window_size):"""計算移動平均值,用于平滑獎勵曲線該函數通過滑動窗口的方式對時間序列數據進行平滑處理,可以有效減少數據中的噪聲,使曲線更加平滑美觀。常用于強化學習中對訓練過程的獎勵曲線進行可視化優化。參數:data (list): 原始數據序列,維度: [num_episodes]包含需要平滑處理的數值數據(如每輪訓練的獎勵值)window_size (int): 移動窗口大小,維度: 標量決定了平滑程度,窗口越大平滑效果越明顯但也會導致更多的數據點丟失返回:list: 移動平均后的數據,維度: [len(data) - window_size + 1]返回的數據長度會比原數據少 window_size - 1 個元素這是因為需要足夠的數據點來計算第一個移動平均值示例:>>> data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] # 維度: [10]>>> smoothed = moving_average(data, 3) # window_size = 3>>> print(smoothed) # 輸出: [2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0] 維度: [8]"""# 邊界檢查:如果數據長度小于窗口大小,直接返回原數據# 這種情況下無法計算移動平均值# data維度: [num_episodes], window_size維度: 標量if len(data) < window_size:return data# 初始化移動平均值列表# moving_avg維度: 最終為[len(data) - window_size + 1]moving_avg = []# 遍歷數據,計算每個窗口的移動平均值# i的取值范圍: 0 到 len(data) - window_size# 循環次數: len(data) - window_size + 1# 每次循環處理一個滑動窗口位置for i in range(len(data) - window_size + 1):# 提取當前窗口內的數據切片# window_data維度: [window_size]# 包含從索引i開始的連續window_size個元素# 例如:當i=0, window_size=3時,提取data[0:3]window_data = data[i:i + window_size]# 計算當前窗口內數據的算術平均值# np.mean(window_data)維度: 標量# 將平均值添加到結果列表中moving_avg.append(np.mean(window_data))# 返回移動平均后的數據列表# moving_avg維度: [len(data) - window_size + 1]return moving_avg
"""
強化學習環境初始化模塊
用于創建和配置OpenAI Gym環境
"""import gym # OpenAI Gym庫,提供標準化的強化學習環境接口
import numpy as np # 數值計算庫,用于處理多維數組和數學運算# 定義環境名稱
# env_name維度: 字符串標量
# 'Pendulum-v1'是一個連續控制任務,倒立擺環境
# 狀態空間: 3維連續空間 (cos(theta), sin(theta), thetadot)
# 動作空間: 1維連續空間,范圍[-2.0, 2.0]
env_name = 'Pendulum-v1'# 創建強化學習環境實例
# env維度: gym.Env對象
# 包含完整的環境狀態、動作空間、獎勵函數等信息
# 該環境支持reset()、step()、render()、close()等標準方法
env = gym.make(env_name)
ReplayBuffer、網絡結構與SAC算法
這部分代碼是SAC算法的核心。
- ReplayBuffer: 經驗回放池,用于存儲和采樣智能體的經驗數據,打破數據相關性,提高學習效率。
- PolicyNetContinuous: 策略網絡(Actor),輸入狀態,輸出動作分布的均值和標準差。這里使用了重參數化技巧(Reparameterization Trick),使得從策略分布中采樣的過程可導,從而能夠利用梯度進行端到端的訓練。動作經過
tanh
函數激活并縮放到環境的動作邊界內。 - QValueNetContinuous: Q值網絡(Critic),輸入狀態和動作,輸出對應的Q值。SAC采用了雙Q網絡的技巧,即構建兩個結構相同的Q網絡,在計算目標Q值時取兩者的較小值,以緩解Q值過高估計的問題。
- SACContinuous: SAC算法的主類,整合了上述所有網絡和組件。它實現了動作選擇、目標Q值計算、網絡參數的軟更新以及策略和價值網絡的更新邏輯。其中,溫度參數α的學習和更新是SAC的核心之一,它通過最大化熵的目標自動調整,平衡探索與利用。
"""
SAC (Soft Actor-Critic) 算法實現
用于連續動作空間的強化學習智能體
"""import torch # PyTorch深度學習框架
import torch.nn as nn # 神經網絡模塊
import torch.nn.functional as F # 神經網絡功能函數
import numpy as np # 數值計算庫
import random # 隨機數生成庫
import collections # 集合數據類型模塊
from torch.distributions import Normal # 正態分布類class ReplayBuffer:"""經驗回放緩沖區類用于存儲和采樣智能體的經驗數據"""def __init__(self, capacity):"""初始化經驗回放緩沖區參數:capacity (int): 緩沖區容量,維度: 標量"""# 使用雙端隊列作為緩沖區存儲結構# self.buffer維度: deque,最大長度為capacity# 存儲格式: (state, action, reward, next_state, done)self.buffer = collections.deque(maxlen=capacity)def add(self, state, action, reward, next_state, done):"""向緩沖區添加一條經驗參數:state (np.array): 當前狀態,維度: [state_dim]action (float): 執行的動作,維度: 標量reward (float): 獲得的獎勵,維度: 標量next_state (np.array): 下一個狀態,維度: [state_dim]done (bool): 是否結束,維度: 標量布爾值"""# 將經驗元組添加到緩沖區# 元組維度: (state[state_dim], action[1], reward[1], next_state[state_dim], done[1])self.buffer.append((state, action, reward, next_state, done))def sample(self, batch_size):"""從緩沖區隨機采樣一批經驗參數:batch_size (int): 批次大小,維度: 標量返回:tuple: 包含狀態、動作、獎勵、下一狀態、完成標志的元組state (np.array): 狀態批次,維度: [batch_size, state_dim]action (tuple): 動作批次,維度: [batch_size]reward (tuple): 獎勵批次,維度: [batch_size]next_state (np.array): 下一狀態批次,維度: [batch_size, state_dim]done (tuple): 完成標志批次,維度: [batch_size]"""# 隨機采樣batch_size個經驗# transitions維度: list,長度為batch_sizetransitions = random.sample(self.buffer, batch_size)# 將經驗元組解包并轉置# 每個元素的維度: state[batch_size個state_dim], action[batch_size], 等等state, action, reward, next_state, done = zip(*transitions)# 將狀態轉換為numpy數組便于后續處理# state維度: [batch_size, state_dim]# next_state維度: [batch_size, state_dim]return np.array(state), action, reward, np.array(next_state), donedef size(self):"""返回緩沖區當前大小返回:int: 緩沖區大小,維度: 標量"""return len(self.buffer)class PolicyNetContinuous(torch.nn.Module):"""連續動作空間的策略網絡輸出動作的均值和標準差,用于生成隨機策略"""def __init__(self, state_dim, hidden_dim, action_dim, action_bound):"""初始化策略網絡參數:state_dim (int): 狀態空間維度,維度: 標量hidden_dim (int): 隱藏層維度,維度: 標量action_dim (int): 動作空間維度,維度: 標量action_bound (float): 動作邊界值,維度: 標量"""super(PolicyNetContinuous, self).__init__()# 第一個全連接層:狀態到隱藏層# 輸入維度: [batch_size, state_dim]# 輸出維度: [batch_size, hidden_dim]self.fc1 = torch.nn.Linear(state_dim, hidden_dim)# 輸出動作均值的全連接層# 輸入維度: [batch_size, hidden_dim]# 輸出維度: [batch_size, action_dim]self.fc_mu = torch.nn.Linear(hidden_dim, action_dim)# 輸出動作標準差的全連接層# 輸入維度: [batch_size, hidden_dim]# 輸出維度: [batch_size, action_dim]self.fc_std = torch.nn.Linear(hidden_dim, action_dim)# 動作邊界值,用于縮放tanh輸出# action_bound維度: 標量self.action_bound = action_bounddef forward(self, x):"""前向傳播函數參數:x (torch.Tensor): 輸入狀態,維度: [batch_size, state_dim]返回:tuple: 包含動作和對數概率的元組action (torch.Tensor): 輸出動作,維度: [batch_size, action_dim]log_prob (torch.Tensor): 動作對數概率,維度: [batch_size, action_dim]"""# 第一層激活# x維度: [batch_size, state_dim] -> [batch_size, hidden_dim]x = F.relu(self.fc1(x))# 計算動作均值# mu維度: [batch_size, action_dim]mu = self.fc_mu(x)# 計算動作標準差,使用softplus確保為正值# std維度: [batch_size, action_dim]std = F.softplus(self.fc_std(x))# 創建正態分布對象# dist維度: Normal分布對象,參數維度均為[batch_size, action_dim]dist = Normal(mu, std)# 重參數化采樣,確保梯度可以反向傳播# normal_sample維度: [batch_size, action_dim]normal_sample = dist.rsample() # rsample()是重參數化采樣"""重參數化采樣是一種用于訓練神經網絡生成模型(Generative Models)的技術,特別是在概率編碼器-解碼器框架中常見,例如變分自編碼器(Variational Autoencoder,VAE)。這種技術的目的是將采樣過程通過神經網絡的可導操作,使得模型可以被端到端地訓練。在普通的采樣過程中,由于采樣操作本身是不可導的,傳統的梯度下降方法無法直接用于訓練神經網絡。為了解決這個問題,引入了重參數化技巧。`dist.rsample()` 是重參數化采樣的一部分。這里的重參數化指的是將采樣操作重新參數化為可導的操作,使得梯度能夠通過網絡反向傳播。通過這種方式,可以有效地訓練生成模型,尤其是概率生成模型。在正態分布的情況下,傳統的采樣操作是直接從標準正態分布中抽取樣本,然后通過線性變換得到最終的樣本。而重參數化采樣則通過在標準正態分布上進行采樣,并通過神經網絡產生的均值和標準差進行變換,使得采樣操作變為可導的。這有助于在訓練過程中通過梯度下降來優化網絡參數。"""# 計算采樣點的對數概率密度# log_prob維度: [batch_size, action_dim]log_prob = dist.log_prob(normal_sample)# 使用tanh函數將動作限制在[-1, 1]范圍內# action維度: [batch_size, action_dim]action = torch.tanh(normal_sample)# 計算tanh_normal分布的對數概率密度# 根據變換的雅可比行列式調整概率密度# 避免數值不穩定性,添加小常數1e-7# log_prob維度: [batch_size, action_dim]log_prob = log_prob - torch.log(1 - torch.tanh(action).pow(2) + 1e-7)# 將動作縮放到實際的動作邊界范圍# action維度: [batch_size, action_dim]action = action * self.action_boundreturn action, log_probclass QValueNetContinuous(torch.nn.Module):"""連續動作空間的Q值網絡輸入狀態和動作,輸出對應的Q值"""def __init__(self, state_dim, hidden_dim, action_dim):"""初始化Q值網絡參數:state_dim (int): 狀態空間維度,維度: 標量hidden_dim (int): 隱藏層維度,維度: 標量action_dim (int): 動作空間維度,維度: 標量"""super(QValueNetContinuous, self).__init__()# 第一個全連接層:拼接狀態和動作后的輸入層# 輸入維度: [batch_size, state_dim + action_dim]# 輸出維度: [batch_size, hidden_dim]self.fc1 = torch.nn.Linear(state_dim + action_dim, hidden_dim)# 第二個隱藏層# 輸入維度: [batch_size, hidden_dim]# 輸出維度: [batch_size, hidden_dim]self.fc2 = torch.nn.Linear(hidden_dim, hidden_dim)# 輸出層:輸出Q值# 輸入維度: [batch_size, hidden_dim]# 輸出維度: [batch_size, 1]self.fc_out = torch.nn.Linear(hidden_dim, 1)def forward(self, x, a):"""前向傳播函數參數:x (torch.Tensor): 輸入狀態,維度: [batch_size, state_dim]a (torch.Tensor): 輸入動作,維度: [batch_size, action_dim]返回:torch.Tensor: Q值,維度: [batch_size, 1]"""# 將狀態和動作拼接作為網絡輸入# cat維度: [batch_size, state_dim + action_dim]cat = torch.cat([x, a], dim=1)# 第一層激活# x維度: [batch_size, state_dim + action_dim] -> [batch_size, hidden_dim]x = F.relu(self.fc1(cat))# 第二層激活# x維度: [batch_size, hidden_dim] -> [batch_size, hidden_dim]x = F.relu(self.fc2(x))# 輸出Q值# 返回值維度: [batch_size, 1]return self.fc_out(x)class SACContinuous:"""SAC (Soft Actor-Critic) 算法實現類處理連續動作空間的強化學習問題SAC 使用兩個 Critic 網絡來使 Actor 的訓練更穩定,而這兩個 Critic 網絡在訓練時則各自需要一個目標價值網絡。因此,SAC 算法一共用到 5 個網絡,分別是一個策略網絡、兩個價值網絡和兩個目標價值網絡。"""def __init__(self, state_dim, hidden_dim, action_dim, action_bound,actor_lr, critic_lr, alpha_lr, target_entropy, tau, gamma,device):"""初始化SAC算法參數:state_dim (int): 狀態空間維度,維度: 標量hidden_dim (int): 隱藏層維度,維度: 標量action_dim (int): 動作空間維度,維度: 標量action_bound (float): 動作邊界值,維度: 標量actor_lr (float): 策略網絡學習率,維度: 標量critic_lr (float): 價值網絡學習率,維度: 標量alpha_lr (float): 溫度參數學習率,維度: 標量target_entropy (float): 目標熵值,維度: 標量tau (float): 軟更新參數,維度: 標量gamma (float): 折扣因子,維度: 標量device (torch.device): 計算設備,維度: 設備對象"""# 策略網絡:輸出動作分布# self.actor維度: PolicyNetContinuous對象self.actor = PolicyNetContinuous(state_dim, hidden_dim, action_dim,action_bound).to(device)# 第一個Q網絡:評估狀態-動作價值# self.critic_1維度: QValueNetContinuous對象self.critic_1 = QValueNetContinuous(state_dim, hidden_dim,action_dim).to(device)# 第二個Q網絡:評估狀態-動作價值# self.critic_2維度: QValueNetContinuous對象self.critic_2 = QValueNetContinuous(state_dim, hidden_dim,action_dim).to(device)# 第一個目標Q網絡:用于計算目標Q值# self.target_critic_1維度: QValueNetContinuous對象self.target_critic_1 = QValueNetContinuous(state_dim,hidden_dim, action_dim).to(device)# 第二個目標Q網絡:用于計算目標Q值# self.target_critic_2維度: QValueNetContinuous對象self.target_critic_2 = QValueNetContinuous(state_dim,hidden_dim, action_dim).to(device)# 令目標Q網絡的初始參數和Q網絡一樣self.target_critic_1.load_state_dict(self.critic_1.state_dict())self.target_critic_2.load_state_dict(self.critic_2.state_dict())# 策略網絡優化器# self.actor_optimizer維度: torch.optim.Adam對象self.actor_optimizer = torch.optim.Adam(self.actor.parameters(),lr