多智能体近端策略优化（MAPPO）算法详解

多智能体近端策略优化（MAPPO）算法

1. 背景与动机

PPO（Proximal Policy Optimization）近年来已成为强化学习领域最流行的算法之一。它通过引入裁剪的策略更新机制，有效解决了传统策略梯度方法（如 TRPO）中因步长过大导致的训练不稳定问题。

在多智能体环境中，多个智能体同时学习策略，彼此的行为会相互影响。因此，需要一个鲁棒且稳定的策略优化方法。MAPPO（Multi-Agent Proximal Policy Optimization）作为 PPO 的扩展，采用集中式 Critic 和去中心化 Actor 的架构，显著提高了多智能体环境下的学习效率和稳定性。

参考论文： The Surprising Effectiveness of PPO in Cooperative, Multi-Agent Games

2. 算法结构

MAPPO 继承了 PPO 的核心思想，并结合多智能体系统的特点，采用了集中式训练，分布式执行（CTDE）的架构：

集中式训练：在训练阶段，所有智能体的 Critic 网络能够访问全局状态和其他智能体的动作信息，从而学习到更准确的价值函数。
分布式执行：在执行阶段，每个智能体仅使用自己观测到的局部状态和策略进行动作选择，保证了系统的分布式控制能力。

3. 核心公式

MAPPO 算法主要包含两部分：策略更新和价值函数估计。

优势函数计算

优势函数 $\hat{A}_t^i$ 用于衡量某个动作 $a_t$ 相对于当前策略下平均动作的优劣程度。通常通过广义优势估计（GAE）进行估算：

$$\hat{A}t^i = \delta_t + (\gamma \lambda) \delta{t+1} + ... + (\gamma \lambda)^{T-t+1} \delta_{T-1}$$

其中 $\delta_t$ 是时间差分误差，定义为：

$$\delta_t = r_t + \gamma V(s_{t+1}) - V(s_t)$$

这里 $\lambda$ 是 GAE 中的权重参数，用于平衡偏差和方差。

价值函数估计

每个智能体 $i$ 的价值函数 $V_i(s)$ 由一个中心化的 Critic 网络估计。Critic 网络利用全局状态 $s$ 和所有智能体的动作 $a_1, a_2, ..., a_N$ 来估计全局价值。其目标是最小化均方误差（MSE）损失函数：

$$L(\phi_i) = \mathbb{E}{s_t, r_t, s{t+1}} \left[ \left( V_i(s_t; \phi_i) - R_t \right)^2 \right]$$

其中 $R_t$ 是从当前时刻 $t$ 到未来的累计回报，通常通过 TD 目标估计：

$$R_t = r_t + \gamma V_i(s_{t+1}; \phi'_i)$$

策略更新

PPO 引入了一个裁剪目标函数来限制每次更新的策略变化幅度。MAPPO 遵循相同原则，但应用在每个智能体 $i$ 的策略上。PPO 的目标函数为：

$$L^{CLIP}(\theta) = \mathbb{E}_t \left[ \min \left( r_t(\theta) \hat{A}_t, \text{clip}(r_t(\theta), 1 - \epsilon, 1 + \epsilon) \hat{A}_t \right) \right]$$

其中 $r_t(\theta) = \frac{\pi_{\theta}(a_t | s_t)}{\pi_{\theta_{\text{old}}}(a_t | s_t)}$ 是当前策略与旧策略的比率，$\epsilon$ 是裁剪阈值。

MAPPO 中每个智能体采用类似的目标函数：

$$L^{CLIP}_i(\theta_i) = \mathbb{E}_t \left[ \min \left( r_t(\theta_i) \hat{A}_t^i, \text{clip}(r_t(\theta_i), 1 - \epsilon, 1 + \epsilon) \hat{A}_t^i \right) \right]$$

4. 算法流程

交互与经验收集：每个智能体根据当前策略与环境交互，存储状态、动作、奖励、下一状态等信息。
：根据损失函数更新 Critic 参数，最小化预测值与真实回报的误差。

import torch import numpy as np from torch.utils.tensorboard import SummaryWriter import argparse from normalization import Normalization, RewardScaling from replay_buffer import ReplayBuffer from mappo_mpe import MAPPO_MPE from environment import Env class Runner_MAPPO_MPE: def __init__(self, args, env_name, number, seed): self.args = args self.env_name = env_name self.number = number self.seed = seed # Set random seed np.random.seed(self.seed) torch.manual_seed(self.seed) # Create env self.env = Env(env_name, discrete=True) self.args.N = self.env.n self.args.obs_dim_n = [self.env.observation_space[i].shape[0] for i in range(self.args.N)] self.args.action_dim_n = [self.env.action_space[i].n for i in range(self.args.N)] # Homogenous agents setup self.args.obs_dim = self.args.obs_dim_n[0] self.args.action_dim = self.args.action_dim_n[0] self.args.state_dim = np.sum(self.args.obs_dim_n) print("observation_space=", self.env.observation_space) print("obs_dim_n={}".format(self.args.obs_dim_n)) print("action_space=", self.env.action_space) print("action_dim_n={}".format(self.args.action_dim_n)) # Create N agents self.agent_n = MAPPO_MPE(self.args) self.replay_buffer = ReplayBuffer(self.args) # Tensorboard self.writer = SummaryWriter(log_dir='runs/MAPPO/MAPPO_env_{}_number_{}_seed_{}'.format( self.env_name, self.number, self.seed)) self.evaluate_rewards = [] self.total_steps = 0 if self.args.use_reward_norm: print("------use reward norm------") self.reward_norm = Normalization(shape=self.args.N) elif self.args.use_reward_scaling: print("------use reward scaling------") self.reward_scaling = RewardScaling(shape=self.args.N, gamma=self.args.gamma) def run(self): evaluate_num = -1 while self.total_steps < self.args.max_train_steps: if self.total_steps // self.args.evaluate_freq > evaluate_num: self.evaluate_policy() evaluate_num += 1 _, episode_steps = self.run_episode_mpe(evaluate=False) self.total_steps += episode_steps if self.replay_buffer.episode_num == self.args.batch_size: self.agent_n.train(self.replay_buffer, self.total_steps) self.replay_buffer.reset_buffer() self.evaluate_policy() self.env.close() def evaluate_policy(self): evaluate_reward = 0 for _ in range(self.args.evaluate_times): episode_reward, _ = self.run_episode_mpe(evaluate=True) evaluate_reward += episode_reward evaluate_reward /= self.args.evaluate_times self.evaluate_rewards.append(evaluate_reward) print("total_steps:{} \t evaluate_reward:{}".format(self.total_steps, evaluate_reward)) self.writer.add_scalar('evaluate_step_rewards_{}'.format(self.env_name), evaluate_reward, global_step=self.total_steps) np.save('./data_train/MAPPO_env_{}_number_{}_seed_{}.npy'.format( self.env_name, self.number, self.seed), np.array(self.evaluate_rewards)) self.agent_n.save_model(self.env_name, self.number, self.seed, self.total_steps) def run_episode_mpe(self, evaluate=False): episode_reward = 0 obs_n = self.env.reset() if self.args.use_reward_scaling: self.reward_scaling.reset() if self.args.use_rnn: self.agent_n.actor.rnn_hidden = None self.agent_n.critic.rnn_hidden = None for episode_step in range(self.args.episode_limit): a_n, a_logprob_n = self.agent_n.choose_action(obs_n, evaluate=evaluate) s = np.array(obs_n).flatten() v_n = self.agent_n.get_value(s) obs_next_n, r_n, done_n, _ = self.env.step(a_n) episode_reward += r_n[0] if not evaluate: if self.args.use_reward_norm: r_n = self.reward_norm(r_n) elif self.args.use_reward_scaling: r_n = self.reward_scaling(r_n) self.replay_buffer.store_transition(episode_step, obs_n, s, v_n, a_n, a_logprob_n, r_n, done_n) obs_n = obs_next_n if all(done_n): break if not evaluate: s = np.array(obs_n).flatten() v_n = self.agent_n.get_value(s) self.replay_buffer.store_last_value(episode_step + 1, v_n) return episode_reward, episode_step + 1 if __name__ == '__main__': parser = argparse.ArgumentParser("Hyperparameters Setting for MAPPO in MPE environment") parser.add_argument("--max_train_steps", type=int, default=int(3e6)) parser.add_argument("--episode_limit", type=int, default=25) parser.add_argument("--evaluate_freq", type=float, default=5000) parser.add_argument("--evaluate_times", type=float, default=3) parser.add_argument("--batch_size", type=int, default=32) parser.add_argument("--mini_batch_size", type=int, default=8) parser.add_argument("--rnn_hidden_dim", type=int, default=64) parser.add_argument("--mlp_hidden_dim", type=int, default=64) parser.add_argument("--lr", type=float, default=5e-4) parser.add_argument("--gamma", type=float, default=0.99) parser.add_argument("--lamda", type=float, default=0.95) parser.add_argument("--epsilon", type=float, default=0.2) parser.add_argument("--K_epochs", type=int, default=15) parser.add_argument("--use_adv_norm", type=bool, default=True) parser.add_argument("--use_reward_norm", type=bool, default=True) parser.add_argument("--use_reward_scaling", type=bool, default=False) parser.add_argument("--entropy_coef", type=float, default=0.01) parser.add_argument("--use_lr_decay", type=bool, default=True) parser.add_argument("--use_grad_clip", type=bool, default=True) parser.add_argument("--use_orthogonal_init", type=bool, default=True) parser.add_argument("--set_adam_eps", type=float, default=True) parser.add_argument("--use_relu", type=float, default=False) parser.add_argument("--use_rnn", type=bool, default=False) parser.add_argument("--add_agent_id", type=float, default=False) parser.add_argument("--use_value_clip", type=float, default=False) args = parser.parse_args() runner = Runner_MAPPO_MPE(args, env_name="simple_spread", number=1, seed=0) runner.run()

多智能体近端策略优化（MAPPO）算法详解