使用 TorchRL 的多智能体强化学习 （PPO） 教程

作者： Matteo Bettini

另请参阅

BenchMARL 库提供最先进的 使用 TorchRL 的 MARL 算法的实现。

本教程演示如何使用 PyTorch 并 解决多智能体强化学习 （MARL） 问题。torchrl

为了便于使用，本教程将遵循 TorchRL 教程中已有的一般结构：强化学习 （PPO） 和 TorchRL 教程。 建议但并非强制性地在开始本教程之前熟悉这一点。

在本教程中，我们将使用 VMAS 中的 Navigation 环境 多机器人模拟器 基于 PyTorch，在设备上运行并行批处理模拟。

在 Navigation （导航） 环境中， 我们需要训练多个机器人（在随机位置生成） 导航到他们的目标（也在随机位置），而 使用 LIDAR 传感器以避免彼此之间的碰撞。

多代理导航场景

主要学习内容：

• 如何在 TorchRL 中创建多代理环境，其规范如何工作，以及它如何与库集成;

• 如何在 TorchRL 中使用 GPU 矢量化环境;

• 如何在 TorchRL 中创建不同的多智能体网络架构（例如，使用参数共享、集中式批评）

• 我们如何用于传输多代理数据;tensordict.TensorDict

• 我们如何在多智能体 MAPPO/IPPO 训练循环中绑定所有库组件（收集器、模块、重放缓冲区和损失）。

如果您在 Google Colab 中运行此程序，请确保安装以下依赖项：

!pip3 install torchrl
!pip3 install vmas
!pip3 install tqdm

近端策略优化 （PPO） 是一种策略梯度算法，其中 正在收集并直接使用批量数据以训练策略以最大化 给定一些近似约束的预期回报。您可以考虑 作为 REINFORCE 的复杂版本， 基础策略优化算法。有关详细信息，请参阅 Proximal Policy Optimization Algorithms 论文。

这种类型的算法通常是根据策略进行训练的。这意味着，在每次学习迭代中，我们都有一个采样和一个训练阶段。在迭代的采样阶段 ，收集卷展栏 使用当前策略在环境中形成座席的交互。 在训练阶段，所有收集的转出都会立即馈送到训练过程以执行 反向传播。这会导致更新的策略，然后再次用于采样。 在循环中执行此过程构成了策略学习。

根据策略学习

在 PPO 算法的训练阶段，使用批评者来估计操作的好坏 由策略采取。批评者学会了近似特定状态的值 （平均贴现回报）。 然后，PPO 损失将保单获得的实际回报与批评者估计的回报进行比较，以确定 所采取操作的优势并指导策略优化。

在多代理设置中，情况略有不同。我们现在有多个策略， 每个代理一个。政策通常是地方性的和分散的。这意味着 单个代理的策略将仅根据该代理的观察结果输出该代理的操作。 在 MARL 文献中，这被称为去中心化执行。 另一方面，批评者存在不同的表述，主要是：

• 在 MAPPO 中，批评者是集中的，并将全局状态作为输入 的系统。这可以是全局观察，也可以只是代理观察的串联。MAPPO公司 可用于执行集中训练的环境，因为它需要访问全球信息。

• 在 IPPO 中，批评家仅将相应主体的观察作为输入， 与政策完全一样。这允许分散式培训，因为批评者和政策都只需要本地 信息来计算其输出。

集中式 Critic 有助于克服多个智能体同时学习的非平稳性，但是， 另一方面，它们可能会受到其较大的 input space 的影响。 在本教程中，我们将能够训练这两种公式，我们还将讨论如何 参数共享（在代理之间共享网络参数的做法）会影响每个 Sensor。

本教程的结构如下：

1. 首先，我们将定义一组我们将使用的超参数。

2. 接下来，我们将使用 TorchRL 的 包装器。

3. 接下来，我们将设计策略和批评者网络，讨论各种选择对 参数共享和 Critic 集中化。

4. 接下来，我们将创建 sampling collector 和 replay buffer。

5. 最后，我们将运行训练循环并分析结果。

如果您在 Colab 或带有 GUI 的计算机中运行此命令，则还可以选择此 在训练前后呈现和可视化您自己的训练策略。

让我们导入我们的依赖项

# Torch
import torch

# Tensordict modules
from tensordict.nn import TensorDictModule
from tensordict.nn.distributions import NormalParamExtractor
from torch import multiprocessing

# Data collection
from torchrl.collectors import SyncDataCollector
from torchrl.data.replay_buffers import ReplayBuffer
from torchrl.data.replay_buffers.samplers import SamplerWithoutReplacement
from torchrl.data.replay_buffers.storages import LazyTensorStorage

# Env
from torchrl.envs import RewardSum, TransformedEnv
from torchrl.envs.libs.vmas import VmasEnv
from torchrl.envs.utils import check_env_specs

# Multi-agent network
from torchrl.modules import MultiAgentMLP, ProbabilisticActor, TanhNormal

# Loss
from torchrl.objectives import ClipPPOLoss, ValueEstimators

# Utils
torch.manual_seed(0)
from matplotlib import pyplot as plt
from tqdm import tqdm

定义超参数

我们为教程设置超参数。 取决于资源 可用，则可以选择在 GPU 或其他 GPU 上执行策略和模拟器 装置。 您可以调整其中一些值来调整计算要求。

# Devices
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  # The device where the simulator is run (VMAS can run on GPU)

# Sampling
frames_per_batch = 6_000  # Number of team frames collected per training iteration
n_iters = 10  # Number of sampling and training iterations
total_frames = frames_per_batch * n_iters

# Training
num_epochs = 30  # Number of optimization steps per training iteration
minibatch_size = 400  # Size of the mini-batches in each optimization step
lr = 3e-4  # Learning rate
max_grad_norm = 1.0  # Maximum norm for the gradients

# PPO
clip_epsilon = 0.2  # clip value for PPO loss
gamma = 0.99  # discount factor
lmbda = 0.9  # lambda for generalised advantage estimation
entropy_eps = 1e-4  # coefficient of the entropy term in the PPO loss

环境

多代理环境模拟多个代理与世界交互。 TorchRL API 允许集成各种类型的多代理环境风格。 一些示例包括具有共享或单个代理奖励、完成标记和观察的环境。 有关多代理环境 API 在 TorchRL 中如何工作的更多信息，您可以查看专门的文档部分。

特别是 VMAS 模拟器，它使用个人奖励、信息、观察和操作对代理进行建模，但 带有 collective done 标志。 此外，它还使用矢量化批量执行仿真。 这意味着它的所有 state 和 physics 是 PyTorch 张量，其第一维度表示批处理中并行环境的数量。 这允许利用 GPU 的单指令多数据 （SIMD） 范式，并且显着 通过利用 GPU Warps 中的并行化来加速并行计算。它还意味着 在 TorchRL 中使用它时，模拟和训练都可以在设备上运行，而无需通过 data 添加到 CPU。

我们今天要解决的多代理任务是 Navigation（参见上面的动画图）。 在 Navigation 中，随机生成的代理 （圆圈周围有点）需要导航 到随机生成的目标（较小的圆圈）。 座席需要使用 LIDAR（它们周围的点）来 避免相互碰撞。 代理在具有阻力和弹性碰撞的 2D 连续世界中起作用。 它们的作用是 2D 连续力，决定了它们的加速度。 奖励由三个术语组成：碰撞惩罚、基于到目标的距离的奖励，以及 当所有代理都达到他们的目标时给予最终共享奖励。 基于距离的项计算为相对距离的差 在代理与其目标之间连续两个时间步长之间。 每个代理都观察其位置， 速度、激光雷达读数和与其目标的相对位置。

现在，我们将实例化环境。 在本教程中，我们将剧集限制为 ，之后设置 done 标志。这是 VMAS 模拟器中已经提供了功能，但也可以使用 TorchRL 转换。 我们还将使用矢量化环境来利用批处理仿真。max_stepsStepCountnum_vmas_envs

max_steps = 100  # Episode steps before done
num_vmas_envs = (
    frames_per_batch // max_steps
)  # Number of vectorized envs. frames_per_batch should be divisible by this number
scenario_name = "navigation"
n_agents = 3

env = VmasEnv(
    scenario=scenario_name,
    num_envs=num_vmas_envs,
    continuous_actions=True,  # VMAS supports both continuous and discrete actions
    max_steps=max_steps,
    device=vmas_device,
    # Scenario kwargs
    n_agents=n_agents,  # These are custom kwargs that change for each VMAS scenario, see the VMAS repo to know more.
)

环境不仅由其模拟器和转换定义，而且还 通过一系列元数据来描述在其 执行。 为了提高效率，TorchRL 在以下方面非常严格 环境规范，但您可以轻松检查您的环境规范是否 足够。 在我们的示例中，负责为您的环境设置适当的规范，因此 你不应该关心这个。VmasEnv

有四个规格可供查看：

• action_spec定义动作空间;

• reward_spec定义奖励域;

• done_spec定义 done 域;

• observation_spec它定义了环境步骤的所有其他输出的域;

print("action_spec:", env.full_action_spec)
print("reward_spec:", env.full_reward_spec)
print("done_spec:", env.full_done_spec)
print("observation_spec:", env.observation_spec)

action_spec: Composite(
    agents: Composite(
        action: BoundedContinuous(
            shape=torch.Size([60, 3, 2]),
            space=ContinuousBox(
                low=Tensor(shape=torch.Size([60, 3, 2]), device=cpu, dtype=torch.float32, contiguous=True),
                high=Tensor(shape=torch.Size([60, 3, 2]), device=cpu, dtype=torch.float32, contiguous=True)),
            device=cpu,
            dtype=torch.float32,
            domain=continuous),
        device=cpu,
        shape=torch.Size([60, 3])),
    device=cpu,
    shape=torch.Size([60]))
reward_spec: Composite(
    agents: Composite(
        reward: UnboundedContinuous(
            shape=torch.Size([60, 3, 1]),
            space=ContinuousBox(
                low=Tensor(shape=torch.Size([60, 3, 1]), device=cpu, dtype=torch.float32, contiguous=True),
                high=Tensor(shape=torch.Size([60, 3, 1]), device=cpu, dtype=torch.float32, contiguous=True)),
            device=cpu,
            dtype=torch.float32,
            domain=continuous),
        device=cpu,
        shape=torch.Size([60, 3])),
    device=cpu,
    shape=torch.Size([60]))
done_spec: Composite(
    done: Categorical(
        shape=torch.Size([60, 1]),
        space=CategoricalBox(n=2),
        device=cpu,
        dtype=torch.bool,
        domain=discrete),
    terminated: Categorical(
        shape=torch.Size([60, 1]),
        space=CategoricalBox(n=2),
        device=cpu,
        dtype=torch.bool,
        domain=discrete),
    device=cpu,
    shape=torch.Size([60]))
observation_spec: Composite(
    agents: Composite(
        observation: UnboundedContinuous(
            shape=torch.Size([60, 3, 18]),
            space=ContinuousBox(
                low=Tensor(shape=torch.Size([60, 3, 18]), device=cpu, dtype=torch.float32, contiguous=True),
                high=Tensor(shape=torch.Size([60, 3, 18]), device=cpu, dtype=torch.float32, contiguous=True)),
            device=cpu,
            dtype=torch.float32,
            domain=continuous),
        info: Composite(
            pos_rew: UnboundedContinuous(
                shape=torch.Size([60, 3, 1]),
                space=ContinuousBox(
                    low=Tensor(shape=torch.Size([60, 3, 1]), device=cpu, dtype=torch.float32, contiguous=True),
                    high=Tensor(shape=torch.Size([60, 3, 1]), device=cpu, dtype=torch.float32, contiguous=True)),
                device=cpu,
                dtype=torch.float32,
                domain=continuous),
            final_rew: UnboundedContinuous(
                shape=torch.Size([60, 3, 1]),
                space=ContinuousBox(
                    low=Tensor(shape=torch.Size([60, 3, 1]), device=cpu, dtype=torch.float32, contiguous=True),
                    high=Tensor(shape=torch.Size([60, 3, 1]), device=cpu, dtype=torch.float32, contiguous=True)),
                device=cpu,
                dtype=torch.float32,
                domain=continuous),
            agent_collisions: UnboundedContinuous(
                shape=torch.Size([60, 3, 1]),
                space=ContinuousBox(
                    low=Tensor(shape=torch.Size([60, 3, 1]), device=cpu, dtype=torch.float32, contiguous=True),
                    high=Tensor(shape=torch.Size([60, 3, 1]), device=cpu, dtype=torch.float32, contiguous=True)),
                device=cpu,
                dtype=torch.float32,
                domain=continuous),
            device=cpu,
            shape=torch.Size([60, 3])),
        device=cpu,
        shape=torch.Size([60, 3])),
    device=cpu,
    shape=torch.Size([60]))

使用刚才显示的命令，我们可以访问每个值的域。 这样做，我们可以看到除了 done 之外的所有规范都有一个 leading shape . 这表示每个单独环境中的每个代理都将存在这些值。 另一方面，done 规范具有 leading shape ，表示 done 在 代理。(num_vmas_envs, n_agents)num_vmas_envs

TorchRL 有一种方法可以跟踪哪些 MARL 规范是共享的，哪些不是。 事实上，具有 additional agent dimension 的 specs （即，它们因每个代理而异）将包含在内部的 “agents” 键中。

如您所见，奖励和操作规范提供了 “agent” 键， 这意味着属于这些规范的 Tensordict 中的条目将嵌套在 “agents” Tensordict 中， 对所有每个代理的值进行分组。

要在 tensordict 中快速访问每个值的键，我们只需向环境请求 各自的键，以及 我们将立即了解哪些是按代理分配的，哪些是共享的。 此信息将有助于告诉所有其他 TorchRL 组件在何处找到每个值

print("action_keys:", env.action_keys)
print("reward_keys:", env.reward_keys)
print("done_keys:", env.done_keys)

action_keys: [('agents', 'action')]
reward_keys: [('agents', 'reward')]
done_keys: ['done', 'terminated']

变换

我们可以将所需的任何 TorchRL 转换附加到我们的环境中。 这些将以某种所需的方式修改其 input/output。 我们强调，在多代理上下文中，明确提供要修改的键至关重要。

例如，在本例中，我们将实例化一个转换，该转换将对整个剧集的奖励求和。 我们将告诉这个转换在哪里可以找到 reward key，在哪里写求和的 episode reward。 转换后的环境将继承 包装环境的设备和元数据，并根据序列转换它们 of transforms。RewardSum

env = TransformedEnv(
    env,
    RewardSum(in_keys=[env.reward_key], out_keys=[("agents", "episode_reward")]),
)

该函数运行一个小的 rollout，并将其输出与环境进行比较 规格。如果没有引发错误，我们可以确信 spec 已正确定义：check_env_specs()

check_env_specs(env)

推出

为了好玩，让我们看看简单的随机推出是什么样子的。您可以 调用 env.rollout（n_steps） 并获取环境输入内容的概览 和输出如下所示。操作将自动从操作规范中随机抽取 域。

n_rollout_steps = 5
rollout = env.rollout(n_rollout_steps)
print("rollout of three steps:", rollout)
print("Shape of the rollout TensorDict:", rollout.batch_size)

rollout of three steps: TensorDict(
    fields={
        agents: TensorDict(
            fields={
                action: Tensor(shape=torch.Size([60, 5, 3, 2]), device=cpu, dtype=torch.float32, is_shared=False),
                episode_reward: Tensor(shape=torch.Size([60, 5, 3, 1]), device=cpu, dtype=torch.float32, is_shared=False),
                info: TensorDict(
                    fields={
                        agent_collisions: Tensor(shape=torch.Size([60, 5, 3, 1]), device=cpu, dtype=torch.float32, is_shared=False),
                        final_rew: Tensor(shape=torch.Size([60, 5, 3, 1]), device=cpu, dtype=torch.float32, is_shared=False),
                        pos_rew: Tensor(shape=torch.Size([60, 5, 3, 1]), device=cpu, dtype=torch.float32, is_shared=False)},
                    batch_size=torch.Size([60, 5, 3]),
                    device=cpu,
                    is_shared=False),
                observation: Tensor(shape=torch.Size([60, 5, 3, 18]), device=cpu, dtype=torch.float32, is_shared=False)},
            batch_size=torch.Size([60, 5, 3]),
            device=cpu,
            is_shared=False),
        done: Tensor(shape=torch.Size([60, 5, 1]), device=cpu, dtype=torch.bool, is_shared=False),
        next: TensorDict(
            fields={
                agents: TensorDict(
                    fields={
                        episode_reward: Tensor(shape=torch.Size([60, 5, 3, 1]), device=cpu, dtype=torch.float32, is_shared=False),
                        info: TensorDict(
                            fields={
                                agent_collisions: Tensor(shape=torch.Size([60, 5, 3, 1]), device=cpu, dtype=torch.float32, is_shared=False),
                                final_rew: Tensor(shape=torch.Size([60, 5, 3, 1]), device=cpu, dtype=torch.float32, is_shared=False),
                                pos_rew: Tensor(shape=torch.Size([60, 5, 3, 1]), device=cpu, dtype=torch.float32, is_shared=False)},
                            batch_size=torch.Size([60, 5, 3]),
                            device=cpu,
                            is_shared=False),
                        observation: Tensor(shape=torch.Size([60, 5, 3, 18]), device=cpu, dtype=torch.float32, is_shared=False),
                        reward: Tensor(shape=torch.Size([60, 5, 3, 1]), device=cpu, dtype=torch.float32, is_shared=False)},
                    batch_size=torch.Size([60, 5, 3]),
                    device=cpu,
                    is_shared=False),
                done: Tensor(shape=torch.Size([60, 5, 1]), device=cpu, dtype=torch.bool, is_shared=False),
                terminated: Tensor(shape=torch.Size([60, 5, 1]), device=cpu, dtype=torch.bool, is_shared=False)},
            batch_size=torch.Size([60, 5]),
            device=cpu,
            is_shared=False),
        terminated: Tensor(shape=torch.Size([60, 5, 1]), device=cpu, dtype=torch.bool, is_shared=False)},
    batch_size=torch.Size([60, 5]),
    device=cpu,
    is_shared=False)
Shape of the rollout TensorDict: torch.Size([60, 5])

我们可以看到，我们的推出已经有 . 这意味着它中的所有张量都将具有这些前导维度。batch_size(num_vmas_envs, n_rollout_steps)

更深入地看，我们可以看到输出的 tensordict 可以按以下方式划分：

• 在根（可通过运行 ）中，我们将找到所有可用的键 after a reset 在第一个时间步调用。我们可以通过索引来了解它们在推出步骤中的演变 维度。在这些键中，我们会找到每个代理不同的键 在 tensordict 中，它将具有 Batch size，表示它正在存储额外的代理维度。此代理 tensordict 之外的那些 将是共享的 （在本例中仅完成）。rollout.exclude("next")n_rollout_stepsrollout["agents"](num_vmas_envs, n_rollout_steps, n_agents)

• 在下一个（可通过运行 访问）。我们将找到与根相同的结构， 但对于仅在步骤后可用的键。rollout.get("next")

在 TorchRL 中，约定是 done 和 observations 将同时出现在 root 和 next 中（因为它们是 在 Reset 时和 step） 后可用。操作仅在 root 中可用（因为没有操作 由步骤产生）和奖励将仅在 Next 中可用（因为在重置时没有奖励）。 这种结构遵循 强化学习：简介（Sutton 和 Barto）中的结构，其中 root 表示时间的数据， next 表示世界步长时的数据。

渲染随机卷展栏

如果您在 Google Colab 上，或者在具有 OpenGL 和 GUI 的计算机上，您实际上可以渲染随机转出。 这将使您了解随机策略在此任务中将实现什么，以便进行比较 有了这个政策，您将训练自己！

要渲染卷展栏，请按照本教程末尾的“渲染”部分中的说明进行操作 ，然后删除 .policy=policyenv.rollout()

政策

PPO 利用随机策略来处理探索。这意味着我们的 神经网络必须输出分布的参数，而不是 而不是与所执行的操作对应的单个值。

由于数据是连续的，因此我们使用 Tanh-Normal 分布来遵循 操作空间边界。TorchRL 提供了这样的发行版，并且唯一的 我们需要关心的是构建一个神经网络，将 正确的参数数量。

在这种情况下，每个代理的动作将由 2 维独立正态分布表示。 为此，我们的神经网络必须为每个动作输出平均值和标准差。 因此，每个代理都将具有 outputs。2 * n_actions_per_agents

我们需要做出的另一个重要决定是，我们是否希望代理共享策略参数。 一方面，共享参数意味着它们都将共享相同的策略，这将使它们能够从 彼此的经历。这也将导致更快的训练。 另一方面，这将使它们在行为上同质化，因为它们实际上将共享相同的模型。 在这个例子中，我们将启用共享，因为我们不介意同质性，并且可以从计算 速度，但重要的是要始终在您自己的问题中考虑这个决定！

我们分三个步骤设计策略。

第一：定义神经网络n_obs_per_agent -> 2 * n_actions_per_agents

为此，我们使用 ，一个专门用于 多个代理，提供大量自定义功能。MultiAgentMLP

share_parameters_policy = True

policy_net = torch.nn.Sequential(
    MultiAgentMLP(
        n_agent_inputs=env.observation_spec["agents", "observation"].shape[
            -1
        ],  # n_obs_per_agent
        n_agent_outputs=2 * env.action_spec.shape[-1],  # 2 * n_actions_per_agents
        n_agents=env.n_agents,
        centralised=False,  # the policies are decentralised (ie each agent will act from its observation)
        share_params=share_parameters_policy,
        device=device,
        depth=2,
        num_cells=256,
        activation_class=torch.nn.Tanh,
    ),
    NormalParamExtractor(),  # this will just separate the last dimension into two outputs: a loc and a non-negative scale
)

第二种：将神经网络包装在TensorDictModule

这只是一个模块，它将从 tensordict 中读取 ，将它们提供给 神经网络，并编写 输出就地位于 .in_keysout_keys

请注意，我们使用 key，因为这些 key 表示数据，并且 additional dimension.("agents", ...)n_agents

policy_module = TensorDictModule(
    policy_net,
    in_keys=[("agents", "observation")],
    out_keys=[("agents", "loc"), ("agents", "scale")],
)

第三：将TensorDictModuleProbabilisticActor

我们现在需要根据 正态分布。为此，我们指示类构建一个 out of the location 和 scale 参数。我们还提供了此 发行版，我们从环境规范中收集。ProbabilisticActorTanhNormal

的名称（因此是 from 的名称 above） 必须以 distribution constructor 关键字参数（loc 和 scale）结尾。in_keysout_keysTensorDictModuleTanhNormal

policy = ProbabilisticActor(
    module=policy_module,
    spec=env.action_spec_unbatched,
    in_keys=[("agents", "loc"), ("agents", "scale")],
    out_keys=[env.action_key],
    distribution_class=TanhNormal,
    distribution_kwargs={
        "low": env.full_action_spec_unbatched[env.action_key].space.low,
        "high": env.full_action_spec_unbatched[env.action_key].space.high,
    },
    return_log_prob=True,
    log_prob_key=("agents", "sample_log_prob"),
)  # we'll need the log-prob for the PPO loss

评论家网络

批评者网络是 PPO 算法的重要组成部分，尽管它 在采样时不使用。此模块将读取 observations 和 返回相应的值 estimates。

和以前一样，人们应该仔细考虑共享 critic 参数的决定。 通常，参数共享将提供更快的训练收敛，但有一些重要的 需要考虑的事项：

• 当代理具有不同的奖励功能时，不建议分享，因为批评者需要学习 为同一状态分配不同的值（例如，在混合合作社竞争环境中）。

• 在分散式训练环境中，如果没有额外的基础设施，就无法执行共享 同步参数。

在所有其他情况下，奖励函数（与奖励区分开来）对所有代理都相同 （与当前情景一样）， 共享可以提高性能。这可能是以代理策略的同质性为代价的。 通常，了解哪个选项更可取的最佳方法是快速试验这两个选项。

这也是我们必须在 MAPPO 和 IPPO 之间进行选择的地方：

• 使用 MAPPO，我们将获得具有完全可观察性的 Central Critic （即，它将所有串联的代理观察值作为输入）。 我们可以这样做，因为我们在模拟器中 并且培训是集中的。

• 有了 IPPO，我们将有一个本地的去中心化批评者，就像政策一样。

无论如何，critic 输出将具有 shape 。 如果批评者是集中和共享的， 沿维度的所有值都将相同。(..., n_agents, 1)n_agents

share_parameters_critic = True
mappo = True  # IPPO if False

critic_net = MultiAgentMLP(
    n_agent_inputs=env.observation_spec["agents", "observation"].shape[-1],
    n_agent_outputs=1,  # 1 value per agent
    n_agents=env.n_agents,
    centralised=mappo,
    share_params=share_parameters_critic,
    device=device,
    depth=2,
    num_cells=256,
    activation_class=torch.nn.Tanh,
)

critic = TensorDictModule(
    module=critic_net,
    in_keys=[("agents", "observation")],
    out_keys=[("agents", "state_value")],
)

让我们试试我们的 policy 和 critic 模块。如前所述，使用 of 可以直接读取输出 运行这些模块，因为它们知道要读取哪些信息 以及写入位置：TensorDictModule

从这时起，多代理特定的组件已经实例化了，我们将简单地使用相同的组件 组件，就像在单代理学习中一样。这不是很棒吗？

print("Running policy:", policy(env.reset()))
print("Running value:", critic(env.reset()))

Running policy: TensorDict(
    fields={
        agents: TensorDict(
            fields={
                action: Tensor(shape=torch.Size([60, 3, 2]), device=cpu, dtype=torch.float32, is_shared=False),
                episode_reward: Tensor(shape=torch.Size([60, 3, 1]), device=cpu, dtype=torch.float32, is_shared=False),
                info: TensorDict(
                    fields={
                        agent_collisions: Tensor(shape=torch.Size([60, 3, 1]), device=cpu, dtype=torch.float32, is_shared=False),
                        final_rew: Tensor(shape=torch.Size([60, 3, 1]), device=cpu, dtype=torch.float32, is_shared=False),
                        pos_rew: Tensor(shape=torch.Size([60, 3, 1]), device=cpu, dtype=torch.float32, is_shared=False)},
                    batch_size=torch.Size([60, 3]),
                    device=cpu,
                    is_shared=False),
                loc: Tensor(shape=torch.Size([60, 3, 2]), device=cpu, dtype=torch.float32, is_shared=False),
                observation: Tensor(shape=torch.Size([60, 3, 18]), device=cpu, dtype=torch.float32, is_shared=False),
                sample_log_prob: Tensor(shape=torch.Size([60, 3]), device=cpu, dtype=torch.float32, is_shared=False),
                scale: Tensor(shape=torch.Size([60, 3, 2]), device=cpu, dtype=torch.float32, is_shared=False)},
            batch_size=torch.Size([60, 3]),
            device=cpu,
            is_shared=False),
        done: Tensor(shape=torch.Size([60, 1]), device=cpu, dtype=torch.bool, is_shared=False),
        terminated: Tensor(shape=torch.Size([60, 1]), device=cpu, dtype=torch.bool, is_shared=False)},
    batch_size=torch.Size([60]),
    device=cpu,
    is_shared=False)
Running value: TensorDict(
    fields={
        agents: TensorDict(
            fields={
                episode_reward: Tensor(shape=torch.Size([60, 3, 1]), device=cpu, dtype=torch.float32, is_shared=False),
                info: TensorDict(
                    fields={
                        agent_collisions: Tensor(shape=torch.Size([60, 3, 1]), device=cpu, dtype=torch.float32, is_shared=False),
                        final_rew: Tensor(shape=torch.Size([60, 3, 1]), device=cpu, dtype=torch.float32, is_shared=False),
                        pos_rew: Tensor(shape=torch.Size([60, 3, 1]), device=cpu, dtype=torch.float32, is_shared=False)},
                    batch_size=torch.Size([60, 3]),
                    device=cpu,
                    is_shared=False),
                observation: Tensor(shape=torch.Size([60, 3, 18]), device=cpu, dtype=torch.float32, is_shared=False),
                state_value: Tensor(shape=torch.Size([60, 3, 1]), device=cpu, dtype=torch.float32, is_shared=False)},
            batch_size=torch.Size([60, 3]),
            device=cpu,
            is_shared=False),
        done: Tensor(shape=torch.Size([60, 1]), device=cpu, dtype=torch.bool, is_shared=False),
        terminated: Tensor(shape=torch.Size([60, 1]), device=cpu, dtype=torch.bool, is_shared=False)},
    batch_size=torch.Size([60]),
    device=cpu,
    is_shared=False)

数据收集器

TorchRL 提供了一组数据收集器类。简而言之，这些 类执行三个操作：重置环境、计算操作 使用策略和最新观察结果，在环境中执行一个步骤，然后重复 最后两个步骤，直到环境发出停止信号（或达到 Done） state） 的 S S

我们将使用最简单的数据收集器，其输出与环境卷展栏相同。 唯一的区别是它会自动重置 DONE 状态，直到收集到所需的帧。

collector = SyncDataCollector(
    env,
    policy,
    device=vmas_device,
    storing_device=device,
    frames_per_batch=frames_per_batch,
    total_frames=total_frames,
)

重放缓冲区

重放缓冲区是非策略 RL 算法的常见构建部分。 在策略上下文中，每次一批 收集数据，并且其数据被重复消耗一定数量 的时代。

对 PPO 使用重放缓冲区不是强制性的，我们可以简单地 在线使用收集的数据，但使用这些类 使我们能够轻松地以可重现的方式构建内部训练循环。

replay_buffer = ReplayBuffer(
    storage=LazyTensorStorage(
        frames_per_batch, device=device
    ),  # We store the frames_per_batch collected at each iteration
    sampler=SamplerWithoutReplacement(),
    batch_size=minibatch_size,  # We will sample minibatches of this size
)

损失函数

为了方便使用类，可以直接从 TorchRL 导入 PPO 损失。这是使用 PPO 的最简单方法： 它隐藏了 PPO 的数学运算和控制流 随它去。

PPO 需要计算一些“优势估计”。简而言之，一个优势 是一个值，该值反映在处理 偏差/方差权衡。 要计算 advantage，只需要 （1） 构建 advantage 模块，该 使用我们的 Value 运算符，并且 （2） 在每个 时代。 GAE 模块将使用 new 和 entries 更新输入。 这是一个无梯度的张量，代表经验 值 network 应与输入观测值表示的值。 这两者都将被用于 退还保单和价值损失。TensorDict"advantage""value_target""value_target"ClipPPOLoss

loss_module = ClipPPOLoss(
    actor_network=policy,
    critic_network=critic,
    clip_epsilon=clip_epsilon,
    entropy_coef=entropy_eps,
    normalize_advantage=False,  # Important to avoid normalizing across the agent dimension
)
loss_module.set_keys(  # We have to tell the loss where to find the keys
    reward=env.reward_key,
    action=env.action_key,
    sample_log_prob=("agents", "sample_log_prob"),
    value=("agents", "state_value"),
    # These last 2 keys will be expanded to match the reward shape
    done=("agents", "done"),
    terminated=("agents", "terminated"),
)


loss_module.make_value_estimator(
    ValueEstimators.GAE, gamma=gamma, lmbda=lmbda
)  # We build GAE
GAE = loss_module.value_estimator

optim = torch.optim.Adam(loss_module.parameters(), lr)

训练循环

现在，我们已具备编写训练循环所需的所有部分。 这些步骤包括：

• 收集数据

  • 计算优势

    • 循环 epochs

      • 循环小批量以计算损失值

        • 反向传播

        • 优化

      • 重复

    • 重复

  • 重复

• 重复

pbar = tqdm(total=n_iters, desc="episode_reward_mean = 0")

episode_reward_mean_list = []
for tensordict_data in collector:
    tensordict_data.set(
        ("next", "agents", "done"),
        tensordict_data.get(("next", "done"))
        .unsqueeze(-1)
        .expand(tensordict_data.get_item_shape(("next", env.reward_key))),
    )
    tensordict_data.set(
        ("next", "agents", "terminated"),
        tensordict_data.get(("next", "terminated"))
        .unsqueeze(-1)
        .expand(tensordict_data.get_item_shape(("next", env.reward_key))),
    )
    # We need to expand the done and terminated to match the reward shape (this is expected by the value estimator)

    with torch.no_grad():
        GAE(
            tensordict_data,
            params=loss_module.critic_network_params,
            target_params=loss_module.target_critic_network_params,
        )  # Compute GAE and add it to the data

    data_view = tensordict_data.reshape(-1)  # Flatten the batch size to shuffle data
    replay_buffer.extend(data_view)

    for _ in range(num_epochs):
        for _ in range(frames_per_batch // minibatch_size):
            subdata = replay_buffer.sample()
            loss_vals = loss_module(subdata)

            loss_value = (
                loss_vals["loss_objective"]
                + loss_vals["loss_critic"]
                + loss_vals["loss_entropy"]
            )

            loss_value.backward()

            torch.nn.utils.clip_grad_norm_(
                loss_module.parameters(), max_grad_norm
            )  # Optional

            optim.step()
            optim.zero_grad()

    collector.update_policy_weights_()

    # Logging
    done = tensordict_data.get(("next", "agents", "done"))
    episode_reward_mean = (
        tensordict_data.get(("next", "agents", "episode_reward"))[done].mean().item()
    )
    episode_reward_mean_list.append(episode_reward_mean)
    pbar.set_description(f"episode_reward_mean = {episode_reward_mean}", refresh=False)
    pbar.update()

episode_reward_mean = 0:   0%|          | 0/10 [00:00<?, ?it/s]
episode_reward_mean = -0.4579917788505554:  10%|█         | 1/10 [00:05<00:46,  5.20s/it]
episode_reward_mean = 0.1870676875114441:  20%|██        | 2/10 [00:10<00:41,  5.21s/it]
episode_reward_mean = 1.1689366102218628:  30%|███       | 3/10 [00:15<00:36,  5.19s/it]
episode_reward_mean = 1.4059897661209106:  40%|████      | 4/10 [00:20<00:31,  5.19s/it]
episode_reward_mean = 1.890636920928955:  50%|█████     | 5/10 [00:25<00:25,  5.19s/it]
episode_reward_mean = 2.1948678493499756:  60%|██████    | 6/10 [00:31<00:20,  5.18s/it]
episode_reward_mean = 2.1456124782562256:  70%|███████   | 7/10 [00:36<00:15,  5.18s/it]
episode_reward_mean = 2.6171820163726807:  80%|████████  | 8/10 [00:41<00:10,  5.23s/it]
episode_reward_mean = 2.5954275131225586:  90%|█████████ | 9/10 [00:46<00:05,  5.19s/it]
episode_reward_mean = 2.713913679122925: 100%|██████████| 10/10 [00:51<00:00,  5.17s/it]

结果

让我们绘制每集获得的平均奖励

要使训练持续时间更长，请增加 hyperparameter 。n_iters

plt.plot(episode_reward_mean_list)
plt.xlabel("Training iterations")
plt.ylabel("Reward")
plt.title("Episode reward mean")
plt.show()

呈现

如果您在具有 GUI 的计算机中运行此操作，则可以通过运行以下命令来呈现经过训练的策略：

with torch.no_grad():
   env.rollout(
       max_steps=max_steps,
       policy=policy,
       callback=lambda env, _: env.render(),
       auto_cast_to_device=True,
       break_when_any_done=False,
   )

如果您在 Google Colab 中运行此操作，则可以通过运行以下命令来呈现经过训练的策略：

!apt-get update
!apt-get install -y x11-utils
!apt-get install -y xvfb
!pip install pyvirtualdisplay

import pyvirtualdisplay
display = pyvirtualdisplay.Display(visible=False, size=(1400, 900))
display.start()
from PIL import Image

def rendering_callback(env, td):
    env.frames.append(Image.fromarray(env.render(mode="rgb_array")))
env.frames = []
with torch.no_grad():
   env.rollout(
       max_steps=max_steps,
       policy=policy,
       callback=rendering_callback,
       auto_cast_to_device=True,
       break_when_any_done=False,
   )
env.frames[0].save(
    f"{scenario_name}.gif",
    save_all=True,
    append_images=env.frames[1:],
   duration=3,
   loop=0,
)

from IPython.display import Image
Image(open(f"{scenario_name}.gif", "rb").read())

结论和下一步

在本教程中，我们看到了：

• 如何在 TorchRL 中创建多代理环境，其规范如何工作，以及它如何与库集成;

• 如何在 TorchRL 中使用 GPU 矢量化环境;

• 如何在 TorchRL 中创建不同的多智能体网络架构（例如，使用参数共享、集中式批评）

• 我们如何用于传输多代理数据;tensordict.TensorDict

• 我们如何在多智能体 MAPPO/IPPO 训练循环中绑定所有库组件（收集器、模块、重放缓冲区和损失）。

现在您已经精通了多智能体 DDPG，您可以在 GitHub 存储库。 这些是许多流行的 MARL 算法的纯代码脚本，例如本教程中所示的算法。 QMIX、MADDPG、IJL 等等！

您还可以查看我们的其他多代理教程，了解如何训练 Competitive PettingZoo/VMAS 中具有多个代理组的 MADDPG/IDDPG：使用 TorchRL 教程的竞争性多代理强化学习 （DDPG）。

如果您有兴趣在 TorchRL 中创建或包装自己的多代理环境， 您可以查看 dedicated doc 部分。

最后，您可以修改本教程的参数以尝试许多其他配置和场景 成为 MARL 母版。 以下是一些视频，介绍了您可以在 VMAS 中尝试的一些可能场景。

VMAS 中可用的方案

估计内存使用量：319 MB