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]])
- 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