StepCounter

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

计算自重置以来的步数，并可选择在一定步数后将截断状态设置为 True。

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

参数:

• max_steps (int, optional) – 一个正整数，指示在将 truncated_key 条目设置为 True 之前的最大步数。

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

• step_count_key (str, optional) – 用于写入步数条目的键。默认为 "step_count"。此参数只能是字符串（不能是嵌套键），因为它将与父环境中的每个叶子 done 键匹配（例如，如果使用 "step_count" 键名，则 ("agent", "done") 键将伴随一个 ("agent", "step_count")）。

• update_done (bool, optional) – 如果为 True，则将更新 "truncated" 级别的 "done" 布尔张量。此信号表示轨迹已到达其末尾，因为任务已完成（"completed" 条目为 True）或因为已被截断（"truncated" 条目为 True）。默认为 True。

注意

为了确保与具有多个 done_key 的环境兼容，此转换将为 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

读取输入 tensordict，并对选定的键应用转换。

默认情况下，此方法

• 直接调用 _apply_transform()。

• 不调用 _step() 或 _call()。

此方法不会在任何时候在 env.step 中调用。但是，它会在 sample() 中调用。

注意

forward 还使用 dispatch 通过将参数名称强制转换为键来处理常规关键字参数。

示例

>>> class TransformThatMeasuresBytes(Transform):
...     '''Measures the number of bytes in the tensordict, and writes it under `"bytes"`.'''
...     def __init__(self):
...         super().__init__(in_keys=[], out_keys=["bytes"])
...
...     def forward(self, tensordict: TensorDictBase) -> TensorDictBase:
...         bytes_in_td = tensordict.bytes()
...         tensordict["bytes"] = bytes
...         return tensordict
>>> t = TransformThatMeasuresBytes()
>>> env = env.append_transform(t) # works within envs
>>> t(TensorDict(a=0))  # Works offline too.

transform_input_spec(input_spec: Composite) → Composite

转换输入规范，使结果规范与转换映射匹配。

参数:

input_spec (TensorSpec) – 转换前的规范

返回:

转换后的预期规范

transform_observation_spec(observation_spec: Composite) → Composite

转换观察规范，使结果规范与转换映射匹配。

参数:

observation_spec (TensorSpec) – 转换前的规范

返回:

转换后的预期规范

transform_output_spec(output_spec: Composite) → Composite

转换输出规范，使结果规范与转换映射匹配。

此方法通常应保持不变。应通过 transform_observation_spec()、transform_reward_spec() 和 transform_full_done_spec() 来实现更改。 :param output_spec: 转换前的规范 :type output_spec: TensorSpec

返回:

转换后的预期规范