快捷方式

StepCounter

class torchrl.envs.transforms.StepCounter(max_steps: Optional[int] = None, truncated_key: str | None = 'truncated', step_count_key: str | None = 'step_count', update_done: bool = True)[source]

从重置开始计数步数,并可选择在一定步数后将截断状态设置为 True

done”状态也会相应调整(因为 done 是任务完成和提前截断的析取)。

参数:
  • max_steps (int, 可选) – 一个正整数,表示在将 truncated_key 条目设置为 True 之前要执行的最大步数。

  • truncated_key (str, 可选) – 应该写入截断条目的键。默认为“truncated”,数据收集器将其识别为重置信号。此参数只能是字符串(而不是嵌套键),因为它将与父环境中的每个叶 done 键匹配(例如,如果使用“truncated”键名,则 ("agent", "done") 键将伴随 ("agent", "truncated"))。

  • step_count_key (str, 可选) – 应该写入步数计数条目的键。默认为“step_count”。此参数只能是字符串(而不是嵌套键),因为它将与父环境中的每个叶 done 键匹配(例如,如果使用“step_count”键名,则 ("agent", "done") 键将伴随 ("agent", "step_count"))。

  • update_done (bool, 可选) – 如果为 True,则将更新“truncated”级别的“done”布尔张量。此信号表示轨迹已到达终点,要么是因为任务已完成(“completed”条目为 True),要么是因为已被截断(“truncated”条目为 True)。默认为 True

注意

为了确保与具有多个 done_key(s) 的环境兼容,此转换将为 tensordict 中的每个 done 条目写入一个 step_count 条目。

示例

>>> import gymnasium
>>> from torchrl.envs import GymWrapper
>>> base_env = GymWrapper(gymnasium.make("Pendulum-v1"))
>>> env = TransformedEnv(base_env,
...     StepCounter(max_steps=5))
>>> rollout = env.rollout(100)
>>> print(rollout)
TensorDict(
    fields={
        action: Tensor(shape=torch.Size([5, 1]), device=cpu, dtype=torch.float32, is_shared=False),
        done: Tensor(shape=torch.Size([5, 1]), device=cpu, dtype=torch.bool, is_shared=False),
        completed: Tensor(shape=torch.Size([5, 1]), device=cpu, dtype=torch.bool, is_shared=False)},
        next: TensorDict(
            fields={
                done: Tensor(shape=torch.Size([5, 1]), device=cpu, dtype=torch.bool, is_shared=False),
                completed: Tensor(shape=torch.Size([5, 1]), device=cpu, dtype=torch.bool, is_shared=False)},
                observation: Tensor(shape=torch.Size([5, 3]), device=cpu, dtype=torch.float32, is_shared=False),
                reward: Tensor(shape=torch.Size([5, 1]), device=cpu, dtype=torch.float32, is_shared=False),
                step_count: Tensor(shape=torch.Size([5, 1]), device=cpu, dtype=torch.int64, is_shared=False),
                truncated: Tensor(shape=torch.Size([5, 1]), device=cpu, dtype=torch.bool, is_shared=False)},
            batch_size=torch.Size([5]),
            device=cpu,
            is_shared=False),
        observation: Tensor(shape=torch.Size([5, 3]), device=cpu, dtype=torch.float32, is_shared=False),
        step_count: Tensor(shape=torch.Size([5, 1]), device=cpu, dtype=torch.int64, is_shared=False),
        truncated: Tensor(shape=torch.Size([5, 1]), device=cpu, dtype=torch.bool, is_shared=False)},
    batch_size=torch.Size([5]),
    device=cpu,
    is_shared=False)
>>> print(rollout["next", "step_count"])
tensor([[1],
        [2],
        [3],
        [4],
        [5]])
forward(tensordict: TensorDictBase) TensorDictBase[source]

读取输入 tensordict,并为选定的键应用转换。

transform_input_spec(input_spec: Composite) Composite[source]

转换输入 spec,使结果 spec 与转换映射匹配。

参数:

input_spec (TensorSpec) – 转换前的 spec

返回:

转换后预期的 spec

transform_observation_spec(observation_spec: Composite) Composite[source]

转换观察 spec,使结果 spec 与转换映射匹配。

参数:

observation_spec (TensorSpec) – 转换前的 spec

返回:

转换后预期的 spec

transform_output_spec(output_spec: Composite) Composite[source]

转换输出 spec,使结果 spec 与转换映射匹配。

此方法通常应保持不变。更改应使用 transform_observation_spec()transform_reward_spec()transformfull_done_spec() 实现。 :param output_spec: 转换前的 spec :type output_spec: TensorSpec

返回:

转换后预期的 spec

文档

访问 PyTorch 的全面开发者文档

查看文档

教程

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

查看教程

资源

查找开发资源并获得解答

查看资源