# MADDPG实战:用Python从零搭建多智能体协作与竞争环境(附避坑指南)
如果你已经对单智能体强化学习(如DDPG)有所了解,并渴望将技能扩展到更复杂、更贴近现实的多智能体世界,那么这篇文章正是为你准备的。多智能体系统无处不在——从协作的机器人编队、自动驾驶车队的协调,到游戏AI中的团队对抗与竞争。然而,当多个智能体同时学习并交互时,环境会变得极不稳定,传统的单智能体算法往往束手无策。这时,**MADDPG** 便闪亮登场。
MADDPG,全称多智能体深度确定性策略梯度,它巧妙地将**集中式训练与分布式执行** 的思想与演员-评论家框架相结合。简单来说,在训练时,每个智能体的“评论家”可以知晓全局信息(所有智能体的观察和动作),从而获得稳定的学习信号;而在执行时,每个智能体仅依靠自身的“演员”根据局部观察做出决策,保持了分布式应用的可行性。本文将手把手带你,从一个空白的Python文件开始,搭建一个完整的MADDPG训练系统,并聚焦于那些官方论文很少提及、但在实际编码中会让你头疼不已的“坑”。我们不仅会复现算法核心,更会深入工程细节:如何设计灵活的网络结构、高效管理经验回放、处理智能体间的异构性,以及调试训练过程中那些令人困惑的失败现象。
本文面向有一定PyTorch和强化学习基础的开发者,目标是交付一套可运行、可扩展、且包含大量实战调试心得的代码库。让我们跳过繁琐的理论复述,直接进入代码的战场。
## 1. 环境搭建与核心依赖管理
在开始编写算法之前,一个稳定、可复现的环境是成功的基石。多智能体强化学习的实验环境往往比单智能体更为复杂,依赖冲突是新手的第一道坎。
**强烈建议使用虚拟环境** 来隔离项目依赖。这里我选择 `conda`,因为它能更好地处理一些科学计算包的二进制依赖。
```bash
# 创建并激活一个名为maddpg_env的Python 3.8环境
conda create -n maddpg_env python=3.8
conda activate maddpg_env
```
接下来安装核心依赖。PyTorch的版本需要特别注意,某些较新的版本(如1.8+)在计算图操作中可能存在兼容性问题,导致我们在计算损失时的原地操作报错。根据社区经验,**PyTorch 1.4.0 到 1.7.1 是比较稳定的选择**。我们以1.7.1为例,通过Conda安装:
```bash
# 安装PyTorch和Torchvision(请根据你的CUDA版本调整)
conda install pytorch==1.7.1 torchvision==0.8.2 cudatoolkit=10.2 -c pytorch
```
对于多智能体环境,OpenAI的 **Multi-Agent Particle Environment** 是一个经典的轻量级测试平台。它包含合作、竞争、混合等多种场景,非常适合算法验证。
```bash
# 克隆MAPE仓库
git clone https://github.com/openai/multiagent-particle-envs.git
cd multiagent-particle-envs
pip install -e .
```
> **注意**:MAPE的依赖(如 `gym`)版本可能较低,直接 `pip install -e .` 可能会与你环境中其他库冲突。一个更稳妥的做法是,先进入MAPE目录,根据其 `requirements.txt` 或 `setup.py` 安装一个最低限度的环境,然后再安装我们算法所需的其他高级包。或者,直接将我们的算法项目目录放在MAPE的同级目录下,在代码中通过相对路径导入。
安装其他必要的工具库:
```bash
pip install numpy matplotlib tensorboard
```
现在,在你的项目根目录下,应该有一个类似这样的结构:
```
your_project/
├── multiagent-particle-envs/ # 环境仓库
├── src/ # 你的代码
│ ├── networks.py
│ ├── buffer.py
│ ├── maddpg.py
│ └── train.py
└── requirements.txt
```
## 2. 智能体核心组件:演员与评论家网络设计
MADDPG的核心是每个智能体拥有一套独立的演员-评论家网络。但“独立”不意味着“重复造轮子”,我们需要设计一个灵活的基础网络模块。
首先定义**演员网络**。它输入单个智能体的局部观察,输出一个确定性的连续动作。通常使用一个多层感知机。
```python
import torch
import torch.nn as nn
import torch.nn.functional as F
class Actor(nn.Module):
"""
确定性策略网络(演员)。
输入:单个智能体的观察 (obs_dim)
输出:该智能体的确定性动作 (action_dim)
"""
def __init__(self, obs_dim, action_dim, hidden_dim=256):
super(Actor, self).__init__()
self.fc1 = nn.Linear(obs_dim, hidden_dim)
self.fc2 = nn.Linear(hidden_dim, hidden_dim)
self.fc3 = nn.Linear(hidden_dim, action_dim)
# 初始化最后一层权重,使初始输出接近零,有利于稳定探索
nn.init.uniform_(self.fc3.weight, -3e-3, 3e-3)
def forward(self, obs):
x = F.relu(self.fc1(obs))
x = F.relu(self.fc2(x))
# 对于连续动作,通常使用tanh将输出限制在[-1, 1],环境会负责映射到实际范围
action = torch.tanh(self.fc3(x))
return action
```
接下来是**评论家网络**,这是MADDPG与DDPG的关键区别。MADDPG的评论家是集中式的,其输入是所有智能体的**联合观察**和**联合动作**。
```python
class Critic(nn.Module):
"""
集中式动作价值网络(评论家)。
输入:所有智能体的联合观察 (total_obs_dim) + 所有智能体的联合动作 (total_action_dim)
输出:单个智能体的Q值估计 (标量)
"""
def __init__(self, total_obs_dim, total_action_dim, hidden_dim=256):
super(Critic, self).__init__()
# 将状态和动作在输入层拼接
input_dim = total_obs_dim + total_action_dim
self.fc1 = nn.Linear(input_dim, hidden_dim)
self.fc2 = nn.Linear(hidden_dim, hidden_dim)
self.fc3 = nn.Linear(hidden_dim, 1)
# 同样初始化最后一层权重
nn.init.uniform_(self.fc3.weight, -3e-3, 3e-3)
def forward(self, global_obs, global_actions):
# global_obs: [batch_size, total_obs_dim]
# global_actions: [batch_size, total_action_dim]
x = torch.cat([global_obs, global_actions], dim=-1)
x = F.relu(self.fc1(x))
x = F.relu(self.fc2(x))
q_value = self.fc3(x)
return q_value
```
> **避坑提示1:输入维度计算**:`total_obs_dim` 和 `total_action_dim` 是 `num_agents * obs_dim` 和 `num_agents * action_dim` 吗?不一定!如果智能体是异构的(观察和动作空间不同),你需要将所有智能体的观察维度相加得到 `total_obs_dim`,动作维度同理。在经典的粒子环境中,智能体通常是同构的,所以可以直接相乘。在代码中明确处理这两种情况能大大提高复用性。
一个常见的优化是使用**分层特征提取**。先将每个智能体的观察和动作分别通过一个小网络,再聚合,这有助于网络学习个体特征。但为了代码清晰,我们先采用上述简单的拼接方式。
## 3. 经验回放缓冲区的多智能体适配
经验回放是深度强化学习稳定训练的关键。在MADDPG中,我们需要存储的每条经验是一个元组,包含:
* `obs`: 所有智能体的联合观察(状态)
* `actions`: 所有智能体采取的动作
* `rewards`: 所有智能体获得的奖励
* `next_obs`: 下一步所有智能体的联合观察
* `dones`: 所有智能体是否终止(通常环境同时终止,但设计上支持独立)
```python
import random
import numpy as np
from collections import deque
class MultiAgentReplayBuffer:
def __init__(self, capacity, obs_dims, action_dims, num_agents):
"""
初始化多智能体经验回放缓冲区。
Args:
capacity: 缓冲区最大容量
obs_dims: 列表,每个智能体的观察维度
action_dims: 列表,每个智能体的动作维度
num_agents: 智能体数量
"""
self.capacity = capacity
self.num_agents = num_agents
self.buffer = deque(maxlen=capacity)
# 预分配存储空间(可选优化,这里用deque更简单)
self.obs_dims = obs_dims
self.action_dims = action_dims
def push(self, obs, actions, rewards, next_obs, dones):
"""
存储一条经验。
所有输入都应是包含所有智能体信息的列表或数组。
"""
experience = (obs, actions, rewards, next_obs, dones)
self.buffer.append(experience)
def sample(self, batch_size):
"""
随机采样一批经验。
返回格式对PyTorch友好。
"""
if len(self.buffer) < batch_size:
return None
batch = random.sample(self.buffer, batch_size)
# 解包并转换为数组
obs_batch, action_batch, reward_batch, next_obs_batch, done_batch = zip(*batch)
# 转换为numpy数组,再转为PyTorch张量(训练脚本中完成转换更高效)
# 这里我们返回numpy数组,让训练循环决定何时转换到设备上
return {
'obs': np.array(obs_batch, dtype=np.float32), # [batch, num_agents, obs_dim]
'actions': np.array(action_batch, dtype=np.float32), # [batch, num_agents, action_dim]
'rewards': np.array(reward_batch, dtype=np.float32), # [batch, num_agents]
'next_obs': np.array(next_obs_batch, dtype=np.float32),
'dones': np.array(done_batch, dtype=np.float32)
}
def __len__(self):
return len(self.buffer)
```
> **避坑提示2:数据形状一致性**:确保你从环境获取的数据形状与缓冲区期望的形状匹配。粒子环境通常返回列表的列表。例如,`obs` 可能是一个长度为 `num_agents` 的列表,每个元素是 `(obs_dim,)` 的数组。在存入缓冲区前,最好将其转换为一个形状为 `(num_agents, obs_dim)` 的NumPy数组,这样在批量采样时更容易处理。
> **避坑提示3:智能体异构性**:上述缓冲区假设所有智能体经验被一起存储。如果智能体完全异构且独立学习,你可能需要为每个智能体维护独立的缓冲区。但MADDPG的集中式评论家需要联合数据,所以通常一个共享缓冲区就够了。
## 4. MADDPG算法主体:训练循环与更新逻辑
这是最核心的部分。我们将构建一个 `MADDPG` 类,它管理所有智能体的网络、优化器,并执行训练步骤。
首先,定义一些超参数。这些参数对训练稳定性影响巨大,需要仔细调校。
```python
# 超参数配置示例(可放在配置文件或类属性中)
DEFAULT_CONFIG = {
'actor_lr': 1e-4, # 演员学习率,通常比评论家小
'critic_lr': 1e-3, # 评论家学习率
'tau': 0.01, # 目标网络软更新系数
'gamma': 0.95, # 折扣因子
'batch_size': 1024, # 批次大小,根据环境复杂度调整
'hidden_dim': 256, # 网络隐藏层维度
'exploration_noise': 0.1, # 动作探索噪声标准差(OU噪声参数之一)
'noise_decay': 0.9995, # 噪声衰减因子
}
```
现在,我们来看 `MADDPG` 类的初始化。它需要知道每个智能体的观察和动作维度。
```python
class MADDPG:
def __init__(self, obs_dims, action_dims, num_agents, config=DEFAULT_CONFIG):
self.num_agents = num_agents
self.obs_dims = obs_dims # 列表
self.action_dims = action_dims # 列表
self.config = config
# 计算全局维度(假设同构智能体简化处理)
self.global_obs_dim = sum(obs_dims)
self.global_action_dim = sum(action_dims)
# 创建智能体网络
self.actors = []
self.critics = []
self.actor_optimizers = []
self.critic_optimizers = []
self.target_actors = []
self.target_critics = []
for i in range(num_agents):
# 演员网络(局部观察 -> 局部动作)
actor = Actor(obs_dims[i], action_dims[i], config['hidden_dim'])
actor_optim = torch.optim.Adam(actor.parameters(), lr=config['actor_lr'])
target_actor = Actor(obs_dims[i], action_dims[i], config['hidden_dim'])
# 目标网络权重初始化为与在线网络相同
target_actor.load_state_dict(actor.state_dict())
# 评论家网络(全局观察+全局动作 -> Q值)
critic = Critic(self.global_obs_dim, self.global_action_dim, config['hidden_dim'])
critic_optim = torch.optim.Adam(critic.parameters(), lr=config['critic_lr'])
target_critic = Critic(self.global_obs_dim, self.global_action_dim, config['hidden_dim'])
target_critic.load_state_dict(critic.state_dict())
self.actors.append(actor)
self.critics.append(critic)
self.actor_optimizers.append(actor_optim)
self.critic_optimizers.append(critic_optim)
self.target_actors.append(target_actor)
self.target_critics.append(target_critic)
# 探索噪声(使用Ornstein-Uhlenbeck过程,适用于惯性系统)
self.noise = OUNoise(action_dims[0], sigma=config['exploration_noise']) # 假设同构
self.noise_scale = config['exploration_noise']
self.noise_decay = config['noise_decay']
def act(self, obs_n, explore=True):
"""
根据当前观察,为所有智能体选择动作。
obs_n: 列表,每个元素是一个智能体的观察 (obs_dim,)
返回: 动作列表
"""
actions = []
for i, obs in enumerate(obs_n):
obs_tensor = torch.FloatTensor(obs).unsqueeze(0) # [1, obs_dim]
with torch.no_grad():
action = self.actors[i](obs_tensor).squeeze(0).numpy() # [action_dim,]
if explore:
noise = self.noise.sample() * self.noise_scale
action = np.clip(action + noise, -1.0, 1.0)
actions.append(action)
return actions
def update(self, batch, agent_id):
"""
更新指定智能体的网络。
batch: 从缓冲区采样的字典
agent_id: 要更新的智能体索引
"""
# 将数据转换为张量并放到设备上(假设为CPU)
obs = torch.FloatTensor(batch['obs']) # [batch, num_agents, obs_dim]
actions = torch.FloatTensor(batch['actions'])
rewards = torch.FloatTensor(batch['rewards'])
next_obs = torch.FloatTensor(batch['next_obs'])
dones = torch.FloatTensor(batch['dones'])
# 重塑以便处理:将智能体维度与特征维度合并
batch_size = obs.shape[0]
global_obs = obs.view(batch_size, -1) # [batch, global_obs_dim]
global_actions = actions.view(batch_size, -1)
global_next_obs = next_obs.view(batch_size, -1)
# --- 更新评论家 ---
with torch.no_grad():
# 计算目标动作:使用目标演员网络
target_next_actions = []
for i in range(self.num_agents):
# 取出智能体i的下一观察
agent_next_obs = next_obs[:, i, :] # [batch, obs_dim_i]
target_action_i = self.target_actors[i](agent_next_obs)
target_next_actions.append(target_action_i)
target_next_actions = torch.stack(target_next_actions, dim=1).view(batch_size, -1) # [batch, global_action_dim]
# 计算目标Q值
target_q = self.target_critics[agent_id](global_next_obs, target_next_actions)
target_q = rewards[:, agent_id].unsqueeze(1) + self.config['gamma'] * target_q * (1 - dones[:, agent_id].unsqueeze(1))
# 当前Q值估计
current_q = self.critics[agent_id](global_obs, global_actions)
# 评论家损失(MSE)
critic_loss = F.mse_loss(current_q, target_q)
# 优化评论家
self.critic_optimizers[agent_id].zero_grad()
critic_loss.backward()
# 可选:梯度裁剪,防止爆炸
torch.nn.utils.clip_grad_norm_(self.critics[agent_id].parameters(), 0.5)
self.critic_optimizers[agent_id].step()
# --- 更新演员 ---
# 重新计算当前动作(使用在线演员),但这次需要梯度
current_actions = []
for i in range(self.num_agents):
agent_obs = obs[:, i, :]
if i == agent_id:
# 对于要更新的智能体,动作需要梯度
action_i = self.actors[i](agent_obs)
else:
# 对于其他智能体,动作是固定的(从缓冲区取出的旧动作),不需要梯度
with torch.no_grad():
action_i = self.actors[i](agent_obs)
current_actions.append(action_i)
current_actions = torch.stack(current_actions, dim=1).view(batch_size, -1)
# 演员损失:最大化Q值(即最小化负Q值)
actor_loss = -self.critics[agent_id](global_obs, current_actions).mean()
self.actor_optimizers[agent_id].zero_grad()
actor_loss.backward()
torch.nn.utils.clip_grad_norm_(self.actors[agent_id].parameters(), 0.5)
self.actor_optimizers[agent_id].step()
# --- 软更新目标网络 ---
self._soft_update(self.critics[agent_id], self.target_critics[agent_id])
self._soft_update(self.actors[agent_id], self.target_actors[agent_id])
return critic_loss.item(), actor_loss.item()
def _soft_update(self, local_model, target_model):
"""
软更新目标网络参数:θ_target = τ * θ_local + (1 - τ) * θ_target
"""
for target_param, local_param in zip(target_model.parameters(), local_model.parameters()):
target_param.data.copy_(self.config['tau'] * local_param.data + (1.0 - self.config['tau']) * target_param.data)
def decay_noise(self):
"""衰减探索噪声"""
self.noise_scale *= self.noise_decay
```
> **避坑提示4:计算图中的动作分离**:在更新演员时,一个极其关键的细节是**动作的梯度流**。我们计算 `current_actions` 时,对于正在更新的智能体 `agent_id`,其动作来自在线演员网络,需要梯度以进行反向传播。但对于其他智能体的动作,我们应该使用它们**当前策略**计算的动作,但**阻止梯度流向它们的网络参数**。为什么?因为我们在更新智能体A时,不应该通过梯度去改变智能体B的网络参数。一个常见的错误是直接使用缓冲区中的旧动作(`actions[:, i, :]`),这会导致策略评估不一致(用旧动作评估新策略)。正确做法是用当前演员网络计算动作,但对其他智能体动作 `detach` 梯度。上面代码中,通过 `with torch.no_grad():` 实现了这一点。
> **避坑提示5:目标Q值的计算**:注意,计算 `target_q` 时,我们使用了 `target_next_actions`,这些动作是由**目标演员网络**基于 `next_obs` 产生的。这是DDPG/MADDPG的标准做法,旨在提高稳定性。千万不要误用在线演员网络来计算目标动作。
为了方便,这里附上 `OUNoise` 的实现,它用于生成时间相关的探索噪声,有助于在惯性环境中进行探索。
```python
class OUNoise:
"""Ornstein-Uhlenbeck 过程噪声"""
def __init__(self, action_dim, mu=0.0, theta=0.15, sigma=0.2):
self.action_dim = action_dim
self.mu = mu
self.theta = theta
self.sigma = sigma
self.state = np.ones(self.action_dim) * self.mu
self.reset()
def reset(self):
self.state = np.ones(self.action_dim) * self.mu
def sample(self):
dx = self.theta * (self.mu - self.state)
dx += self.sigma * np.random.randn(self.action_dim)
self.state += dx
return self.state
```
## 5. 训练流程整合与可视化调试
将以上所有组件串联起来,形成完整的训练循环。我们还需要集成TensorBoard来监控训练过程,这是调试的利器。
```python
import os
from torch.utils.tensorboard import SummaryWriter
from multiagent.environment import MultiAgentEnv
import multiagent.scenarios as scenarios
def make_env(scenario_name):
"""
创建多智能体环境。
"""
scenario = scenarios.load(scenario_name + ".py").Scenario()
world = scenario.make_world()
env = MultiAgentEnv(world, scenario.reset_world, scenario.reward, scenario.observation)
return env
def train(cfg):
# 创建环境
env = make_env(cfg['scenario']) # 例如 "simple_adversary"
num_agents = env.n
obs_dims = [env.observation_space[i].shape[0] for i in range(num_agents)]
action_dims = [env.action_space[i].n for i in range(num_agents)] # 注意:粒子环境离散动作需特殊处理
# 初始化MADDPG智能体和缓冲区
maddpg = MADDPG(obs_dims, action_dims, num_agents, cfg)
buffer = MultiAgentReplayBuffer(cfg['buffer_size'], obs_dims, action_dims, num_agents)
# 日志记录器
writer = SummaryWriter(log_dir=cfg['log_dir'])
total_steps = 0
episode_rewards = []
for episode in range(cfg['max_episodes']):
obs_n = env.reset()
episode_reward = np.zeros(num_agents)
maddpg.noise.reset()
for step in range(cfg['max_episode_len']):
# 选择动作
action_n = maddpg.act(obs_n, explore=True)
# 环境步进
next_obs_n, reward_n, done_n, _ = env.step(action_n)
# 存储经验
# 注意:环境返回的done_n可能是单个布尔值或列表,需统一处理
buffer.push(np.array(obs_n), np.array(action_n), np.array(reward_n), np.array(next_obs_n), np.array(done_n))
obs_n = next_obs_n
episode_reward += reward_n
total_steps += 1
# 如果缓冲区有足够样本,开始训练
if len(buffer) >= cfg['batch_size'] and total_steps % cfg['train_interval'] == 0:
for agent_id in range(num_agents):
batch = buffer.sample(cfg['batch_size'])
if batch:
critic_loss, actor_loss = maddpg.update(batch, agent_id)
# 记录损失
writer.add_scalar(f'Agent{agent_id}/critic_loss', critic_loss, total_steps)
writer.add_scalar(f'Agent{agent_id}/actor_loss', actor_loss, total_steps)
# 衰减噪声
if total_steps % 100 == 0:
maddpg.decay_noise()
if all(done_n):
break
# 记录回合奖励
mean_episode_reward = np.mean(episode_reward)
episode_rewards.append(mean_episode_reward)
writer.add_scalar('Reward/mean_episode_reward', mean_episode_reward, episode)
print(f"Episode {episode:4d}, Total Steps {total_steps:6d}, Mean Reward: {mean_episode_reward:6.2f}, Noise Scale: {maddpg.noise_scale:.4f}")
# 定期保存模型
if episode % cfg['save_interval'] == 0:
save_path = os.path.join(cfg['model_dir'], f"maddpg_ep{episode}.pth")
torch.save({
'actors_state_dict': [actor.state_dict() for actor in maddpg.actors],
'critics_state_dict': [critic.state_dict() for critic in maddpg.critics],
'episode': episode,
'reward': mean_episode_reward,
}, save_path)
env.close()
writer.close()
```
> **避坑提示6:环境动作空间处理**:上面的代码中有一个潜在问题。粒子环境 `MultiAgentEnv` 的 `action_space` 通常是 `Discrete` 类型(离散动作),而我们MADDPG的演员网络输出连续动作。这需要适配。一种常见做法是,将离散动作视为连续动作空间的一个特殊子集,或者修改网络输出层以适应离散分布(如Gumbel-Softmax)。在OpenAI的官方MADDPG实现中,他们确实处理了离散通信动作。对于物理动作,粒子环境通常期望一个连续力向量。你需要仔细检查所选 `scenario` 的 `action_space` 类型,并相应调整网络和动作选择逻辑。一个快速的测试方法是打印 `env.action_space`。
> **避坑提示7:训练间隔与批量更新**:不要每一步都更新网络!这会导致训练不稳定且低效。通常每隔几步(例如 `train_interval=100`)采样一个批次进行更新。同时,确保在更新前缓冲区已有足够样本(`len(buffer) >= batch_size`)。
最后,提供一个配置字典示例,并启动训练:
```python
if __name__ == '__main__':
config = {
'scenario': 'simple_adversary',
'actor_lr': 1e-4,
'critic_lr': 1e-3,
'tau': 0.01,
'gamma': 0.95,
'batch_size': 1024,
'hidden_dim': 128,
'exploration_noise': 0.2,
'noise_decay': 0.999,
'buffer_size': int(1e6),
'max_episodes': 5000,
'max_episode_len': 25,
'train_interval': 100,
'save_interval': 500,
'log_dir': './logs/maddpg_simple_adversary',
'model_dir': './models',
}
# 创建目录
os.makedirs(config['log_dir'], exist_ok=True)
os.makedirs(config['model_dir'], exist_ok=True)
train(config)
```
## 6. 进阶技巧与性能调优
当你的基础版本能够运行后,以下技巧可以帮助你提升性能、稳定性和收敛速度。
**技巧一:梯度归一化与裁剪**
如代码所示,在优化器 `step()` 之前,对网络参数的梯度进行裁剪(`clip_grad_norm_`)可以防止梯度爆炸,这在训练初期尤其重要。
**技巧二:学习率调度**
随着训练进行,逐渐降低学习率有助于收敛到更精细的策略。可以使用PyTorch的 `lr_scheduler`。
```python
from torch.optim.lr_scheduler import StepLR
# 在初始化优化器后
actor_scheduler = StepLR(actor_optimizer, step_size=1000, gamma=0.95)
# 在每个epoch或一定步数后调用 actor_scheduler.step()
```
**技巧三:参数共享**
如果智能体是同构的(角色、观察动作空间相同),可以让它们共享演员和评论家网络的底层参数,只保留独立的输出层或批归一化层。这能大幅减少参数数量,加速训练,并促进知识共享。但要注意,在竞争性环境中,共享参数可能限制策略的多样性。
**技巧四:集成训练**
如原论文所述,为每个智能体训练一个策略集合(多个子策略),在每回合随机选择一个执行。这能提高策略的鲁棒性,防止过拟合到其他智能体的特定策略。实现上,你需要为每个智能体维护K个演员网络,并相应地调整经验收集和更新逻辑。
**技巧五:注意力机制**
当智能体数量较多时,全局拼接的输入维度会很大。可以使用注意力机制让每个智能体的评论家只关注与其相关的其他智能体的信息,从而提高可扩展性。这属于对MADDPG的改进,如 **MAAC** 算法。
调试时,密切关注TensorBoard中的曲线:
* `Reward/mean_episode_reward`:应总体呈上升趋势,但会有波动。
* 各智能体的 `critic_loss`:应逐渐下降并趋于平稳。如果critic_loss持续上升或剧烈震荡,可能是学习率太高、批次大小太小或网络结构不合适。
* 各智能体的 `actor_loss`:由于是负Q值,其绝对值可能减小。
如果奖励长时间不增长,尝试:
1. 增大探索噪声 `exploration_noise` 或放慢其衰减。
2. 检查环境奖励尺度,必要时对奖励进行归一化。
3. 简化环境(从 `simple` 场景开始),确保智能体能在其中学到有意义的行为。
4. 调整网络深度和宽度:有时更小的网络反而更容易训练。
多智能体强化学习的训练就像调教一支乐队,每个乐手(智能体)都在即兴发挥,同时又需要聆听他人。MADDPG提供的集中式“指挥”(评论家)在排练时指导大家,而演出时则依靠每个乐手自己的技艺。从零实现这个过程,你会对算法内部的微妙平衡有更深的理解——例如,探索与利用的权衡在多个智能体间如何相互影响,以及为什么一个智能体策略的微小变化会像蝴蝶效应般扰动整个系统的学习轨迹。我自己的实验表明,在竞争性环境中,先让智能体进行一段时间的完全随机探索,再启动MADDPG训练,往往能避免早期陷入次优均衡。另外,别忘了定期用渲染模式可视化智能体的行为,这比任何数字都更能直观地告诉你它们是否学会了协作或竞争策略。