做搜狗手机网站快速排,网站页面设计最宽可做多宽,用ftp上传wordpress,网站备案号 英文大家好#xff0c;今天和各位分享一下多智能体深度强化学习算法 ippo#xff0c;并基于 gym 环境完成一个小案例。完整代码可以从我的 GitHub 中获得#xff1a;https://github.com/LiSir-HIT/Reinforcement-Learning/tree/main/Model 1. 算法原理
多智能体的情形相比于单智…大家好今天和各位分享一下多智能体深度强化学习算法 ippo并基于 gym 环境完成一个小案例。完整代码可以从我的 GitHub 中获得https://github.com/LiSir-HIT/Reinforcement-Learning/tree/main/Model 1. 算法原理
多智能体的情形相比于单智能体更加复杂因为每个智能体在和环境交互的同时也在和其他智能体进行直接或者间接的交互。因此多智能体强化学习要比单智能体更困难其难点主要体现在以下几点
1由于多个智能体在环境中进行实时动态交互并且每个智能体在不断学习并更新自身策略因此在每个智能体的视角下环境是非稳态的即对于一个智能体而言即使在相同的状态下采取相同的动作得到的状态转移和奖励信号的分布可能在不断改变
2多个智能体的训练可能是多目标的不同智能体需要最大化自己的利益
3训练评估的复杂度会增加可能需要大规模分布式训练来提高效率。
iPPO 算法的模型部分和 PPO 类似可以看我下面这篇博文
https://blog.csdn.net/dgvv4/article/details/129496576?spm1001.2014.3001.5501 IPPOIndependent PPO是一种完全去中心化的算法此类算法被称为独立学习。由于对于每个智能体使用单智能体算法 PPO 进行训练所因此这个算法叫作独立 PPO 算法。
这里使用的 PPO 算法版本为 PPO-截断其算法流程如下 2. 代码实现
代码和 ppo 离散模型基本相同
# 和PPO离散模型基本一致
import torch
from torch import nn
from torch.nn import functional as F
import numpy as np# ----------------------------------------- #
# 策略网络--actor
# ----------------------------------------- #class PolicyNet(nn.Module): # 输入当前状态输出动作的概率分布def __init__(self, n_states, n_hiddens, n_actions):super(PolicyNet, self).__init__()self.fc1 nn.Linear(n_states, n_hiddens)self.fc2 nn.Linear(n_hiddens, n_hiddens)self.fc3 nn.Linear(n_hiddens, n_actions)def forward(self, x): # [b,n_states]x self.fc1(x) # [b,n_states]--[b,n_hiddens]x F.relu(x)x self.fc2(x) # [b,n_hiddens]--[b,n_hiddens]x F.relu(x)x self.fc3(x) # [b,n_hiddens]--[b,n_actions]x F.softmax(x, dim1) # 每种动作选择的概率return x# ----------------------------------------- #
# 价值网络--critic
# ----------------------------------------- #class ValueNet(nn.Module): # 评价当前状态的价值def __init__(self, n_states, n_hiddens):super(ValueNet, self).__init__()self.fc1 nn.Linear(n_states, n_hiddens)self.fc2 nn.Linear(n_hiddens, n_hiddens)self.fc3 nn.Linear(n_hiddens, 1)def forward(self, x): # [b,n_states]x self.fc1(x) # [b,n_states]--[b,n_hiddens]x F.relu(x)x self.fc2(x) # [b,n_hiddens]--[b,n_hiddens]x F.relu(x)x self.fc3(x) # [b,n_hiddens]--[b,1]return x# ----------------------------------------- #
# 模型构建
# ----------------------------------------- #class PPO:def __init__(self, n_states, n_hiddens, n_actions,actor_lr, critic_lr, lmbda, eps, gamma, device):# 属性分配self.n_hiddens n_hiddensself.actor_lr actor_lr # 策略网络的学习率self.critic_lr critic_lr # 价值网络的学习率self.lmbda lmbda # 优势函数的缩放因子self.eps eps # ppo截断范围缩放因子self.gamma gamma # 折扣因子self.device device# 网络实例化self.actor PolicyNet(n_states, n_hiddens, n_actions).to(device) # 策略网络self.critic ValueNet(n_states, n_hiddens).to(device) # 价值网络# 优化器self.actor_optimizer torch.optim.Adam(self.actor.parameters(), lractor_lr)self.critic_optimizer torch.optim.Adam(self.critic.parameters(), lrcritic_lr)# 动作选择def take_action(self, state): # [n_states]state torch.tensor([state], dtypetorch.float).to(self.device) # [1,n_states]probs self.actor(state) # 当前状态的动作概率 [b,n_actions]action_dist torch.distributions.Categorical(probs) # 构造概率分布action action_dist.sample().item() # 从概率分布中随机取样 intreturn action# 训练def update(self, transition_dict):# 取出数据集states torch.tensor(transition_dict[states], dtypetorch.float).to(self.device) # [b,n_states]actions torch.tensor(transition_dict[actions]).view(-1,1).to(self.device) # [b,1]next_states torch.tensor(transition_dict[next_states], dtypetorch.float).to(self.device) # [b,n_states]dones torch.tensor(transition_dict[dones], dtypetorch.float).view(-1,1).to(self.device) # [b,1]rewards torch.tensor(transition_dict[rewards], dtypetorch.float).view(-1,1).to(self.device) # [b,1]# 价值网络next_state_value self.critic(next_states) # 下一时刻的state_value [b,1]td_target rewards self.gamma * next_state_value * (1-dones) # 目标--当前时刻的state_value [b,1]td_value self.critic(states) # 预测--当前时刻的state_value [b,1]td_delta td_value - td_target # 时序差分 # [b,1]# 计算GAE优势函数当前状态下某动作相对于平均的优势advantage 0 # 累计一个序列上的优势函数advantage_list [] # 存放每个时序的优势函数值td_delta td_delta.cpu().detach().numpy() # gpu--numpyfor delta in td_delta[::-1]: # 逆序取出时序差分值advantage self.gamma * self.lmbda * advantage deltaadvantage_list.append(advantage) # 保存每个时刻的优势函数advantage_list.reverse() # 正序advantage torch.tensor(advantage_list, dtypetorch.float).to(self.device)# 计算当前策略下状态s的行为概率 / 在之前策略下状态s的行为概率old_log_probs torch.log(self.actor(states).gather(1,actions)) # [b,1]log_probs torch.log(self.actor(states).gather(1,actions))ratio log_probs / old_log_probs# clip截断surr1 ratio * advantagesurr2 torch.clamp(ratio, 1-self.eps, 1self.eps) * advantage# 损失计算actor_loss torch.mean(-torch.min(surr1, surr2)) # clip截断critic_loss torch.mean(F.mse_loss(td_value, td_target)) # # 梯度更新self.actor_optimizer.zero_grad()self.critic_optimizer.zero_grad()actor_loss.backward()critic_loss.backward()self.actor_optimizer.step()self.critic_optimizer.step() 3. 案例演示
ma-gym 库中的 Combat 环境。Combat 是一个在二维的格子世界上进行的两个队伍的对战模拟游戏每个智能体的动作集合为向四周移动格攻击周围格范围内其他敌对智能体或者不采取任何行动。起初每个智能体有 3 点生命值如果智能体在敌人的攻击范围内被攻击到了则会扣 1 生命值生命值掉为 0 则死亡最后存活的队伍获胜。每个智能体的攻击有一轮的冷却时间。
IPPO 代码实践的最主要部分。值得注意的是在训练时使用了参数共享parameter sharing的技巧即对于所有智能体使用同一套策略参数这样做的好处是能够使得模型训练数据更多同时训练更稳定。能够这样做的前提是两个智能体是同质的homogeneous即它们的状态空间和动作空间是完全一致的并且它们的优化目标也完全一致。感兴趣的读者也可以自行实现非参数共享版本的 IPPO此时每个智能体就是一个独立的 PPO 的实例。
import numpy as np
import matplotlib.pyplot as plt
import torch
from ma_gym.envs.combat.combat import Combat
from RL_brain import PPO
import time# ----------------------------------------- #
# 参数设置
# ----------------------------------------- #n_hiddens 64 # 隐含层数量
actor_lr 3e-4
critic_lr 1e-3
gamma 0.9
lmbda 0.97
eps 0.2
device torch.device(cuda) if torch.cuda.is_available() \else torch.device(cpu)
num_episodes 10 # 回合数
team_size 2 # 智能体数量
grid_size (15, 15)# ----------------------------------------- #
# 环境设置--onpolicy
# ----------------------------------------- ## 创建Combat环境格子世界的大小为15x15己方智能体和敌方智能体数量都为2
env Combat(grid_shapegrid_size, n_agentsteam_size, n_opponentsteam_size)
n_states env.observation_space[0].shape[0] # 状态数
n_actions env.action_space[0].n # 动作数# 两个智能体共享同一个策略
agent PPO(n_states n_states,n_hiddens n_hiddens,n_actions n_actions,actor_lr actor_lr,critic_lr critic_lr,lmbda lmbda,eps eps,gamma gamma,device device,)# ----------------------------------------- #
# 模型训练
# ----------------------------------------- #for i in range(num_episodes):# 每回合开始前初始化两支队伍的数据集transition_dict_1 {states: [],actions: [],next_states: [],rewards: [],dones: [],}transition_dict_2 {states: [],actions: [],next_states: [],rewards: [],dones: [],}s env.reset() # 状态初始化terminal False # 结束标记while not terminal:env.render()# 动作选择a_1 agent.take_action(s[0])a_2 agent.take_action(s[1])# 环境更新next_s, r, done, info env.step([a_1, a_2])# 构造数据集transition_dict_1[states].append(s[0])transition_dict_1[actions].append(a_1)transition_dict_1[next_states].append(next_s[0])transition_dict_1[dones].append(False)transition_dict_1[rewards].append(r[0])transition_dict_2[states].append(s[1])transition_dict_2[actions].append(a_2)transition_dict_2[next_states].append(next_s[1])transition_dict_2[dones].append(False)transition_dict_2[rewards].append(r[1])s next_s # 状态更新terminal all(done) # 判断当前回合是否都为True是返回True不是返回Falsetime.sleep(0.1)print(epoch:, i)# 回合训练agent.update(transition_dict_1)agent.update(transition_dict_2)
