使用Actor-Critic的DDPG強化學習算法控制雙關節機械臂
在本文中，我們將介紹在 Reacher 環境中訓練智能代理控制雙關節機械臂，這是一種使用 Unity ML-Agents 工具包開發的基于 Unity 的模擬程序。我們的目標是高精度的到達目標位置，所以這里我們可以使用專為連續狀態和動作空間設計的最先進的Deep Deterministic Policy Gradient (DDPG) 算法。
現實世界的應用程序
機械臂在制造業、生產設施、空間探索和搜救行動中發揮著關鍵作用。控制機械臂的高精度和靈活性是非常重要的。通過采用強化學習技術，可以使這些機器人系統實時學習和調整其行為，從而提高性能和靈活性。強化學習的進步不僅有助于我們對人工智能的理解，而且有可能徹底改變行業并對社會產生有意義的影響。
而Reacher是一種機械臂模擬器，常用于控制算法的開發和測試。它提供了一個虛擬環境，模擬了機械臂的物理特性和運動規律，使得開發者可以在不需要實際硬件的情況下進行控制算法的研究和實驗。
Reacher的環境主要由以下幾個部分組成：
- 機械臂：Reacher模擬了一個雙關節機械臂，包括一個固定基座和兩個可動關節。開發者可以通過控制機械臂的兩個關節來改變機械臂的姿態和位置。
- 目標點：在機械臂的運動范圍內，Reacher提供了一個目標點，目標點的位置是隨機生成的。開發者的任務是控制機械臂，使得機械臂的末端能夠接觸到目標點。
- 物理引擎：Reacher使用物理引擎來模擬機械臂的物理特性和運動規律。開發者可以通過調整物理引擎的參數來模擬不同的物理環境。
- 視覺界面：Reacher提供了一個可視化的界面，可以顯示機械臂和目標點的位置，以及機械臂的姿態和運動軌跡。開發者可以通過視覺界面來調試和優化控制算法。
Reacher模擬器是一個非常實用的工具，可以幫助開發者在不需要實際硬件的情況下，快速測試和優化控制算法。
模擬環境
Reacher 使用 Unity ML-Agents 工具包構建，我們的代理可以控制雙關節機械臂。目標是引導手臂朝向目標位置并盡可能長時間地保持其在目標區域內的位置。該環境具有 20 個同步代理，每個代理獨立運行，這有助于在訓練期間有效地收集經驗。
狀態和動作空間
了解狀態和動作空間對于設計有效的強化學習算法至關重要。在 Reacher 環境中，狀態空間由 33 個連續變量組成，這些變量提供有關機械臂的信息，例如其位置、旋轉、速度和角速度。動作空間也是連續的，四個變量對應于施加在機械臂兩個關節上的扭矩。每個動作變量都是一個介于 -1 和 1 之間的實數。
任務類型和成功標準
Reacher 任務被認為是片段式的，每個片段都包含固定數量的時間步長。代理的目標是在這些步驟中最大化其總獎勵。手臂末端執行器保持在目標位置的每一步都會獲得 +0.1 的獎勵。當代理在連續 100 次操作中的平均得分達到 30 分或以上時，就認為成功。
了解了環境，下面我們將探討 DDPG 算法、它的實現，以及它如何有效地解決這種環境中的連續控制問題。
連續控制的算法選擇：DDPG
當涉及到像Reacher問題這樣的連續控制任務時，算法的選擇對于實現最佳性能至關重要。在這個項目中，我們選擇了DDPG算法，因為這是一種專門設計用于處理連續狀態和動作空間的actor-critic方法。
DDPG算法通過結合兩個神經網絡，結合了基于策略和基于值的方法的優勢:行動者網絡(Actor network)決定給定當前狀態下的最佳行為，批評家網絡(Critic network)估計狀態-行為值函數(Q-function)。這兩種網絡都有目標網絡，通過在更新過程中提供一個固定的目標來穩定學習過程。
通過使用Critic網絡估計q函數，使用Actor網絡確定最優行為，DDPG算法有效地融合了策略梯度方法和DQN的優點。這種混合方法允許代理在連續控制環境中有效地學習。
import random
from collections import deque
import torch
import torch.nn as nn
import numpy as np
from actor_critic import Actor, Critic
class ReplayBuffer:
def __init__(self, buffer_size, batch_size):
self.memory = deque(maxlen=buffer_size)
self.batch_size = batch_size
def add(self, state, action, reward, next_state, done):
self.memory.append((state, action, reward, next_state, done))
def sample(self):
batch = random.sample(self.memory, self.batch_size)
states, actions, rewards, next_states, dones = zip(*batch)
return states, actions, rewards, next_states, dones
def __len__(self):
return len(self.memory)
class DDPG:
def __init__(self, state_dim, action_dim, hidden_dim, buffer_size, batch_size, actor_lr, critic_lr, tau, gamma):
self.actor = Actor(state_dim, hidden_dim, action_dim, actor_lr)
self.actor_target = Actor(state_dim, hidden_dim, action_dim, actor_lr)
self.critic = Critic(state_dim, action_dim, hidden_dim, critic_lr)
self.critic_target = Critic(state_dim, action_dim, hidden_dim, critic_lr)
self.memory = ReplayBuffer(buffer_size, batch_size)
self.batch_size = batch_size
self.tau = tau
self.gamma = gamma
self._update_target_networks(tau=1) # initialize target networks
def act(self, state, noise=0.0):
state = torch.tensor(state, dtype=torch.float32).unsqueeze(0)
action = self.actor(state).detach().numpy()[0]
return np.clip(action + noise, -1, 1)
def store_transition(self, state, action, reward, next_state, done):
self.memory.add(state, action, reward, next_state, done)
def learn(self):
if len(self.memory) < self.batch_size:
return
states, actions, rewards, next_states, dones = self.memory.sample()
states = torch.tensor(states, dtype=torch.float32)
actions = torch.tensor(actions, dtype=torch.float32)
rewards = torch.tensor(rewards, dtype=torch.float32).unsqueeze(1)
next_states = torch.tensor(next_states, dtype=torch.float32)
dones = torch.tensor(dones, dtype=torch.float32).unsqueeze(1)
# Update Critic
self.critic.optimizer.zero_grad()
with torch.no_grad():
next_actions = self.actor_target(next_states)
target_q_values = self.critic_target(next_states, next_actions)
target_q_values = rewards + (1 - dones) * self.gamma * target_q_values
current_q_values = self.critic(states, actions)
critic_loss = nn.MSELoss()(current_q_values, target_q_values)
critic_loss.backward()
self.critic.optimizer.step()
# Update Actor
self.actor.optimizer.zero_grad()
actor_loss = -self.critic(states, self.actor(states)).mean()
actor_loss.backward()
self.actor.optimizer.step()
# Update target networks
self._update_target_networks()
def _update_target_networks(self, tau=None):
if tau is None:
tau = self.tau
for target_param, param in zip(self.actor_target.parameters(), self.actor.parameters()):
target_param.data.copy_(tau * param.data + (1 - tau) * target_param.data)
for target_param, param in zip(self.critic_target.parameters(), self.critic.parameters()):
target_param.data.copy_(tau * param.data + (1 - tau) * target_param.data)上面的代碼還使用了Replay Buffer，這可以提高學習效率和穩定性。Replay Buffer本質上是一種存儲固定數量的過去經驗或過渡的內存數據結構，由狀態、動作、獎勵、下一狀態和完成信息組成。使用它的主要優點是使代理能夠打破連續經驗之間的相關性，從而減少有害的時間相關性的影響。
通過從緩沖區中抽取隨機的小批量經驗，代理可以從一組不同的轉換中學習，這有助于穩定和概括學習過程。Replay Buffer還可以讓代理多次重用過去的經驗，從而提高數據效率并促進從與環境的有限交互中更有效地學習。
DDPG算法是一個很好的選擇，因為它能夠有效地處理連續的動作空間，這是這個環境的一個關鍵方面。該算法的設計允許有效地利用多個代理收集的并行經驗，從而實現更快的學習和更好的收斂。就像上面介紹的Reacher 可以同時運行20個代理，所以我們可以使用這20個代理進行分享經驗，集體學習，提高學習速度。
完成了算法，下面我們將介紹、超參數選擇和訓練過程。
DDPG算法在Reacher 環境中工作
為了更好地理解算法在環境中的有效性，我們需要仔細研究學習過程中涉及的關鍵組件和步驟。
網絡架構
DDPG算法采用兩個神經網絡，Actor 和Critic。兩個網絡都包含兩個隱藏層，每個隱藏層包含400個節點。隱藏層使用ReLU (Rectified Linear Unit)激活函數，而Actor網絡的輸出層使用tanh激活函數產生范圍為-1到1的動作。Critic網絡的輸出層沒有激活函數，因為它直接估計q函數。
以下是網絡的代碼:
import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim
class Actor(nn.Module):
def __init__(self, input_dim, hidden_dim, output_dim, learning_rate=1e-4):
super(Actor, self).__init__()
self.fc1 = nn.Linear(input_dim, hidden_dim)
self.fc2 = nn.Linear(hidden_dim, hidden_dim)
self.fc3 = nn.Linear(hidden_dim, output_dim)
self.tanh = nn.Tanh()
self.optimizer = optim.Adam(self.parameters(), lr=learning_rate)
def forward(self, state):
x = torch.relu(self.fc1(state))
x = torch.relu(self.fc2(x))
x = self.tanh(self.fc3(x))
return x
class Critic(nn.Module):
def __init__(self, state_dim, action_dim, hidden_dim, learning_rate=1e-4):
super(Critic, self).__init__()
self.fc1 = nn.Linear(state_dim, hidden_dim)
self.fc2 = nn.Linear(hidden_dim + action_dim, hidden_dim)
self.fc3 = nn.Linear(hidden_dim, 1)
self.optimizer = optim.Adam(self.parameters(), lr=learning_rate)
def forward(self, state, action):
x = torch.relu(self.fc1(state))
x = torch.relu(self.fc2(torch.cat([x, action], dim=1)))
x = self.fc3(x)
return x超參數選擇
選擇的超參數對于高效學習至關重要。在這個項目中，我們的Replay Buffer大小為200,000，批大小為256。演員Actor的學習率為5e-4，Critic的學習率為1e-3，soft update參數(tau)為5e-3，gamma為0.995。最后還加入了動作噪聲，初始噪聲標度為0.5，噪聲衰減率為0.998。
訓練過程
訓練過程涉及兩個網絡之間的持續交互，并且20個平行代理共享相同的網絡，模型會從所有代理收集的經驗中集體學習。這種設置加快了學習過程，提高了效率。
from collections import deque
import numpy as np
import torch
from ddpg import DDPG
def train_ddpg(env, agent, episodes, max_steps, num_agents, noise_scale=0.1, noise_decay=0.99):
scores_window = deque(maxlen=100)
scores = []
for episode in range(1, episodes + 1):
env_info = env.reset(train_mode=True)[brain_name]
states = env_info.vector_observations
agent_scores = np.zeros(num_agents)
for step in range(max_steps):
actions = agent.act(states, noise_scale)
env_info = env.step(actions)[brain_name]
next_states = env_info.vector_observations
rewards = env_info.rewards
dones = env_info.local_done
for i in range(num_agents):
agent.store_transition(states[i], actions[i], rewards[i], next_states[i], dones[i])
agent.learn()
states = next_states
agent_scores += rewards
noise_scale *= noise_decay
if np.any(dones):
break
avg_score = np.mean(agent_scores)
scores_window.append(avg_score)
scores.append(avg_score)
if episode % 10 == 0:
print(f"Episode: {episode}, Score: {avg_score:.2f}, Avg Score: {np.mean(scores_window):.2f}")
# Saving trained Networks
torch.save(agent.actor.state_dict(), "actor_final.pth")
torch.save(agent.critic.state_dict(), "critic_final.pth")
return scores
if __name__ == "__main__":
env = UnityEnvironment(file_name='Reacher_20.app')
brain_name = env.brain_names[0]
brain = env.brains[brain_name]
state_dim = 33
action_dim = brain.vector_action_space_size
num_agents = 20
# Hyperparameter suggestions
hidden_dim = 400
batch_size = 256
actor_lr = 5e-4
critic_lr = 1e-3
tau = 5e-3
gamma = 0.995
noise_scale = 0.5
noise_decay = 0.998
agent = DDPG(state_dim, action_dim, hidden_dim=hidden_dim, buffer_size=200000, batch_size=batch_size,
actor_lr=actor_lr, critic_lr=critic_lr, tau=tau, gamma=gamma)
episodes = 200
max_steps = 1000
scores = train_ddpg(env, agent, episodes, max_steps, num_agents, noise_scale=0.2, noise_decay=0.995)訓練過程中的關鍵步驟如下所示：
初始化網絡：代理使用隨機權重初始化共享的 Actor 和 Critic 網絡及其各自的目標網絡。目標網絡在更新期間提供穩定的學習目標。
- 與環境交互：每個代理使用共享的 Actor 網絡，通過根據其當前狀態選擇動作來與環境交互。為了鼓勵探索，在訓練的初始階段還將噪聲項添加到動作中。采取行動后，每個代理都會觀察由此產生的獎勵和下一個狀態。
- 存儲經驗：每個代理將觀察到的經驗（狀態、動作、獎勵、next_state）存儲在共享重放緩沖區中。該緩沖區包含固定數量的近期經驗，這樣每個代理能夠從所有代理收集的各種轉換中學習。
- 從經驗中學習：定期從共享重放緩沖區中抽取一批經驗。通過最小化預測 Q 值和目標 Q 值之間的均方誤差，使用采樣經驗來更新共享 Critic 網絡。
- 更新 Actor 網絡：共享 Actor 網絡使用策略梯度進行更新，策略梯度是通過采用共享 Critic 網絡關于所選動作的輸出梯度來計算的。共享 Actor 網絡學習選擇最大化預期 Q 值的動作。
- 更新目標網絡：共享的 Actor 和 Critic 目標網絡使用當前和目標網絡權重的混合進行軟更新。這確保了穩定的學習過程。
結果展示
我們的agent使用DDPG算法成功地學會了在Racher環境下控制雙關節機械臂。在整個訓練過程中，我們根據所有20個代理的平均得分來監控代理的表現。隨著智能體探索環境和收集經驗，其預測獎勵最大化最佳行為的能力顯著提高。
可以看到代理在任務中表現出了顯著的熟練程度，平均得分超過了解決環境所需的閾值(30+)，雖然代理的表現在整個訓練過程中有所不同，但總體趨勢呈上升趨勢，表明學習過程是成功的。
下圖顯示了20個代理的平均得分:
可以看到我們實現的DDPG算法，有效地解決了Racher環境的問題。代理能夠調整自己的行為，并在任務中達到預期的性能。
下一步工作
本項目中的超參數是根據文獻和實證測試的建議組合選擇的。還可以通過系統超參數調優的進一步優化可能會帶來更好的性能。
多agent并行訓練:在這個項目中，我們使用20個agent同時收集經驗。使用更多代理對整個學習過程的影響可能會導致更快的收斂或提高性能。
批歸一化：為了進一步增強學習過程，在神經網絡架構中實現批歸一化是值得探索的。通過在訓練過程中對每一層的輸入特征進行歸一化，批歸一化可以幫助減少內部協變量移位，加速學習，并潛在地提高泛化。將批處理歸一化加入到Actor和Critic網絡可能會導致更穩定和有效的訓練，但是這個需要進一步測試。
