Composite¶
- class torchrl.data.Composite(*args, **kwargs)[源代码]¶
TensorSpecs 的组合。
如果一个
TensorSpec
是张量类别的集合描述,那么Composite
类类似于TensorDict
类。像TensorDict
一样,它有一个shape
(类似于TensorDict
的batch_size
)和一个可选的device
。- 参数:
- 变量:
device (torch.device or None) – 如果未指定,复合 spec 的 device 为
None
(TensorDict 的情况也是如此)。非 None 的 device 会约束所有叶子节点位于同一 device。另一方面,None
device 允许叶子节点拥有不同的 device。默认为None
。shape (torch.Size) – 所有叶子节点的前导 shape。等同于相应 tensordicts 的批大小 (batch-size)。
示例
>>> pixels_spec = Bounded( ... low=torch.zeros(4, 3, 32, 32), ... high=torch.ones(4, 3, 32, 32), ... dtype=torch.uint8 ... ) >>> observation_vector_spec = Bounded( ... low=torch.zeros(4, 33), ... high=torch.ones(4, 33), ... dtype=torch.float) >>> composite_spec = Composite( ... pixels=pixels_spec, ... observation_vector=observation_vector_spec, ... shape=(4,) ... ) >>> composite_spec Composite( pixels: BoundedDiscrete( shape=torch.Size([4, 3, 32, 32]), space=ContinuousBox( low=Tensor(shape=torch.Size([4, 3, 32, 32]), device=cpu, dtype=torch.uint8, contiguous=True), high=Tensor(shape=torch.Size([4, 3, 32, 32]), device=cpu, dtype=torch.uint8, contiguous=True)), device=cpu, dtype=torch.uint8, domain=discrete), observation_vector: BoundedContinuous( shape=torch.Size([4, 33]), space=ContinuousBox( low=Tensor(shape=torch.Size([4, 33]), device=cpu, dtype=torch.float32, contiguous=True), high=Tensor(shape=torch.Size([4, 33]), device=cpu, dtype=torch.float32, contiguous=True)), device=cpu, dtype=torch.float32, domain=continuous), device=None, shape=torch.Size([4])) >>> td = composite_spec.rand() >>> td TensorDict( fields={ observation_vector: Tensor(shape=torch.Size([4, 33]), device=cpu, dtype=torch.float32, is_shared=False), pixels: Tensor(shape=torch.Size([4, 3, 32, 32]), device=cpu, dtype=torch.uint8, is_shared=False)}, batch_size=torch.Size([4]), device=None, is_shared=False) >>> # we can build a nested composite spec using unnamed arguments >>> print(Composite({("a", "b"): None, ("a", "c"): None})) Composite( a: Composite( b: None, c: None, device=None, shape=torch.Size([])), device=None, shape=torch.Size([]))
- assert_is_in(value: 张量) None ¶
断言张量是否属于该框,否则引发异常。
- 参数:
value (torch.Tensor) – 要检查的值。
- contains(item: torch.Tensor | TensorDictBase) bool ¶
如果值
val
可以由TensorSpec
生成,则返回True
,否则返回False
。更多信息请参见
is_in()
。
- cpu()¶
将 TensorSpec 转换为 ‘cpu’ device。
- cuda(device=None)¶
将 TensorSpec 转换为 ‘cuda’ device。
- property device: Union[device, str, int]¶
spec 的 device。
只有
Composite
spec 可以有None
device。所有叶子节点必须有非空 device。
- encode(vals: Dict[str, Any], *, ignore_device: bool = False) Dict[str, 张量] [源代码]¶
根据指定的 spec 编码值,并返回相应的张量。
此方法用于返回一个值(例如 numpy 数组)的环境中,该值可以轻松映射到 TorchRL 所需的域。如果该值已经是张量,则 spec 不会更改其值并按原样返回。
- 参数:
val (np.ndarray or torch.Tensor) – 要编码为张量的值。
- 关键字参数:
ignore_device (bool, optional) – 如果为
True
,则 spec device 将被忽略。这用于在调用TensorDict(..., device="cuda")
中对张量进行分组转换,这样更快。- 返回:
与所需张量 spec 匹配的 torch.Tensor。
- expand(*shape)[源代码]¶
返回一个具有扩展 shape 的新 Spec。
- 参数:
*shape (tuple or iterable of int) – Spec 的新 shape。必须与当前 shape 可广播:其长度必须至少与当前 shape 长度一样长,并且其最后一个值也必须兼容;也就是说,它们只能在当前维度为单例时与当前 shape 不同。
- classmethod implements_for_spec(torch_function: Callable) Callable ¶
为 TensorSpec 注册一个 torch 函数覆盖。
- index(index: INDEX_TYPING, tensor_to_index: torch.Tensor | TensorDictBase) torch.Tensor | TensorDictBase [源代码]¶
对输入张量进行索引。
此方法用于编码一个或多个分类变量的 spec(例如
OneHot
或Categorical
),以便可以用样本对张量进行索引,而无需关心索引的实际表示。- 参数:
index (int, torch.Tensor, slice or list) – 张量的索引
tensor_to_index – 要索引的张量
- 返回:
索引后的张量
- 示例
>>> from torchrl.data import OneHot >>> import torch >>> >>> one_hot = OneHot(n=100) >>> categ = one_hot.to_categorical_spec() >>> idx_one_hot = torch.zeros((100,), dtype=torch.bool) >>> idx_one_hot[50] = 1 >>> print(one_hot.index(idx_one_hot, torch.arange(100))) tensor(50) >>> idx_categ = one_hot.to_categorical(idx_one_hot) >>> print(categ.index(idx_categ, torch.arange(100))) tensor(50)
- is_empty(recurse: bool | None =False)[源代码]¶
复合 spec 是否包含 specs。
- 参数:
recurse (bool) – 是否递归地评估 spec 是否为空。如果为
True
,如果不存在叶子节点则返回True
。如果为False
(默认值),则返回根级别是否定义了任何 spec。
- is_in(val: Union[dict, TensorDictBase]) bool [源代码]¶
如果值
val
可以由TensorSpec
生成,则返回True
,否则返回False
。更精确地说,
is_in
方法检查值val
是否在由space
属性(即边界)定义的限制内,以及dtype
、device
、shape
和其他可能的元数据是否与 spec 的元数据匹配。如果这些检查中的任何一个失败,is_in
方法将返回False
。- 参数:
val (torch.Tensor) – 要检查的值。
- 返回:
指示值是否属于 TensorSpec 边界的布尔值。
- items(include_nested: bool = False, leaves_only: bool = False, *, is_leaf: Callable[[type], bool] | None =None) _CompositeSpecItemsView [源代码]¶
Composite 的 items。
- 参数:
include_nested (bool, optional) – 如果为
False
,返回的键将不是嵌套的。它们只表示根节点的直接子节点,而不是整个嵌套序列,例如Composite(next=Composite(obs=None))
将产生键["next"]. 默认值为 ``False``
,即不返回嵌套键。leaves_only (bool, optional) – 如果为
False
,返回的值将包含每一层嵌套,例如Composite(next=Composite(obs=None))
将产生键["next", ("next", "obs")]
。默认值为False
。
- 关键字参数:
is_leaf (callable, optional) – 读取一个类型并返回一个布尔值,指示该类型是否应被视为叶子节点。默认情况下,所有非 Composite 节点都被视为叶子节点。
- keys(include_nested: bool = False, leaves_only: bool = False, *, is_leaf: Callable[[type], bool] | None =None) _CompositeSpecKeysView [源代码]¶
Composite 的 keys。
keys 参数反映了
tensordict.TensorDict
的 keys。- 参数:
include_nested (bool, optional) – 如果为
False
,返回的键将不是嵌套的。它们只表示根节点的直接子节点,而不是整个嵌套序列,例如Composite(next=Composite(obs=None))
将产生键["next"]. 默认值为 ``False``
,即不返回嵌套键。leaves_only (bool, optional) – 如果为
False
,返回的值将包含每一层嵌套,例如Composite(next=Composite(obs=None))
将产生键["next", ("next", "obs")]
。默认值为False
。
- 关键字参数:
is_leaf (callable, optional) – 读取一个类型并返回一个布尔值,指示该类型是否应被视为叶子节点。默认情况下,所有非 Composite 节点都被视为叶子节点。
- lock_(recurse: bool | None =None) T [源代码]¶
锁定 Composite 并阻止对其内容的修改。
recurse 参数控制锁定是否传播到子 spec。当前的默认值是
False
,但在 v0.8 中为了与 TensorDict API 保持一致,它将变为True
。示例
>>> shape = [3, 4, 5] >>> spec = Composite( ... a=Composite( ... b=Composite(shape=shape[:3], device="cpu"), shape=shape[:2] ... ), ... shape=shape[:1], ... ) >>> spec["a"] = spec["a"].clone() >>> recurse = False >>> spec.lock_(recurse=recurse) >>> try: ... spec["a"] = spec["a"].clone() ... except RuntimeError: ... print("failed!") failed! >>> try: ... spec["a", "b"] = spec["a", "b"].clone() ... print("succeeded!") ... except RuntimeError: ... print("failed!") succeeded! >>> recurse = True >>> spec.lock_(recurse=recurse) >>> try: ... spec["a", "b"] = spec["a", "b"].clone() ... print("succeeded!") ... except RuntimeError: ... print("failed!") failed!
- make_neg_dim(dim: int) T ¶
将指定维度转换为
-1
。
- property ndim¶
spec shape 的维度数量。
是
len(spec.shape)
的快捷方式。
- one(shape: torch.Size = None) torch.Tensor | TensorDictBase ¶
返回在该范围内填充为 1 的张量。
注意
即使无法保证
1
属于规范域,当此条件违反时,此方法也不会引发异常。one
的主要用例是生成空数据缓冲区,而不是有意义的数据。- 参数:
shape (torch.Size) – 填充为 1 的张量的形状
- 返回:
在 TensorSpec 范围内抽样得到的填充为 1 的张量。
- ones(shape: torch.Size = None) torch.Tensor | TensorDictBase ¶
是
one()
的代理。
- pop(key: NestedKey, default: Any = _NoDefault.ZERO) Any [源]¶
从复合规范中移除并返回与指定键关联的值。
此方法在复合规范中搜索给定的键,将其移除,并返回其关联的值。如果未找到该键,则返回指定的默认值(如果已指定),否则引发 KeyError。
- 参数:
key (NestedKey) – 要从复合规范中移除的键。它可以是单个键或嵌套键。
default (Any, optional) – 如果在复合规范中未找到指定的键,则返回的值。如果未提供且未找到该键,则会引发 KeyError。
- 返回:
从复合规范中移除的与指定键关联的值。
- 返回类型:
Any
- 引发:
KeyError – 如果在复合规范中未找到指定的键且未提供默认值。
- project(val: TensorDictBase) TensorDictBase [源]¶
如果输入张量不在 TensorSpec 范围内,则根据定义的启发式方法将其映射回该范围。
- 参数:
val (torch.Tensor) – 要映射到该范围的张量。
- 返回:
属于 TensorSpec 范围的 torch.Tensor。
- rand(shape: Optional[Size] = None) TensorDictBase [源]¶
返回在规范定义的空间内的随机张量。
抽样将在空间内均匀进行,除非该范围无界,在这种情况下将抽取正态分布值。
- 参数:
shape (torch.Size) – 随机张量的形状
- 返回:
在 TensorSpec 范围内抽样得到的随机张量。
- sample(shape: torch.Size = None) torch.Tensor | TensorDictBase ¶
返回在规范定义的空间内的随机张量。
详情请参阅
rand()
。
- separates(*keys: NestedKey, default: Optional[Any] = None) Composite [源]¶
通过提取指定的键及其关联的值,将复合规范分割成一个新的复合规范。
此方法遍历提供的键,将其从当前复合规范中移除,并将其添加到新的复合规范中。如果未找到键,则使用指定的默认值。返回新的复合规范。
- 参数:
*keys (NestedKey) – 要从复合规范中提取的一个或多个键。每个键可以是单个键或嵌套键。
default (Any, optional) – 如果在复合规范中未找到指定的键时使用的值。默认为 None。
- 返回:
包含提取的键及其关联值的新复合规范。
- 返回类型:
注意
如果找不到任何指定的键,则此方法返回 None。
- set(name: str, spec: TensorSpec) Composite [源]¶
在复合规范中设置一个规范。
- squeeze(dim: int | None = None)[源]¶
返回一个新的规范,其中所有大小为
1
的维度都被移除。当给定
dim
时,仅在该维度上执行压缩操作。- 参数:
dim (int or None) – 要应用压缩操作的维度
- to(dest: Union[dtype, device, str, int]) Composite [源]¶
将 TensorSpec 转换为指定设备或数据类型。
如果未进行更改,则返回相同的规范。
- to_numpy(val: TensorDict, safe: Optional[bool] = None) dict [源]¶
返回输入张量对应的
np.ndarray
。这旨在作为
encode()
的逆操作。- 参数:
val (torch.Tensor) – 要转换为 numpy 的张量。
safe (bool) – 布尔值,指示是否应对值对照规范域执行检查。默认为
CHECK_SPEC_ENCODE
环境变量的值。
- 返回:
一个 np.ndarray。
- type_check(value: Union[Tensor, TensorDictBase], selected_keys: Optional[Union[str, Sequence[str]]] = None)[源]¶
检查输入值的
dtype
是否与TensorSpec
的dtype
匹配,如果不匹配则引发异常。- 参数:
value (torch.Tensor) – 需要检查其数据类型的张量。
key (str, optional) – 如果 TensorSpec 有键,则值的 dtype 将对照由指定键指向的规范进行检查。
- unflatten(dim: int, sizes: Tuple[int]) T ¶
解展平一个
TensorSpec
。请查阅
unflatten()
以获取有关此方法的更多信息。
- unsqueeze(dim: int)[源]¶
返回一个新的规范,其中增加一个大小为 1 的维度(位于由
dim
指定的位置)。- 参数:
dim (int or None) – 要应用扩展维度操作的维度。
- values(include_nested: bool = False, leaves_only: bool = False, *, is_leaf: Callable[[type], bool] | None = None) _CompositeSpecValuesView [源]¶
复合对象的值。
- 参数:
include_nested (bool, optional) – 如果为
False
,返回的键将不是嵌套的。它们只表示根节点的直接子节点,而不是整个嵌套序列,例如Composite(next=Composite(obs=None))
将产生键["next"]. 默认值为 ``False``
,即不返回嵌套键。leaves_only (bool, optional) – 如果为
False
,返回的值将包含每一层嵌套,例如Composite(next=Composite(obs=None))
将产生键["next", ("next", "obs")]
。默认值为False
。
- 关键字参数:
is_leaf (callable, optional) – 读取一个类型并返回一个布尔值,指示该类型是否应被视为叶子节点。默认情况下,所有非 Composite 节点都被视为叶子节点。
- zero(shape: Optional[Size] = None) TensorDictBase [源]¶
返回在该范围内填充为 0 的张量。
注意
即使无法保证
0
属于规范域,当此条件违反时,此方法也不会引发异常。zero
的主要用例是生成空数据缓冲区,而不是有意义的数据。- 参数:
shape (torch.Size) – 填充为 0 的张量的形状
- 返回:
在 TensorSpec 范围内抽样得到的填充为 0 的张量。
- zeros(shape: torch.Size = None) torch.Tensor | TensorDictBase ¶
是
zero()
的代理。