step_mdp¶
- torchrl.envs.utils.step_mdp(tensordict: TensorDictBase, next_tensordict: Optional[TensorDictBase] = None, keep_other: bool = True, exclude_reward: bool = True, exclude_done: bool = False, exclude_action: bool = True, reward_keys: Union[NestedKey, List[NestedKey]] = 'reward', done_keys: Union[NestedKey, List[NestedKey]] = 'done', action_keys: Union[NestedKey, List[NestedKey]] = 'action') TensorDictBase [source]¶
创建一个新的 tensordict,反映输入 tensordict 的时间步进。
给定一个在步进后检索的 tensordict,返回
"next"
索引的 tensordict。 这些参数允许精确控制应保留哪些内容以及应从"next"
条目复制哪些内容。 默认行为是:将观察条目、奖励和完成状态移动到根目录,排除当前操作并保留所有额外的键(非操作、非完成、非奖励)。- 参数:
tensordict (TensorDictBase) – 要重命名的键的 tensordict
next_tensordict (TensorDictBase, optional) – 目标 tensordict
keep_other (bool, optional) – 如果为
True
,则所有不以'next_'
开头的键都将被保留。 默认为True
。exclude_reward (bool, optional) – 如果为
True
,则"reward"
键将从结果 tensordict 中丢弃。 如果为False
,它将从"next"
条目(如果存在)中复制(和替换)。 默认为True
。exclude_done (bool, optional) – 如果为
True
,则"done"
键将从结果 tensordict 中丢弃。 如果为False
,它将从"next"
条目(如果存在)中复制(和替换)。 默认为False
。exclude_action (bool, optional) – 如果为
True
,则"action"
键将从结果 tensordict 中丢弃。 如果为False
,它将保留在根 tensordict 中(因为它不应存在于"next"
条目中)。 默认为True
。reward_keys (NestedKey or list of NestedKey, optional) – 写入奖励的键。 默认为 “reward”。
done_keys (NestedKey or list of NestedKey, optional) – 写入完成状态的键。 默认为 “done”。
action_keys (NestedKey or list of NestedKey, optional) – 写入操作的键。 默认为 “action”。
- 返回:
一个新的 tensordict(或 next_tensordict),包含 t+1 步的张量。
示例:此函数允许使用这种循环
>>> from tensordict import TensorDict >>> import torch >>> td = TensorDict({ ... "done": torch.zeros((), dtype=torch.bool), ... "reward": torch.zeros(()), ... "extra": torch.zeros(()), ... "next": TensorDict({ ... "done": torch.zeros((), dtype=torch.bool), ... "reward": torch.zeros(()), ... "obs": torch.zeros(()), ... }, []), ... "obs": torch.zeros(()), ... "action": torch.zeros(()), ... }, []) >>> print(step_mdp(td)) TensorDict( fields={ done: Tensor(shape=torch.Size([]), device=cpu, dtype=torch.bool, is_shared=False), extra: Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, is_shared=False), obs: Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, is_shared=False)}, batch_size=torch.Size([]), device=None, is_shared=False) >>> print(step_mdp(td, exclude_done=True)) # "done" is dropped TensorDict( fields={ extra: Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, is_shared=False), obs: Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, is_shared=False)}, batch_size=torch.Size([]), device=None, is_shared=False) >>> print(step_mdp(td, exclude_reward=False)) # "reward" is kept TensorDict( fields={ done: Tensor(shape=torch.Size([]), device=cpu, dtype=torch.bool, is_shared=False), extra: Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, is_shared=False), obs: Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, is_shared=False), reward: Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, is_shared=False)}, batch_size=torch.Size([]), device=None, is_shared=False) >>> print(step_mdp(td, exclude_action=False)) # "action" persists at the root TensorDict( fields={ action: Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, is_shared=False), done: Tensor(shape=torch.Size([]), device=cpu, dtype=torch.bool, is_shared=False), extra: Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, is_shared=False), obs: Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, is_shared=False)}, batch_size=torch.Size([]), device=None, is_shared=False) >>> print(step_mdp(td, keep_other=False)) # "extra" is missing TensorDict( fields={ done: Tensor(shape=torch.Size([]), device=cpu, dtype=torch.bool, is_shared=False), obs: Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, is_shared=False)}, batch_size=torch.Size([]), device=None, is_shared=False)
警告
如果排除奖励键时,奖励键也是输入键的一部分,则此函数将无法正常工作。 这就是为什么
RewardSum
转换默认将 episode reward 注册在 observation 中,而不是 reward spec 中。 当使用此函数的快速缓存版本 (_StepMDP
) 时,不应观察到此问题。