目录
目录

注意

转到末尾下载完整的示例代码。

开始使用 TorchRL 的模块

作者： Vincent Moens

注意

要在笔记本中运行本教程，请添加安装单元 开头包含：

!pip install tensordict
!pip install torchrl

强化学习旨在创建可以有效 处理特定任务。策略可以采用多种形式，从可微分的 map 从 observation space 过渡到 action space，再到 more 临时方法，如 argmax 对为每个可能的值计算的列表 行动。策略可以是确定性的或随机的，并且可以包含 复杂元素，例如递归神经网络 （RNN） 或变压器。

适应所有这些情况可能非常复杂。在这个简洁的 教程中，我们将深入研究 TorchRL 的核心功能 策略构建。我们将主要关注随机指标和 Q 值 策略：使用多层感知器 （MLP） 或 卷积神经网络 （CNN） 作为主干。

TensorDict 模块

与环境与 的实例交互的方式类似，用于表示策略和 value 函数也执行相同的操作。核心思想很简单：封装一个 standard （或任何其他函数）中的类 知道需要读取哪些条目并将其传递给模块，然后 将结果与分配的条目一起记录。为了说明这一点，我们将 使用尽可能简单的策略：来自观察的确定性映射 space 设置为 action space。为了获得最大的通用性，我们将使用一个带有 Pendulum 环境的模块 we 在上一个教程中实例化。TensorDict

import torch

from tensordict.nn import TensorDictModule
from torchrl.envs import GymEnv

env = GymEnv("Pendulum-v1")
module = torch.nn.LazyLinear(out_features=env.action_spec.shape[-1])
policy = TensorDictModule(
    module,
    in_keys=["observation"],
    out_keys=["action"],
)

这就是执行我们的政策所需要的全部内容！使用 lazy 模块 允许我们绕过获取观察空间形状的需要，因为 模块将自动确定它。此策略现已准备就绪 在环境中运行：

rollout = env.rollout(max_steps=10, policy=policy)
print(rollout)

TensorDict(
    fields={
        action: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.float32, is_shared=False),
        done: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.bool, is_shared=False),
        next: TensorDict(
            fields={
                done: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.bool, is_shared=False),
                observation: Tensor(shape=torch.Size([10, 3]), device=cpu, dtype=torch.float32, is_shared=False),
                reward: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.float32, is_shared=False),
                terminated: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.bool, is_shared=False),
                truncated: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.bool, is_shared=False)},
            batch_size=torch.Size([10]),
            device=None,
            is_shared=False),
        observation: Tensor(shape=torch.Size([10, 3]), device=cpu, dtype=torch.float32, is_shared=False),
        terminated: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.bool, is_shared=False),
        truncated: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.bool, is_shared=False)},
    batch_size=torch.Size([10]),
    device=None,
    is_shared=False)

专用包装器

为了简化 的合并， # ， # 或 # . 例如，提供 和 的默认值，进行集成 许多常见环境都很简单：Actorin_keysout_keys

from torchrl.modules import Actor

policy = Actor(module)
rollout = env.rollout(max_steps=10, policy=policy)
print(rollout)

TensorDict(
    fields={
        action: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.float32, is_shared=False),
        done: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.bool, is_shared=False),
        next: TensorDict(
            fields={
                done: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.bool, is_shared=False),
                observation: Tensor(shape=torch.Size([10, 3]), device=cpu, dtype=torch.float32, is_shared=False),
                reward: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.float32, is_shared=False),
                terminated: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.bool, is_shared=False),
                truncated: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.bool, is_shared=False)},
            batch_size=torch.Size([10]),
            device=None,
            is_shared=False),
        observation: Tensor(shape=torch.Size([10, 3]), device=cpu, dtype=torch.float32, is_shared=False),
        terminated: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.bool, is_shared=False),
        truncated: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.bool, is_shared=False)},
    batch_size=torch.Size([10]),
    device=None,
    is_shared=False)

API 参考中提供了可用的专用 TensorDictModules 列表。

网络

TorchRL 还提供了常规模块，无需重复使用 tensordict 功能。您将遇到的两个最常见的网络是 和 （CNN） 模块。我们可以用以下之一替换我们的策略模块：

from torchrl.modules import MLP

module = MLP(
    out_features=env.action_spec.shape[-1],
    num_cells=[32, 64],
    activation_class=torch.nn.Tanh,
)
policy = Actor(module)
rollout = env.rollout(max_steps=10, policy=policy)

TorchRL 还支持基于 RNN 的策略。由于这是一个技术性更强的 主题，它将在单独的教程中处理。

概率策略

PPO 等策略优化算法要求策略 stochastic：与上面的示例不同，该模块现在对来自 观察空间转换为参数空间，该参数空间编码在 可能的操作。TorchRL 通过分组来促进此类模块的设计 在单个类下，各种操作（例如构建分布 从参数中，从该分布中采样并检索 对数概率。在这里，我们将构建一个依赖于常规 使用三个分量的正态分布：

  • 一个主干读取 size 的观察值并输出单个 size 的张量[3][2];

  • 一个 将此输出拆分为两个块，平均值和标准差为 大小NormalParamExtractor[1];

  • A 将 将这些参数读取为 ，使用它们创建分配，然后 用样本和对数概率填充我们的 Tensordict。in_keys

from tensordict.nn.distributions import NormalParamExtractor
from torch.distributions import Normal
from torchrl.modules import ProbabilisticActor

backbone = MLP(in_features=3, out_features=2)
extractor = NormalParamExtractor()
module = torch.nn.Sequential(backbone, extractor)
td_module = TensorDictModule(module, in_keys=["observation"], out_keys=["loc", "scale"])
policy = ProbabilisticActor(
    td_module,
    in_keys=["loc", "scale"],
    out_keys=["action"],
    distribution_class=Normal,
    return_log_prob=True,
)

rollout = env.rollout(max_steps=10, policy=policy)
print(rollout)

TensorDict(
    fields={
        action: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.float32, is_shared=False),
        done: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.bool, is_shared=False),
        loc: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.float32, is_shared=False),
        next: TensorDict(
            fields={
                done: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.bool, is_shared=False),
                observation: Tensor(shape=torch.Size([10, 3]), device=cpu, dtype=torch.float32, is_shared=False),
                reward: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.float32, is_shared=False),
                terminated: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.bool, is_shared=False),
                truncated: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.bool, is_shared=False)},
            batch_size=torch.Size([10]),
            device=None,
            is_shared=False),
        observation: Tensor(shape=torch.Size([10, 3]), device=cpu, dtype=torch.float32, is_shared=False),
        sample_log_prob: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.float32, is_shared=False),
        scale: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.float32, is_shared=False),
        terminated: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.bool, is_shared=False),
        truncated: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.bool, is_shared=False)},
    batch_size=torch.Size([10]),
    device=None,
    is_shared=False)

关于此推出，有几点需要注意：

  • 由于我们在构建 actor 时要求它， 给定当时分布的动作的对数概率为 也写了。这对于 PPO 等算法是必需的。

  • 分配的参数在输出中返回 tensordict 也位于 and 条目下。"loc""scale"

您可以控制操作的采样以使用预期值或 分布的其他属性，而不是使用随机样本，如果 您的应用程序需要它。这可以通过以下功能进行控制：

from torchrl.envs.utils import ExplorationType, set_exploration_type

with set_exploration_type(ExplorationType.DETERMINISTIC):
    # takes the mean as action
    rollout = env.rollout(max_steps=10, policy=policy)
with set_exploration_type(ExplorationType.RANDOM):
    # Samples actions according to the dist
    rollout = env.rollout(max_steps=10, policy=policy)

检查 keyword 参数 要了解更多信息的文档字符串。default_interaction_type

勘探

像这样的随机策略在某种程度上自然而然地权衡了探索和 剥削，但确定性策略不会。幸运的是，TorchRL 可以 也可以通过它的探索模块来缓解这一点。 我们将以 exploration 模块为例（另请检查 和 ）。 为了查看此模块的实际操作，让我们恢复到确定性策略：

from tensordict.nn import TensorDictSequential
from torchrl.modules import EGreedyModule

policy = Actor(MLP(3, 1, num_cells=[32, 64]))

我们的 -greedy 探索模块通常会被定制 替换为多个退火帧和参数的初始值。值 表示每个 action taken 是随机的，而表示没有 探索。要退火（即减少）探测因子，请调用 to 是必需的（有关示例，请参阅上一个教程）。

exploration_module = EGreedyModule(
    spec=env.action_spec, annealing_num_steps=1000, eps_init=0.5
)

为了构建我们的探索性策略，我们只需要将 确定性策略模块与模块中的 Exploration 模块（类似于 to 在 tensordict 领域中）。TensorDictSequential

exploration_policy = TensorDictSequential(policy, exploration_module)

with set_exploration_type(ExplorationType.DETERMINISTIC):
    # Turns off exploration
    rollout = env.rollout(max_steps=10, policy=exploration_policy)
with set_exploration_type(ExplorationType.RANDOM):
    # Turns on exploration
    rollout = env.rollout(max_steps=10, policy=exploration_policy)

因为它必须能够在动作空间中对随机动作进行采样，所以必须配备从环境中知道使用什么策略 随机抽样操作。action_space

Q 值参与者

在某些设置中，策略不是一个独立的模块，而是在 另一个模块的顶部。Q-Value 参与者就是这种情况。简而言之，这些 Actor 需要对 action value 的估计（大多数时候是离散的） 并且会贪婪地选取价值最高的操作。在一些 设置（有限离散动作空间和有限离散状态空间）、 可以只存储一个状态-动作对的 2D 表，然后选取 操作。DQN 带来的创新是将其扩展到连续 状态空间。让我们考虑另一个具有离散操作空间的环境 为了更清楚地理解：Q(s, a)

env = GymEnv("CartPole-v1")
print(env.action_spec)

OneHot(
    shape=torch.Size([2]),
    space=CategoricalBox(n=2),
    device=cpu,
    dtype=torch.int64,
    domain=discrete)

我们构建了一个价值网络，当它读取 state 从环境中获取：

num_actions = 2
value_net = TensorDictModule(
    MLP(out_features=num_actions, num_cells=[32, 32]),
    in_keys=["observation"],
    out_keys=["action_value"],
)

我们可以通过在值后添加 Q 值 actor 来轻松构建 Q 值 actor 网络：

from torchrl.modules import QValueModule

policy = TensorDictSequential(
    value_net,  # writes action values in our tensordict
    QValueModule(spec=env.action_spec),  # Reads the "action_value" entry by default
)

让我们来看看吧！我们运行策略几个步骤，并查看 输出。我们应该在我们获得的 rollout 中找到 an 和 a 条目："action_value""chosen_action_value"

rollout = env.rollout(max_steps=3, policy=policy)
print(rollout)

TensorDict(
    fields={
        action: Tensor(shape=torch.Size([3, 2]), device=cpu, dtype=torch.int64, is_shared=False),
        action_value: Tensor(shape=torch.Size([3, 2]), device=cpu, dtype=torch.float32, is_shared=False),
        chosen_action_value: Tensor(shape=torch.Size([3, 1]), device=cpu, dtype=torch.float32, is_shared=False),
        done: Tensor(shape=torch.Size([3, 1]), device=cpu, dtype=torch.bool, is_shared=False),
        next: TensorDict(
            fields={
                done: Tensor(shape=torch.Size([3, 1]), device=cpu, dtype=torch.bool, is_shared=False),
                observation: Tensor(shape=torch.Size([3, 4]), device=cpu, dtype=torch.float32, is_shared=False),
                reward: Tensor(shape=torch.Size([3, 1]), device=cpu, dtype=torch.float32, is_shared=False),
                terminated: Tensor(shape=torch.Size([3, 1]), device=cpu, dtype=torch.bool, is_shared=False),
                truncated: Tensor(shape=torch.Size([3, 1]), device=cpu, dtype=torch.bool, is_shared=False)},
            batch_size=torch.Size([3]),
            device=None,
            is_shared=False),
        observation: Tensor(shape=torch.Size([3, 4]), device=cpu, dtype=torch.float32, is_shared=False),
        terminated: Tensor(shape=torch.Size([3, 1]), device=cpu, dtype=torch.bool, is_shared=False),
        truncated: Tensor(shape=torch.Size([3, 1]), device=cpu, dtype=torch.bool, is_shared=False)},
    batch_size=torch.Size([3]),
    device=None,
    is_shared=False)

因为它依赖于 operator，所以这个策略是确定性的。 在数据收集期间，我们需要探索环境。为此， 我们再次使用：argmaxEGreedyModule

policy_explore = TensorDictSequential(policy, EGreedyModule(env.action_spec))

with set_exploration_type(ExplorationType.RANDOM):
    rollout_explore = env.rollout(max_steps=3, policy=policy_explore)

这就是我们关于使用 TorchRL 构建策略的简短教程！

您可以使用库执行更多操作。一个好的起点 是查看模块的 API 参考。

后续步骤：

  • 检查如何在 动作是复合的（例如，离散动作和连续动作是 env 要求）;CompositeDistribution

  • 了解如何在策略中使用 RNN（教程);

  • 将此与 Decision Transformer 的 transformer 用法进行比较 示例（请参阅 GitHub 上的目录）。example

脚本总运行时间：（0 分 46.376 秒）

估计内存使用量：320 MB

由 Sphinx-Gallery 生成的图库

文档

访问 PyTorch 的全面开发人员文档

查看文档

教程

获取面向初学者和高级开发人员的深入教程

查看教程

资源

查找开发资源并解答您的问题

查看资源