使用 Python TorchRL 進(jìn)行多代理強(qiáng)化學(xué)習(xí)
隨著多代理系統(tǒng)的出現(xiàn),強(qiáng)化學(xué)習(xí)的復(fù)雜性不斷增加。為了管理這種復(fù)雜性,像 TorchRL 這樣的專門工具提供了一個(gè)強(qiáng)大的框架,可以開發(fā)和實(shí)驗(yàn)多代理強(qiáng)化學(xué)習(xí)(MARL)算法。本文將深入探討如何使用 TorchRL 解決 MARL 問題,重點(diǎn)關(guān)注多代理環(huán)境中的近端策略優(yōu)化(PPO)。
我們將使用 VMAS 模擬器,這是一個(gè)多機(jī)器人模擬器并且可以在 GPU 上進(jìn)行并行訓(xùn)練。他的主要目標(biāo)多個(gè)機(jī)器人必須導(dǎo)航到各自的目標(biāo),同時(shí)避免碰撞。
依賴
在開始之前,請(qǐng)確保安裝以下依賴項(xiàng):
!pip3 install torchrl
!pip3 install vmas
!pip3 install tqdm理解近端策略優(yōu)化 (PPO)
PPO 是一種策略梯度算法,它迭代地從環(huán)境中采樣數(shù)據(jù),并直接使用這些數(shù)據(jù)來優(yōu)化策略。這個(gè)過程包括采樣和訓(xùn)練兩個(gè)階段,數(shù)據(jù)在收集后立即進(jìn)行訓(xùn)練更新。這種在線方法確保策略根據(jù)與環(huán)境最近的交互持續(xù)改進(jìn)。
在線學(xué)習(xí)
在 PPO 中,學(xué)習(xí)過程依賴于一個(gè)評(píng)論家(critic),它評(píng)估策略所采取行動(dòng)的質(zhì)量。評(píng)論家估計(jì)給定狀態(tài)的價(jià)值,通過比較預(yù)期回報(bào)與實(shí)際結(jié)果來指導(dǎo)策略優(yōu)化。
在多代理設(shè)置中,我們部署多個(gè)策略,每個(gè)代理一個(gè),通常以分散的方式運(yùn)作。每個(gè)代理的策略僅根據(jù)其局部觀察來決定其行動(dòng)。但是評(píng)論家可以是集中的或分散的:
- MAPPO : 評(píng)論家是集中的,以全局觀察或連接的代理觀察作為輸入。這種方法在可獲得全局狀態(tài)信息的集中式訓(xùn)練場(chǎng)景中有益。
- IPPO : 評(píng)論家是分散的,僅依賴于局部觀察。這種設(shè)置支持分散式訓(xùn)練,代理只需要局部信息。
集中式評(píng)論家有助于緩解多個(gè)代理同時(shí)學(xué)習(xí)時(shí)出現(xiàn)的非平穩(wěn)性問題,但可能因輸入的高維度性而面臨挑戰(zhàn)。
TorchRL
TorchRL是一個(gè)基于PyTorch的強(qiáng)化學(xué)習(xí)（Reinforcement Learning, RL）庫，專為研究人員和開發(fā)者設(shè)計(jì)，旨在提供一個(gè)靈活、高效的框架來實(shí)現(xiàn)和實(shí)驗(yàn)各種RL算法。
- 與PyTorch深度集成：TorchRL充分利用了PyTorch的生態(tài)系統(tǒng)，使用戶能夠無縫地將RL算法與深度學(xué)習(xí)模型結(jié)合。
- 模塊化設(shè)計(jì)：庫提供了可組合的組件，允許用戶輕松構(gòu)建和定制RL算法。
- 高性能：TorchRL注重效率，支持GPU加速和并行化，以加快訓(xùn)練和推理速度。
- 多環(huán)境支持：兼容多種RL環(huán)境，包括OpenAI Gym、DeepMind Control Suite等。
- 豐富的算法實(shí)現(xiàn)：內(nèi)置多種流行的RL算法，如DQN、PPO、SAC等。
- 擴(kuò)展性：易于擴(kuò)展和添加新的算法、環(huán)境和功能。
下面代碼我們將使用TorchRL來完成我們的目標(biāo)
1、設(shè)置超參數(shù)
我們從定義 MARL 設(shè)置的超參數(shù)開始。這些參數(shù)控制模擬和訓(xùn)練過程的各個(gè)方面,如設(shè)備類型、批量大小、學(xué)習(xí)率和 PPO 特定設(shè)置。
import torch
from torch import multiprocessing
# 設(shè)置設(shè)備
is_fork = multiprocessing.get_start_method() == "fork"
device = torch.device(0) if torch.cuda.is_available() and not is_fork else torch.device("cpu")
vmas_device = device # 運(yùn)行 VMAS 模擬器的設(shè)備
# 采樣和訓(xùn)練參數(shù)
frames_per_batch = 6000
n_iters = 10
total_frames = frames_per_batch * n_iters
# 訓(xùn)練細(xì)節(jié)
num_epochs = 30
minibatch_size = 400
lr = 3e-4
max_grad_norm = 1.0
# PPO 參數(shù)
clip_epsilon = 0.2
gamma = 0.99
lmbda = 0.9
entropy_eps = 1e-42、創(chuàng)建環(huán)境
TorchRL 與 VMAS 的集成允許我們高效地創(chuàng)建和管理多代理環(huán)境。在我們環(huán)境中多個(gè)代理必須在 LIDAR 傳感器的引導(dǎo)下導(dǎo)航到各自的目標(biāo),同時(shí)避免碰撞。
from torchrl.envs.libs.vmas import VmasEnv
max_steps = 100
num_vmas_envs = frames_per_batch // max_steps
scenario_name = "navigation"
n_agents = 3
env = VmasEnv(
scenario=scenario_name,
num_envs=num_vmas_envs,
continuous_actions=True,
max_steps=max_steps,
device=vmas_device,
n_agents=n_agents,
)3、策略設(shè)計(jì)
策略網(wǎng)絡(luò)在 PPO 中至關(guān)重要,它負(fù)責(zé)根據(jù)代理觀察生成動(dòng)作。鑒于環(huán)境中的連續(xù)動(dòng)作空間,我們將使用 Tanh-Normal 分布來模擬動(dòng)作，這樣還可以決定是否在代理之間共享參數(shù),在計(jì)算效率和行為多樣性之間權(quán)衡。
from torch.nn import Sequential, Tanh
from tensordict.nn import TensorDictModule
from torchrl.modules import MultiAgentMLP, ProbabilisticActor, TanhNormal
from tensordict.nn.distributions import NormalParamExtractor
share_parameters_policy = True
# 定義策略網(wǎng)絡(luò)
policy_net = Sequential(
MultiAgentMLP(
n_agent_inputs=env.observation_spec["agents", "observation"].shape[-1],
n_agent_outputs=2 * env.action_spec.shape[-1],
n_agents=env.n_agents,
centralised=False,
share_params=share_parameters_policy,
device=device,
depth=2,
num_cells=256,
activation_class=Tanh,
),
NormalParamExtractor(),
)
# 將網(wǎng)絡(luò)包裝在 TensorDictModule 中
policy_module = TensorDictModule(
policy_net,
in_keys=[("agents", "observation")],
out_keys=[("agents", "loc"), ("agents", "scale")],
)
# 創(chuàng)建概率性行動(dòng)者
policy = ProbabilisticActor(
module=policy_module,
spec=env.unbatched_action_spec,
in_keys=[("agents", "loc"), ("agents", "scale")],
out_keys=[env.action_key],
distribution_class=TanhNormal,
distribution_kwargs={
"low": env.unbatched_action_spec[env.action_key].space.low,
"high": env.unbatched_action_spec[env.action_key].space.high,
},
return_log_prob=True,
log_prob_key=("agents", "sample_log_prob"),
)4、評(píng)論家網(wǎng)絡(luò)設(shè)計(jì)
評(píng)論家網(wǎng)絡(luò)評(píng)估狀態(tài)值,指導(dǎo)策略更新?？梢愿鶕?jù)使用 MAPPO 還是 IPPO 來選擇集中式或分散式評(píng)論家。在代理之間共享參數(shù)可以加速訓(xùn)練,但是可能導(dǎo)致同質(zhì)化策略。
share_parameters_critic = True
mappo = True # 設(shè)置為 False 以使用 IPPO
critic_net = MultiAgentMLP(
n_agent_inputs=env.observation_spec["agents", "observation"].shape[-1],
n_agent_outputs=1,
n_agents=env.n_agents,
centralised=mappo,
share_params=share_parameters_critic,
device=device,
depth=2,
num_cells=256,
activation_class=Tanh,
)
critic = TensorDictModule(
module=critic_net,
in_keys=[("agents", "observation")],
out_keys=[("agents", "state_value")],
)5、數(shù)據(jù)收集
TorchRL 中的數(shù)據(jù)收集通過設(shè)計(jì)用于自動(dòng)處理環(huán)境重置、動(dòng)作計(jì)算和環(huán)境步進(jìn)的類得到簡(jiǎn)化。所以我們可以直接使用 SyncDataCollector 來收集訓(xùn)練數(shù)據(jù)。
from torchrl.collectors import SyncDataCollector
collector = SyncDataCollector(
env,
policy,
device=vmas_device,
storing_device=device,
frames_per_batch=frames_per_batch,
total_frames=total_frames,
)6、訓(xùn)練循環(huán)
訓(xùn)練循環(huán)將環(huán)境、策略、評(píng)論家和數(shù)據(jù)收集器結(jié)合在一起,通過采樣和訓(xùn)練階段的迭代來優(yōu)化代理的性能。
from torchrl.data.replay_buffers import ReplayBuffer
from torchrl.data.replay_buffers.samplers import SamplerWithoutReplacement
from torchrl.data.replay_buffers.storages import LazyTensorStorage
from torchrl.objectives import ClipPPOLoss, ValueEstimators
from tqdm import tqdm
# 重放緩沖區(qū)設(shè)置
replay_buffer = ReplayBuffer(
storage=LazyTensorStorage(
size=frames_per_batch,
device=device,
),
sampler=SamplerWithoutReplacement(),
)
# PPO 損失設(shè)置
ppo_loss = ClipPPOLoss(
actor=policy,
critic=critic,
clip_epsilon=clip_epsilon,
entropy_bonus=entropy_eps,
value_loss_coef=0.5,
gamma=gamma,
lam=lmbda,
advantage_normalization=True,
)
# 訓(xùn)練循環(huán)
for i in tqdm(range(n_iters)):
batch = collector.next()
replay_buffer.extend(batch)
for _ in range(num_epochs):
for minibatch in replay_buffer.sample(minibatch_size):
loss = ppo_loss(minibatch)
loss.backward()
torch.nn.utils.clip_grad_norm_(ppo_loss.parameters(), max_grad_norm)
optimizer.step()
optimizer.zero_grad()這樣我們完整的代碼就完成了，可以看到通過TorchRL，可以減少我們很多的代碼開發(fā)工作。
總結(jié)
本文提供了使用 TorchRL 和 PPO 實(shí)現(xiàn) MARL 解決方案的全面指南。通過這些步驟,可以在多代理環(huán)境中訓(xùn)練代理以導(dǎo)航復(fù)雜任務(wù),同時(shí)利用 GPU 加速模擬和并行計(jì)算的力量。無論選擇集中式還是分散式評(píng)論家,TorchRL 都提供了設(shè)計(jì)和實(shí)驗(yàn)不同 MARL 架構(gòu),可以幫助你解決多代理強(qiáng)化學(xué)習(xí)的復(fù)雜性。
