ding.rl_utils

a2c

请参考 ding/rl_utils/a2c 获取更多详细信息。

a2c_error

ding.rl_utils.a2c_error(data: namedtuple) → namedtuple

Overview:

A2C（优势演员-评论家）的实现（arXiv:1602.01783）用于离散动作空间

Arguments:

• 数据 (namedtuple): a2c 输入数据，包含 a2c_data 中显示的字段

Returns:

• a2c_loss (namedtuple): a2c损失项，它们都是可微的0维张量

Shapes:

• logit (torch.FloatTensor): \((B, N)\), 其中 B 是批量大小，N 是动作维度

• 动作 (torch.LongTensor): \((B, )\)

• 值 (torch.FloatTensor): \((B, )\)

• adv (torch.FloatTensor): \((B, )\)

• 返回 (torch.FloatTensor): \((B, )\)

• 权重 (torch.FloatTensor 或 None): \((B, )\)

• policy_loss (torch.FloatTensor): \(()\), 0维张量

• value_loss (torch.FloatTensor): \(()\)

• entropy_loss (torch.FloatTensor): \(()\)

Examples:

>>> data = a2c_data(
>>>     logit=torch.randn(2, 3),
>>>     action=torch.randint(0, 3, (2, )),
>>>     value=torch.randn(2, ),
>>>     adv=torch.randn(2, ),
>>>     return_=torch.randn(2, ),
>>>     weight=torch.ones(2, ),
>>> )
>>> loss = a2c_error(data)

a2c_error_continuous

ding.rl_utils.a2c_error_continuous(data: namedtuple) → namedtuple

Overview:

A2C（优势演员-评论家）的实现（arXiv:1602.01783）用于连续动作空间

Arguments:

• 数据 (namedtuple): a2c 输入数据，包含在 a2c_data 中显示的字段

Returns:

• a2c_loss (namedtuple): a2c损失项，它们都是可微的0维张量

Shapes:

• logit (torch.FloatTensor): \((B, N)\), 其中 B 是批量大小，N 是动作维度

• 动作 (torch.LongTensor): \((B, N)\)

• 值 (torch.FloatTensor): \((B, )\)

• adv (torch.FloatTensor): \((B, )\)

• 返回 (torch.FloatTensor): \((B, )\)

• 权重 (torch.FloatTensor 或 None): \((B, )\)

• policy_loss (torch.FloatTensor): \(()\), 0维张量

• value_loss (torch.FloatTensor): \(()\)

• entropy_loss (torch.FloatTensor): \(()\)

Examples:

>>> data = a2c_data(
>>>     logit={'mu': torch.randn(2, 3), 'sigma': torch.sqrt(torch.randn(2, 3)**2)},
>>>     action=torch.randn(2, 3),
>>>     value=torch.randn(2, ),
>>>     adv=torch.randn(2, ),
>>>     return_=torch.randn(2, ),
>>>     weight=torch.ones(2, ),
>>> )
>>> loss = a2c_error_continuous(data)

宏碁

请参考 ding/rl_utils/acer 获取更多详细信息。

acer_policy_error

ding.rl_utils.acer_policy_error(q_values: Tensor, q_retraces: Tensor, v_pred: Tensor, target_logit: Tensor, actions: Tensor, ratio: Tensor, c_clip_ratio: float = 10.0) → Tuple[Tensor, Tensor]

Overview:

获取ACER策略损失。

Arguments:

• q_values (torch.Tensor): Q值

• q_retraces (torch.Tensor): Q值（通过retrace方法计算）

• v_pred (torch.Tensor): V值

• target_pi (torch.Tensor): 新策略的概率

• 动作 (torch.Tensor): 回放缓冲区中的动作

• 比率 (torch.Tensor): 新策略与行为策略的比率

• c_clip_ratio (float): 比率的裁剪值

Returns:

• actor_loss (torch.Tensor): 来自q_retrace的策略损失

• bc_loss (torch.Tensor): 正确的策略损失

Shapes:

• q_values (torch.FloatTensor): \((T, B, N)\), 其中 B 是批量大小，N 是动作维度

• q_retraces (torch.FloatTensor): \((T, B, 1)\)

• v_pred (torch.FloatTensor): \((T, B, 1)\)

• target_pi (torch.FloatTensor): \((T, B, N)\)

• 动作 (torch.LongTensor): \((T, B)\)

• 比率 (torch.FloatTensor): \((T, B, N)\)

• actor_loss (torch.FloatTensor): \((T, B, 1)\)

• bc_loss (torch.FloatTensor): \((T, B, 1)\)

Examples:

>>> q_values=torch.randn(2, 3, 4),
>>> q_retraces=torch.randn(2, 3, 1),
>>> v_pred=torch.randn(2, 3, 1),
>>> target_pi=torch.randn(2, 3, 4),
>>> actions=torch.randint(0, 4, (2, 3)),
>>> ratio=torch.randn(2, 3, 4),
>>> loss = acer_policy_error(q_values, q_retraces, v_pred, target_pi, actions, ratio)

acer_value_error

ding.rl_utils.acer_value_error(q_values, q_retraces, actions)

Overview:

获取ACER批评损失。

Arguments:

• q_values (torch.Tensor): Q值

• q_retraces (torch.Tensor): Q值（通过retrace方法计算）

• 动作 (torch.Tensor): 回放缓冲区中的动作

• 比率 (torch.Tensor): 新策略与行为策略的比率

Returns:

• critic_loss (torch.Tensor): 评论者损失

Shapes:

• q_values (torch.FloatTensor): \((T, B, N)\), 其中 B 是批量大小，N 是动作维度

• q_retraces (torch.FloatTensor): \((T, B, 1)\)

• 动作 (torch.LongTensor): \((T, B)\)

• critic_loss (torch.FloatTensor): \((T, B, 1)\)

Examples:

>>> q_values=torch.randn(2, 3, 4)
>>> q_retraces=torch.randn(2, 3, 1)
>>> actions=torch.randint(0, 4, (2, 3))
>>> loss = acer_value_error(q_values, q_retraces, actions)

acer_trust_region_update

ding.rl_utils.acer_trust_region_update(actor_gradients: List[Tensor], target_logit: Tensor, avg_logit: Tensor, trust_region_value: float) → List[Tensor]

Overview:

使用信任区域约束计算梯度

Arguments:

• actor_gradients (list(torch.Tensor)): 不同部分的梯度值

• target_pi (torch.Tensor): 新策略的概率

• avg_pi (torch.Tensor): 平均策略的概率

• trust_region_value (float): 信任区域的范围

Returns:

• update_gradients (list(torch.Tensor)): 带有信任区域约束的梯度

Shapes:

• target_pi (torch.FloatTensor): \((T, B, N)\)

• avg_pi (torch.FloatTensor): \((T, B, N)\)

• update_gradients (list(torch.FloatTensor)): \((T, B, N)\)

Examples:

>>> actor_gradients=[torch.randn(2, 3, 4)]
>>> target_pi=torch.randn(2, 3, 4)
>>> avg_pi=torch.randn(2, 3, 4)
>>> loss = acer_trust_region_update(actor_gradients, target_pi, avg_pi, 0.1)

加法器

请参考 ding/rl_utils/adder 获取更多详细信息。

加法器

class ding.rl_utils.adder.Adder

Overview:

Adder 是一个组件，用于处理 Collector 模块（数据生成和处理）中转换的不同变换和计算，例如 GAE、n-step return、转换采样等。

Interface:

__init__, get_gae, get_gae_with_default_last_value, get_nstep_return_data, get_train_sample

classmethod _get_null_transition(template: dict, null_transition: dict | None = None) → dict

Overview:

获取用于填充的空转换。如果 cls._null_transition 为 None，则返回输入的 template。

Arguments:

• 模板 (dict): 用于空转换的模板。

• null_transition (Optional[dict]): 字典类型的空转换，用于 null_padding

Returns:

• null_transition (dict): 深拷贝的空转换。

classmethod get_gae(data: List[Dict[str, Any]], last_value: Tensor, gamma: float, gae_lambda: float, cuda: bool) → List[Dict[str, Any]]

Overview:

获取堆叠转换的GAE优势（T时间步长，1批次）。调用gae进行计算。

Arguments:

• 数据 (list): 转换列表，每个元素是一个至少包含 ['value', 'reward'] 的转换字典。

• last_value (torch.Tensor): 最后一个值（即：T+1 时间步）

• gamma (float): 未来的折扣因子，应在 [0, 1] 范围内，默认为 0.99。

• gae_lambda (float): GAE lambda 参数，应在 [0, 1] 范围内，默认值为 0.97，当 lambda -> 0 时，会引入偏差，但当 lambda -> 1 时，由于项的总和，会有高方差。

• cuda (bool): 是否在GAE计算中使用cuda

Returns:

• 数据 (list): 类似于输入的转换列表，但每个元素都拥有额外的优势键 'adv'

Examples:

>>> B, T = 2, 3 # batch_size, timestep
>>> data = [dict(value=torch.randn(B), reward=torch.randn(B)) for _ in range(T)]
>>> last_value = torch.randn(B)
>>> gamma = 0.99
>>> gae_lambda = 0.95
>>> cuda = False
>>> data = Adder.get_gae(data, last_value, gamma, gae_lambda, cuda)

classmethod get_gae_with_default_last_value(data: deque, done: bool, gamma: float, gae_lambda: float, cuda: bool) → List[Dict[str, Any]]

Overview:

类似于上面的get_gae来获取堆叠转换的GAE优势。然而，此函数设计用于在未传递last_value的情况下使用。如果转换尚未完成，它将把data中的最后一个值分配为last_value，丢弃data中的最后一个元素（即len(data)将减少1），然后调用get_gae。否则，它将使last_value等于0。

Arguments:

• 数据 (deque): 转换列表，每个元素是一个转换字典，至少包含['value', 'reward']

• 完成 (bool): 转换是否达到一个情节的结束（即环境是否完成）

• gamma (float): 未来的折扣因子，应在 [0, 1] 范围内，默认为 0.99。

• gae_lambda (float): GAE lambda 参数，应在 [0, 1] 范围内，默认值为 0.97，当 lambda -> 0 时，会引入偏差，但当 lambda -> 1 时，由于项的总和，会有高方差。

• cuda (bool): 是否在GAE计算中使用cuda

Returns:

• 数据 (List[Dict[str, Any]]): 类似于输入的转换列表，但每个元素都拥有额外的优势键 'adv'

Examples:

>>> B, T = 2, 3 # batch_size, timestep
>>> data = [dict(value=torch.randn(B), reward=torch.randn(B)) for _ in range(T)]
>>> done = False
>>> gamma = 0.99
>>> gae_lambda = 0.95
>>> cuda = False
>>> data = Adder.get_gae_with_default_last_value(data, done, gamma, gae_lambda, cuda)

classmethod get_nstep_return_data(data: deque, nstep: int, cum_reward=False, correct_terminate_gamma=True, gamma=0.99) → deque

Overview:

通过更新数据字典元素中的键 ['next_obs', 'reward', 'done'] 来处理原始轨迹数据。

Arguments:

• 数据 (deque): 转换列表，每个元素是一个转换字典

• nstep (int): 步数。如果等于1，直接返回data；否则使用nstep值进行更新。

Returns:

• 数据 (deque): 类似于输入的转换列表，但每个元素都更新了nstep值。

Examples:

>>> data = [dict(
>>>     obs=torch.randn(B),
>>>     reward=torch.randn(1),
>>>     next_obs=torch.randn(B),
>>>     done=False) for _ in range(T)]
>>> nstep = 2
>>> data = Adder.get_nstep_return_data(data, nstep)

classmethod get_train_sample(data: List[Dict[str, Any]], unroll_len: int, last_fn_type: str = 'last', null_transition: dict | None = None) → List[Dict[str, Any]]

Overview:

通过更新数据字典元素中的键 ['next_obs', 'reward', 'done'] 来处理原始轨迹数据。 如果 unroll_len 等于 1，这意味着不需要处理，可以直接返回 data。 否则，data 将根据 unroll_len 进行分割，根据 last_fn_type 处理剩余部分，并调用 lists_to_dicts 来形成采样的训练数据。

Arguments:

• 数据 (List[Dict[str, Any]]): 转换列表，每个元素是一个转换字典

• unroll_len (int): 学习训练展开长度

• last_fn_type (str): 处理拆分后轨迹中最后剩余数据的方法类型名称，应在 ['last', 'drop', 'null_padding'] 中

• null_transition (Optional[dict]): 字典类型的空转换，用于 null_padding

Returns:

• data (List[Dict[str, Any]]): 展开后处理的转换列表

get_gae

ding.rl_utils.adder.get_gae(data: List[Dict[str, Any]], last_value: Tensor, gamma: float, gae_lambda: float, cuda: bool) → List[Dict[str, Any]]

Overview:

获取堆叠转换的GAE优势（T时间步长，1批次）。调用gae进行计算。

Arguments:

• 数据 (list): 转换列表，每个元素是一个至少包含 ['value', 'reward'] 的转换字典。

• last_value (torch.Tensor): 最后一个值（即：T+1 时间步）

• gamma (float): 未来的折扣因子，应在 [0, 1] 范围内，默认为 0.99。

• gae_lambda (float): GAE lambda 参数，应在 [0, 1] 范围内，默认值为 0.97，当 lambda -> 0 时，会引入偏差，但当 lambda -> 1 时，由于项的总和，会有高方差。

• cuda (bool): 是否在GAE计算中使用cuda

Returns:

• 数据 (list): 类似于输入的转换列表，但每个元素都拥有额外的优势键 'adv'

Examples:

>>> B, T = 2, 3 # batch_size, timestep
>>> data = [dict(value=torch.randn(B), reward=torch.randn(B)) for _ in range(T)]
>>> last_value = torch.randn(B)
>>> gamma = 0.99
>>> gae_lambda = 0.95
>>> cuda = False
>>> data = Adder.get_gae(data, last_value, gamma, gae_lambda, cuda)

get_gae_with_default_last_value

ding.rl_utils.adder.get_gae_with_default_last_value(data: deque, done: bool, gamma: float, gae_lambda: float, cuda: bool) → List[Dict[str, Any]]

Overview:

类似于上面的get_gae来获取堆叠转换的GAE优势。然而，此函数设计用于在未传递last_value的情况下使用。如果转换尚未完成，它将把data中的最后一个值分配为last_value，丢弃data中的最后一个元素（即len(data)将减少1），然后调用get_gae。否则，它将使last_value等于0。

Arguments:

• 数据 (deque): 转换列表，每个元素是一个转换字典，至少包含['value', 'reward']

• 完成 (bool): 转换是否达到一个情节的结束（即环境是否完成）

• gamma (float): 未来的折扣因子，应在 [0, 1] 范围内，默认为 0.99。

• gae_lambda (float): GAE lambda 参数，应在 [0, 1] 范围内，默认值为 0.97，当 lambda -> 0 时，会引入偏差，但当 lambda -> 1 时，由于项的总和，会有高方差。

• cuda (bool): 是否在GAE计算中使用cuda

Returns:

• 数据 (List[Dict[str, Any]]): 类似于输入的转换列表，但每个元素都拥有额外的优势键 'adv'

Examples:

>>> B, T = 2, 3 # batch_size, timestep
>>> data = [dict(value=torch.randn(B), reward=torch.randn(B)) for _ in range(T)]
>>> done = False
>>> gamma = 0.99
>>> gae_lambda = 0.95
>>> cuda = False
>>> data = Adder.get_gae_with_default_last_value(data, done, gamma, gae_lambda, cuda)

获取n步回报数据

ding.rl_utils.adder.get_nstep_return_data(data: deque, nstep: int, cum_reward=False, correct_terminate_gamma=True, gamma=0.99) → deque

Overview:

通过更新数据字典元素中的键 ['next_obs', 'reward', 'done'] 来处理原始轨迹数据。

Arguments:

• 数据 (deque): 转换列表，每个元素是一个转换字典

• nstep (int): 步数。如果等于1，直接返回data；否则使用nstep值进行更新。

Returns:

• 数据 (deque): 类似于输入的转换列表，但每个元素都更新了nstep值。

Examples:

>>> data = [dict(
>>>     obs=torch.randn(B),
>>>     reward=torch.randn(1),
>>>     next_obs=torch.randn(B),
>>>     done=False) for _ in range(T)]
>>> nstep = 2
>>> data = Adder.get_nstep_return_data(data, nstep)

获取训练样本

ding.rl_utils.adder.get_train_sample(data: List[Dict[str, Any]], unroll_len: int, last_fn_type: str = 'last', null_transition: dict | None = None) → List[Dict[str, Any]]

Overview:

通过更新数据字典元素中的键 ['next_obs', 'reward', 'done'] 来处理原始轨迹数据。 如果 unroll_len 等于1，这意味着不需要处理，可以直接返回 data。 否则，data 将根据 unroll_len 进行分割，根据 last_fn_type 处理剩余部分，并调用 lists_to_dicts 来形成采样的训练数据。

Arguments:

• 数据 (List[Dict[str, Any]]): 转换列表，每个元素是一个转换字典

• unroll_len (int): 学习训练展开长度

• last_fn_type (str): 处理拆分后轨迹中最后剩余数据的方法类型名称，应在 ['last', 'drop', 'null_padding'] 中

• null_transition (Optional[dict]): 字典类型的空转换，用于 null_padding

Returns:

• data (List[Dict[str, Any]]): 展开后处理的转换列表

beta_function

请参考 ding/rl_utils/beta_function 获取更多详细信息。

cpw

ding.rl_utils.beta_function.cpw(x: Tensor | float, eta: float = 0.71) → Tensor | float

Overview:

CPW函数的实现。

Arguments:

• x (Union[torch.Tensor, float]): 输入值。

• eta (float): CPW函数的超参数。

Returns:

• 输出 (Union[torch.Tensor, float]): 输出值。

条件风险价值

ding.rl_utils.beta_function.CVaR(x: Tensor | float, eta: float = 0.71) → Tensor | float

Overview:

CVaR函数的实现，这是一个风险规避函数。

Arguments:

• x (Union[torch.Tensor, float]): 输入值。

• eta (float): CVaR函数的超参数。

Returns:

• 输出 (Union[torch.Tensor, float]): 输出值。

beta_function_map

rl_utils.beta_function_map = {'CPW': <function cpw>, 'CVaR': <function CVaR>, 'Pow': <function Pow>, 'uniform': <function <lambda>>}

昏迷

请参考 ding/rl_utils/coma 获取更多详细信息。

coma_error

ding.rl_utils.coma_error(data: namedtuple, gamma: float, lambda_: float) → namedtuple

Overview:

COMA的实现

Arguments:

• 数据 (namedtuple): 包含在 coma_data 中显示的字段的coma输入数据

Returns:

• coma_loss (namedtuple): 昏迷损失项，它们都是可微的0维张量

Shapes:

• logit (torch.FloatTensor): \((T, B, A, N)\), 其中 B 是批量大小，A 是代理数量，N 是动作维度

• 动作 (torch.LongTensor): \((T, B, A)\)

• q_value (torch.FloatTensor): \((T, B, A, N)\)

• 目标Q值 (torch.FloatTensor): \((T, B, A, N)\)

• 奖励 (torch.FloatTensor): \((T, B)\)

• 权重 (torch.FloatTensor 或 None): \((T ,B, A)\)

• policy_loss (torch.FloatTensor): \(()\), 0维张量

• value_loss (torch.FloatTensor): \(()\)

• entropy_loss (torch.FloatTensor): \(()\)

Examples:

>>> action_dim = 4
>>> agent_num = 3
>>> data = coma_data(
>>>     logit=torch.randn(2, 3, agent_num, action_dim),
>>>     action=torch.randint(0, action_dim, (2, 3, agent_num)),
>>>     q_value=torch.randn(2, 3, agent_num, action_dim),
>>>     target_q_value=torch.randn(2, 3, agent_num, action_dim),
>>>     reward=torch.randn(2, 3),
>>>     weight=torch.ones(2, 3, agent_num),
>>> )
>>> loss = coma_error(data, 0.99, 0.99)

探索

请参考 ding/rl_utils/exploration 获取更多详细信息。

get_epsilon_greedy_fn

ding.rl_utils.exploration.get_epsilon_greedy_fn(start: float, end: float, decay: int, type_: str = 'exp') → Callable

Overview:

生成一个带有衰减的epsilon_greedy函数，该函数输入当前时间步并输出当前的epsilon。

Arguments:

• 开始值 (float): Epsilon 的起始值。对于 linear，它应该是 1.0。

• 结束值 (float): Epsilon 结束值。

• 衰减 (int): 控制 epsilon 从 start 到 end 减少的速度。我们建议 epsilon 根据环境步骤而不是迭代来衰减。

• 类型 (str): epsilon 如何衰减，目前支持 ['linear', 'exp'(指数)] 。

Returns:

• eps_fn (function): 带有衰减的epsilon贪婪函数。

基础噪声

class ding.rl_utils.exploration.BaseNoise

Overview:

动作噪声的基类

Interface:

__init__, __call__

Examples:

>>> noise_generator = OUNoise()  # init one type of noise
>>> noise = noise_generator(action.shape, action.device)  # generate noise

abstract __call__(shape: tuple, device: str) → Tensor

Overview:

根据动作张量的形状和设备生成噪声。

Arguments:

• 形状 (tuple): 动作张量的大小，输出噪声的大小应与之相同。

• 设备 (str): 动作张量的设备，输出噪声的设备应与之相同。

Returns:

• 噪声 (torch.Tensor): 生成的动作噪声，与输入动作张量具有相同的形状和设备。

__init__() → None

Overview:

初始化方法。

高斯噪声

class ding.rl_utils.exploration.GaussianNoise(mu: float = 0.0, sigma: float = 1.0)

Overview:

用于生成高斯噪声的派生类，满足 \(X \sim N(\mu, \sigma^2)\)

Interface:

__init__, __call__

__call__(shape: tuple, device: str) → Tensor

Overview:

根据动作张量的形状和设备生成高斯噪声

Arguments:

• 形状 (tuple): 动作张量的大小，输出噪声的大小应与之相同

• 设备 (str): 动作张量的设备，输出噪声的设备应与之相同

Returns:

• 噪声 (torch.Tensor): 生成的动作噪声，与输入动作张量具有相同的形状和设备

__init__(mu: float = 0.0, sigma: float = 1.0) → None

Overview:

初始化高斯分布中的\(\mu\)和\(\sigma\)。

Arguments:

• mu (float): \(\mu\) , 平均值。

• sigma (float): \(\sigma\) , 标准差, 应为正数。

OU噪声

class ding.rl_utils.exploration.OUNoise(mu: float = 0.0, sigma: float = 0.3, theta: float = 0.15, dt: float = 0.01, x0: float | Tensor | None = 0.0)

Overview:

用于生成Ornstein-Uhlenbeck过程噪声的派生类。 满足 \(dx_t=\theta(\mu-x_t)dt + \sigma dW_t\)， 其中 \(W_t\) 表示Weiner过程，作为随机扰动项。

Interface:

__init__, reset, __call__

__call__(shape: tuple, device: str, mu: float | None = None) → Tensor

Overview:

根据动作张量的形状和设备生成高斯噪声。

Arguments:

• 形状 (tuple): 动作张量的大小，输出噪声的大小应与之相同。

• 设备 (str): 动作张量的设备，输出噪声的设备应与之相同。

• mu (float): 新的平均值 \(\mu\)，如果不需要可以设置为 None。

Returns:

• 噪声 (torch.Tensor): 生成的动作噪声，与输入动作张量具有相同的形状和设备。

__init__(mu: float = 0.0, sigma: float = 0.3, theta: float = 0.15, dt: float = 0.01, x0: float | Tensor | None = 0.0) → None

Overview:

初始化 _alpha \(= heta * dt\`, ``beta`\) \(= \sigma * \sqrt{dt}\), 在Ornstein-Uhlenbeck过程中。

Arguments:

• mu (float): \(\mu\) , 平均值。

• sigma (float): \(\sigma\) , 扰动噪声的标准差。

• theta (float): 噪声对扰动的反应强度，值越大表示反应越强。

• dt (float): 时间 t 的导数。

• x0 (Union[float, torch.Tensor]): 噪声的初始状态，应该是一个标量或与动作张量形状相同的张量。

reset() → None

Overview:

将 _x 重置为初始状态 _x0。

create_noise_generator

ding.rl_utils.exploration.create_noise_generator(noise_type: str, noise_kwargs: dict) → BaseNoise

Overview:

给定键（noise_type），如果在noise_mapping的值中，则创建一个新的噪声生成器实例，否则抛出KeyError。换句话说，派生的噪声生成器必须首先注册，然后调用create_noise generator来获取实例对象。

Arguments:

• 噪声类型 (str): 要创建的噪声生成器的类型。

Returns:

• 噪声 (BaseNoise): 创建的新噪声生成器，应该是 noise_mapping 值之一的实例。

gae

请参考 ding/rl_utils/gae 获取更多详细信息。

gae_data

class ding.rl_utils.gae.gae_data(value, next_value, reward, done, traj_flag)

shape_fn_gae

ding.rl_utils.gae.shape_fn_gae(args, kwargs)

Overview:

返回用于HPC的GAE形状

Returns:

形状: [T, B]

gae

ding.rl_utils.gae.gae(data: namedtuple, gamma: float = 0.99, lambda_: float = 0.97) → FloatTensor

Overview:

广义优势估计器的实现 (arXiv:1506.02438)

Arguments:

• 数据 (namedtuple): gae 输入数据包含字段 ['value', 'reward']，其中包含一些片段或轨迹数据。

• gamma (float): 未来的折扣因子，应在 [0, 1] 范围内，默认为 0.99。

• lambda (float): GAE参数lambda，应在[0, 1]范围内，默认值为0.97，当lambda -> 0时，会引入偏差，但当lambda -> 1时，由于项的总和，它具有高方差。

Returns:

• adv (torch.FloatTensor): 计算得到的优势

Shapes:

• 值 (torch.FloatTensor): \((T, B)\), 其中 T 是轨迹长度，B 是批量大小

• next_value (torch.FloatTensor): \((T, B)\)

• 奖励 (torch.FloatTensor): \((T, B)\)

• adv (torch.FloatTensor): \((T, B)\)

Examples:

>>> value = torch.randn(2, 3)
>>> next_value = torch.randn(2, 3)
>>> reward = torch.randn(2, 3)
>>> data = gae_data(value, next_value, reward, None, None)
>>> adv = gae(data)

isw

请参考 ding/rl_utils/isw 获取更多详细信息。

计算重要性权重

ding.rl_utils.isw.compute_importance_weights(target_output: Tensor | dict, behaviour_output: Tensor | dict, action: Tensor, action_space_type: str = 'discrete', requires_grad: bool = False)

Overview:

使用给定的输出和动作计算重要性采样权重

Arguments:

• target_output (Union[torch.Tensor,dict]): 当前策略网络采取行动的输出，通常如果动作空间是离散的，这个输出是网络输出的logit，如果动作空间是连续的，则是一个包含动作分布参数的字典。

• 行为输出 (Union[torch.Tensor,dict]): 行为策略网络采取行动的输出，通常这个输出是网络输出的logit，如果动作空间是离散的，或者是一个包含动作分布参数的字典，如果动作空间是连续的。

• 动作 (torch.Tensor): 在轨迹中选择的动作（离散动作空间的索引），即：行为动作

• action_space_type (str): 动作空间类型在 ['discrete', 'continuous'] 中

• requires_grad (bool): 是否需要计算梯度

Returns:

• rhos (torch.Tensor): 重要性采样权重

Shapes:

• target_output (Union[torch.FloatTensor,dict]): \((T, B, N)\), 其中 T 是时间步长，B 是批量大小，N 是动作维度

• 行为输出 (Union[torch.FloatTensor,dict]): \((T, B, N)\)

• 动作 (torch.LongTensor): \((T, B)\)

• rhos (torch.FloatTensor): \((T, B)\)

Examples:

>>> target_output = torch.randn(2, 3, 4)
>>> behaviour_output = torch.randn(2, 3, 4)
>>> action = torch.randint(0, 4, (2, 3))
>>> rhos = compute_importance_weights(target_output, behaviour_output, action)

ppg

请参考 ding/rl_utils/ppg 获取更多详细信息。

ppg_data

class ding.rl_utils.ppg.ppg_data(logit_new, logit_old, action, value_new, value_old, return_, weight)

ppg_joint_loss

class ding.rl_utils.ppg.ppg_joint_loss(auxiliary_loss, behavioral_cloning_loss)

ppg_joint_error

ding.rl_utils.ppg.ppg_joint_error(data: namedtuple, clip_ratio: float = 0.2, use_value_clip: bool = True) → Tuple[namedtuple, namedtuple]

Overview:

获取PPG联合损失

Arguments:

• 数据 (namedtuple): ppg 输入数据，字段显示在 ppg_data 中

• clip_ratio (float): 比率的裁剪值

• use_value_clip (bool): 是否使用值裁剪

Returns:

• ppg_joint_loss (namedtuple): ppg损失项，它们都是可微的0维张量

Shapes:

• logit_new (torch.FloatTensor): \((B, N)\), 其中 B 是批量大小，N 是动作维度

• logit_old (torch.FloatTensor): \((B, N)\)

• 动作 (torch.LongTensor): \((B,)\)

• value_new (torch.FloatTensor): \((B, 1)\)

• value_old (torch.FloatTensor): \((B, 1)\)

• 返回 (torch.FloatTensor): \((B, 1)\)

• 权重 (torch.FloatTensor): \((B,)\)

• 辅助损失 (torch.FloatTensor): \(()\), 0维张量

• 行为克隆损失 (torch.FloatTensor): \(()\)

Examples:

>>> action_dim = 4
>>> data = ppg_data(
>>>     logit_new=torch.randn(3, action_dim),
>>>     logit_old=torch.randn(3, action_dim),
>>>     action=torch.randint(0, action_dim, (3,)),
>>>     value_new=torch.randn(3, 1),
>>>     value_old=torch.randn(3, 1),
>>>     return_=torch.randn(3, 1),
>>>     weight=torch.ones(3),
>>> )
>>> loss = ppg_joint_error(data, 0.99, 0.99)

ppo

请参考 ding/rl_utils/ppo 获取更多详细信息。

ppo_data

class ding.rl_utils.ppo.ppo_data(logit_new, logit_old, action, value_new, value_old, adv, return_, weight)

ppo_policy_data

class ding.rl_utils.ppo.ppo_policy_data(logit_new, logit_old, action, adv, weight)

ppo_value_data

class ding.rl_utils.ppo.ppo_value_data(value_new, value_old, return_, weight)

ppo_loss

class ding.rl_utils.ppo.ppo_loss(policy_loss, value_loss, entropy_loss)

ppo_policy_loss

class ding.rl_utils.ppo.ppo_policy_loss(policy_loss, entropy_loss)

ppo信息

class ding.rl_utils.ppo.ppo_info(approx_kl, clipfrac)

shape_fn_ppo

ding.rl_utils.ppo.shape_fn_ppo(args, kwargs)

Overview:

返回hpc的ppo形状

Returns:

形状: [B, N]

ppo_error

ding.rl_utils.ppo.ppo_error(data: namedtuple, clip_ratio: float = 0.2, use_value_clip: bool = True, dual_clip: float | None = None) → Tuple[namedtuple, namedtuple]

Overview:

实现近端策略优化（arXiv:1707.06347），包含value_clip和dual_clip

Arguments:

• 数据 (namedtuple): 包含在 ppo_data 中显示的字段的 ppo 输入数据

• clip_ratio (float): 用于策略更新约束的PPO剪裁比率，默认为0.2

• use_value_clip (bool): 是否在值损失中使用与策略相同的比例的裁剪

• dual_clip (float): 在arXiv:1912.09729 公式5中提到的参数c，应在[1, inf)范围内，默认值为5.0，如果你不想使用它，将此参数设置为None

Returns:

• ppo_loss (namedtuple): PPO损失项，所有这些都是可微的0维张量

• ppo_info (namedtuple): 用于监控的ppo优化信息，所有信息都是Python标量

Shapes:

• logit_new (torch.FloatTensor): \((B, N)\), 其中 B 是批量大小，N 是动作维度

• logit_old (torch.FloatTensor): \((B, N)\)

• 动作 (torch.LongTensor): \((B, )\)

• value_new (torch.FloatTensor): \((B, )\)

• value_old (torch.FloatTensor): \((B, )\)

• adv (torch.FloatTensor): \((B, )\)

• 返回 (torch.FloatTensor): \((B, )\)

• 权重 (torch.FloatTensor 或 None): \((B, )\)

• policy_loss (torch.FloatTensor): \(()\), 0维张量

• value_loss (torch.FloatTensor): \(()\)

• entropy_loss (torch.FloatTensor): \(()\)

Examples:

>>> action_dim = 4
>>> data = ppo_data(
>>>     logit_new=torch.randn(3, action_dim),
>>>     logit_old=torch.randn(3, action_dim),
>>>     action=torch.randint(0, action_dim, (3,)),
>>>     value_new=torch.randn(3),
>>>     value_old=torch.randn(3),
>>>     adv=torch.randn(3),
>>>     return_=torch.randn(3),
>>>     weight=torch.ones(3),
>>> )
>>> loss, info = ppo_error(data)

注意

adv 已经是归一化的值 (adv - adv.mean()) / (adv.std() + 1e-8)，并且有许多方法可以计算这个均值和标准差，比如在数据缓冲区或训练批次中，所以我们不将这部分耦合到 ppo_error 中，你可以参考我们的示例了解不同的方法。

ppo_policy_error

ding.rl_utils.ppo.ppo_policy_error(data: namedtuple, clip_ratio: float = 0.2, dual_clip: float | None = None) → Tuple[namedtuple, namedtuple]

Overview:

获取PPO策略损失

Arguments:

• 数据 (namedtuple): ppo 输入数据，包含 ppo_policy_data 中显示的字段

• clip_ratio (float): 比率的裁剪值

• dual_clip (float): 在arXiv:1912.09729 公式5中提到的参数c，应在[1, inf)范围内，默认值为5.0，如果你不想使用它，将此参数设置为None

Returns:

• ppo_policy_loss (namedtuple): PPO策略损失项，所有这些都是可微的0维张量

• ppo_info (namedtuple): 用于监控的ppo优化信息，所有信息都是Python标量

Shapes:

• logit_new (torch.FloatTensor): \((B, N)\), 其中 B 是批量大小，N 是动作维度

• logit_old (torch.FloatTensor): \((B, N)\)

• 动作 (torch.LongTensor): \((B, )\)

• adv (torch.FloatTensor): \((B, )\)

• 权重 (torch.FloatTensor 或 None): \((B, )\)

• policy_loss (torch.FloatTensor): \(()\), 0维张量

• entropy_loss (torch.FloatTensor): \(()\)

Examples:

>>> action_dim = 4
>>> data = ppo_policy_data(
>>>     logit_new=torch.randn(3, action_dim),
>>>     logit_old=torch.randn(3, action_dim),
>>>     action=torch.randint(0, action_dim, (3,)),
>>>     adv=torch.randn(3),
>>>     weight=torch.ones(3),
>>> )
>>> loss, info = ppo_policy_error(data)

ppo_value_error

ding.rl_utils.ppo.ppo_value_error(data: namedtuple, clip_ratio: float = 0.2, use_value_clip: bool = True) → Tensor

Overview:

获取PPO值损失

Arguments:

• 数据 (namedtuple): ppo 输入数据，包含 ppo_value_data 中显示的字段

• clip_ratio (float): 比率的裁剪值

• use_value_clip (bool): 是否使用值裁剪

Returns:

• value_loss (torch.FloatTensor): PPO值损失项，所有这些都是可微的0维张量

Shapes:

• value_new (torch.FloatTensor): \((B, )\), 其中 B 是批量大小

• value_old (torch.FloatTensor): \((B, )\)

• 返回 (torch.FloatTensor): \((B, )\)

• 权重 (torch.FloatTensor 或 None): \((B, )\)

• value_loss (torch.FloatTensor): \(()\), 0维张量

Examples:

>>> action_dim = 4
>>> data = ppo_value_data(
>>>     value_new=torch.randn(3),
>>>     value_old=torch.randn(3),
>>>     return_=torch.randn(3),
>>>     weight=torch.ones(3),
>>> )
>>> loss, info = ppo_value_error(data)

ppo_error_continuous

ding.rl_utils.ppo.ppo_error_continuous(data: namedtuple, clip_ratio: float = 0.2, use_value_clip: bool = True, dual_clip: float | None = None) → Tuple[namedtuple, namedtuple]

Overview:

实现近端策略优化（arXiv:1707.06347），包含value_clip和dual_clip

Arguments:

• 数据 (namedtuple): 包含在 ppo_data 中显示的字段的 ppo 输入数据

• clip_ratio (float): 用于策略更新约束的PPO剪裁比率，默认为0.2

• use_value_clip (bool): 是否在值损失中使用与策略相同的比例的裁剪

• dual_clip (float): 在arXiv:1912.09729 公式5中提到的参数c，应在[1, inf)范围内，默认值为5.0，如果你不想使用它，将此参数设置为None

Returns:

• ppo_loss (namedtuple): ppo损失项，所有项都是可微的0维张量

• ppo_info (namedtuple): 用于监控的ppo优化信息，所有信息都是Python标量

Shapes:

• mu_sigma_new (tuple): \(((B, N), (B, N))\), 其中 B 是批量大小，N 是动作维度

• mu_sigma_old (tuple): \(((B, N), (B, N))\), 其中 B 是批量大小，N 是动作维度

• 动作 (torch.LongTensor): \((B, )\)

• value_new (torch.FloatTensor): \((B, )\)

• value_old (torch.FloatTensor): \((B, )\)

• adv (torch.FloatTensor): \((B, )\)

• 返回 (torch.FloatTensor): \((B, )\)

• 权重 (torch.FloatTensor 或 None): \((B, )\)

• policy_loss (torch.FloatTensor): \(()\), 0维张量

• value_loss (torch.FloatTensor): \(()\)

• entropy_loss (torch.FloatTensor): \(()\)

Examples:

>>> action_dim = 4
>>> data = ppo_data_continuous(
>>>     mu_sigma_new= dict(mu=torch.randn(3, action_dim), sigma=torch.randn(3, action_dim)**2),
>>>     mu_sigma_old= dict(mu=torch.randn(3, action_dim), sigma=torch.randn(3, action_dim)**2),
>>>     action=torch.randn(3, action_dim),
>>>     value_new=torch.randn(3),
>>>     value_old=torch.randn(3),
>>>     adv=torch.randn(3),
>>>     return_=torch.randn(3),
>>>     weight=torch.ones(3),
>>> )
>>> loss, info = ppo_error(data)

注意

adv 已经是归一化的值 (adv - adv.mean()) / (adv.std() + 1e-8)，并且有许多方法可以计算这个均值和标准差，比如在数据缓冲区或训练批次中，所以我们不将这部分耦合到 ppo_error 中，你可以参考我们的示例了解不同的方法。

ppo_policy_error_continuous

ding.rl_utils.ppo.ppo_policy_error_continuous(data: namedtuple, clip_ratio: float = 0.2, dual_clip: float | None = None) → Tuple[namedtuple, namedtuple]

Overview:

实现带有dual_clip的近端策略优化（arXiv:1707.06347）

Arguments:

• 数据 (namedtuple): 包含在 ppo_data 中显示的字段的 ppo 输入数据

• clip_ratio (float): 用于策略更新约束的PPO剪裁比率，默认为0.2

• dual_clip (float): 在arXiv:1912.09729 公式5中提到的参数c，应在[1, inf)范围内，默认值为5.0，如果你不想使用它，将此参数设置为None

Returns:

• ppo_loss (namedtuple): ppo损失项，它们都是可微的0维张量

• ppo_info (namedtuple): 用于监控的ppo优化信息，所有信息都是Python标量

Shapes:

• mu_sigma_new (tuple): \(((B, N), (B, N))\), 其中 B 是批量大小，N 是动作维度

• mu_sigma_old (tuple): \(((B, N), (B, N))\), 其中 B 是批量大小，N 是动作维度

• 动作 (torch.LongTensor): \((B, )\)

• adv (torch.FloatTensor): \((B, )\)

• 权重 (torch.FloatTensor 或 None): \((B, )\)

• policy_loss (torch.FloatTensor): \(()\), 0维张量

• entropy_loss (torch.FloatTensor): \(()\)

Examples:

>>> action_dim = 4
>>> data = ppo_policy_data_continuous(
>>>     mu_sigma_new=dict(mu=torch.randn(3, action_dim), sigma=torch.randn(3, action_dim)**2),
>>>     mu_sigma_old=dict(mu=torch.randn(3, action_dim), sigma=torch.randn(3, action_dim)**2),
>>>     action=torch.randn(3, action_dim),
>>>     adv=torch.randn(3),
>>>     weight=torch.ones(3),
>>> )
>>> loss, info = ppo_policy_error_continuous(data)

回撤

请参考 ding/rl_utils/retrace 获取更多详细信息。

计算Q回放

ding.rl_utils.retrace.compute_q_retraces(q_values: Tensor, v_pred: Tensor, rewards: Tensor, actions: Tensor, weights: Tensor, ratio: Tensor, gamma: float = 0.9) → Tensor

Shapes:

• q_values (torch.Tensor): \((T + 1, B, N)\), 其中 T 是 unroll_len, B 是批量大小, N 是离散动作维度。

• v_pred (torch.Tensor): \((T + 1, B, 1)\)

• 奖励 (torch.Tensor): \((T, B)\)

• 动作 (torch.Tensor): \((T, B)\)

• 权重 (torch.Tensor): \((T, B)\)

• 比率 (torch.Tensor): \((T, B, N)\)

• q_retraces (torch.Tensor): \((T + 1, B, 1)\)

Examples:

>>> T=2
>>> B=3
>>> N=4
>>> q_values=torch.randn(T+1, B, N)
>>> v_pred=torch.randn(T+1, B, 1)
>>> rewards=torch.randn(T, B)
>>> actions=torch.randint(0, N, (T, B))
>>> weights=torch.ones(T, B)
>>> ratio=torch.randn(T, B, N)
>>> q_retraces = compute_q_retraces(q_values, v_pred, rewards, actions, weights, ratio)

注意

q_retrace 操作不需要计算梯度，只需执行前向计算。

采样器

请参考 ding/rl_utils/sampler 获取更多详细信息。

ArgmaxSampler

class ding.rl_utils.sampler.ArgmaxSampler

Overview:

Argmax采样器，返回最大值的索引

__call__(logit: Tensor) → Tensor

Overview:

返回最大值的索引

Arguments:

• logit (torch.Tensor): 输入张量

Returns:

• action (torch.Tensor): 最大值的索引

多项式采样器

class ding.rl_utils.sampler.MultinomialSampler

Overview:

多项式采样器，返回采样值的索引

__call__(logit: Tensor) → Tensor

Overview:

返回采样值的索引

Arguments:

• logit (torch.Tensor): 输入张量

Returns:

• 动作 (torch.Tensor): 采样值的索引

MuSampler

class ding.rl_utils.sampler.MuSampler

Overview:

Mu采样器，返回输入张量的mu

__call__(logit: Tensor) → Tensor

Overview:

返回输入张量的 mu

Arguments:

• logit (ttorch.Tensor): 输入张量

Returns:

• 动作 (torch.Tensor): 输入张量的均值

重参数化采样器

class ding.rl_utils.sampler.ReparameterizationSampler

Overview:

重新参数化采样器，返回输入张量的重新参数化值

__call__(logit: Tensor) → Tensor

Overview:

返回输入张量的重新参数化值

Arguments:

• logit (ttorch.Tensor): 输入张量

Returns:

• action (torch.Tensor): 输入张量的重新参数化值

混合随机采样器

class ding.rl_utils.sampler.HybridStochasticSampler

Overview:

混合随机采样器，返回采样的动作类型和重新参数化的动作参数

__call__(logit: Tensor) → Tensor

Overview:

返回采样的动作类型和重新参数化的动作参数

Arguments:

• logit (ttorch.Tensor): 输入张量

Returns:

• 动作 (ttorch.Tensor): 采样的动作类型和重新参数化的动作参数

混合确定性采样器

class ding.rl_utils.sampler.HybridDeterminsticSampler

Overview:

混合确定性采样器，返回argmax动作类型和mu动作参数

__call__(logit: Tensor) → Tensor

Overview:

返回argmax动作类型和mu动作参数

Arguments:

• logit (ttorch.Tensor): 输入张量

Returns:

• action (ttorch.Tensor): 最大动作类型和mu动作参数

td

请参考 ding/rl_utils/td 获取更多详细信息。

q_1step_td_data

class ding.rl_utils.td.q_1step_td_data(q, next_q, act, next_act, reward, done, weight)

q_1step_td_error

ding.rl_utils.td.q_1step_td_error(data: ~collections.namedtuple, gamma: float, criterion: <module 'torch.nn.modules' from '/home/docs/checkouts/readthedocs.org/user_builds/di-engine-docs/envs/latest/lib/python3.9/site-packages/torch/nn/modules/__init__.py'> = MSELoss()) → Tensor

Overview:

1步td_error，支持单代理情况和多代理情况。

Arguments:

• 数据 (q_1step_td_data): 输入数据，q_1step_td_data 用于计算损失

• gamma (float): 折扣因子

• criterion (torch.nn.modules): 损失函数准则

Returns:

• 损失 (torch.Tensor): 一步时间差分误差

Shapes:

• 数据 (q_1step_td_data): 包含 ['q', 'next_q', 'act', 'next_act', 'reward', 'done', 'weight'] 的 q_1step_td_data

• q (torch.FloatTensor): \((B, N)\) 即 [批量大小, 动作维度]

• next_q (torch.FloatTensor): \((B, N)\) 即 [batch_size, action_dim]

• act (torch.LongTensor): \((B, )\)

• next_act (torch.LongTensor): \((B, )\)

• 奖励 (torch.FloatTensor): \(( , B)\)

• 完成 (torch.BoolTensor) \((B, )\), 是否在最后一步完成

• 权重 (torch.FloatTensor 或 None): \((B, )\), 训练样本的权重

Examples:

>>> action_dim = 4
>>> data = q_1step_td_data(
>>>     q=torch.randn(3, action_dim),
>>>     next_q=torch.randn(3, action_dim),
>>>     act=torch.randint(0, action_dim, (3,)),
>>>     next_act=torch.randint(0, action_dim, (3,)),
>>>     reward=torch.randn(3),
>>>     done=torch.randint(0, 2, (3,)).bool(),
>>>     weight=torch.ones(3),
>>> )
>>> loss = q_1step_td_error(data, 0.99)

m_q_1step_td_data

class ding.rl_utils.td.m_q_1step_td_data(q, target_q, next_q, act, reward, done, weight)

m_q_1step_td_error

ding.rl_utils.td.m_q_1step_td_error(data: ~collections.namedtuple, gamma: float, tau: float, alpha: float, criterion: <module 'torch.nn.modules' from '/home/docs/checkouts/readthedocs.org/user_builds/di-engine-docs/envs/latest/lib/python3.9/site-packages/torch/nn/modules/__init__.py'> = MSELoss()) → Tensor

Overview:

DQN算法的Munchausen td_error，支持1步td误差。

Arguments:

• 数据 (m_q_1step_td_data): 输入数据，m_q_1step_td_data 用于计算损失

• gamma (float): 折扣因子

• tau (float): Munchausen DQN的熵因子

• alpha (float): Munchausen项的折扣因子

• criterion (torch.nn.modules): 损失函数准则

Returns:

• 损失 (torch.Tensor): 1步时间差分误差，0维张量

Shapes:

• 数据 (m_q_1step_td_data): 包含 ['q', 'target_q', 'next_q', 'act', 'reward', 'done', 'weight'] 的 m_q_1step_td_data

• q (torch.FloatTensor): \((B, N)\) 即 [批量大小, 动作维度]

• target_q (torch.FloatTensor): \((B, N)\) 即 [batch_size, action_dim]

• next_q (torch.FloatTensor): \((B, N)\) 即 [batch_size, action_dim]

• act (torch.LongTensor): \((B, )\)

• 奖励 (torch.FloatTensor): \(( , B)\)

• 完成 (torch.BoolTensor) \((B, )\), 是否在最后一步完成

• 权重 (torch.FloatTensor 或 None): \((B, )\), 训练样本的权重

Examples:

>>> action_dim = 4
>>> data = m_q_1step_td_data(
>>>     q=torch.randn(3, action_dim),
>>>     target_q=torch.randn(3, action_dim),
>>>     next_q=torch.randn(3, action_dim),
>>>     act=torch.randint(0, action_dim, (3,)),
>>>     reward=torch.randn(3),
>>>     done=torch.randint(0, 2, (3,)),
>>>     weight=torch.ones(3),
>>> )
>>> loss = m_q_1step_td_error(data, 0.99, 0.01, 0.01)

q_v_1step_td_data

class ding.rl_utils.td.q_v_1step_td_data(q, v, act, reward, done, weight)

q_v_1step_td_error

ding.rl_utils.td.q_v_1step_td_error(data: ~collections.namedtuple, gamma: float, criterion: <module 'torch.nn.modules' from '/home/docs/checkouts/readthedocs.org/user_builds/di-engine-docs/envs/latest/lib/python3.9/site-packages/torch/nn/modules/__init__.py'> = MSELoss()) → Tensor

Overview:

SAC算法中q和v值之间的td_error，支持1步td误差。

Arguments:

• 数据 (q_v_1step_td_data): 输入数据，q_v_1step_td_data 用于计算损失

• gamma (float): 折扣因子

• criterion (torch.nn.modules): 损失函数准则

Returns:

• 损失 (torch.Tensor): 1步时间差分误差，0维张量

Shapes:

• 数据 (q_v_1step_td_data): 包含 ['q', 'v', 'act', 'reward', 'done', 'weight'] 的 q_v_1step_td_data

• q (torch.FloatTensor): \((B, N)\) 即 [批量大小, 动作维度]

• v (torch.FloatTensor): \((B, )\)

• act (torch.LongTensor): \((B, )\)

• 奖励 (torch.FloatTensor): \(( , B)\)

• 完成 (torch.BoolTensor) \((B, )\), 是否在最后一步完成

• 权重 (torch.FloatTensor 或 None): \((B, )\), 训练样本的权重

Examples:

>>> action_dim = 4
>>> data = q_v_1step_td_data(
>>>     q=torch.randn(3, action_dim),
>>>     v=torch.randn(3),
>>>     act=torch.randint(0, action_dim, (3,)),
>>>     reward=torch.randn(3),
>>>     done=torch.randint(0, 2, (3,)),
>>>     weight=torch.ones(3),
>>> )
>>> loss = q_v_1step_td_error(data, 0.99)

nstep_return_data

class ding.rl_utils.td.nstep_return_data(reward, next_value, done)

nstep_return

ding.rl_utils.td.nstep_return(data: namedtuple, gamma: float | list, nstep: int, value_gamma: Tensor | None = None)

Overview:

计算DQN算法的n步回报，支持单智能体情况和多智能体情况。

Arguments:

• 数据 (nstep_return_data): 输入数据，nstep_return_data 用于计算损失

• gamma (float): 折扣因子

• nstep (int): nstep 数量

• value_gamma (torch.Tensor): 价值的折扣因子

Returns:

• 返回 (torch.Tensor): n步返回

Shapes:

• 数据 (nstep_return_data): 包含 ['reward', 'next_value', 'done'] 的 nstep_return_data

• 奖励 (torch.FloatTensor): \((T, B)\), 其中 T 是时间步长（nstep）

• next_value (torch.FloatTensor): \((, B)\)

• 完成 (torch.BoolTensor) \((B, )\), 是否在最后一步完成

Examples:

>>> data = nstep_return_data(
>>>     reward=torch.randn(3, 3),
>>>     next_value=torch.randn(3),
>>>     done=torch.randint(0, 2, (3,)),
>>> )
>>> loss = nstep_return(data, 0.99, 3)

dist_1step_td_data

class ding.rl_utils.td.dist_1step_td_data(dist, next_dist, act, next_act, reward, done, weight)

dist_1step_td_error

ding.rl_utils.td.dist_1step_td_error(data: namedtuple, gamma: float, v_min: float, v_max: float, n_atom: int) → Tensor

Overview:

1步td_error用于基于分布式q学习的算法

Arguments:

• 数据 (dist_1step_td_data): 输入数据，dist_nstep_td_data 用于计算损失

• gamma (float): 折扣因子

• v_min (float): 支持的最小值

• v_max (float): 支持的最大值

• n_atom (int): 原子数量

Returns:

• 损失 (torch.Tensor): n步时间差分误差，0维张量

Shapes:

• 数据 (dist_1step_td_data): 包含 ['dist', 'next_n_dist', 'act', 'reward', 'done', 'weight'] 的 dist_1step_td_data

• dist (torch.FloatTensor): \((B, N, n_atom)\) 即 [batch_size, action_dim, n_atom]

• next_dist (torch.FloatTensor): \((B, N, n_atom)\)

• act (torch.LongTensor): \((B, )\)

• next_act (torch.LongTensor): \((B, )\)

• 奖励 (torch.FloatTensor): \((, B)\)

• 完成 (torch.BoolTensor) \((B, )\), 是否在最后一步完成

• 权重 (torch.FloatTensor 或 None): \((B, )\), 训练样本的权重

Examples:

>>> dist = torch.randn(4, 3, 51).abs().requires_grad_(True)
>>> next_dist = torch.randn(4, 3, 51).abs()
>>> act = torch.randint(0, 3, (4,))
>>> next_act = torch.randint(0, 3, (4,))
>>> reward = torch.randn(4)
>>> done = torch.randint(0, 2, (4,))
>>> data = dist_1step_td_data(dist, next_dist, act, next_act, reward, done, None)
>>> loss = dist_1step_td_error(data, 0.99, -10.0, 10.0, 51)

dist_nstep_td_data

ding.rl_utils.td.dist_nstep_td_data

dist_1step_td_data 的别名

shape_fn_dntd

ding.rl_utils.td.shape_fn_dntd(args, kwargs)

Overview:

返回用于HPC的dntd形状

Returns:

形状: [T, B, N, n_atom]

dist_nstep_td_error

ding.rl_utils.td.dist_nstep_td_error(data: namedtuple, gamma: float, v_min: float, v_max: float, n_atom: int, nstep: int = 1, value_gamma: Tensor | None = None) → Tensor

Overview:

多步（1步或n步）td_error用于基于分布式q学习的算法，支持单代理情况和多代理情况。

Arguments:

• 数据 (dist_nstep_td_data): 输入数据，dist_nstep_td_data 用于计算损失

• gamma (float): 折扣因子

• nstep (int): nstep 数量，默认设置为 1

Returns:

• 损失 (torch.Tensor): n步时间差分误差，0维张量

Shapes:

• 数据 (dist_nstep_td_data): 包含 ['dist', 'next_n_dist', 'act', 'reward', 'done', 'weight'] 的 dist_nstep_td_data

• dist (torch.FloatTensor): \((B, N, n_atom)\) 即 [batch_size, action_dim, n_atom]

• next_n_dist (torch.FloatTensor): \((B, N, n_atom)\)

• act (torch.LongTensor): \((B, )\)

• next_n_act (torch.LongTensor): \((B, )\)

• 奖励 (torch.FloatTensor): \((T, B)\), 其中 T 是时间步长（nstep）

• 完成 (torch.BoolTensor) \((B, )\), 是否在最后一步完成

Examples:

>>> dist = torch.randn(4, 3, 51).abs().requires_grad_(True)
>>> next_n_dist = torch.randn(4, 3, 51).abs()
>>> done = torch.randn(4)
>>> action = torch.randint(0, 3, size=(4, ))
>>> next_action = torch.randint(0, 3, size=(4, ))
>>> reward = torch.randn(5, 4)
>>> data = dist_nstep_td_data(dist, next_n_dist, action, next_action, reward, done, None)
>>> loss, _ = dist_nstep_td_error(data, 0.95, -10.0, 10.0, 51, 5)

v_1step_td_data

class ding.rl_utils.td.v_1step_td_data(v, next_v, reward, done, weight)

v_1step_td_error

ding.rl_utils.td.v_1step_td_error(data: ~collections.namedtuple, gamma: float, criterion: <module 'torch.nn.modules' from '/home/docs/checkouts/readthedocs.org/user_builds/di-engine-docs/envs/latest/lib/python3.9/site-packages/torch/nn/modules/__init__.py'> = MSELoss()) → Tensor

Overview:

1步td_error用于分布式基于值的算法

Arguments:

• 数据 (v_1step_td_data): 输入数据，v_1step_td_data 用于计算损失

• gamma (float): 折扣因子

• criterion (torch.nn.modules): 损失函数准则

Returns:

• 损失 (torch.Tensor): 1步时间差分误差，0维张量

Shapes:

• 数据 (v_1step_td_data): 包含 ['v', 'next_v', 'reward', 'done', 'weight'] 的 v_1step_td_data

• v (torch.FloatTensor): \((B, )\) 即 [batch_size, ]

• next_v (torch.FloatTensor): \((B, )\)

• 奖励 (torch.FloatTensor): \((, B)\)

• 完成 (torch.BoolTensor) \((B, )\), 是否在最后一步完成

• 权重 (torch.FloatTensor 或 None): \((B, )\), 训练样本的权重

Examples:

>>> v = torch.randn(5).requires_grad_(True)
>>> next_v = torch.randn(5)
>>> reward = torch.rand(5)
>>> done = torch.zeros(5)
>>> data = v_1step_td_data(v, next_v, reward, done, None)
>>> loss, td_error_per_sample = v_1step_td_error(data, 0.99)

v_nstep_td_data

class ding.rl_utils.td.v_nstep_td_data(v, next_n_v, reward, done, weight, value_gamma)

v_nstep_td_error

ding.rl_utils.td.v_nstep_td_error(data: ~collections.namedtuple, gamma: float, nstep: int = 1, criterion: <module 'torch.nn.modules' from '/home/docs/checkouts/readthedocs.org/user_builds/di-engine-docs/envs/latest/lib/python3.9/site-packages/torch/nn/modules/__init__.py'> = MSELoss()) → Tensor

Overview:

分布式基于值算法的多步（n步）td_error

Arguments:

• 数据 (dist_nstep_td_data): 输入数据，v_nstep_td_data 用于计算损失

• gamma (float): 折扣因子

• nstep (int): nstep 数量，默认设置为 1

Returns:

• 损失 (torch.Tensor): n步时间差分误差，0维张量

Shapes:

• 数据 (dist_nstep_td_data): 包含 ['v', 'next_n_v', 'reward', 'done', 'weight', 'value_gamma'] 的 v_nstep_td_data

• v (torch.FloatTensor): \((B, )\) 即 [batch_size, ]

• next_v (torch.FloatTensor): \((B, )\)

• 奖励 (torch.FloatTensor): \((T, B)\), 其中 T 是时间步长（nstep）

• 完成 (torch.BoolTensor) \((B, )\), 是否在最后一步完成

• 权重 (torch.FloatTensor 或 None): \((B, )\), 训练样本的权重

• value_gamma (torch.Tensor): 如果缓冲区中剩余的数据少于n_step，我们使用value_gamma作为next_v的gamma折扣值，而不是gamma**n_step

Examples:

>>> v = torch.randn(5).requires_grad_(True)
>>> next_v = torch.randn(5)
>>> reward = torch.rand(5, 5)
>>> done = torch.zeros(5)
>>> data = v_nstep_td_data(v, next_v, reward, done, 0.9, 0.99)
>>> loss, td_error_per_sample = v_nstep_td_error(data, 0.99, 5)

q_nstep_td_data

class ding.rl_utils.td.q_nstep_td_data(q, next_n_q, action, next_n_action, reward, done, weight)

dqfd_nstep_td_data

class ding.rl_utils.td.dqfd_nstep_td_data(q, next_n_q, action, next_n_action, reward, done, done_one_step, weight, new_n_q_one_step, next_n_action_one_step, is_expert)

shape_fn_qntd

ding.rl_utils.td.shape_fn_qntd(args, kwargs)

Overview:

返回hpc的qntd形状

Returns:

形状: [T, B, N]

q_nstep_td_error

ding.rl_utils.td.q_nstep_td_error(data: ~collections.namedtuple, gamma: float | list, nstep: int = 1, cum_reward: bool = False, value_gamma: ~torch.Tensor | None = None, criterion: <module 'torch.nn.modules' from '/home/docs/checkouts/readthedocs.org/user_builds/di-engine-docs/envs/latest/lib/python3.9/site-packages/torch/nn/modules/__init__.py'> = MSELoss()) → Tensor

Overview:

基于q-learning算法的多步（1步或n步）td_error

Arguments:

• 数据 (q_nstep_td_data): 输入数据，q_nstep_td_data 用于计算损失

• gamma (float): 折扣因子

• cum_reward (bool): 是否使用累积的n步奖励，这是在收集数据时计算出来的

• value_gamma (torch.Tensor): 目标q值的Gamma折扣值

• criterion (torch.nn.modules): 损失函数准则

• nstep (int): nstep 数量，默认设置为 1

Returns:

• 损失 (torch.Tensor): n步时间差分误差，0维张量

• td_error_per_sample (torch.Tensor): n步时间差分误差，一维张量

Shapes:

• 数据 (q_nstep_td_data): 包含 ['q', 'next_n_q', 'action', 'reward', 'done'] 的 q_nstep_td_data

• q (torch.FloatTensor): \((B, N)\) 即 [批量大小, 动作维度]

• next_n_q (torch.FloatTensor): \((B, N)\)

• 动作 (torch.LongTensor): \((B, )\)

• next_n_action (torch.LongTensor): \((B, )\)

• 奖励 (torch.FloatTensor): \((T, B)\), 其中 T 是时间步长（nstep）

• 完成 (torch.BoolTensor) \((B, )\), 是否在最后一步完成

• td_error_per_sample (torch.FloatTensor): \((B, )\)

Examples:

>>> next_q = torch.randn(4, 3)
>>> done = torch.randn(4)
>>> action = torch.randint(0, 3, size=(4, ))
>>> next_action = torch.randint(0, 3, size=(4, ))
>>> nstep =3
>>> q = torch.randn(4, 3).requires_grad_(True)
>>> reward = torch.rand(nstep, 4)
>>> data = q_nstep_td_data(q, next_q, action, next_action, reward, done, None)
>>> loss, td_error_per_sample = q_nstep_td_error(data, 0.95, nstep=nstep)

bdq_nstep_td_error

ding.rl_utils.td.bdq_nstep_td_error(data: ~collections.namedtuple, gamma: float | list, nstep: int = 1, cum_reward: bool = False, value_gamma: ~torch.Tensor | None = None, criterion: <module 'torch.nn.modules' from '/home/docs/checkouts/readthedocs.org/user_builds/di-engine-docs/envs/latest/lib/python3.9/site-packages/torch/nn/modules/__init__.py'> = MSELoss()) → Tensor

Overview:

BDQ算法的多步（1步或n步）td_error，参考论文“深度强化学习的动作分支架构”，链接：https://arxiv.org/pdf/1711.08946。 实际上，原始论文仅提供了1步TD-error的计算方法，这里我们扩展了n步的计算方法，即TD-error：

Arguments:

• 数据 (q_nstep_td_data): 输入数据，q_nstep_td_data 用于计算损失

• gamma (float): 折扣因子

• cum_reward (bool): 是否使用累积的n步奖励，这是在收集数据时计算出来的

• value_gamma (torch.Tensor): 目标q值的Gamma折扣值

• criterion (torch.nn.modules): 损失函数准则

• nstep (int): nstep 数量，默认设置为 1

Returns:

• 损失 (torch.Tensor): n步时间差分误差，0维张量

• td_error_per_sample (torch.Tensor): n步时间差分误差，一维张量

Shapes:

• 数据 (q_nstep_td_data): 包含 ['q', 'next_n_q', 'action', 'reward', 'done'] 的 q_nstep_td_data

• q (torch.FloatTensor): \((B, D, N)\) 即 [batch_size, branch_num, action_bins_per_branch]

• next_n_q (torch.FloatTensor): \((B, D, N)\)

• 动作 (torch.LongTensor): \((B, D)\)

• next_n_action (torch.LongTensor): \((B, D)\)

• 奖励 (torch.FloatTensor): \((T, B)\), 其中 T 是时间步长（nstep）

• 完成 (torch.BoolTensor) \((B, )\), 是否在最后一步完成

• td_error_per_sample (torch.FloatTensor): \((B, )\)

Examples:

>>> action_per_branch = 3
>>> next_q = torch.randn(8, 6, action_per_branch)
>>> done = torch.randn(8)
>>> action = torch.randint(0, action_per_branch, size=(8, 6))
>>> next_action = torch.randint(0, action_per_branch, size=(8, 6))
>>> nstep =3
>>> q = torch.randn(8, 6, action_per_branch).requires_grad_(True)
>>> reward = torch.rand(nstep, 8)
>>> data = q_nstep_td_data(q, next_q, action, next_action, reward, done, None)
>>> loss, td_error_per_sample = bdq_nstep_td_error(data, 0.95, nstep=nstep)

shape_fn_qntd_rescale

ding.rl_utils.td.shape_fn_qntd_rescale(args, kwargs)

Overview:

返回hpc的qntd_rescale形状

Returns:

形状: [T, B, N]

q_nstep_td_error_with_rescale

ding.rl_utils.td.q_nstep_td_error_with_rescale(data: ~collections.namedtuple, gamma: float | list, nstep: int = 1, value_gamma: ~torch.Tensor | None = None, criterion: <module 'torch.nn.modules' from '/home/docs/checkouts/readthedocs.org/user_builds/di-engine-docs/envs/latest/lib/python3.9/site-packages/torch/nn/modules/__init__.py'> = MSELoss(), trans_fn: ~typing.Callable = <function value_transform>, inv_trans_fn: ~typing.Callable = <function value_inv_transform>) → Tensor

Overview:

多步（1步或n步）td_error，带有值重新缩放

Arguments:

• 数据 (q_nstep_td_data): 输入数据，q_nstep_td_data 用于计算损失

• gamma (float): 折扣因子

• nstep (int): nstep 数量，默认设置为 1

• criterion (torch.nn.modules): 损失函数准则

• trans_fn (Callable): 值转换函数，默认为value_transform (参考rl_utils/value_rescale.py)

• inv_trans_fn (Callable): 值逆变换函数，默认为 value_inv_transform (参考 rl_utils/value_rescale.py)

Returns:

• 损失 (torch.Tensor): n步时间差分误差，0维张量

Shapes:

• 数据 (q_nstep_td_data): 包含 ['q', 'next_n_q', 'action', 'reward', 'done'] 的 q_nstep_td_data

• q (torch.FloatTensor): \((B, N)\) 即 [批量大小, 动作维度]

• next_n_q (torch.FloatTensor): \((B, N)\)

• 动作 (torch.LongTensor): \((B, )\)

• next_n_action (torch.LongTensor): \((B, )\)

• 奖励 (torch.FloatTensor): \((T, B)\), 其中 T 是时间步长（nstep）

• 完成 (torch.BoolTensor) \((B, )\), 是否在最后一步完成

Examples:

>>> next_q = torch.randn(4, 3)
>>> done = torch.randn(4)
>>> action = torch.randint(0, 3, size=(4, ))
>>> next_action = torch.randint(0, 3, size=(4, ))
>>> nstep =3
>>> q = torch.randn(4, 3).requires_grad_(True)
>>> reward = torch.rand(nstep, 4)
>>> data = q_nstep_td_data(q, next_q, action, next_action, reward, done, None)
>>> loss, _ = q_nstep_td_error_with_rescale(data, 0.95, nstep=nstep)

dqfd_nstep_td_error

ding.rl_utils.td.dqfd_nstep_td_error(data: ~collections.namedtuple, gamma: float, lambda_n_step_td: float, lambda_supervised_loss: float, margin_function: float, lambda_one_step_td: float = 1.0, nstep: int = 1, cum_reward: bool = False, value_gamma: ~torch.Tensor | None = None, criterion: <module 'torch.nn.modules' from '/home/docs/checkouts/readthedocs.org/user_builds/di-engine-docs/envs/latest/lib/python3.9/site-packages/torch/nn/modules/__init__.py'> = MSELoss()) → Tensor

Overview:

多步n步td_error + 1步td_error + 监督边缘损失或dqfd

Arguments:

• 数据 (dqfd_nstep_td_data): 输入数据，dqfd_nstep_td_data 用于计算损失

• gamma (float): 折扣因子

• cum_reward (bool): 是否使用累积的n步奖励，这是在收集数据时计算出来的

• value_gamma (torch.Tensor): 目标q值的Gamma折扣值

• criterion (torch.nn.modules): 损失函数准则

• nstep (int): nstep 数量，默认设置为 10

Returns:

• 损失 (torch.Tensor): 多步n步td_error + 1步td_error + 监督边缘损失，0维张量

• td_error_per_sample (torch.Tensor): 多步n步td_error + 1步td_error + 监督边缘损失，1维张量

Shapes:

• 数据 (q_nstep_td_data): 包含以下内容的 q_nstep_td_data [‘q’, ‘next_n_q’, ‘action’, ‘next_n_action’, ‘reward’, ‘done’, ‘weight’ , ‘new_n_q_one_step’, ‘next_n_action_one_step’, ‘is_expert’]

• q (torch.FloatTensor): \((B, N)\) 即 [批量大小, 动作维度]

• next_n_q (torch.FloatTensor): \((B, N)\)

• 动作 (torch.LongTensor): \((B, )\)

• next_n_action (torch.LongTensor): \((B, )\)

• 奖励 (torch.FloatTensor): \((T, B)\), 其中 T 是时间步长（nstep）

• 完成 (torch.BoolTensor) \((B, )\), 是否在最后一步完成

• td_error_per_sample (torch.FloatTensor): \((B, )\)

• new_n_q_one_step (torch.FloatTensor): \((B, N)\)

• next_n_action_one_step (torch.LongTensor): \((B, )\)

• is_expert (int) : 0 或 1

Examples:

>>> next_q = torch.randn(4, 3)
>>> done = torch.randn(4)
>>> done_1 = torch.randn(4)
>>> next_q_one_step = torch.randn(4, 3)
>>> action = torch.randint(0, 3, size=(4, ))
>>> next_action = torch.randint(0, 3, size=(4, ))
>>> next_action_one_step = torch.randint(0, 3, size=(4, ))
>>> is_expert = torch.ones((4))
>>> nstep = 3
>>> q = torch.randn(4, 3).requires_grad_(True)
>>> reward = torch.rand(nstep, 4)
>>> data = dqfd_nstep_td_data(
>>>     q, next_q, action, next_action, reward, done, done_1, None,
>>>     next_q_one_step, next_action_one_step, is_expert
>>> )
>>> loss, td_error_per_sample, loss_statistics = dqfd_nstep_td_error(
>>>     data, 0.95, lambda_n_step_td=1, lambda_supervised_loss=1,
>>>     margin_function=0.8, nstep=nstep
>>> )

dqfd_nstep_td_error_with_rescale

ding.rl_utils.td.dqfd_nstep_td_error_with_rescale(data: ~collections.namedtuple, gamma: float, lambda_n_step_td: float, lambda_supervised_loss: float, lambda_one_step_td: float, margin_function: float, nstep: int = 1, cum_reward: bool = False, value_gamma: ~torch.Tensor | None = None, criterion: <module 'torch.nn.modules' from '/home/docs/checkouts/readthedocs.org/user_builds/di-engine-docs/envs/latest/lib/python3.9/site-packages/torch/nn/modules/__init__.py'> = MSELoss(), trans_fn: ~typing.Callable = <function value_transform>, inv_trans_fn: ~typing.Callable = <function value_inv_transform>) → Tensor

Overview:

多步n步td_error + 1步td_error + 监督边缘损失或dqfd

Arguments:

• 数据 (dqfd_nstep_td_data): 输入数据，dqfd_nstep_td_data 用于计算损失

• gamma (float): 折扣因子

• cum_reward (bool): 是否使用累积的n步奖励，这是在收集数据时计算出来的

• value_gamma (torch.Tensor): 目标q值的Gamma折扣值

• criterion (torch.nn.modules): 损失函数准则

• nstep (int): nstep 数量，默认设置为 10

Returns:

• 损失 (torch.Tensor): 多步n步td_error + 1步td_error + 监督边缘损失，0维张量

• td_error_per_sample (torch.Tensor): 多步n步td_error + 1步td_error + 监督边缘损失，1维张量

Shapes:

• 数据 (q_nstep_td_data): 包含以下内容的 q_nstep_td_data [‘q’, ‘next_n_q’, ‘action’, ‘next_n_action’, ‘reward’, ‘done’, ‘weight’ , ‘new_n_q_one_step’, ‘next_n_action_one_step’, ‘is_expert’]

• q (torch.FloatTensor): \((B, N)\) 即 [批量大小, 动作维度]

• next_n_q (torch.FloatTensor): \((B, N)\)

• 动作 (torch.LongTensor): \((B, )\)

• next_n_action (torch.LongTensor): \((B, )\)

• 奖励 (torch.FloatTensor): \((T, B)\), 其中 T 是时间步长（nstep）

• 完成 (torch.BoolTensor) \((B, )\), 是否在最后一步完成

• td_error_per_sample (torch.FloatTensor): \((B, )\)

• new_n_q_one_step (torch.FloatTensor): \((B, N)\)

• next_n_action_one_step (torch.LongTensor): \((B, )\)

• is_expert (int) : 0 或 1

qrdqn_nstep_td_data

class ding.rl_utils.td.qrdqn_nstep_td_data(q, next_n_q, action, next_n_action, reward, done, tau, weight)

qrdqn_nstep_td_error

ding.rl_utils.td.qrdqn_nstep_td_error(data: namedtuple, gamma: float, nstep: int = 1, value_gamma: Tensor | None = None) → Tensor

Overview:

多步（1步或n步）在QRDQN中的td_error

Arguments:

• 数据 (qrdqn_nstep_td_data): 输入数据，qrdqn_nstep_td_data 用于计算损失

• gamma (float): 折扣因子

• nstep (int): nstep 数量，默认设置为 1

Returns:

• 损失 (torch.Tensor): n步时间差分误差，0维张量

Shapes:

• 数据 (q_nstep_td_data): 包含 ['q', 'next_n_q', 'action', 'reward', 'done'] 的 q_nstep_td_data

• q (torch.FloatTensor): \((tau, B, N)\) 即 [tau x batch_size, action_dim]

• next_n_q (torch.FloatTensor): \((tau', B, N)\)

• 动作 (torch.LongTensor): \((B, )\)

• next_n_action (torch.LongTensor): \((B, )\)

• 奖励 (torch.FloatTensor): \((T, B)\), 其中 T 是时间步长（nstep）

• 完成 (torch.BoolTensor) \((B, )\), 是否在最后一步完成

Examples:

>>> next_q = torch.randn(4, 3, 3)
>>> done = torch.randn(4)
>>> action = torch.randint(0, 3, size=(4, ))
>>> next_action = torch.randint(0, 3, size=(4, ))
>>> nstep = 3
>>> q = torch.randn(4, 3, 3).requires_grad_(True)
>>> reward = torch.rand(nstep, 4)
>>> data = qrdqn_nstep_td_data(q, next_q, action, next_action, reward, done, 3, None)
>>> loss, td_error_per_sample = qrdqn_nstep_td_error(data, 0.95, nstep=nstep)

q_nstep_sql_td_error

ding.rl_utils.td.q_nstep_sql_td_error(data: ~collections.namedtuple, gamma: float, alpha: float, nstep: int = 1, cum_reward: bool = False, value_gamma: ~torch.Tensor | None = None, criterion: <module 'torch.nn.modules' from '/home/docs/checkouts/readthedocs.org/user_builds/di-engine-docs/envs/latest/lib/python3.9/site-packages/torch/nn/modules/__init__.py'> = MSELoss()) → Tensor

Overview:

基于q-learning算法的多步（1步或n步）td_error

Arguments:

• 数据 (q_nstep_td_data): 输入数据，q_nstep_sql_td_data 用于计算损失

• gamma (float): 折扣因子

• Alpha (:obj:｀float`): 一个用于在策略方程中加权熵项的参数

• cum_reward (bool): 是否使用累积的n步奖励，这是在收集数据时计算出来的

• value_gamma (torch.Tensor): 目标soft_q_value的Gamma折扣值

• criterion (torch.nn.modules): 损失函数准则

• nstep (int): nstep 数量，默认设置为 1

Returns:

• 损失 (torch.Tensor): n步时间差分误差，0维张量

• td_error_per_sample (torch.Tensor): n步时间差分误差，一维张量

Shapes:

• 数据 (q_nstep_td_data): 包含 ['q', 'next_n_q', 'action', 'reward', 'done'] 的 q_nstep_td_data

• q (torch.FloatTensor): \((B, N)\) 即 [批量大小, 动作维度]

• next_n_q (torch.FloatTensor): \((B, N)\)

• 动作 (torch.LongTensor): \((B, )\)

• next_n_action (torch.LongTensor): \((B, )\)

• 奖励 (torch.FloatTensor): \((T, B)\), 其中 T 是时间步长（nstep）

• 完成 (torch.BoolTensor) \((B, )\), 是否在最后一步完成

• td_error_per_sample (torch.FloatTensor): \((B, )\)

Examples:

>>> next_q = torch.randn(4, 3)
>>> done = torch.randn(4)
>>> action = torch.randint(0, 3, size=(4, ))
>>> next_action = torch.randint(0, 3, size=(4, ))
>>> nstep = 3
>>> q = torch.randn(4, 3).requires_grad_(True)
>>> reward = torch.rand(nstep, 4)
>>> data = q_nstep_td_data(q, next_q, action, next_action, reward, done, None)
>>> loss, td_error_per_sample, record_target_v = q_nstep_sql_td_error(data, 0.95, 1.0, nstep=nstep)

iqn_nstep_td_data

class ding.rl_utils.td.iqn_nstep_td_data(q, next_n_q, action, next_n_action, reward, done, replay_quantiles, weight)

iqn_nstep_td_error

ding.rl_utils.td.iqn_nstep_td_error(data: namedtuple, gamma: float, nstep: int = 1, kappa: float = 1.0, value_gamma: Tensor | None = None) → Tensor

Overview:

多步（1步或n步）td_error在IQN中的应用，参考论文《隐式分位数网络用于分布强化学习》<https://arxiv.org/pdf/1806.06923.pdf>

Arguments:

• 数据 (iqn_nstep_td_data): 输入数据，iqn_nstep_td_data 用于计算损失

• gamma (float): 折扣因子

• nstep (int): nstep 数量，默认设置为 1

• criterion (torch.nn.modules): 损失函数准则

• beta_function (Callable): 风险函数

Returns:

• 损失 (torch.Tensor): n步时间差分误差，0维张量

Shapes:

• 数据 (q_nstep_td_data): 包含 ['q', 'next_n_q', 'action', 'reward', 'done'] 的 q_nstep_td_data

• q (torch.FloatTensor): \((tau, B, N)\) 即 [tau x batch_size, action_dim]

• next_n_q (torch.FloatTensor): \((tau', B, N)\)

• 动作 (torch.LongTensor): \((B, )\)

• next_n_action (torch.LongTensor): \((B, )\)

• 奖励 (torch.FloatTensor): \((T, B)\), 其中 T 是时间步长（nstep）

• 完成 (torch.BoolTensor) \((B, )\), 是否在最后一步完成

Examples:

>>> next_q = torch.randn(3, 4, 3)
>>> done = torch.randn(4)
>>> action = torch.randint(0, 3, size=(4, ))
>>> next_action = torch.randint(0, 3, size=(4, ))
>>> nstep = 3
>>> q = torch.randn(3, 4, 3).requires_grad_(True)
>>> replay_quantile = torch.randn([3, 4, 1])
>>> reward = torch.rand(nstep, 4)
>>> data = iqn_nstep_td_data(q, next_q, action, next_action, reward, done, replay_quantile, None)
>>> loss, td_error_per_sample = iqn_nstep_td_error(data, 0.95, nstep=nstep)

fqf_nstep_td_data

class ding.rl_utils.td.fqf_nstep_td_data(q, next_n_q, action, next_n_action, reward, done, quantiles_hats, weight)

fqf_nstep_td_error

ding.rl_utils.td.fqf_nstep_td_error(data: namedtuple, gamma: float, nstep: int = 1, kappa: float = 1.0, value_gamma: Tensor | None = None) → Tensor

Overview:

多步（1步或n步）td_error在FQF中的应用，参考论文《完全参数化的分位数函数用于分布强化学习》<https://arxiv.org/pdf/1911.02140.pdf>

Arguments:

• 数据 (fqf_nstep_td_data): 输入数据，fqf_nstep_td_data 用于计算损失

• gamma (float): 折扣因子

• nstep (int): nstep 数量，默认设置为 1

• criterion (torch.nn.modules): 损失函数准则

• beta_function (Callable): 风险函数

Returns:

• 损失 (torch.Tensor): n步时间差分误差，0维张量

Shapes:

• 数据 (q_nstep_td_data): 包含 ['q', 'next_n_q', 'action', 'reward', 'done'] 的 q_nstep_td_data

• q (torch.FloatTensor): \((B, tau, N)\) 即 [batch_size, tau, action_dim]

• next_n_q (torch.FloatTensor): \((B, tau', N)\)

• 动作 (torch.LongTensor): \((B, )\)

• next_n_action (torch.LongTensor): \((B, )\)

• 奖励 (torch.FloatTensor): \((T, B)\), 其中 T 是时间步长（nstep）

• 完成 (torch.BoolTensor) \((B, )\), 是否在最后一步完成

• quantiles_hats (torch.FloatTensor): \((B, tau)\)

Examples:

>>> next_q = torch.randn(4, 3, 3)
>>> done = torch.randn(4)
>>> action = torch.randint(0, 3, size=(4, ))
>>> next_action = torch.randint(0, 3, size=(4, ))
>>> nstep = 3
>>> q = torch.randn(4, 3, 3).requires_grad_(True)
>>> quantiles_hats = torch.randn([4, 3])
>>> reward = torch.rand(nstep, 4)
>>> data = fqf_nstep_td_data(q, next_q, action, next_action, reward, done, quantiles_hats, None)
>>> loss, td_error_per_sample = fqf_nstep_td_error(data, 0.95, nstep=nstep)

评估动作的分位数

ding.rl_utils.td.evaluate_quantile_at_action(q_s, actions)

fqf_calculate_fraction_loss

ding.rl_utils.td.fqf_calculate_fraction_loss(q_tau_i, q_value, quantiles, actions)

Overview:

计算FQF中的分数损失，参考论文《用于分布强化学习的全参数化分位数函数》<https://arxiv.org/pdf/1911.02140.pdf>

Arguments:

• q_tau_i (torch.FloatTensor): \((batch_size, num_quantiles-1, action_dim)\)

• q_value (torch.FloatTensor): \((batch_size, num_quantiles, action_dim)\)

• 分位数 (torch.FloatTensor): \((batch_size, num_quantiles+1)\)

• 动作 (torch.LongTensor): \((batch_size, )\)

Returns:

• fraction_loss (torch.Tensor): 分数损失，0维张量

td_lambda_data

class ding.rl_utils.td.td_lambda_data(value, reward, weight)

shape_fn_td_lambda

ding.rl_utils.td.shape_fn_td_lambda(args, kwargs)

Overview:

返回用于高性能计算的td_lambda形状

Returns:

形状: [T, B]

td_lambda_error

ding.rl_utils.td.td_lambda_error(data: namedtuple, gamma: float = 0.9, lambda_: float = 0.8) → Tensor

Overview:

计算给定常数gamma和lambda的TD(lambda)损失。 对于终端状态值没有特殊处理， 如果某个状态已达到终端，只需在终端之后的值和奖励中填入零 （包括终端状态，values[terminal]也应为0）

Arguments:

• 数据 (namedtuple): td_lambda 输入数据，包含字段 ['value', 'reward', 'weight']

• gamma (float): 常数折扣因子 gamma，应在 [0, 1] 范围内，默认为 0.9

• lambda (float): 常数 lambda，应在 [0, 1] 范围内，默认为 0.8

Returns:

• 损失 (torch.Tensor): 计算出的MSE损失，在批次上平均

Shapes:

• 值 (torch.FloatTensor): \((T+1, B)\), 其中 T 是轨迹长度，B 是批次大小，这是从步骤 0 到 T 的状态值的估计

• 奖励 (torch.FloatTensor): \((T, B)\), 从时间步0到T-1的回报

• 权重 (torch.FloatTensor 或 None): \((B, )\), 训练样本的权重

• 损失 (torch.FloatTensor): \(()\), 0维张量

Examples:

>>> T, B = 8, 4
>>> value = torch.randn(T + 1, B).requires_grad_(True)
>>> reward = torch.rand(T, B)
>>> loss = td_lambda_error(td_lambda_data(value, reward, None))

广义lambda返回

ding.rl_utils.td.generalized_lambda_returns(bootstrap_values: Tensor, rewards: Tensor, gammas: float, lambda_: float, done: Tensor | None = None) → Tensor

Overview:

功能等同于 trfl.value_ops.generalized_lambda_returns https://github.com/deepmind/trfl/blob/2c07ac22512a16715cc759f0072be43a5d12ae45/trfl/value_ops.py#L74 传入一个数字而不是张量，以使该值在批次中的所有样本中保持恒定

Arguments:

• bootstrap_values (torch.Tensor 或 float): 在步骤0到T的值的估计，大小为[T_traj+1, batchsize]

• 奖励 (torch.Tensor): 从0到T-1的回报，大小为[T_traj, batchsize]

• gammas (torch.Tensor 或 float): 每一步的折扣因子（从0到T-1），大小为[T_traj, batchsize]

• lambda (torch.Tensor 或 float): 确定在每个时间步长上自举与多步回报进一步累积的混合比例，大小为 [T_traj, batchsize]

• 完成 (torch.Tensor 或 float): 当前步骤（从0到T-1）是否完成，大小为 [T_traj, batchsize]

Returns:

• 返回 (torch.Tensor): 计算从状态0到T-1的每个状态的lambda返回值，大小为[T_traj, batchsize]

多步前向视图

ding.rl_utils.td.multistep_forward_view(bootstrap_values: Tensor, rewards: Tensor, gammas: float, lambda_: float, done: Tensor | None = None) → Tensor

Overview:

与 trfl.sequence_ops.multistep_forward_view 相同，它实现了 Sutton & Barto 中的 (12.18)。 假设输入张量的第一个维度对应于批次中的索引。

注意

result[T-1] = rewards[T-1] + gammas[T-1] * bootstrap_values[T] 对于 t 在 0…T-2 范围内： result[t] = rewards[t] + gammas[t]*(lambdas[t]*result[t+1] + (1-lambdas[t])*bootstrap_values[t+1])

Arguments:

• bootstrap_values (torch.Tensor): 估计在步骤1到T的值，大小为[T_traj, batchsize]

• 奖励 (torch.Tensor): 从0到T-1的回报，大小为[T_traj, batchsize]

• gammas (torch.Tensor): 每个步骤的折扣因子（从0到T-1），大小为[T_traj, batchsize]

• lambda (torch.Tensor): 确定在每个时间步长 [T_traj, batchsize] 上自举与多步回报进一步累积的混合比例，T-1 的元素被忽略并实际上设置为 0，因为没有关于未来奖励的信息。

• 完成 (torch.Tensor 或 float): 当前步骤（从0到T-1）是否完成，大小为 [T_traj, batchsize]

Returns:

• ret (torch.Tensor): 计算从0到T-1每个状态的lambda返回值，大小为[T_traj, batchsize]

upgo

请参考 ding/rl_utils/upgo 获取更多详细信息。

upgo_returns

ding.rl_utils.upgo.upgo_returns(rewards: Tensor, bootstrap_values: Tensor) → Tensor

Overview:

计算UPGO回报目标。同时注意，对于终端状态没有特殊处理。

Arguments:

• 奖励 (torch.Tensor): 从时间步0到T-1的回报，大小为[T_traj, batchsize]

• bootstrap_values (torch.Tensor): 估计从步骤0到T的状态值，大小为[T_traj+1, batchsize]

Returns:

• ret (torch.Tensor): 计算从0到T-1每个状态的lambda返回值，大小为[T_traj, batchsize]

Examples:

>>> T, B, N, N2 = 4, 8, 5, 7
>>> rewards = torch.randn(T, B)
>>> bootstrap_values = torch.randn(T + 1, B).requires_grad_(True)
>>> returns = upgo_returns(rewards, bootstrap_values)

upgo_loss

ding.rl_utils.upgo.upgo_loss(target_output: Tensor, rhos: Tensor, action: Tensor, rewards: Tensor, bootstrap_values: Tensor, mask=None) → Tensor

Overview:

计算给定常数gamma和lambda的UPGO损失。对于终端状态值没有特殊处理，如果轨迹中的最后一个状态是终端，只需传递0作为bootstrap_terminal_value。

Arguments:

• target_output (torch.Tensor): 目标策略网络计算的输出，大小为 [T_traj, batchsize, n_output]

• rhos (torch.Tensor): 重要性采样比率，大小为 [T_traj, batchsize]

• 动作 (torch.Tensor): 采取的动作，大小为 [T_traj, batchsize]

• 奖励 (torch.Tensor): 从时间步0到T-1的回报，大小为[T_traj, batchsize]

• bootstrap_values (torch.Tensor): 估计从步骤0到T的状态值，大小为[T_traj+1, batchsize]

Returns:

• 损失 (torch.Tensor): 计算的重要性采样UPGO损失，在样本上平均，大小为[]

Examples:

>>> T, B, N, N2 = 4, 8, 5, 7
>>> rhos = torch.randn(T, B)
>>> loss = upgo_loss(logit, rhos, action, rewards, bootstrap_values)

value_rescale

请参考 ding/rl_utils/value_rescale 获取更多详细信息。

value_transform

ding.rl_utils.value_rescale.value_transform(x: Tensor, eps: float = 0.01) → Tensor

Overview:

一个用于减少动作值函数规模的函数。 :math: h(x) = sign(x)(sqrt{(abs(x)+1)} - 1) + epsilon * x .

Arguments:

• x: (torch.Tensor) 需要标准化的输入张量。

• eps: (float) 加性正则化项的系数，以确保逆函数是Lipschitz连续的

Returns:

• (torch.Tensor) 归一化张量。

注意

观察并看得更远：在Atari上实现一致的性能 (https://arxiv.org/abs/1805.11593).

value_inv_transform

ding.rl_utils.value_rescale.value_inv_transform(x: Tensor, eps: float = 0.01) → Tensor

Overview:

值的逆缩放形式。 :math: `h^{-1}(x) = sign(x)({(

rac{sqrt{1+4epsilon(|x|+1+epsilon)}-1}{2epsilon})}^2-1)` .

Arguments:

• x: (torch.Tensor) 需要反归一化的输入张量。

• eps: (float) 加性正则化项的系数，以确保逆函数是Lipschitz连续的

Returns:

• (torch.Tensor) 未归一化的张量。

symlog

ding.rl_utils.value_rescale.symlog(x: Tensor) → Tensor

Overview:

一个用于标准化目标的函数。 :math: symlog(x) = sign(x)(ln{|x|+1}) 。

Arguments:

• x: (torch.Tensor) 需要标准化的输入张量。

Returns:

• (torch.Tensor) 归一化张量。

注意

通过世界模型掌握多样化领域 (https://arxiv.org/abs/2301.04104)

inv_symlog

ding.rl_utils.value_rescale.inv_symlog(x: Tensor) → Tensor

Overview:

symlog 的反函数形式。 :math: symexp(x) = sign(x)(exp{|x|}-1) 。

Arguments:

• x: (torch.Tensor) 需要反归一化的输入张量。

Returns:

• (torch.Tensor) 未归一化的张量。

vtrace

请参考 ding/rl_utils/vtrace 获取更多详细信息。

vtrace_nstep_return

ding.rl_utils.vtrace.vtrace_nstep_return(clipped_rhos, clipped_cs, reward, bootstrap_values, gamma=0.99, lambda_=0.95)

Overview:

计算vtrace返回值。

Returns:

• vtrace_return (torch.FloatTensor): vtrace损失项，所有这些都是可微的0维张量

Shapes:

• clipped_rhos (torch.FloatTensor): \((T, B)\), 其中 T 是时间步长，B 是批量大小

• clipped_cs (torch.FloatTensor): \((T, B)\)

• 奖励 (torch.FloatTensor): \((T, B)\)

• bootstrap_values (torch.FloatTensor): \((T+1, B)\)

• vtrace_return (torch.FloatTensor): \((T, B)\)

vtrace_advantage

ding.rl_utils.vtrace.vtrace_advantage(clipped_pg_rhos, reward, return_, bootstrap_values, gamma)

Overview:

计算vtrace优势。

Returns:

• vtrace_advantage (namedtuple): vtrace损失项，它们都是可微的0维张量

Shapes:

• clipped_pg_rhos (torch.FloatTensor): \((T, B)\), 其中 T 是时间步长，B 是批量大小

• 奖励 (torch.FloatTensor): \((T, B)\)

• 返回 (torch.FloatTensor): \((T, B)\)

• bootstrap_values (torch.FloatTensor): \((T, B)\)

• vtrace_advantage (torch.FloatTensor): \((T, B)\)

vtrace_data

class ding.rl_utils.vtrace.vtrace_data(target_output, behaviour_output, action, value, reward, weight)

vtrace_loss

class ding.rl_utils.vtrace.vtrace_loss(policy_loss, value_loss, entropy_loss)

vtrace_error_discrete_action

ding.rl_utils.vtrace.vtrace_error_discrete_action(data: namedtuple, gamma: float = 0.99, lambda_: float = 0.95, rho_clip_ratio: float = 1.0, c_clip_ratio: float = 1.0, rho_pg_clip_ratio: float = 1.0)

Overview:

实现vtrace（IMPALA：具有重要性加权演员-学习者架构的可扩展分布式深度强化学习），(arXiv:1802.01561)

Arguments:

• data (namedtuple): input data with fields shown in vtrace_data

  • target_output (torch.Tensor): 当前策略网络采取行动后的输出，通常此输出是网络的输出logit

  • 行为输出 (torch.Tensor): 由行为策略网络采取行动的输出，通常这个输出是网络的输出logit，用于生成轨迹（收集器）

  • 动作 (torch.Tensor): 在轨迹中选择的动作（离散动作空间的索引），即：行为动作

• gamma: (float): 未来的折扣因子，默认为0.95

• lambda: (float): 1步（lambda_=0）和n步之间的混合因子，默认为1.0

• rho_clip_ratio (float): 在计算基线目标 (vs) 时，重要性权重 (rho) 的裁剪阈值

• c_clip_ratio (float): 在计算基线目标 (vs) 时，重要性权重 (c) 的裁剪阈值

• rho_pg_clip_ratio (float): 在计算策略梯度优势时，重要性权重（rho）的裁剪阈值

Returns:

• trace_loss (namedtuple): vtrace损失项，所有这些都是可微的0维张量

Shapes:

• target_output (torch.FloatTensor): \((T, B, N)\), 其中 T 是时间步长，B 是批量大小，N 是动作维度

• 行为输出 (torch.FloatTensor): \((T, B, N)\)

• 动作 (torch.LongTensor): \((T, B)\)

• 值 (torch.FloatTensor): \((T+1, B)\)

• 奖励 (torch.LongTensor): \((T, B)\)

• 权重 (torch.LongTensor): \((T, B)\)

Examples:

>>> T, B, N = 4, 8, 16
>>> value = torch.randn(T + 1, B).requires_grad_(True)
>>> reward = torch.rand(T, B)
>>> target_output = torch.randn(T, B, N).requires_grad_(True)
>>> behaviour_output = torch.randn(T, B, N)
>>> action = torch.randint(0, N, size=(T, B))
>>> data = vtrace_data(target_output, behaviour_output, action, value, reward, None)
>>> loss = vtrace_error_discrete_action(data, rho_clip_ratio=1.1)

vtrace_error_continuous_action

ding.rl_utils.vtrace.vtrace_error_continuous_action(data: namedtuple, gamma: float = 0.99, lambda_: float = 0.95, rho_clip_ratio: float = 1.0, c_clip_ratio: float = 1.0, rho_pg_clip_ratio: float = 1.0)

Overview:

实现vtrace（IMPALA：具有重要性加权演员-学习者架构的可扩展分布式深度强化学习），(arXiv:1802.01561)

Arguments:

• data (namedtuple): input data with fields shown in vtrace_data

  • target_output (dict{key:torch.Tensor}): 当前策略网络采取行动后的输出，通常这个输出是网络的输出，它通过重参数化技巧表示分布。

  • 行为输出 (dict{key:torch.Tensor}): 通过行为策略网络采取行动的输出，通常此输出是网络输出的logit，它通过重参数化技巧表示分布。

  • 动作 (torch.Tensor): 在轨迹中选择的动作（离散动作空间的索引），即：行为动作

• gamma: (float): 未来的折扣因子，默认为0.95

• lambda: (float): 1步（lambda_=0）和n步之间的混合因子，默认为1.0

• rho_clip_ratio (float): 在计算基线目标 (vs) 时，重要性权重 (rho) 的裁剪阈值

• c_clip_ratio (float): 在计算基线目标 (vs) 时，重要性权重 (c) 的裁剪阈值

• rho_pg_clip_ratio (float): 在计算策略梯度优势时，重要性权重（rho）的裁剪阈值

Returns:

• trace_loss (namedtuple): vtrace损失项，所有这些都是可微的0维张量

Shapes:

• target_output (dict{key:torch.FloatTensor}): \((T, B, N)\), 其中 T 是时间步长，B 是批量大小，N 是动作维度。键通常是重参数化技巧的参数。

• 行为输出 (dict{key:torch.FloatTensor}): \((T, B, N)\)

• 动作 (torch.LongTensor): \((T, B)\)

• 值 (torch.FloatTensor): \((T+1, B)\)

• 奖励 (torch.LongTensor): \((T, B)\)

• 权重 (torch.LongTensor): \((T, B)\)

Examples:

>>> T, B, N = 4, 8, 16
>>> value = torch.randn(T + 1, B).requires_grad_(True)
>>> reward = torch.rand(T, B)
>>> target_output = dict(
>>>     'mu': torch.randn(T, B, N).requires_grad_(True),
>>>     'sigma': torch.exp(torch.randn(T, B, N).requires_grad_(True)),
>>> )
>>> behaviour_output = dict(
>>>     'mu': torch.randn(T, B, N),
>>>     'sigma': torch.exp(torch.randn(T, B, N)),
>>> )
>>> action = torch.randn((T, B, N))
>>> data = vtrace_data(target_output, behaviour_output, action, value, reward, None)
>>> loss = vtrace_error_continuous_action(data, rho_clip_ratio=1.1)