政策

AlphaZeroPolicy

class lzero.policy.alphazero.AlphaZeroPolicy(cfg: EasyDict, model: Module | None = None, enable_field: List[str] | None = None)

基类：Policy

概述：

AlphaZero 的策略类。

__init__(cfg: EasyDict, model: Module | None = None, enable_field: List[str] | None = None) → None

概述：

根据输入配置和模型初始化策略实例。此方法将初始化策略中的不同字段，包括 learn、collect、eval。learn 字段用于训练策略，collect 字段用于收集训练数据，eval 字段用于评估策略。enable_field 用于指定要初始化的字段，如果为 None，则将初始化所有字段。

参数:

• cfg (-) – 用于初始化策略的最终合并配置。对于默认配置，请参阅策略类的 config 属性及其注释。

• model (-) – 用于初始化策略的神经网络模型。如果它是 None，则模型将根据 default_model 方法和 cfg.model 字段创建。否则，模型将被设置为由外部调用者创建的 model 实例。

• enable_field (-) – 要初始化的字段列表。如果为 None，则所有字段都将被初始化。否则，只有 enable_field 中的字段将被初始化，这有助于节省资源。

备注

对于派生的策略类，它应该实现 _init_learn、_init_collect、_init_eval 方法来初始化相应的字段。

_abc_impl = <_abc._abc_data object>

_create_model(cfg: EasyDict, model: Module | None = None) → Module

概述：

根据输入配置和模型创建或验证神经网络模型。如果输入模型为 None，则将根据 default_model 方法和 cfg.model 字段创建模型。否则，模型将被验证为 torch.nn.Module 的实例，并设置为由外部调用者创建的 model 实例。

参数:

• cfg (-) – 用于初始化策略的最终合并配置。

• model (-) – 用于初始化策略的神经网络模型。用户可以参考相应策略中定义的默认模型来自定义自己的模型。

返回:

创建的神经网络模型。策略的不同模式将向模型添加不同的包装器和插件，这些包装器和插件用于训练、收集和评估。

返回类型:

• model (torch.nn.Module)

抛出:

- RuntimeError – 如果输入模型不是 None 且不是 torch.nn.Module 的实例。

_forward_collect(obs: Dict, temperature: float = 1) → Dict[str, Tensor]

概述：

在收集模式下收集数据的前向函数。使用真实环境执行MCTS搜索。

参数:

• obs (-) – obs 的字典，键是 env_id，值是该时间步对应的 obs。

• temperature (-) – MCTS搜索的温度。

返回:

输出字典，键是 env_id，值是该时间步对应的策略输出，包括动作、概率等。

返回类型:

• output (Dict[str, torch.Tensor])

_forward_eval(obs: Dict) → Dict[str, Tensor]

概述：

用于在评估模式下评估当前策略的前向函数，类似于 self._forward_collect。

参数:

obs (-) – obs 的字典，键是 env_id，值是该时间步对应的 obs。

返回:

输出字典，键是 env_id，值是该时间步对应的策略输出，包括动作、概率等。

返回类型:

• output (Dict[str, torch.Tensor])

_forward_learn(inputs: Dict[str, Tensor]) → Dict[str, float]

概述：

学习模式下的策略前向函数（训练策略和更新参数）。前向意味着策略从回放缓冲区输入一些训练批次数据，然后返回输出结果，包括各种训练信息，如损失值、策略熵、q值、优先级等。此方法留给子类实现，如果需要，可以在``data``项中添加更多参数。

参数:

data (-) – 用于策略前向传播的输入数据，包括一批训练样本。对于列表中的每个元素，字典的键是数据项的名称，值是对应的数据。通常，在 _forward_learn 方法中，数据应通过一些实用函数（如 default_preprocess_learn）在批次维度上堆叠。

返回:

策略前向传播的训练信息，包括一些用于监控训练的指标，如损失、优先级、Q值、策略熵，以及一些用于下一步训练的数据，如优先级。注意输出数据项应该是Python原生标量，而不是PyTorch张量，这样对外部使用更方便。

返回类型:

• output (Dict[int, Any])

_get_attribute(name: str) → Any

概述：

为了控制策略属性的访问，我们对外暴露不同的模式，而不是直接使用策略实例。我们还提供了一种方法来获取不同模式下的策略属性。

参数:

name (-) – 属性的名称。

返回:

属性的值。

返回类型:

• value (Any)

备注

DI-engine 的策略将首先尝试访问 _get_{name} 方法，然后尝试访问 _{name} 属性。如果两者都未找到，它将引发一个 NotImplementedError。

_get_batch_size() → int | Dict[str, int]

_get_n_episode() → int | None

_get_n_sample() → int | None

_get_simulation_env()

_get_train_sample(data)

概述：

对于给定的轨迹（过渡，过渡列表）数据，将其处理为可以直接用于训练的样本列表。一个训练样本可以是一个经过处理的过渡（DQN 使用 n步 TD）或者一些多时间步的过渡（DRQN）。这个方法通常在收集器中使用，以在训练前执行必要的强化学习数据预处理，这可以帮助学习者分摊相关的时间消耗。此外，你也可以将这个方法实现为一个恒等函数，并在 self._forward_learn 方法中进行数据处理。

参数:

transitions (-) – 轨迹数据（一个过渡列表），每个元素的格式与 self._process_transition 方法的返回值相同。

返回:

处理后的训练样本，每个元素的格式与输入转换相似，但可能包含更多用于训练的数据，如n步奖励、优势等。

返回类型:

• samples (List[Dict[str, Any]])

备注

我们将在下一个发布版本中向量化 process_transition 和 get_train_sample 方法。用户可以通过重写这两个方法和收集器本身来自定义此数据处理过程。

_init_collect() → None

概述：

收集模式初始化方法。由 self.__init__ 调用。初始化收集模型和MCTS工具。

_init_eval() → None

概述：

评估模式初始化方法。由 self.__init__ 调用。初始化评估模型和MCTS工具。

_init_learn() → None

概述：

初始化策略的学习模式，包括相关属性和模块。如果 learn 字段在 enable_field 中，此方法将在 __init__ 方法中被调用。几乎不同的策略都有其自己的学习模式，因此此方法必须在子类中被重写。

备注

对于需要保存和加载的成员变量，请参考 _state_dict_learn 和 _load_state_dict_learn 方法。

备注

对于需要监控的成员变量，请参考 _monitor_vars_learn 方法。

备注

如果你想在 _init_learn 方法中设置一些特殊的成员变量，最好用前缀 _learn_ 来命名它们，以避免与其他模式发生冲突，例如 self._learn_attr1。

_init_multi_gpu_setting(model: Module, bp_update_sync: bool) → None

概述：

初始化多GPU数据并行训练设置，包括在训练开始时广播模型参数，并为所有模型参数梯度准备hook函数以进行allreduce操作。

参数:

• model (-) – 要训练的神经网络模型。

• bp_update_sync (-) – 是否在所有reduce模型参数的梯度后同步更新模型参数。异步更新可以在不同的网络层中并行进行，如流水线，从而节省时间。

_load_state_dict_collect(state_dict: Dict[str, Any]) → None

概述：

将 state_dict 变量加载到策略收集模式中，例如加载预训练的 state_dict、自动恢复检查点，或在分布式训练场景中从学习者复制模型。

参数:

state_dict (-) – 之前保存的策略收集状态的字典。

小技巧

如果你想只加载模型的一部分，你可以简单地将 load_state_dict 中的 strict 参数设置为 False，或者参考 ding.torch_utils.checkpoint_helper 以进行更复杂的操作。

_load_state_dict_eval(state_dict: Dict[str, Any]) → None

概述：

将 state_dict 变量加载到策略评估模式中，例如加载自动恢复检查点，或在分布式训练场景中从学习者加载模型副本。

参数:

state_dict (-) – 保存之前的策略评估状态的字典。

小技巧

如果你想只加载模型的一部分，你可以简单地将 load_state_dict 中的 strict 参数设置为 False，或者参考 ding.torch_utils.checkpoint_helper 以进行更复杂的操作。

_load_state_dict_learn(state_dict: Dict[str, Any]) → None

概述：

将 state_dict 变量加载到策略学习模式中。

参数:

state_dict (-) – 之前保存的策略学习状态的字典。

小技巧

如果你想只加载模型的一部分，你可以简单地将 load_state_dict 中的 strict 参数设置为 False，或者参考 ding.torch_utils.checkpoint_helper 以进行更复杂的操作。

_monitor_vars_learn() → List[str]

概述：

在学习模式中注册要监控的变量。根据返回值 _forward_learn，注册的变量将被记录在 tensorboard 中。

_policy_value_fn(env: Env) → Tuple[Dict[int, ndarray], float]

_process_transition(obs: Dict, model_output: Dict[str, Tensor], timestep: namedtuple) → Dict

概述：

从策略学习中生成字典类型转换（一个时间步长）数据。

_reset_collect(data_id: List[int] | None = None) → None

概述：

在必要时重置收集模式下的一些有状态变量，例如RNN的隐藏状态或某些特殊算法的内存库。如果``data_id``为None，则意味着重置所有有状态变量。否则，它将根据``data_id``重置有状态变量。例如，在``data_id``中收集的不同环境/情节在RNN中将具有不同的隐藏状态。

参数:

data_id (-) – 数据的id，用于重置由 data_id 指定的状态变量。

备注

此方法不是必须实现的。子类可以根据需要重写此方法。

_reset_eval(data_id: List[int] | None = None) → None

概述：

在必要时重置一些有状态变量以适应评估模式，例如RNN的隐藏状态或某些特殊算法的内存库。如果 data_id 为None，则意味着重置所有有状态变量。否则，它将根据 data_id 重置有状态变量。例如，在 data_id 中的不同环境/情节在评估中将具有RNN中的不同隐藏状态。

参数:

data_id (-) – 数据的id，用于重置由 data_id 指定的状态变量。

备注

此方法不是必须实现的。子类可以根据需要重写此方法。

_reset_learn(data_id: List[int] | None = None) → None

概述：

在必要时重置学习模式下的一些有状态变量，例如RNN的隐藏状态或某些特殊算法的状态库。如果 data_id 为None，则意味着重置所有有状态变量。否则，它将根据 data_id 重置有状态变量。例如， data_id 中的不同轨迹在RNN中将具有不同的隐藏状态。

参数:

data_id (-) – 数据的id，用于重置由 data_id 指定的状态变量。

备注

此方法不是必须实现的。子类可以根据需要重写此方法。

_set_attribute(name: str, value: Any) → None

概述：

为了控制策略属性的访问，我们对外暴露不同的模式，而不是直接使用策略实例。我们还提供了一种方法，用于在不同模式下设置策略的属性。新属性将被命名为 _{name}。

参数:

• name (-) – 属性的名称。

• value (-) – 属性的值。

_state_dict_collect() → Dict[str, Any]

概述：

返回收集模式的 state_dict，仅包括通常情况下的模型，这对于分布式训练场景中自动恢复收集器是必要的。

返回:

当前策略收集状态的字典，用于保存和恢复。

返回类型:

• state_dict (Dict[str, Any])

小技巧

并非所有场景都需要自动恢复收集器，有时，我们可以直接关闭崩溃的收集器并重新启动一个新的。

_state_dict_eval() → Dict[str, Any]

概述：

返回评估模式下的 state_dict，通常只包括模型，这对于分布式训练场景是必要的，以便自动恢复评估器。

返回:

当前策略评估状态的字典，用于保存和恢复。

返回类型:

• state_dict (Dict[str, Any])

小技巧

并非所有场景都需要自动恢复评估器，有时，我们可以直接关闭崩溃的评估器并重新启动一个新的评估器。

_state_dict_learn() → Dict[str, Any]

概述：

返回学习模式的 state_dict，通常包括模型和优化器。

返回:

当前策略学习状态的字典，用于保存和恢复。

返回类型:

• state_dict (Dict[str, Any])

property cfg: EasyDict

class collect_function(forward, process_transition, get_train_sample, reset, get_attribute, set_attribute, state_dict, load_state_dict)

基类：tuple

_asdict()

返回一个新字典，该字典将字段名称映射到它们的值。

_field_defaults = {}

_fields = ('forward', 'process_transition', 'get_train_sample', 'reset', 'get_attribute', 'set_attribute', 'state_dict', 'load_state_dict')

classmethod _make(iterable)

从序列或可迭代对象创建一个新的 collect_function 对象

_replace(**kwds)

返回一个新的 collect_function 对象，用新值替换指定字段

count(value, /)

返回值出现的次数。

forward

字段编号 0 的别名

get_attribute

字段编号4的别名

get_train_sample

字段编号2的别名

index(value, start=0, stop=9223372036854775807, /)

返回值的第一个索引。

如果值不存在，则引发 ValueError。

load_state_dict

字段编号 7 的别名

process_transition

字段编号1的别名

reset

字段编号 3 的别名

set_attribute

字段编号 5 的别名

state_dict

字段编号 6 的别名

property collect_mode: collect_function

概述：

返回策略的收集模式的接口，该接口用于训练模型。这里我们使用 namedtuple 来定义不可变的接口，并限制策略在不同模式下的使用。此外，派生子类可以重写这些接口以自定义其自己的收集模式。

返回:

策略的收集模式的接口，它是一个 namedtuple，其不同字段的值是不同的内部方法。

返回类型:

• interfaces (Policy.collect_function)

示例

>>> policy = Policy(cfg, model)
>>> policy_collect = policy.collect_mode
>>> obs = env_manager.ready_obs
>>> inference_output = policy_collect.forward(obs)
>>> next_obs, rew, done, info = env_manager.step(inference_output.action)

config = {'batch_size': 256, 'collector_env_num': 8, 'cuda': False, 'evaluator_env_num': 3, 'fixed_temperature_value': 0.25, 'grad_clip_value': 10, 'gumbel_algo': False, 'learning_rate': 0.2, 'manual_temperature_decay': False, 'mcts': {'max_moves': 512, 'num_simulations': 50, 'pb_c_base': 19652, 'pb_c_init': 1.25, 'root_dirichlet_alpha': 0.3, 'root_noise_weight': 0.25}, 'model': {'num_channels': 32, 'num_res_blocks': 1, 'observation_shape': (3, 6, 6)}, 'momentum': 0.9, 'multi_gpu': False, 'optim_type': 'SGD', 'other': {'replay_buffer': {'replay_buffer_size': 1000000, 'save_episode': False}}, 'piecewise_decay_lr_scheduler': True, 'replay_ratio': 0.25, 'sampled_algo': False, 'tensor_float_32': False, 'threshold_training_steps_for_final_lr': 500000, 'threshold_training_steps_for_final_temperature': 100000, 'torch_compile': False, 'update_per_collect': None, 'value_weight': 1.0, 'weight_decay': 0.0001}

classmethod default_config() → EasyDict

概述：

获取策略的默认配置。此方法用于创建策略的默认配置。

返回:

对应策略的默认配置。对于派生策略类，它将递归合并基类的默认配置及其自身的默认配置。

返回类型:

• cfg (EasyDict)

小技巧

此方法将深度复制类的 config 属性并返回结果。因此，用户无需担心返回配置的修改。

default_model() → Tuple[str, List[str]]

概述：

返回此算法的默认模型设置以供演示。

返回:

The model type used in this algorithm, which is registered in ModelRegistry. - import_names (List[str]): The model class path list used in this algorithm.

返回类型:

• model_type (str)

class eval_function(forward, reset, get_attribute, set_attribute, state_dict, load_state_dict)

基类：tuple

_asdict()

返回一个新字典，该字典将字段名称映射到它们的值。

_field_defaults = {}

_fields = ('forward', 'reset', 'get_attribute', 'set_attribute', 'state_dict', 'load_state_dict')

classmethod _make(iterable)

从序列或可迭代对象创建一个新的 eval_function 对象

_replace(**kwds)

返回一个新的 eval_function 对象，用新值替换指定字段。

count(value, /)

返回值出现的次数。

forward

字段编号 0 的别名

get_attribute

字段编号2的别名

index(value, start=0, stop=9223372036854775807, /)

返回值的第一个索引。

如果值不存在，则引发 ValueError。

load_state_dict

字段编号 5 的别名

reset

字段编号1的别名

set_attribute

字段编号 3 的别名

state_dict

字段编号4的别名

property eval_mode: eval_function

概述：

返回策略的评估模式的接口，该接口用于训练模型。这里我们使用 namedtuple 来定义不可变的接口，并限制策略在不同模式下的使用。此外，派生子类可以重写这些接口以自定义其自己的评估模式。

返回:

策略的eval模式的接口，它是一个namedtuple，其不同字段的值是不同的内部方法。

返回类型:

• interfaces (Policy.eval_function)

示例

>>> policy = Policy(cfg, model)
>>> policy_eval = policy.eval_mode
>>> obs = env_manager.ready_obs
>>> inference_output = policy_eval.forward(obs)
>>> next_obs, rew, done, info = env_manager.step(inference_output.action)

class learn_function(forward, reset, info, monitor_vars, get_attribute, set_attribute, state_dict, load_state_dict)

基类：tuple

_asdict()

返回一个新字典，该字典将字段名称映射到它们的值。

_field_defaults = {}

_fields = ('forward', 'reset', 'info', 'monitor_vars', 'get_attribute', 'set_attribute', 'state_dict', 'load_state_dict')

classmethod _make(iterable)

从序列或可迭代对象创建一个新的 learn_function 对象

_replace(**kwds)

返回一个新的 learn_function 对象，用新值替换指定的字段

count(value, /)

返回值出现的次数。

forward

字段编号 0 的别名

get_attribute

字段编号4的别名

index(value, start=0, stop=9223372036854775807, /)

返回值的第一个索引。

如果值不存在，则引发 ValueError。

info

字段编号2的别名

load_state_dict

字段编号 7 的别名

monitor_vars

字段编号 3 的别名

reset

字段编号1的别名

set_attribute

字段编号 5 的别名

state_dict

字段编号 6 的别名

property learn_mode: learn_function

概述：

返回策略学习模式的接口，该接口用于训练模型。这里我们使用 namedtuple 来定义不可变的接口，并限制策略在不同模式下的使用。此外，派生子类可以重写这些接口以自定义其自己的学习模式。

返回:

策略的学习模式的接口，它是一个 namedtuple，其不同字段的值是不同的内部方法。

返回类型:

• interfaces (Policy.learn_function)

示例

>>> policy = Policy(cfg, model)
>>> policy_learn = policy.learn_mode
>>> train_output = policy_learn.forward(data)
>>> state_dict = policy_learn.state_dict()

sync_gradients(model: Module) → None

概述：

在数据并行的多GPU训练中，同步（allreduce）模型参数的梯度。

参数:

model (-) – 用于同步梯度的模型。

备注

此方法仅在多GPU训练中使用，并且应在 backward 方法之后和 step 方法之前调用。用户还可以使用 bp_update_sync 配置来控制是否同步梯度allreduce和优化器更新。

total_field = {'collect', 'eval', 'learn'}

MuZero策略

class lzero.policy.muzero.MuZeroPolicy(cfg: EasyDict, model: Module | None = None, enable_field: List[str] | None = None)

基类：Policy

概述：

如果 self._cfg.model.model_type 是 [‘conv’, ‘mlp’] 之一：

MuZero 的策略类。

如果 self._cfg.model.model_type 等于 [“conv_context”, “mlp_context”]：

MuZero w/ Context 的策略类，这是 MuZero 的一个变体。此变体保留了与 MuZero 相同的训练设置，但在推理过程中通过在根节点处采用 k 步递归预测的潜在表示而有所不同，该方法在 UniZero 论文 https://arxiv.org/abs/2406.10667 中提出。

__init__(cfg: EasyDict, model: Module | None = None, enable_field: List[str] | None = None) → None

概述：

根据输入配置和模型初始化策略实例。此方法将初始化策略中的不同字段，包括 learn、collect、eval。learn 字段用于训练策略，collect 字段用于收集训练数据，eval 字段用于评估策略。enable_field 用于指定要初始化的字段，如果为 None，则将初始化所有字段。

参数:

• cfg (-) – 用于初始化策略的最终合并配置。对于默认配置，请参阅策略类的 config 属性及其注释。

• model (-) – 用于初始化策略的神经网络模型。如果它是 None，则模型将根据 default_model 方法和 cfg.model 字段创建。否则，模型将被设置为由外部调用者创建的 model 实例。

• enable_field (-) – 要初始化的字段列表。如果为 None，则所有字段都将被初始化。否则，只有 enable_field 中的字段将被初始化，这有助于节省资源。

备注

对于派生的策略类，它应该实现 _init_learn、_init_collect、_init_eval 方法来初始化相应的字段。

_abc_impl = <_abc._abc_data object>

_create_model(cfg: EasyDict, model: Module | None = None) → Module

概述：

根据输入配置和模型创建或验证神经网络模型。如果输入模型为 None，则将根据 default_model 方法和 cfg.model 字段创建模型。否则，模型将被验证为 torch.nn.Module 的实例，并设置为由外部调用者创建的 model 实例。

参数:

• cfg (-) – 用于初始化策略的最终合并配置。

• model (-) – 用于初始化策略的神经网络模型。用户可以参考相应策略中定义的默认模型来自定义自己的模型。

返回:

创建的神经网络模型。策略的不同模式将向模型添加不同的包装器和插件，这些包装器和插件用于训练、收集和评估。

返回类型:

• model (torch.nn.Module)

抛出:

- RuntimeError – 如果输入模型不是 None 且不是 torch.nn.Module 的实例。

_forward_collect(data: Tensor, action_mask: list = None, temperature: float = 1, to_play: List = [-1], epsilon: float = 0.25, ready_env_id: array = None) → Dict

概述：

在收集模式下收集数据的正向函数。使用模型执行MCTS搜索。在收集模式期间通过采样选择动作。

参数:

• data (-) – 输入数据，即观察结果。

• action_mask (-) – 动作掩码，即不能被选择的动作。

• temperature (-) – 政策的温度。

• to_play (-) – 要播放的玩家。

• epsilon (-) – eps 贪婪探索的 epsilon 值。

• ready_env_id (-) – 准备好收集的环境的id。

形状：

• data (torch.Tensor):

  • 对于Atari，\((N, C*S, H, W)\)，其中N是collect_env的数量，C是通道的数量，S是堆叠帧的数量，H是图像的高度，W是图像的宽度。

  • 对于 lunarlander，\((N, O)\)，其中 N 是 collect_env 的数量，O 是观察空间的大小。

• action_mask: \((N, action_space_size)\)，其中 N 是 collect_env 的数量。

• 温度: \((1, )\).

• to_play: \((N, 1)\)，其中 N 是 collect_env 的数量。

• epsilon: \((1, )\).

• ready_env_id: None

返回:

字典类型数据，包含的键有 action、distributions、visit_count_distribution_entropy、value、pred_value、policy_logits。

返回类型:

• output (Dict[int, Any])

_forward_eval(data: Tensor, action_mask: list, to_play: int = -1, ready_env_id: array = None) → Dict

概述：

用于评估当前策略在评估模式下的前向函数。使用模型执行MCTS搜索。在评估模式下选择值最高的动作（argmax）而不是采样。

参数:

• data (-) – 输入数据，即观察结果。

• action_mask (-) – 动作掩码，即不能被选择的动作。

• to_play (-) – 要播放的玩家。

• ready_env_id (-) – 准备好收集的环境的id。

形状：

• data (torch.Tensor):

  • 对于Atari，\((N, C*S, H, W)\)，其中N是collect_env的数量，C是通道的数量，S是堆叠帧的数量，H是图像的高度，W是图像的宽度。

  • 对于 lunarlander，\((N, O)\)，其中 N 是 collect_env 的数量，O 是观察空间的大小。

• action_mask: \((N, action_space_size)\)，其中 N 是 collect_env 的数量。

• to_play: \((N, 1)\)，其中 N 是 collect_env 的数量。

• ready_env_id: None

返回:

字典类型数据，包含的键有 action、distributions、visit_count_distribution_entropy、value、pred_value、policy_logits。

返回类型:

• output (Dict[int, Any])

_forward_learn(data: Tuple[Tensor]) → Dict[str, float | int]

概述：

学习模式下用于学习策略的前向函数，这是学习过程的核心。数据从回放缓冲区中采样。通过损失函数计算损失，并将损失反向传播以更新模型。

参数:

data (-) – 从回放缓冲区中采样的数据，这是一个张量元组。第一个张量是当前批次，第二个张量是目标批次。

返回:

要记录的信息字典，其中包含当前的学习损失和学习统计数据。

返回类型:

• info_dict (Dict[str, Union[float, int]])

_get_attribute(name: str) → Any

概述：

为了控制策略属性的访问，我们对外暴露不同的模式，而不是直接使用策略实例。我们还提供了一种方法来获取不同模式下的策略属性。

参数:

name (-) – 属性的名称。

返回:

属性的值。

返回类型:

• value (Any)

备注

DI-engine 的策略将首先尝试访问 _get_{name} 方法，然后尝试访问 _{name} 属性。如果两者都未找到，它将引发一个 NotImplementedError。

_get_batch_size() → int | Dict[str, int]

_get_n_episode() → int | None

_get_n_sample() → int | None

_get_target_obs_index_in_step_k(step)

概述：

获取步骤 k 中目标 obs 的起始索引和结束索引。

参数:

step (-) – 当前步骤 k。

返回:

The begin index of the target obs in step k. - end_index (int): The end index of the target obs in step k.

返回类型:

• beg_index (int)

示例

>>> self._cfg.model.model_type = 'conv'
>>> self._cfg.model.image_channel = 3
>>> self._cfg.model.frame_stack_num = 4
>>> self._get_target_obs_index_in_step_k(0)
>>> (0, 12)

_get_train_sample(data)

概述：

对于给定的轨迹（过渡，过渡列表）数据，将其处理为可以直接用于训练的样本列表。一个训练样本可以是一个经过处理的过渡（DQN 使用 n步 TD）或者一些多时间步的过渡（DRQN）。这个方法通常在收集器中使用，以在训练前执行必要的强化学习数据预处理，这可以帮助学习者分摊相关的时间消耗。此外，你也可以将这个方法实现为一个恒等函数，并在 self._forward_learn 方法中进行数据处理。

参数:

transitions (-) – 轨迹数据（一个过渡列表），每个元素的格式与 self._process_transition 方法的返回值相同。

返回:

处理后的训练样本，每个元素的格式与输入转换相似，但可能包含更多用于训练的数据，如n步奖励、优势等。

返回类型:

• samples (List[Dict[str, Any]])

备注

我们将在下一个发布版本中向量化 process_transition 和 get_train_sample 方法。用户可以通过重写这两个方法和收集器本身来自定义此数据处理过程。

_init_collect() → None

概述：

收集模式初始化方法。由 self.__init__ 调用。初始化收集模型和MCTS工具。

_init_eval() → None

概述：

评估模式初始化方法。由 self.__init__ 调用。初始化评估模型和MCTS工具。

_init_learn() → None

概述：

学习模式初始化方法。由 self.__init__ 调用。初始化学习模型、优化器和MCTS工具。

_init_multi_gpu_setting(model: Module, bp_update_sync: bool) → None

概述：

初始化多GPU数据并行训练设置，包括在训练开始时广播模型参数，并为所有模型参数梯度准备hook函数以进行allreduce操作。

参数:

• model (-) – 要训练的神经网络模型。

• bp_update_sync (-) – 是否在所有reduce模型参数的梯度后同步更新模型参数。异步更新可以在不同的网络层中并行进行，如流水线，从而节省时间。

_load_state_dict_collect(state_dict: Dict[str, Any]) → None

概述：

将 state_dict 变量加载到策略收集模式中，例如加载预训练的 state_dict、自动恢复检查点，或在分布式训练场景中从学习者复制模型。

参数:

state_dict (-) – 之前保存的策略收集状态的字典。

小技巧

如果你想只加载模型的一部分，你可以简单地将 load_state_dict 中的 strict 参数设置为 False，或者参考 ding.torch_utils.checkpoint_helper 以进行更复杂的操作。

_load_state_dict_eval(state_dict: Dict[str, Any]) → None

概述：

将 state_dict 变量加载到策略评估模式中，例如加载自动恢复检查点，或在分布式训练场景中从学习者加载模型副本。

参数:

state_dict (-) – 保存之前的策略评估状态的字典。

小技巧

如果你想只加载模型的一部分，你可以简单地将 load_state_dict 中的 strict 参数设置为 False，或者参考 ding.torch_utils.checkpoint_helper 以进行更复杂的操作。

_load_state_dict_learn(state_dict: Dict[str, Any]) → None

概述：

将 state_dict 变量加载到策略学习模式中。

参数:

state_dict (-) – 之前保存的策略学习状态的字典。

_monitor_vars_learn() → List[str]

概述：

在学习模式中注册要监控的变量。根据返回值 _forward_learn，注册的变量将被记录在 tensorboard 中。

_process_transition(obs, policy_output, timestep)

概述：

将一个时间步的过渡数据处理并打包成一个字典，例如 <s, a, r, s’, done>。一些策略需要进行一些特殊处理并打包其自身的必要属性（例如隐藏状态和logit），因此此方法留给子类实现。

参数:

• obs (-) – 当前时间步的观察。

• policy_output (-) – 以观察为输入的策略网络的输出。通常，它包含动作和动作的对数概率。

• timestep (-) – 环境步骤方法返回的名为namedtuple的执行结果，除了所有元素都已转换为张量数据。通常，它包含下一个观察值、奖励、完成状态、信息等。

返回:

当前时间步的已处理过渡数据。

返回类型:

• transition (Dict[str, torch.Tensor])

_reset_collect(data_id: List[int] | None = None) → None

概述：

重置收集器环境的观察和动作。

参数:

data_id (-) – 要重置的数据ID列表（在此实现中未使用）。

_reset_eval(data_id: List[int] | None = None) → None

概述：

重置评估环境的观察和动作。

参数:

data_id (-) – 要重置的数据ID列表（在此实现中未使用）。

_reset_learn(data_id: List[int] | None = None) → None

概述：

在必要时重置学习模式下的一些有状态变量，例如RNN的隐藏状态或某些特殊算法的状态库。如果 data_id 为None，则意味着重置所有有状态变量。否则，它将根据 data_id 重置有状态变量。例如， data_id 中的不同轨迹在RNN中将具有不同的隐藏状态。

参数:

data_id (-) – 数据的id，用于重置由 data_id 指定的状态变量。

备注

此方法不是必须实现的。子类可以根据需要重写此方法。

_set_attribute(name: str, value: Any) → None

概述：

为了控制策略属性的访问，我们对外暴露不同的模式，而不是直接使用策略实例。我们还提供了一种方法，用于在不同模式下设置策略的属性。新属性将被命名为 _{name}。

参数:

• name (-) – 属性的名称。

• value (-) – 属性的值。

_state_dict_collect() → Dict[str, Any]

概述：

返回收集模式的 state_dict，仅包括通常情况下的模型，这对于分布式训练场景中自动恢复收集器是必要的。

返回:

当前策略收集状态的字典，用于保存和恢复。

返回类型:

• state_dict (Dict[str, Any])

小技巧

并非所有场景都需要自动恢复收集器，有时，我们可以直接关闭崩溃的收集器并重新启动一个新的。

_state_dict_eval() → Dict[str, Any]

概述：

返回评估模式下的 state_dict，通常只包括模型，这对于分布式训练场景是必要的，以便自动恢复评估器。

返回:

当前策略评估状态的字典，用于保存和恢复。

返回类型:

• state_dict (Dict[str, Any])

小技巧

并非所有场景都需要自动恢复评估器，有时，我们可以直接关闭崩溃的评估器并重新启动一个新的评估器。

_state_dict_learn() → Dict[str, Any]

概述：

返回学习模式的 state_dict，通常包括模型、目标模型和优化器。

返回:

当前策略学习状态的字典，用于保存和恢复。

返回类型:

• state_dict (Dict[str, Any])

property cfg: EasyDict

class collect_function(forward, process_transition, get_train_sample, reset, get_attribute, set_attribute, state_dict, load_state_dict)

基类：tuple

_asdict()

返回一个新字典，该字典将字段名称映射到它们的值。

_field_defaults = {}

_fields = ('forward', 'process_transition', 'get_train_sample', 'reset', 'get_attribute', 'set_attribute', 'state_dict', 'load_state_dict')

classmethod _make(iterable)

从序列或可迭代对象创建一个新的 collect_function 对象

_replace(**kwds)

返回一个新的 collect_function 对象，用新值替换指定字段

count(value, /)

返回值出现的次数。

forward

字段编号 0 的别名

get_attribute

字段编号4的别名

get_train_sample

字段编号2的别名

index(value, start=0, stop=9223372036854775807, /)

返回值的第一个索引。

如果值不存在，则引发 ValueError。

load_state_dict

字段编号 7 的别名

process_transition

字段编号1的别名

reset

字段编号 3 的别名

set_attribute

字段编号 5 的别名

state_dict

字段编号 6 的别名

property collect_mode: collect_function

概述：

返回策略的收集模式的接口，该接口用于训练模型。这里我们使用 namedtuple 来定义不可变的接口，并限制策略在不同模式下的使用。此外，派生子类可以重写这些接口以自定义其自己的收集模式。

返回:

策略的收集模式的接口，它是一个 namedtuple，其不同字段的值是不同的内部方法。

返回类型:

• interfaces (Policy.collect_function)

示例

>>> policy = Policy(cfg, model)
>>> policy_collect = policy.collect_mode
>>> obs = env_manager.ready_obs
>>> inference_output = policy_collect.forward(obs)
>>> next_obs, rew, done, info = env_manager.step(inference_output.action)

config = {'action_type': 'fixed_action_space', 'analysis_dormant_ratio': False, 'analysis_sim_norm': False, 'augmentation': ['shift', 'intensity'], 'batch_size': 256, 'battle_mode': 'play_with_bot_mode', 'cal_dormant_ratio': False, 'collect_with_pure_policy': False, 'collector_env_num': 8, 'cuda': True, 'discount_factor': 0.997, 'env_type': 'not_board_games', 'eps': {'decay': 100000, 'end': 0.05, 'eps_greedy_exploration_in_collect': False, 'start': 1.0, 'type': 'linear'}, 'eval_offline': False, 'evaluator_env_num': 3, 'fixed_temperature_value': 0.25, 'game_segment_length': 200, 'grad_clip_value': 10, 'gray_scale': False, 'gumbel_algo': False, 'ignore_done': False, 'learning_rate': 0.2, 'manual_temperature_decay': False, 'mcts_ctree': True, 'model': {'analysis_dormant_ratio': False, 'analysis_sim_norm': False, 'bias': True, 'categorical_distribution': True, 'continuous_action_space': False, 'discrete_action_encoding_type': 'one_hot', 'frame_stack_num': 1, 'harmony_balance': False, 'image_channel': 1, 'model_type': 'conv', 'norm_type': 'BN', 'num_channels': 64, 'num_res_blocks': 1, 'observation_shape': (4, 96, 96), 'res_connection_in_dynamics': True, 'self_supervised_learning_loss': False, 'support_scale': 300}, 'momentum': 0.9, 'monitor_extra_statistics': True, 'multi_gpu': False, 'n_episode': 8, 'num_segments': 8, 'num_simulations': 50, 'num_unroll_steps': 5, 'optim_type': 'SGD', 'piecewise_decay_lr_scheduler': True, 'policy_entropy_weight': 0, 'policy_loss_weight': 1, 'priority_prob_alpha': 0.6, 'priority_prob_beta': 0.4, 'random_collect_episode_num': 0, 'reanalyze_noise': True, 'replay_ratio': 0.25, 'reuse_search': False, 'reward_loss_weight': 1, 'root_dirichlet_alpha': 0.3, 'root_noise_weight': 0.25, 'sampled_algo': False, 'ssl_loss_weight': 0, 'target_update_freq': 100, 'target_update_freq_for_intrinsic_reward': 1000, 'td_steps': 5, 'threshold_training_steps_for_final_lr': 50000, 'threshold_training_steps_for_final_temperature': 100000, 'transform2string': False, 'update_per_collect': None, 'use_augmentation': False, 'use_priority': False, 'use_rnd_model': False, 'use_ture_chance_label_in_chance_encoder': False, 'use_wandb': False, 'value_loss_weight': 0.25, 'weight_decay': 0.0001}

classmethod default_config() → EasyDict

概述：

获取策略的默认配置。此方法用于创建策略的默认配置。

返回:

对应策略的默认配置。对于派生策略类，它将递归合并基类的默认配置及其自身的默认配置。

返回类型:

• cfg (EasyDict)

小技巧

此方法将深度复制类的 config 属性并返回结果。因此，用户无需担心返回配置的修改。

default_model() → Tuple[str, List[str]]

概述：

返回此算法的默认模型设置以供演示。

返回:

模型名称和模型导入名称。

• model_type (str): The model type used in this algorithm, which is registered in ModelRegistry.

• import_names (List[str]): The model class path list used in this algorithm.

返回类型:

• model_info (Tuple[str, List[str]])

备注

用户可以定义和使用自定义的网络模型，但必须遵守由 import_names 路径指示的相同接口定义。对于 MuZero，使用 lzero.model.muzero_model.MuZeroModel。

class eval_function(forward, reset, get_attribute, set_attribute, state_dict, load_state_dict)

基类：tuple

_asdict()

返回一个新字典，该字典将字段名称映射到它们的值。

_field_defaults = {}

_fields = ('forward', 'reset', 'get_attribute', 'set_attribute', 'state_dict', 'load_state_dict')

classmethod _make(iterable)

从序列或可迭代对象创建一个新的 eval_function 对象

_replace(**kwds)

返回一个新的 eval_function 对象，用新值替换指定字段。

count(value, /)

返回值出现的次数。

forward

字段编号 0 的别名

get_attribute

字段编号2的别名

index(value, start=0, stop=9223372036854775807, /)

返回值的第一个索引。

如果值不存在，则引发 ValueError。

load_state_dict

字段编号 5 的别名

reset

字段编号1的别名

set_attribute

字段编号 3 的别名

state_dict

字段编号4的别名

property eval_mode: eval_function

概述：

返回策略的评估模式的接口，该接口用于训练模型。这里我们使用 namedtuple 来定义不可变的接口，并限制策略在不同模式下的使用。此外，派生子类可以重写这些接口以自定义其自己的评估模式。

返回:

策略的eval模式的接口，它是一个namedtuple，其不同字段的值是不同的内部方法。

返回类型:

• interfaces (Policy.eval_function)

示例

>>> policy = Policy(cfg, model)
>>> policy_eval = policy.eval_mode
>>> obs = env_manager.ready_obs
>>> inference_output = policy_eval.forward(obs)
>>> next_obs, rew, done, info = env_manager.step(inference_output.action)

class learn_function(forward, reset, info, monitor_vars, get_attribute, set_attribute, state_dict, load_state_dict)

基类：tuple

_asdict()

返回一个新字典，该字典将字段名称映射到它们的值。

_field_defaults = {}

_fields = ('forward', 'reset', 'info', 'monitor_vars', 'get_attribute', 'set_attribute', 'state_dict', 'load_state_dict')

classmethod _make(iterable)

从序列或可迭代对象创建一个新的 learn_function 对象

_replace(**kwds)

返回一个新的 learn_function 对象，用新值替换指定的字段

count(value, /)

返回值出现的次数。

forward

字段编号 0 的别名

get_attribute

字段编号4的别名

index(value, start=0, stop=9223372036854775807, /)

返回值的第一个索引。

如果值不存在，则引发 ValueError。

info

字段编号2的别名

load_state_dict

字段编号 7 的别名

monitor_vars

字段编号 3 的别名

reset

字段编号1的别名

set_attribute

字段编号 5 的别名

state_dict

字段编号 6 的别名

property learn_mode: learn_function

概述：

返回策略学习模式的接口，该接口用于训练模型。这里我们使用 namedtuple 来定义不可变的接口，并限制策略在不同模式下的使用。此外，派生子类可以重写这些接口以自定义其自己的学习模式。

返回:

策略的学习模式的接口，它是一个 namedtuple，其不同字段的值是不同的内部方法。

返回类型:

• interfaces (Policy.learn_function)

示例

>>> policy = Policy(cfg, model)
>>> policy_learn = policy.learn_mode
>>> train_output = policy_learn.forward(data)
>>> state_dict = policy_learn.state_dict()

set_train_iter_env_step(train_iter, env_step) → None

概述：

设置策略的 train_iter 和 env_step。

参数:

• train_iter (-) – 策略的 train_iter。

• env_step (-) – 策略的 env_step。

sync_gradients(model: Module) → None

概述：

在数据并行的多GPU训练中，同步（allreduce）模型参数的梯度。

参数:

model (-) – 用于同步梯度的模型。

备注

此方法仅在多GPU训练中使用，并且应在 backward 方法之后和 step 方法之前调用。用户还可以使用 bp_update_sync 配置来控制是否同步梯度allreduce和优化器更新。

total_field = {'collect', 'eval', 'learn'}

EfficientZeroPolicy

class lzero.policy.efficientzero.EfficientZeroPolicy(cfg: EasyDict, model: Module | None = None, enable_field: List[str] | None = None)

基类：MuZeroPolicy

概述：

论文 https://arxiv.org/abs/2111.00210 中提出的 EfficientZero 的策略类。

__init__(cfg: EasyDict, model: Module | None = None, enable_field: List[str] | None = None) → None

概述：

根据输入配置和模型初始化策略实例。此方法将初始化策略中的不同字段，包括 learn、collect、eval。learn 字段用于训练策略，collect 字段用于收集训练数据，eval 字段用于评估策略。enable_field 用于指定要初始化的字段，如果为 None，则将初始化所有字段。

参数:

• cfg (-) – 用于初始化策略的最终合并配置。对于默认配置，请参阅策略类的 config 属性及其注释。

• model (-) – 用于初始化策略的神经网络模型。如果它是 None，则模型将根据 default_model 方法和 cfg.model 字段创建。否则，模型将被设置为由外部调用者创建的 model 实例。

• enable_field (-) – 要初始化的字段列表。如果为 None，则所有字段都将被初始化。否则，只有 enable_field 中的字段将被初始化，这有助于节省资源。

备注

对于派生的策略类，它应该实现 _init_learn、_init_collect、_init_eval 方法来初始化相应的字段。

_abc_impl = <_abc._abc_data object>

_create_model(cfg: EasyDict, model: Module | None = None) → Module

概述：

根据输入配置和模型创建或验证神经网络模型。如果输入模型为 None，则将根据 default_model 方法和 cfg.model 字段创建模型。否则，模型将被验证为 torch.nn.Module 的实例，并设置为由外部调用者创建的 model 实例。

参数:

• cfg (-) – 用于初始化策略的最终合并配置。

• model (-) – 用于初始化策略的神经网络模型。用户可以参考相应策略中定义的默认模型来自定义自己的模型。

返回:

创建的神经网络模型。策略的不同模式将向模型添加不同的包装器和插件，这些包装器和插件用于训练、收集和评估。

返回类型:

• model (torch.nn.Module)

抛出:

- RuntimeError – 如果输入模型不是 None 且不是 torch.nn.Module 的实例。

_forward_collect(data: Tensor, action_mask: list = None, temperature: float = 1, to_play: List = [-1], epsilon: float = 0.25, ready_env_id: array = None)

概述：

在收集模式下收集数据的正向函数。使用模型执行MCTS搜索。在收集模式期间通过采样选择动作。

参数:

• data (-) – 输入数据，即观察结果。

• action_mask (-) – 动作掩码，即不能被选择的动作。

• temperature (-) – 政策的温度。

• to_play (-) – 要播放的玩家。

• ready_env_id (-) – 准备好收集的环境的id。

形状：

• data (torch.Tensor):

  • 对于Atari，\((N, C*S, H, W)\)，其中N是collect_env的数量，C是通道的数量，S是堆叠帧的数量，H是图像的高度，W是图像的宽度。

  • 对于 lunarlander，\((N, O)\)，其中 N 是 collect_env 的数量，O 是观察空间的大小。

• action_mask: \((N, action_space_size)\)，其中 N 是 collect_env 的数量。

• 温度: \((1, )\).

• to_play: \((N, 1)\)，其中 N 是 collect_env 的数量。

• ready_env_id: None

返回:

字典类型数据，包含的键有 action、distributions、visit_count_distribution_entropy、value、pred_value、policy_logits。

返回类型:

• output (Dict[int, Any])

_forward_eval(data: Tensor, action_mask: list, to_play: -1, ready_env_id: array = None)

概述：

用于评估当前策略在评估模式下的前向函数。使用模型执行MCTS搜索。在评估模式下选择值最高的动作（argmax）而不是采样。

参数:

• data (-) – 输入数据，即观察结果。

• action_mask (-) – 动作掩码，即不能被选择的动作。

• to_play (-) – 要播放的玩家。

• ready_env_id (-) – 准备好收集的环境的id。

形状：

• data (torch.Tensor):

  • 对于Atari，\((N, C*S, H, W)\)，其中N是collect_env的数量，C是通道的数量，S是堆叠帧的数量，H是图像的高度，W是图像的宽度。

  • 对于 lunarlander，\((N, O)\)，其中 N 是 collect_env 的数量，O 是观察空间的大小。

• action_mask: \((N, action_space_size)\)，其中 N 是 collect_env 的数量。

• to_play: \((N, 1)\)，其中 N 是 collect_env 的数量。

• ready_env_id: None

返回:

字典类型数据，包含的键有 action、distributions、visit_count_distribution_entropy、value、pred_value、policy_logits。

返回类型:

• output (Dict[int, Any])

_forward_learn(data: Tensor) → Dict[str, float | int]

概述：

学习模式下用于学习策略的前向函数，这是学习过程的核心。数据从回放缓冲区中采样。通过损失函数计算损失，并将损失反向传播以更新模型。

参数:

data (-) – 从回放缓冲区中采样的数据，这是一个张量元组。第一个张量是 current_batch，第二个张量是 target_batch。

返回:

要记录的信息字典，其中包含当前的学习损失和学习统计数据。

返回类型:

• info_dict (Dict[str, Union[float, int]])

_get_attribute(name: str) → Any

概述：

为了控制策略属性的访问，我们对外暴露不同的模式，而不是直接使用策略实例。我们还提供了一种方法来获取不同模式下的策略属性。

参数:

name (-) – 属性的名称。

返回:

属性的值。

返回类型:

• value (Any)

备注

DI-engine 的策略将首先尝试访问 _get_{name} 方法，然后尝试访问 _{name} 属性。如果两者都未找到，它将引发一个 NotImplementedError。

_get_batch_size() → int | Dict[str, int]

_get_n_episode() → int | None

_get_n_sample() → int | None

_get_target_obs_index_in_step_k(step)

概述：

获取步骤 k 中目标 obs 的起始索引和结束索引。

参数:

step (-) – 当前步骤 k。

返回:

The begin index of the target obs in step k. - end_index (int): The end index of the target obs in step k.

返回类型:

• beg_index (int)

示例

>>> self._cfg.model.model_type = 'conv'
>>> self._cfg.model.image_channel = 3
>>> self._cfg.model.frame_stack_num = 4
>>> self._get_target_obs_index_in_step_k(0)
>>> (0, 12)

_get_train_sample(data)

概述：

对于给定的轨迹（过渡，过渡列表）数据，将其处理为可以直接用于训练的样本列表。一个训练样本可以是一个经过处理的过渡（DQN 使用 n步 TD）或者一些多时间步的过渡（DRQN）。这个方法通常在收集器中使用，以在训练前执行必要的强化学习数据预处理，这可以帮助学习者分摊相关的时间消耗。此外，你也可以将这个方法实现为一个恒等函数，并在 self._forward_learn 方法中进行数据处理。

参数:

transitions (-) – 轨迹数据（一个过渡列表），每个元素的格式与 self._process_transition 方法的返回值相同。

返回:

处理后的训练样本，每个元素的格式与输入转换相似，但可能包含更多用于训练的数据，如n步奖励、优势等。

返回类型:

• samples (List[Dict[str, Any]])

备注

我们将在下一个发布版本中向量化 process_transition 和 get_train_sample 方法。用户可以通过重写这两个方法和收集器本身来自定义此数据处理过程。

_init_collect() → None

概述：

收集模式初始化方法。由 self.__init__ 调用。初始化收集模型和MCTS工具。

_init_eval() → None

概述：

评估模式初始化方法。由 self.__init__ 调用。初始化评估模型和MCTS工具。

_init_learn() → None

概述：

学习模式初始化方法。由 self.__init__ 调用。初始化学习模型、优化器和MCTS工具。

_init_multi_gpu_setting(model: Module, bp_update_sync: bool) → None

概述：

初始化多GPU数据并行训练设置，包括在训练开始时广播模型参数，并为所有模型参数梯度准备hook函数以进行allreduce操作。

参数:

• model (-) – 要训练的神经网络模型。

• bp_update_sync (-) – 是否在所有reduce模型参数的梯度后同步更新模型参数。异步更新可以在不同的网络层中并行进行，如流水线，从而节省时间。

_load_state_dict_collect(state_dict: Dict[str, Any]) → None

概述：

将 state_dict 变量加载到策略收集模式中，例如加载预训练的 state_dict、自动恢复检查点，或在分布式训练场景中从学习者复制模型。

参数:

state_dict (-) – 之前保存的策略收集状态的字典。

小技巧

如果你想只加载模型的一部分，你可以简单地将 load_state_dict 中的 strict 参数设置为 False，或者参考 ding.torch_utils.checkpoint_helper 以进行更复杂的操作。

_load_state_dict_eval(state_dict: Dict[str, Any]) → None

概述：

将 state_dict 变量加载到策略评估模式中，例如加载自动恢复检查点，或在分布式训练场景中从学习者加载模型副本。

参数:

state_dict (-) – 保存之前的策略评估状态的字典。

小技巧

如果你想只加载模型的一部分，你可以简单地将 load_state_dict 中的 strict 参数设置为 False，或者参考 ding.torch_utils.checkpoint_helper 以进行更复杂的操作。

_load_state_dict_learn(state_dict: Dict[str, Any]) → None

概述：

将 state_dict 变量加载到策略学习模式中。

参数:

state_dict (-) – 之前保存的策略学习状态的字典。

_monitor_vars_learn() → List[str]

概述：

在学习模式中注册要监控的变量。根据返回值 _forward_learn，注册的变量将被记录在 tensorboard 中。

_process_transition(obs, policy_output, timestep)

概述：

将一个时间步的过渡数据处理并打包成一个字典，例如 <s, a, r, s’, done>。一些策略需要进行一些特殊处理并打包其自身的必要属性（例如隐藏状态和logit），因此此方法留给子类实现。

参数:

• obs (-) – 当前时间步的观察。

• policy_output (-) – 以观察为输入的策略网络的输出。通常，它包含动作和动作的对数概率。

• timestep (-) – 环境步骤方法返回的名为namedtuple的执行结果，除了所有元素都已转换为张量数据。通常，它包含下一个观察值、奖励、完成状态、信息等。

返回:

当前时间步的已处理过渡数据。

返回类型:

• transition (Dict[str, torch.Tensor])

_reset_collect(data_id: List[int] | None = None) → None

概述：

重置收集器环境的观察和动作。

参数:

data_id (-) – 要重置的数据ID列表（在此实现中未使用）。

_reset_eval(data_id: List[int] | None = None) → None

概述：

重置评估环境的观察和动作。

参数:

data_id (-) – 要重置的数据ID列表（在此实现中未使用）。

_reset_learn(data_id: List[int] | None = None) → None

概述：

在必要时重置学习模式下的一些有状态变量，例如RNN的隐藏状态或某些特殊算法的状态库。如果 data_id 为None，则意味着重置所有有状态变量。否则，它将根据 data_id 重置有状态变量。例如， data_id 中的不同轨迹在RNN中将具有不同的隐藏状态。

参数:

data_id (-) – 数据的id，用于重置由 data_id 指定的状态变量。

备注

此方法不是必须实现的。子类可以根据需要重写此方法。

_set_attribute(name: str, value: Any) → None

概述：

为了控制策略属性的访问，我们对外暴露不同的模式，而不是直接使用策略实例。我们还提供了一种方法，用于在不同模式下设置策略的属性。新属性将被命名为 _{name}。

参数:

• name (-) – 属性的名称。

• value (-) – 属性的值。

_state_dict_collect() → Dict[str, Any]

概述：

返回收集模式的 state_dict，仅包括通常情况下的模型，这对于分布式训练场景中自动恢复收集器是必要的。

返回:

当前策略收集状态的字典，用于保存和恢复。

返回类型:

• state_dict (Dict[str, Any])

小技巧

并非所有场景都需要自动恢复收集器，有时，我们可以直接关闭崩溃的收集器并重新启动一个新的。

_state_dict_eval() → Dict[str, Any]

概述：

返回评估模式下的 state_dict，通常只包括模型，这对于分布式训练场景是必要的，以便自动恢复评估器。

返回:

当前策略评估状态的字典，用于保存和恢复。

返回类型:

• state_dict (Dict[str, Any])

小技巧

并非所有场景都需要自动恢复评估器，有时，我们可以直接关闭崩溃的评估器并重新启动一个新的评估器。

_state_dict_learn() → Dict[str, Any]

概述：

返回学习模式的 state_dict，通常包括模型和优化器。

返回:

当前策略学习状态的字典，用于保存和恢复。

返回类型:

• state_dict (Dict[str, Any])

property cfg: EasyDict

class collect_function(forward, process_transition, get_train_sample, reset, get_attribute, set_attribute, state_dict, load_state_dict)

基类：tuple

_asdict()

返回一个新字典，该字典将字段名称映射到它们的值。

_field_defaults = {}

_fields = ('forward', 'process_transition', 'get_train_sample', 'reset', 'get_attribute', 'set_attribute', 'state_dict', 'load_state_dict')

classmethod _make(iterable)

从序列或可迭代对象创建一个新的 collect_function 对象

_replace(**kwds)

返回一个新的 collect_function 对象，用新值替换指定字段

count(value, /)

返回值出现的次数。

forward

字段编号 0 的别名

get_attribute

字段编号4的别名

get_train_sample

字段编号2的别名

index(value, start=0, stop=9223372036854775807, /)

返回值的第一个索引。

如果值不存在，则引发 ValueError。

load_state_dict

字段编号 7 的别名

process_transition

字段编号1的别名

reset

字段编号 3 的别名

set_attribute

字段编号 5 的别名

state_dict

字段编号 6 的别名

property collect_mode: collect_function

概述：

返回策略的收集模式的接口，该接口用于训练模型。这里我们使用 namedtuple 来定义不可变的接口，并限制策略在不同模式下的使用。此外，派生子类可以重写这些接口以自定义其自己的收集模式。

返回:

策略的收集模式的接口，它是一个 namedtuple，其不同字段的值是不同的内部方法。

返回类型:

• interfaces (Policy.collect_function)

示例

>>> policy = Policy(cfg, model)
>>> policy_collect = policy.collect_mode
>>> obs = env_manager.ready_obs
>>> inference_output = policy_collect.forward(obs)
>>> next_obs, rew, done, info = env_manager.step(inference_output.action)

config = {'action_type': 'fixed_action_space', 'augmentation': ['shift', 'intensity'], 'batch_size': 256, 'battle_mode': 'play_with_bot_mode', 'collect_with_pure_policy': False, 'collector_env_num': 8, 'cuda': True, 'discount_factor': 0.997, 'env_type': 'not_board_games', 'eps': {'decay': 100000, 'end': 0.05, 'eps_greedy_exploration_in_collect': False, 'start': 1.0, 'type': 'linear'}, 'eval_offline': False, 'evaluator_env_num': 3, 'fixed_temperature_value': 0.25, 'game_segment_length': 200, 'grad_clip_value': 10, 'gray_scale': False, 'gumbel_algo': False, 'ignore_done': False, 'learning_rate': 0.2, 'lstm_horizon_len': 5, 'manual_temperature_decay': False, 'mcts_ctree': True, 'model': {'bias': True, 'categorical_distribution': True, 'continuous_action_space': False, 'discrete_action_encoding_type': 'one_hot', 'frame_stack_num': 1, 'image_channel': 1, 'lstm_hidden_size': 512, 'model_type': 'conv', 'norm_type': 'BN', 'observation_shape': (4, 96, 96), 'res_connection_in_dynamics': True, 'self_supervised_learning_loss': True, 'support_scale': 300}, 'momentum': 0.9, 'monitor_extra_statistics': True, 'multi_gpu': False, 'n_episode': 8, 'num_simulations': 50, 'num_unroll_steps': 5, 'optim_type': 'SGD', 'piecewise_decay_lr_scheduler': True, 'policy_loss_weight': 1, 'priority_prob_alpha': 0.6, 'priority_prob_beta': 0.4, 'random_collect_episode_num': 0, 'reanalyze_noise': True, 'replay_ratio': 0.25, 'reuse_search': False, 'reward_loss_weight': 1, 'root_dirichlet_alpha': 0.3, 'root_noise_weight': 0.25, 'sampled_algo': False, 'ssl_loss_weight': 2, 'target_update_freq': 100, 'td_steps': 5, 'threshold_training_steps_for_final_lr': 50000, 'threshold_training_steps_for_final_temperature': 100000, 'transform2string': False, 'update_per_collect': None, 'use_augmentation': False, 'use_priority': False, 'use_ture_chance_label_in_chance_encoder': False, 'value_loss_weight': 0.25, 'weight_decay': 0.0001}

classmethod default_config() → EasyDict

概述：

获取策略的默认配置。此方法用于创建策略的默认配置。

返回:

对应策略的默认配置。对于派生策略类，它将递归合并基类的默认配置及其自身的默认配置。

返回类型:

• cfg (EasyDict)

小技巧

此方法将深度复制类的 config 属性并返回结果。因此，用户无需担心返回配置的修改。

default_model() → Tuple[str, List[str]]

概述：

返回此算法的默认模型设置。

返回:

模型名称和模型导入名称。

• model_type (str): The model type used in this algorithm, which is registered in ModelRegistry.

• import_names (List[str]): The model class path list used in this algorithm.

返回类型:

• model_info (Tuple[str, List[str]])

备注

用户可以定义和使用自定义的网络模型，但必须遵守由 import_names 路径指示的相同接口定义。对于 EfficientZero，使用 lzero.model.efficientzero_model.EfficientZeroModel。

class eval_function(forward, reset, get_attribute, set_attribute, state_dict, load_state_dict)

基类：tuple

_asdict()

返回一个新字典，该字典将字段名称映射到它们的值。

_field_defaults = {}

_fields = ('forward', 'reset', 'get_attribute', 'set_attribute', 'state_dict', 'load_state_dict')

classmethod _make(iterable)

从序列或可迭代对象创建一个新的 eval_function 对象

_replace(**kwds)

返回一个新的 eval_function 对象，用新值替换指定字段。

count(value, /)

返回值出现的次数。

forward

字段编号 0 的别名

get_attribute

字段编号2的别名

index(value, start=0, stop=9223372036854775807, /)

返回值的第一个索引。

如果值不存在，则引发 ValueError。

load_state_dict

字段编号 5 的别名

reset

字段编号1的别名

set_attribute

字段编号 3 的别名

state_dict

字段编号4的别名

property eval_mode: eval_function

概述：

返回策略的评估模式的接口，该接口用于训练模型。这里我们使用 namedtuple 来定义不可变的接口，并限制策略在不同模式下的使用。此外，派生子类可以重写这些接口以自定义其自己的评估模式。

返回:

策略的eval模式的接口，它是一个namedtuple，其不同字段的值是不同的内部方法。

返回类型:

• interfaces (Policy.eval_function)

示例

>>> policy = Policy(cfg, model)
>>> policy_eval = policy.eval_mode
>>> obs = env_manager.ready_obs
>>> inference_output = policy_eval.forward(obs)
>>> next_obs, rew, done, info = env_manager.step(inference_output.action)

class learn_function(forward, reset, info, monitor_vars, get_attribute, set_attribute, state_dict, load_state_dict)

基类：tuple

_asdict()

返回一个新字典，该字典将字段名称映射到它们的值。

_field_defaults = {}

_fields = ('forward', 'reset', 'info', 'monitor_vars', 'get_attribute', 'set_attribute', 'state_dict', 'load_state_dict')

classmethod _make(iterable)

从序列或可迭代对象创建一个新的 learn_function 对象

_replace(**kwds)

返回一个新的 learn_function 对象，用新值替换指定的字段

count(value, /)

返回值出现的次数。

forward

字段编号 0 的别名

get_attribute

字段编号4的别名

index(value, start=0, stop=9223372036854775807, /)

返回值的第一个索引。

如果值不存在，则引发 ValueError。

info

字段编号2的别名

load_state_dict

字段编号 7 的别名

monitor_vars

字段编号 3 的别名

reset

字段编号1的别名

set_attribute

字段编号 5 的别名

state_dict

字段编号 6 的别名

property learn_mode: learn_function

概述：

返回策略学习模式的接口，该接口用于训练模型。这里我们使用 namedtuple 来定义不可变的接口，并限制策略在不同模式下的使用。此外，派生子类可以重写这些接口以自定义其自己的学习模式。

返回:

策略的学习模式的接口，它是一个 namedtuple，其不同字段的值是不同的内部方法。

返回类型:

• interfaces (Policy.learn_function)

示例

>>> policy = Policy(cfg, model)
>>> policy_learn = policy.learn_mode
>>> train_output = policy_learn.forward(data)
>>> state_dict = policy_learn.state_dict()

set_train_iter_env_step(train_iter, env_step) → None

概述：

设置策略的 train_iter 和 env_step。

参数:

• train_iter (-) – 策略的 train_iter。

• env_step (-) – 策略的 env_step。

sync_gradients(model: Module) → None

概述：

在数据并行的多GPU训练中，同步（allreduce）模型参数的梯度。

参数:

model (-) – 用于同步梯度的模型。

备注

此方法仅在多GPU训练中使用，并且应在 backward 方法之后和 step 方法之前调用。用户还可以使用 bp_update_sync 配置来控制是否同步梯度allreduce和优化器更新。

total_field = {'collect', 'eval', 'learn'}

Gumbel AlphaZero策略

class lzero.policy.gumbel_alphazero.GumbelAlphaZeroPolicy(cfg: EasyDict, model: Module | None = None, enable_field: List[str] | None = None)

基类：Policy

概述：

GumbelAlphaZero 的策略类。

__init__(cfg: EasyDict, model: Module | None = None, enable_field: List[str] | None = None) → None

概述：

根据输入配置和模型初始化策略实例。此方法将初始化策略中的不同字段，包括 learn、collect、eval。learn 字段用于训练策略，collect 字段用于收集训练数据，eval 字段用于评估策略。enable_field 用于指定要初始化的字段，如果为 None，则将初始化所有字段。

参数:

• cfg (-) – 用于初始化策略的最终合并配置。对于默认配置，请参阅策略类的 config 属性及其注释。

• model (-) – 用于初始化策略的神经网络模型。如果它是 None，则模型将根据 default_model 方法和 cfg.model 字段创建。否则，模型将被设置为由外部调用者创建的 model 实例。

• enable_field (-) – 要初始化的字段列表。如果为 None，则所有字段都将被初始化。否则，只有 enable_field 中的字段将被初始化，这有助于节省资源。

备注

对于派生的策略类，它应该实现 _init_learn、_init_collect、_init_eval 方法来初始化相应的字段。

_abc_impl = <_abc._abc_data object>

_create_model(cfg: EasyDict, model: Module | None = None) → Module

概述：

根据输入配置和模型创建或验证神经网络模型。如果输入模型为 None，则将根据 default_model 方法和 cfg.model 字段创建模型。否则，模型将被验证为 torch.nn.Module 的实例，并设置为由外部调用者创建的 model 实例。

参数:

• cfg (-) – 用于初始化策略的最终合并配置。

• model (-) – 用于初始化策略的神经网络模型。用户可以参考相应策略中定义的默认模型来自定义自己的模型。

返回:

创建的神经网络模型。策略的不同模式将向模型添加不同的包装器和插件，这些包装器和插件用于训练、收集和评估。

返回类型:

• model (torch.nn.Module)

抛出:

- RuntimeError – 如果输入模型不是 None 且不是 torch.nn.Module 的实例。

_forward_collect(obs: Dict, temperature: float = 1) → Dict[str, Tensor]

概述：

在收集模式下收集数据的前向函数。使用真实环境执行MCTS搜索。

参数:

• obs (-) – obs 的字典，键是 env_id，值是该时间步对应的 obs。

• temperature (-) – MCTS搜索的温度。

返回:

输出字典，键是 env_id，值是该时间步对应的策略输出，包括动作、概率等。

返回类型:

• output (Dict[str, torch.Tensor])

_forward_eval(obs: Dict) → Dict[str, Tensor]

概述：

用于在评估模式下评估当前策略的前向函数，类似于 self._forward_collect。

参数:

obs (-) – obs 的字典，键是 env_id，值是该时间步对应的 obs。

返回:

输出字典，键是 env_id，值是该时间步对应的策略输出，包括动作、概率等。

返回类型:

• output (Dict[str, torch.Tensor])

_forward_learn(inputs: Dict[str, Tensor]) → Dict[str, float]

概述：

学习模式下的策略前向函数（训练策略和更新参数）。前向意味着策略从回放缓冲区输入一些训练批次数据，然后返回输出结果，包括各种训练信息，如损失值、策略熵、q值、优先级等。此方法留给子类实现，如果需要，可以在``data``项中添加更多参数。

参数:

data (-) – 用于策略前向传播的输入数据，包括一批训练样本。对于列表中的每个元素，字典的键是数据项的名称，值是对应的数据。通常，在 _forward_learn 方法中，数据应通过一些实用函数（如 default_preprocess_learn）在批次维度上堆叠。

返回:

策略前向传播的训练信息，包括一些用于监控训练的指标，如损失、优先级、Q值、策略熵，以及一些用于下一步训练的数据，如优先级。注意输出数据项应该是Python原生标量，而不是PyTorch张量，这样对外部使用更方便。

返回类型:

• output (Dict[int, Any])

_get_attribute(name: str) → Any

概述：

为了控制策略属性的访问，我们对外暴露不同的模式，而不是直接使用策略实例。我们还提供了一种方法来获取不同模式下的策略属性。

参数:

name (-) – 属性的名称。

返回:

属性的值。

返回类型:

• value (Any)

备注

DI-engine 的策略将首先尝试访问 _get_{name} 方法，然后尝试访问 _{name} 属性。如果两者都未找到，它将引发一个 NotImplementedError。

_get_batch_size() → int | Dict[str, int]

_get_n_episode() → int | None

_get_n_sample() → int | None

_get_simulation_env()

_get_train_sample(data)

概述：

对于给定的轨迹（过渡，过渡列表）数据，将其处理为可以直接用于训练的样本列表。一个训练样本可以是一个经过处理的过渡（DQN 使用 n步 TD）或者一些多时间步的过渡（DRQN）。这个方法通常在收集器中使用，以在训练前执行必要的强化学习数据预处理，这可以帮助学习者分摊相关的时间消耗。此外，你也可以将这个方法实现为一个恒等函数，并在 self._forward_learn 方法中进行数据处理。

参数:

transitions (-) – 轨迹数据（一个过渡列表），每个元素的格式与 self._process_transition 方法的返回值相同。

返回:

处理后的训练样本，每个元素的格式与输入转换相似，但可能包含更多用于训练的数据，如n步奖励、优势等。

返回类型:

• samples (List[Dict[str, Any]])

备注

我们将在下一个发布版本中向量化 process_transition 和 get_train_sample 方法。用户可以通过重写这两个方法和收集器本身来自定义此数据处理过程。

_init_collect() → None

概述：

收集模式初始化方法。由 self.__init__ 调用。初始化收集模型和MCTS工具。

_init_eval() → None

概述：

评估模式初始化方法。由 self.__init__ 调用。初始化评估模型和MCTS工具。

_init_learn() → None

概述：

初始化策略的学习模式，包括相关属性和模块。如果 learn 字段在 enable_field 中，此方法将在 __init__ 方法中被调用。几乎不同的策略都有其自己的学习模式，因此此方法必须在子类中被重写。

备注

对于需要保存和加载的成员变量，请参考 _state_dict_learn 和 _load_state_dict_learn 方法。

备注

对于需要监控的成员变量，请参考 _monitor_vars_learn 方法。

备注

如果你想在 _init_learn 方法中设置一些特殊的成员变量，最好用前缀 _learn_ 来命名它们，以避免与其他模式发生冲突，例如 self._learn_attr1。

_init_multi_gpu_setting(model: Module, bp_update_sync: bool) → None

概述：

初始化多GPU数据并行训练设置，包括在训练开始时广播模型参数，并为所有模型参数梯度准备hook函数以进行allreduce操作。

参数:

• model (-) – 要训练的神经网络模型。

• bp_update_sync (-) – 是否在所有reduce模型参数的梯度后同步更新模型参数。异步更新可以在不同的网络层中并行进行，如流水线，从而节省时间。

_load_state_dict_collect(state_dict: Dict[str, Any]) → None

概述：

将 state_dict 变量加载到策略收集模式中，例如加载预训练的 state_dict、自动恢复检查点，或在分布式训练场景中从学习者复制模型。

参数:

state_dict (-) – 之前保存的策略收集状态的字典。

小技巧

如果你想只加载模型的一部分，你可以简单地将 load_state_dict 中的 strict 参数设置为 False，或者参考 ding.torch_utils.checkpoint_helper 以进行更复杂的操作。

_load_state_dict_eval(state_dict: Dict[str, Any]) → None

概述：

将 state_dict 变量加载到策略评估模式中，例如加载自动恢复检查点，或在分布式训练场景中从学习者加载模型副本。

参数:

state_dict (-) – 保存之前的策略评估状态的字典。

小技巧

如果你想只加载模型的一部分，你可以简单地将 load_state_dict 中的 strict 参数设置为 False，或者参考 ding.torch_utils.checkpoint_helper 以进行更复杂的操作。

_load_state_dict_learn(state_dict: Dict[str, Any]) → None

概述：

将 state_dict 变量加载到策略学习模式中。

参数:

state_dict (-) – 之前保存的策略学习状态的字典。

小技巧

如果你想只加载模型的一部分，你可以简单地将 load_state_dict 中的 strict 参数设置为 False，或者参考 ding.torch_utils.checkpoint_helper 以进行更复杂的操作。

_monitor_vars_learn() → List[str]

概述：

在学习模式中注册要监控的变量。根据返回值 _forward_learn，注册的变量将被记录在 tensorboard 中。

_policy_value_fn(env: Env) → Tuple[Dict[int, ndarray], float]

_process_transition(obs: Dict, model_output: Dict[str, Tensor], timestep: namedtuple) → Dict

概述：

从策略学习中生成字典类型转换（一个时间步长）数据。

_reset_collect(data_id: List[int] | None = None) → None

概述：

在必要时重置收集模式下的一些有状态变量，例如RNN的隐藏状态或某些特殊算法的内存库。如果``data_id``为None，则意味着重置所有有状态变量。否则，它将根据``data_id``重置有状态变量。例如，在``data_id``中收集的不同环境/情节在RNN中将具有不同的隐藏状态。

参数:

data_id (-) – 数据的id，用于重置由 data_id 指定的状态变量。

备注

此方法不是必须实现的。子类可以根据需要重写此方法。

_reset_eval(data_id: List[int] | None = None) → None

概述：

在必要时重置一些有状态变量以适应评估模式，例如RNN的隐藏状态或某些特殊算法的内存库。如果 data_id 为None，则意味着重置所有有状态变量。否则，它将根据 data_id 重置有状态变量。例如，在 data_id 中的不同环境/情节在评估中将具有RNN中的不同隐藏状态。

参数:

data_id (-) – 数据的id，用于重置由 data_id 指定的状态变量。

备注

此方法不是必须实现的。子类可以根据需要重写此方法。

_reset_learn(data_id: List[int] | None = None) → None

概述：

在必要时重置学习模式下的一些有状态变量，例如RNN的隐藏状态或某些特殊算法的状态库。如果 data_id 为None，则意味着重置所有有状态变量。否则，它将根据 data_id 重置有状态变量。例如， data_id 中的不同轨迹在RNN中将具有不同的隐藏状态。

参数:

data_id (-) – 数据的id，用于重置由 data_id 指定的状态变量。

备注

此方法不是必须实现的。子类可以根据需要重写此方法。

_set_attribute(name: str, value: Any) → None

概述：

为了控制策略属性的访问，我们对外暴露不同的模式，而不是直接使用策略实例。我们还提供了一种方法，用于在不同模式下设置策略的属性。新属性将被命名为 _{name}。

参数:

• name (-) – 属性的名称。

• value (-) – 属性的值。

_state_dict_collect() → Dict[str, Any]

概述：

返回收集模式的 state_dict，仅包括通常情况下的模型，这对于分布式训练场景中自动恢复收集器是必要的。

返回:

当前策略收集状态的字典，用于保存和恢复。

返回类型:

• state_dict (Dict[str, Any])

小技巧

并非所有场景都需要自动恢复收集器，有时，我们可以直接关闭崩溃的收集器并重新启动一个新的。

_state_dict_eval() → Dict[str, Any]

概述：

返回评估模式下的 state_dict，通常只包括模型，这对于分布式训练场景是必要的，以便自动恢复评估器。

返回:

当前策略评估状态的字典，用于保存和恢复。

返回类型:

• state_dict (Dict[str, Any])

小技巧

并非所有场景都需要自动恢复评估器，有时，我们可以直接关闭崩溃的评估器并重新启动一个新的评估器。

_state_dict_learn() → Dict[str, Any]

概述：

返回学习模式的 state_dict，通常包括模型和优化器。

返回:

当前策略学习状态的字典，用于保存和恢复。

返回类型:

• state_dict (Dict[str, Any])

property cfg: EasyDict

class collect_function(forward, process_transition, get_train_sample, reset, get_attribute, set_attribute, state_dict, load_state_dict)

基类：tuple

_asdict()

返回一个新字典，该字典将字段名称映射到它们的值。

_field_defaults = {}

_fields = ('forward', 'process_transition', 'get_train_sample', 'reset', 'get_attribute', 'set_attribute', 'state_dict', 'load_state_dict')

classmethod _make(iterable)

从序列或可迭代对象创建一个新的 collect_function 对象

_replace(**kwds)

返回一个新的 collect_function 对象，用新值替换指定字段

count(value, /)

返回值出现的次数。

forward

字段编号 0 的别名

get_attribute

字段编号4的别名

get_train_sample

字段编号2的别名

index(value, start=0, stop=9223372036854775807, /)

返回值的第一个索引。

如果值不存在，则引发 ValueError。

load_state_dict

字段编号 7 的别名

process_transition

字段编号1的别名

reset

字段编号 3 的别名

set_attribute

字段编号 5 的别名

state_dict

字段编号 6 的别名

property collect_mode: collect_function

概述：

返回策略的收集模式的接口，该接口用于训练模型。这里我们使用 namedtuple 来定义不可变的接口，并限制策略在不同模式下的使用。此外，派生子类可以重写这些接口以自定义其自己的收集模式。

返回:

策略的收集模式的接口，它是一个 namedtuple，其不同字段的值是不同的内部方法。

返回类型:

• interfaces (Policy.collect_function)

示例

>>> policy = Policy(cfg, model)
>>> policy_collect = policy.collect_mode
>>> obs = env_manager.ready_obs
>>> inference_output = policy_collect.forward(obs)
>>> next_obs, rew, done, info = env_manager.step(inference_output.action)

config = {'batch_size': 256, 'collector_env_num': 8, 'cuda': False, 'evaluator_env_num': 3, 'fixed_temperature_value': 0.25, 'grad_clip_value': 10, 'learning_rate': 0.2, 'manual_temperature_decay': False, 'mcts': {'action_space_size': 9, 'continuous_action_space': False, 'gumbel_rng': 0.0, 'gumbel_scale': 10.0, 'legal_actions': None, 'max_moves': 512, 'max_num_considered_actions': 6, 'maxvisit_init': 50, 'num_of_sampled_actions': 2, 'num_simulations': 50, 'pb_c_base': 19652, 'pb_c_init': 1.25, 'root_dirichlet_alpha': 0.3, 'root_noise_weight': 0.25, 'value_scale': 0.1}, 'mcts_ctree': True, 'model': {'num_channels': 32, 'num_res_blocks': 1, 'observation_shape': (3, 6, 6)}, 'momentum': 0.9, 'optim_type': 'SGD', 'other': {'replay_buffer': {'replay_buffer_size': 1000000, 'save_episode': False}}, 'piecewise_decay_lr_scheduler': True, 'replay_ratio': 0.25, 'sampled_algo': False, 'tensor_float_32': False, 'threshold_training_steps_for_final_lr': 500000, 'threshold_training_steps_for_final_temperature': 100000, 'torch_compile': False, 'type': 'gumbel_alphazero', 'update_per_collect': None, 'value_weight': 1.0, 'weight_decay': 0.0001}

classmethod default_config() → EasyDict

概述：

获取策略的默认配置。此方法用于创建策略的默认配置。

返回:

对应策略的默认配置。对于派生策略类，它将递归合并基类的默认配置及其自身的默认配置。

返回类型:

• cfg (EasyDict)

小技巧

此方法将深度复制类的 config 属性并返回结果。因此，用户无需担心返回配置的修改。

default_model() → Tuple[str, List[str]]

概述：

返回此算法的默认模型设置以供演示。

返回:

The model type used in this algorithm, which is registered in ModelRegistry. - import_names (List[str]): The model class path list used in this algorithm.

返回类型:

• model_type (str)

class eval_function(forward, reset, get_attribute, set_attribute, state_dict, load_state_dict)

基类：tuple

_asdict()

返回一个新字典，该字典将字段名称映射到它们的值。

_field_defaults = {}

_fields = ('forward', 'reset', 'get_attribute', 'set_attribute', 'state_dict', 'load_state_dict')

classmethod _make(iterable)

从序列或可迭代对象创建一个新的 eval_function 对象

_replace(**kwds)

返回一个新的 eval_function 对象，用新值替换指定字段。

count(value, /)

返回值出现的次数。

forward

字段编号 0 的别名

get_attribute

字段编号2的别名

index(value, start=0, stop=9223372036854775807, /)

返回值的第一个索引。

如果值不存在，则引发 ValueError。

load_state_dict

字段编号 5 的别名

reset

字段编号1的别名

set_attribute

字段编号 3 的别名

state_dict

字段编号4的别名

property eval_mode: eval_function

概述：

返回策略的评估模式的接口，该接口用于训练模型。这里我们使用 namedtuple 来定义不可变的接口，并限制策略在不同模式下的使用。此外，派生子类可以重写这些接口以自定义其自己的评估模式。

返回:

策略的eval模式的接口，它是一个namedtuple，其不同字段的值是不同的内部方法。

返回类型:

• interfaces (Policy.eval_function)

示例

>>> policy = Policy(cfg, model)
>>> policy_eval = policy.eval_mode
>>> obs = env_manager.ready_obs
>>> inference_output = policy_eval.forward(obs)
>>> next_obs, rew, done, info = env_manager.step(inference_output.action)

class learn_function(forward, reset, info, monitor_vars, get_attribute, set_attribute, state_dict, load_state_dict)

基类：tuple

_asdict()

返回一个新字典，该字典将字段名称映射到它们的值。

_field_defaults = {}

_fields = ('forward', 'reset', 'info', 'monitor_vars', 'get_attribute', 'set_attribute', 'state_dict', 'load_state_dict')

classmethod _make(iterable)

从序列或可迭代对象创建一个新的 learn_function 对象

_replace(**kwds)

返回一个新的 learn_function 对象，用新值替换指定的字段

count(value, /)

返回值出现的次数。

forward

字段编号 0 的别名

get_attribute

字段编号4的别名

index(value, start=0, stop=9223372036854775807, /)

返回值的第一个索引。

如果值不存在，则引发 ValueError。

info

字段编号2的别名

load_state_dict

字段编号 7 的别名

monitor_vars

字段编号 3 的别名

reset

字段编号1的别名

set_attribute

字段编号 5 的别名

state_dict

字段编号 6 的别名

property learn_mode: learn_function

概述：

返回策略学习模式的接口，该接口用于训练模型。这里我们使用 namedtuple 来定义不可变的接口，并限制策略在不同模式下的使用。此外，派生子类可以重写这些接口以自定义其自己的学习模式。

返回:

策略的学习模式的接口，它是一个 namedtuple，其不同字段的值是不同的内部方法。

返回类型:

• interfaces (Policy.learn_function)

示例

>>> policy = Policy(cfg, model)
>>> policy_learn = policy.learn_mode
>>> train_output = policy_learn.forward(data)
>>> state_dict = policy_learn.state_dict()

sync_gradients(model: Module) → None

概述：

在数据并行的多GPU训练中，同步（allreduce）模型参数的梯度。

参数:

model (-) – 用于同步梯度的模型。

备注

此方法仅在多GPU训练中使用，并且应在 backward 方法之后和 step 方法之前调用。用户还可以使用 bp_update_sync 配置来控制是否同步梯度allreduce和优化器更新。

total_field = {'collect', 'eval', 'learn'}

Gumbel MuZeroPolicy

class lzero.policy.gumbel_muzero.GumbelMuZeroPolicy(cfg: EasyDict, model: Module | None = None, enable_field: List[str] | None = None)

基类：MuZeroPolicy

概述：

论文 https://openreview.net/forum?id=bERaNdoegnO 中提出的Gumbel MuZero的策略类。

__init__(cfg: EasyDict, model: Module | None = None, enable_field: List[str] | None = None) → None

概述：

根据输入配置和模型初始化策略实例。此方法将初始化策略中的不同字段，包括 learn、collect、eval。learn 字段用于训练策略，collect 字段用于收集训练数据，eval 字段用于评估策略。enable_field 用于指定要初始化的字段，如果为 None，则将初始化所有字段。

参数:

• cfg (-) – 用于初始化策略的最终合并配置。对于默认配置，请参阅策略类的 config 属性及其注释。

• model (-) – 用于初始化策略的神经网络模型。如果它是 None，则模型将根据 default_model 方法和 cfg.model 字段创建。否则，模型将被设置为由外部调用者创建的 model 实例。

• enable_field (-) – 要初始化的字段列表。如果为 None，则所有字段都将被初始化。否则，只有 enable_field 中的字段将被初始化，这有助于节省资源。

备注

对于派生的策略类，它应该实现 _init_learn、_init_collect、_init_eval 方法来初始化相应的字段。

_abc_impl = <_abc._abc_data object>

_create_model(cfg: EasyDict, model: Module | None = None) → Module

概述：

根据输入配置和模型创建或验证神经网络模型。如果输入模型为 None，则将根据 default_model 方法和 cfg.model 字段创建模型。否则，模型将被验证为 torch.nn.Module 的实例，并设置为由外部调用者创建的 model 实例。

参数:

• cfg (-) – 用于初始化策略的最终合并配置。

• model (-) – 用于初始化策略的神经网络模型。用户可以参考相应策略中定义的默认模型来自定义自己的模型。

返回:

创建的神经网络模型。策略的不同模式将向模型添加不同的包装器和插件，这些包装器和插件用于训练、收集和评估。

返回类型:

• model (torch.nn.Module)

抛出:

- RuntimeError – 如果输入模型不是 None 且不是 torch.nn.Module 的实例。

_forward_collect(data: Tensor, action_mask: list = None, temperature: float = 1, to_play: List = [-1], epsilon: float = 0.25, ready_env_id: array = None) → Dict

概述：

在收集模式下收集数据的正向函数。使用模型执行MCTS搜索。在收集模式期间通过采样选择动作。

参数:

• data (-) – 输入数据，即观察结果。

• action_mask (-) – 动作掩码，即不能被选择的动作。

• temperature (-) – 政策的温度。

• to_play (-) – 要播放的玩家。

• ready_env_id (-) – 准备好收集的环境的id。

形状：

• data (torch.Tensor):

  • 对于Atari，\((N, C*S, H, W)\)，其中N是collect_env的数量，C是通道的数量，S是堆叠帧的数量，H是图像的高度，W是图像的宽度。

  • 对于 lunarlander，\((N, O)\)，其中 N 是 collect_env 的数量，O 是观察空间的大小。

• action_mask: \((N, action_space_size)\)，其中 N 是 collect_env 的数量。

• 温度: \((1, )\).

• to_play: \((N, 1)\)，其中 N 是 collect_env 的数量。

• ready_env_id: None

返回:

字典类型数据，包含键如 action、distributions、visit_count_distribution_entropy、value、roots_completed_value、improved_policy_probs、pred_value、policy_logits。

返回类型:

• output (Dict[int, Any])

_forward_eval(data: Tensor, action_mask: list, to_play: int = -1, ready_env_id: array = None) → Dict

概述：

用于评估当前策略在评估模式下的前向函数。使用模型执行MCTS搜索。在评估模式下选择值最高的动作（argmax）而不是采样。

参数:

• data (-) – 输入数据，即观察结果。

• action_mask (-) – 动作掩码，即不能被选择的动作。

• to_play (-) – 要播放的玩家。

• ready_env_id (-) – 准备好收集的环境的id。

形状：

• data (torch.Tensor):

  • 对于Atari，\((N, C*S, H, W)\)，其中N是collect_env的数量，C是通道的数量，S是堆叠帧的数量，H是图像的高度，W是图像的宽度。

  • 对于 lunarlander，\((N, O)\)，其中 N 是 collect_env 的数量，O 是观察空间的大小。

• action_mask: \((N, action_space_size)\)，其中 N 是 collect_env 的数量。

• to_play: \((N, 1)\)，其中 N 是 collect_env 的数量。

• ready_env_id: None

返回:

字典类型数据，包含的键有 action、distributions、visit_count_distribution_entropy、value、pred_value、policy_logits。

返回类型:

• output (Dict[int, Any])

_forward_learn(data: Tensor) → Dict[str, float | int]

概述：

学习模式下用于学习策略的前向函数，这是学习过程的核心。数据从回放缓冲区中采样。通过损失函数计算损失，并将损失反向传播以更新模型。

参数:

data (-) – 从回放缓冲区中采样的数据，这是一个张量元组。第一个张量是当前批次，第二个张量是目标批次。

返回:

要记录的信息字典，其中包含当前的学习损失和学习统计数据。

返回类型:

• info_dict (Dict[str, Union[float, int]])

_get_attribute(name: str) → Any

概述：

为了控制策略属性的访问，我们对外暴露不同的模式，而不是直接使用策略实例。我们还提供了一种方法来获取不同模式下的策略属性。

参数:

name (-) – 属性的名称。

返回:

属性的值。

返回类型:

• value (Any)

备注

DI-engine 的策略将首先尝试访问 _get_{name} 方法，然后尝试访问 _{name} 属性。如果两者都未找到，它将引发一个 NotImplementedError。

_get_batch_size() → int | Dict[str, int]

_get_n_episode() → int | None

_get_n_sample() → int | None

_get_target_obs_index_in_step_k(step)

概述：

获取步骤 k 中目标 obs 的起始索引和结束索引。

参数:

step (-) – 当前步骤 k。

返回:

The begin index of the target obs in step k. - end_index (int): The end index of the target obs in step k.

返回类型:

• beg_index (int)

示例

>>> self._cfg.model.model_type = 'conv'
>>> self._cfg.model.image_channel = 3
>>> self._cfg.model.frame_stack_num = 4
>>> self._get_target_obs_index_in_step_k(0)
>>> (0, 12)

_get_train_sample(data)

概述：

对于给定的轨迹（过渡，过渡列表）数据，将其处理为可以直接用于训练的样本列表。一个训练样本可以是一个经过处理的过渡（DQN 使用 n步 TD）或者一些多时间步的过渡（DRQN）。这个方法通常在收集器中使用，以在训练前执行必要的强化学习数据预处理，这可以帮助学习者分摊相关的时间消耗。此外，你也可以将这个方法实现为一个恒等函数，并在 self._forward_learn 方法中进行数据处理。

参数:

transitions (-) – 轨迹数据（一个过渡列表），每个元素的格式与 self._process_transition 方法的返回值相同。

返回:

处理后的训练样本，每个元素的格式与输入转换相似，但可能包含更多用于训练的数据，如n步奖励、优势等。

返回类型:

• samples (List[Dict[str, Any]])

备注

我们将在下一个发布版本中向量化 process_transition 和 get_train_sample 方法。用户可以通过重写这两个方法和收集器本身来自定义此数据处理过程。

_init_collect() → None

概述：

收集模式初始化方法。由 self.__init__ 调用。初始化收集模型和MCTS工具。

_init_eval() → None

概述：

评估模式初始化方法。由 self.__init__ 调用。初始化评估模型和MCTS工具。

_init_learn() → None

概述：

学习模式初始化方法。由 self.__init__ 调用。初始化学习模型、优化器和MCTS工具。

_init_multi_gpu_setting(model: Module, bp_update_sync: bool) → None

概述：

初始化多GPU数据并行训练设置，包括在训练开始时广播模型参数，并为所有模型参数梯度准备hook函数以进行allreduce操作。

参数:

• model (-) – 要训练的神经网络模型。

• bp_update_sync (-) – 是否在所有reduce模型参数的梯度后同步更新模型参数。异步更新可以在不同的网络层中并行进行，如流水线，从而节省时间。

_load_state_dict_collect(state_dict: Dict[str, Any]) → None

概述：

将 state_dict 变量加载到策略收集模式中，例如加载预训练的 state_dict、自动恢复检查点，或在分布式训练场景中从学习者复制模型。

参数:

state_dict (-) – 之前保存的策略收集状态的字典。

小技巧

如果你想只加载模型的一部分，你可以简单地将 load_state_dict 中的 strict 参数设置为 False，或者参考 ding.torch_utils.checkpoint_helper 以进行更复杂的操作。

_load_state_dict_eval(state_dict: Dict[str, Any]) → None

概述：

将 state_dict 变量加载到策略评估模式中，例如加载自动恢复检查点，或在分布式训练场景中从学习者加载模型副本。

参数:

state_dict (-) – 保存之前的策略评估状态的字典。

小技巧

如果你想只加载模型的一部分，你可以简单地将 load_state_dict 中的 strict 参数设置为 False，或者参考 ding.torch_utils.checkpoint_helper 以进行更复杂的操作。

_load_state_dict_learn(state_dict: Dict[str, Any]) → None

概述：

将 state_dict 变量加载到策略学习模式中。

参数:

state_dict (-) – 之前保存的策略学习状态的字典。

_monitor_vars_learn() → List[str]

概述：

在学习模式中注册要监控的变量。根据返回值 _forward_learn，注册的变量将被记录在 tensorboard 中。

_process_transition(obs, policy_output, timestep)

概述：

将一个时间步的过渡数据处理并打包成一个字典，例如 <s, a, r, s’, done>。一些策略需要进行一些特殊处理并打包其自身的必要属性（例如隐藏状态和logit），因此此方法留给子类实现。

参数:

• obs (-) – 当前时间步的观察。

• policy_output (-) – 以观察为输入的策略网络的输出。通常，它包含动作和动作的对数概率。

• timestep (-) – 环境步骤方法返回的名为namedtuple的执行结果，除了所有元素都已转换为张量数据。通常，它包含下一个观察值、奖励、完成状态、信息等。

返回:

当前时间步的已处理过渡数据。

返回类型:

• transition (Dict[str, torch.Tensor])

_reset_collect(data_id: List[int] | None = None) → None

概述：

重置收集器环境的观察和动作。

参数:

data_id (-) – 要重置的数据ID列表（在此实现中未使用）。

_reset_eval(data_id: List[int] | None = None) → None

概述：

重置评估环境的观察和动作。

参数:

data_id (-) – 要重置的数据ID列表（在此实现中未使用）。

_reset_learn(data_id: List[int] | None = None) → None

概述：

在必要时重置学习模式下的一些有状态变量，例如RNN的隐藏状态或某些特殊算法的状态库。如果 data_id 为None，则意味着重置所有有状态变量。否则，它将根据 data_id 重置有状态变量。例如， data_id 中的不同轨迹在RNN中将具有不同的隐藏状态。

参数:

data_id (-) – 数据的id，用于重置由 data_id 指定的状态变量。

备注

此方法不是必须实现的。子类可以根据需要重写此方法。

_set_attribute(name: str, value: Any) → None

概述：

为了控制策略属性的访问，我们对外暴露不同的模式，而不是直接使用策略实例。我们还提供了一种方法，用于在不同模式下设置策略的属性。新属性将被命名为 _{name}。

参数:

• name (-) – 属性的名称。

• value (-) – 属性的值。

_state_dict_collect() → Dict[str, Any]

概述：

返回收集模式的 state_dict，仅包括通常情况下的模型，这对于分布式训练场景中自动恢复收集器是必要的。

返回:

当前策略收集状态的字典，用于保存和恢复。

返回类型:

• state_dict (Dict[str, Any])

小技巧

并非所有场景都需要自动恢复收集器，有时，我们可以直接关闭崩溃的收集器并重新启动一个新的。

_state_dict_eval() → Dict[str, Any]

概述：

返回评估模式下的 state_dict，通常只包括模型，这对于分布式训练场景是必要的，以便自动恢复评估器。

返回:

当前策略评估状态的字典，用于保存和恢复。

返回类型:

• state_dict (Dict[str, Any])

小技巧

并非所有场景都需要自动恢复评估器，有时，我们可以直接关闭崩溃的评估器并重新启动一个新的评估器。

_state_dict_learn() → Dict[str, Any]

概述：

返回学习模式的 state_dict，通常包括模型、目标模型和优化器。

返回:

当前策略学习状态的字典，用于保存和恢复。

返回类型:

• state_dict (Dict[str, Any])

property cfg: EasyDict

class collect_function(forward, process_transition, get_train_sample, reset, get_attribute, set_attribute, state_dict, load_state_dict)

基类：tuple

_asdict()

返回一个新字典，该字典将字段名称映射到它们的值。

_field_defaults = {}

_fields = ('forward', 'process_transition', 'get_train_sample', 'reset', 'get_attribute', 'set_attribute', 'state_dict', 'load_state_dict')

classmethod _make(iterable)

从序列或可迭代对象创建一个新的 collect_function 对象

_replace(**kwds)

返回一个新的 collect_function 对象，用新值替换指定字段

count(value, /)

返回值出现的次数。

forward

字段编号 0 的别名

get_attribute

字段编号4的别名

get_train_sample

字段编号2的别名

index(value, start=0, stop=9223372036854775807, /)

返回值的第一个索引。

如果值不存在，则引发 ValueError。

load_state_dict

字段编号 7 的别名

process_transition

字段编号1的别名

reset

字段编号 3 的别名

set_attribute

字段编号 5 的别名

state_dict

字段编号 6 的别名

property collect_mode: collect_function

概述：

返回策略的收集模式的接口，该接口用于训练模型。这里我们使用 namedtuple 来定义不可变的接口，并限制策略在不同模式下的使用。此外，派生子类可以重写这些接口以自定义其自己的收集模式。

返回:

策略的收集模式的接口，它是一个 namedtuple，其不同字段的值是不同的内部方法。

返回类型:

• interfaces (Policy.collect_function)

示例

>>> policy = Policy(cfg, model)
>>> policy_collect = policy.collect_mode
>>> obs = env_manager.ready_obs
>>> inference_output = policy_collect.forward(obs)
>>> next_obs, rew, done, info = env_manager.step(inference_output.action)

config = {'action_type': 'fixed_action_space', 'augmentation': ['shift', 'intensity'], 'batch_size': 256, 'battle_mode': 'play_with_bot_mode', 'collector_env_num': 8, 'cuda': True, 'discount_factor': 0.997, 'env_type': 'not_board_games', 'eps': {'decay': 100000, 'end': 0.05, 'eps_greedy_exploration_in_collect': False, 'start': 1.0, 'type': 'linear'}, 'eval_offline': False, 'evaluator_env_num': 3, 'fixed_temperature_value': 0.25, 'game_segment_length': 200, 'grad_clip_value': 10, 'gray_scale': False, 'gumbel_algo': True, 'ignore_done': False, 'learning_rate': 0.2, 'manual_temperature_decay': False, 'max_num_considered_actions': 4, 'mcts_ctree': True, 'model': {'bias': True, 'categorical_distribution': True, 'continuous_action_space': False, 'discrete_action_encoding_type': 'one_hot', 'frame_stack_num': 1, 'image_channel': 1, 'model_type': 'conv', 'norm_type': 'BN', 'num_channels': 64, 'num_res_blocks': 1, 'observation_shape': (4, 96, 96), 'res_connection_in_dynamics': True, 'self_supervised_learning_loss': False, 'support_scale': 300}, 'momentum': 0.9, 'monitor_extra_statistics': True, 'multi_gpu': False, 'n_episode': 8, 'num_simulations': 50, 'num_unroll_steps': 5, 'optim_type': 'SGD', 'piecewise_decay_lr_scheduler': True, 'policy_loss_weight': 1, 'priority_prob_alpha': 0.6, 'priority_prob_beta': 0.4, 'random_collect_episode_num': 0, 'reanalyze_noise': True, 'replay_ratio': 0.25, 'reward_loss_weight': 1, 'root_dirichlet_alpha': 0.3, 'root_noise_weight': 0.25, 'sampled_algo': False, 'ssl_loss_weight': 0, 'target_update_freq': 100, 'td_steps': 5, 'threshold_training_steps_for_final_lr': 50000, 'threshold_training_steps_for_final_temperature': 100000, 'transform2string': False, 'update_per_collect': None, 'use_augmentation': False, 'use_priority': True, 'value_loss_weight': 0.25, 'weight_decay': 0.0001}

classmethod default_config() → EasyDict

概述：

获取策略的默认配置。此方法用于创建策略的默认配置。

返回:

对应策略的默认配置。对于派生策略类，它将递归合并基类的默认配置及其自身的默认配置。

返回类型:

• cfg (EasyDict)

小技巧

此方法将深度复制类的 config 属性并返回结果。因此，用户无需担心返回配置的修改。

default_model() → Tuple[str, List[str]]

概述：

返回此算法的默认模型设置以供演示。

返回:

模型名称和模型导入名称。

• model_type (str): The model type used in this algorithm, which is registered in ModelRegistry.

• import_names (List[str]): The model class path list used in this algorithm.

返回类型:

• model_info (Tuple[str, List[str]])

备注

用户可以定义和使用自定义的网络模型，但必须遵守由 import_names 路径指示的相同接口定义。对于 MuZero，使用 lzero.model.muzero_model.MuZeroModel。

class eval_function(forward, reset, get_attribute, set_attribute, state_dict, load_state_dict)

基类：tuple

_asdict()

返回一个新字典，该字典将字段名称映射到它们的值。

_field_defaults = {}

_fields = ('forward', 'reset', 'get_attribute', 'set_attribute', 'state_dict', 'load_state_dict')

classmethod _make(iterable)

从序列或可迭代对象创建一个新的 eval_function 对象

_replace(**kwds)

返回一个新的 eval_function 对象，用新值替换指定字段。

count(value, /)

返回值出现的次数。

forward

字段编号 0 的别名

get_attribute

字段编号2的别名

index(value, start=0, stop=9223372036854775807, /)

返回值的第一个索引。

如果值不存在，则引发 ValueError。

load_state_dict

字段编号 5 的别名

reset

字段编号1的别名

set_attribute

字段编号 3 的别名

state_dict

字段编号4的别名

property eval_mode: eval_function

概述：

返回策略的评估模式的接口，该接口用于训练模型。这里我们使用 namedtuple 来定义不可变的接口，并限制策略在不同模式下的使用。此外，派生子类可以重写这些接口以自定义其自己的评估模式。

返回:

策略的eval模式的接口，它是一个namedtuple，其不同字段的值是不同的内部方法。

返回类型:

• interfaces (Policy.eval_function)

示例

>>> policy = Policy(cfg, model)
>>> policy_eval = policy.eval_mode
>>> obs = env_manager.ready_obs
>>> inference_output = policy_eval.forward(obs)
>>> next_obs, rew, done, info = env_manager.step(inference_output.action)

class learn_function(forward, reset, info, monitor_vars, get_attribute, set_attribute, state_dict, load_state_dict)

基类：tuple

_asdict()

返回一个新字典，该字典将字段名称映射到它们的值。

_field_defaults = {}

_fields = ('forward', 'reset', 'info', 'monitor_vars', 'get_attribute', 'set_attribute', 'state_dict', 'load_state_dict')

classmethod _make(iterable)

从序列或可迭代对象创建一个新的 learn_function 对象

_replace(**kwds)

返回一个新的 learn_function 对象，用新值替换指定的字段

count(value, /)

返回值出现的次数。

forward

字段编号 0 的别名

get_attribute

字段编号4的别名

index(value, start=0, stop=9223372036854775807, /)

返回值的第一个索引。

如果值不存在，则引发 ValueError。

info

字段编号2的别名

load_state_dict

字段编号 7 的别名

monitor_vars

字段编号 3 的别名

reset

字段编号1的别名

set_attribute

字段编号 5 的别名

state_dict

字段编号 6 的别名

property learn_mode: learn_function

概述：

返回策略学习模式的接口，该接口用于训练模型。这里我们使用 namedtuple 来定义不可变的接口，并限制策略在不同模式下的使用。此外，派生子类可以重写这些接口以自定义其自己的学习模式。

返回:

策略的学习模式的接口，它是一个 namedtuple，其不同字段的值是不同的内部方法。

返回类型:

• interfaces (Policy.learn_function)

示例

>>> policy = Policy(cfg, model)
>>> policy_learn = policy.learn_mode
>>> train_output = policy_learn.forward(data)
>>> state_dict = policy_learn.state_dict()

set_train_iter_env_step(train_iter, env_step) → None

概述：

设置策略的 train_iter 和 env_step。

参数:

• train_iter (-) – 策略的 train_iter。

• env_step (-) – 策略的 env_step。

sync_gradients(model: Module) → None

概述：

在数据并行的多GPU训练中，同步（allreduce）模型参数的梯度。

参数:

model (-) – 用于同步梯度的模型。

备注

此方法仅在多GPU训练中使用，并且应在 backward 方法之后和 step 方法之前调用。用户还可以使用 bp_update_sync 配置来控制是否同步梯度allreduce和优化器更新。

total_field = {'collect', 'eval', 'learn'}

Sampled AlphaZero策略

class lzero.policy.sampled_alphazero.SampledAlphaZeroPolicy(cfg: EasyDict, model: Module | None = None, enable_field: List[str] | None = None)

基类：Policy

概述：

Sampled AlphaZero 的策略类。

__init__(cfg: EasyDict, model: Module | None = None, enable_field: List[str] | None = None) → None

概述：

根据输入配置和模型初始化策略实例。此方法将初始化策略中的不同字段，包括 learn、collect、eval。learn 字段用于训练策略，collect 字段用于收集训练数据，eval 字段用于评估策略。enable_field 用于指定要初始化的字段，如果为 None，则将初始化所有字段。

参数:

• cfg (-) – 用于初始化策略的最终合并配置。对于默认配置，请参阅策略类的 config 属性及其注释。

• model (-) – 用于初始化策略的神经网络模型。如果它是 None，则模型将根据 default_model 方法和 cfg.model 字段创建。否则，模型将被设置为由外部调用者创建的 model 实例。

• enable_field (-) – 要初始化的字段列表。如果为 None，则所有字段都将被初始化。否则，只有 enable_field 中的字段将被初始化，这有助于节省资源。

备注

对于派生的策略类，它应该实现 _init_learn、_init_collect、_init_eval 方法来初始化相应的字段。

_abc_impl = <_abc._abc_data object>

_calculate_policy_loss_disc(policy_probs: Tensor, target_policy: Tensor, target_sampled_actions: Tensor, valid_action_lengths: Tensor) → Tensor

_create_model(cfg: EasyDict, model: Module | None = None) → Module

概述：

根据输入配置和模型创建或验证神经网络模型。如果输入模型为 None，则将根据 default_model 方法和 cfg.model 字段创建模型。否则，模型将被验证为 torch.nn.Module 的实例，并设置为由外部调用者创建的 model 实例。

参数:

• cfg (-) – 用于初始化策略的最终合并配置。

• model (-) – 用于初始化策略的神经网络模型。用户可以参考相应策略中定义的默认模型来自定义自己的模型。

返回:

创建的神经网络模型。策略的不同模式将向模型添加不同的包装器和插件，这些包装器和插件用于训练、收集和评估。

返回类型:

• model (torch.nn.Module)

抛出:

- RuntimeError – 如果输入模型不是 None 且不是 torch.nn.Module 的实例。

_forward_collect(obs: Dict, temperature: float = 1) → Dict[str, Tensor]

概述：

在收集模式下收集数据的前向函数。使用真实环境执行MCTS搜索。

参数:

• obs (-) – obs 的字典，键是 env_id，值是该时间步对应的 obs。

• temperature (-) – MCTS搜索的温度。

返回:

输出字典，键是 env_id，值是该时间步对应的策略输出，包括动作、概率等。

返回类型:

• output (Dict[str, torch.Tensor])

_forward_eval(obs: Dict) → Dict[str, Tensor]

概述：

用于在评估模式下评估当前策略的前向函数，类似于 self._forward_collect。

参数:

obs (-) – obs 的字典，键是 env_id，值是该时间步对应的 obs。

返回:

输出字典，键是 env_id，值是该时间步对应的策略输出，包括动作、概率等。

返回类型:

• output (Dict[str, torch.Tensor])

_forward_learn(inputs: Dict[str, Tensor]) → Dict[str, float]

概述：

学习模式下的策略前向函数（训练策略和更新参数）。前向意味着策略从回放缓冲区输入一些训练批次数据，然后返回输出结果，包括各种训练信息，如损失值、策略熵、q值、优先级等。此方法留给子类实现，如果需要，可以在``data``项中添加更多参数。

参数:

data (-) – 用于策略前向传播的输入数据，包括一批训练样本。对于列表中的每个元素，字典的键是数据项的名称，值是对应的数据。通常，在 _forward_learn 方法中，数据应通过一些实用函数（如 default_preprocess_learn）在批次维度上堆叠。

返回:

策略前向传播的训练信息，包括一些用于监控训练的指标，如损失、优先级、Q值、策略熵，以及一些用于下一步训练的数据，如优先级。注意输出数据项应该是Python原生标量，而不是PyTorch张量，这样对外部使用更方便。

返回类型:

• output (Dict[int, Any])

_get_attribute(name: str) → Any

概述：

为了控制策略属性的访问，我们对外暴露不同的模式，而不是直接使用策略实例。我们还提供了一种方法来获取不同模式下的策略属性。

参数:

name (-) – 属性的名称。

返回:

属性的值。

返回类型:

• value (Any)

备注

DI-engine 的策略将首先尝试访问 _get_{name} 方法，然后尝试访问 _{name} 属性。如果两者都未找到，它将引发一个 NotImplementedError。

_get_batch_size() → int | Dict[str, int]

_get_n_episode() → int | None

_get_n_sample() → int | None

_get_simulation_env()

_get_train_sample(data)

概述：

对于给定的轨迹（过渡，过渡列表）数据，将其处理为可以直接用于训练的样本列表。一个训练样本可以是一个经过处理的过渡（DQN 使用 n步 TD）或者一些多时间步的过渡（DRQN）。这个方法通常在收集器中使用，以在训练前执行必要的强化学习数据预处理，这可以帮助学习者分摊相关的时间消耗。此外，你也可以将这个方法实现为一个恒等函数，并在 self._forward_learn 方法中进行数据处理。

参数:

transitions (-) – 轨迹数据（一个过渡列表），每个元素的格式与 self._process_transition 方法的返回值相同。

返回:

处理后的训练样本，每个元素的格式与输入转换相似，但可能包含更多用于训练的数据，如n步奖励、优势等。

返回类型:

• samples (List[Dict[str, Any]])

备注

我们将在下一个发布版本中向量化 process_transition 和 get_train_sample 方法。用户可以通过重写这两个方法和收集器本身来自定义此数据处理过程。

_init_collect() → None

概述：

收集模式初始化方法。由 self.__init__ 调用。初始化收集模型和MCTS工具。

_init_eval() → None

概述：

评估模式初始化方法。由 self.__init__ 调用。初始化评估模型和MCTS工具。

_init_learn() → None

概述：

初始化策略的学习模式，包括相关属性和模块。如果 learn 字段在 enable_field 中，此方法将在 __init__ 方法中被调用。几乎不同的策略都有其自己的学习模式，因此此方法必须在子类中被重写。

备注

对于需要保存和加载的成员变量，请参考 _state_dict_learn 和 _load_state_dict_learn 方法。

备注

对于需要监控的成员变量，请参考 _monitor_vars_learn 方法。

备注

如果你想在 _init_learn 方法中设置一些特殊的成员变量，最好用前缀 _learn_ 来命名它们，以避免与其他模式发生冲突，例如 self._learn_attr1。

_init_multi_gpu_setting(model: Module, bp_update_sync: bool) → None

概述：

初始化多GPU数据并行训练设置，包括在训练开始时广播模型参数，并为所有模型参数梯度准备hook函数以进行allreduce操作。

参数:

• model (-) – 要训练的神经网络模型。

• bp_update_sync (-) – 是否在所有reduce模型参数的梯度后同步更新模型参数。异步更新可以在不同的网络层中并行进行，如流水线，从而节省时间。

_load_state_dict_collect(state_dict: Dict[str, Any]) → None

概述：

将 state_dict 变量加载到策略收集模式中，例如加载预训练的 state_dict、自动恢复检查点，或在分布式训练场景中从学习者复制模型。

参数:

state_dict (-) – 之前保存的策略收集状态的字典。

小技巧

如果你想只加载模型的一部分，你可以简单地将 load_state_dict 中的 strict 参数设置为 False，或者参考 ding.torch_utils.checkpoint_helper 以进行更复杂的操作。

_load_state_dict_eval(state_dict: Dict[str, Any]) → None

概述：

将 state_dict 变量加载到策略评估模式中，例如加载自动恢复检查点，或在分布式训练场景中从学习者加载模型副本。

参数:

state_dict (-) – 保存之前的策略评估状态的字典。

小技巧

如果你想只加载模型的一部分，你可以简单地将 load_state_dict 中的 strict 参数设置为 False，或者参考 ding.torch_utils.checkpoint_helper 以进行更复杂的操作。

_load_state_dict_learn(state_dict: Dict[str, Any]) → None

概述：

将 state_dict 变量加载到策略学习模式中。

参数:

state_dict (-) – 之前保存的策略学习状态的字典。

小技巧

如果你想只加载模型的一部分，你可以简单地将 load_state_dict 中的 strict 参数设置为 False，或者参考 ding.torch_utils.checkpoint_helper 以进行更复杂的操作。

_monitor_vars_learn() → List[str]

概述：

在学习模式中注册要监控的变量。根据返回值 _forward_learn，注册的变量将被记录在 tensorboard 中。

_policy_value_func(environment: Environment) → Tuple[Dict[int, ndarray], float]

_process_transition(obs: Dict, model_output: Dict[str, Tensor], timestep: namedtuple) → Dict

概述：

从策略学习中生成字典类型转换（一个时间步长）数据。

_reset_collect(data_id: List[int] | None = None) → None

概述：

在必要时重置收集模式下的一些有状态变量，例如RNN的隐藏状态或某些特殊算法的内存库。如果``data_id``为None，则意味着重置所有有状态变量。否则，它将根据``data_id``重置有状态变量。例如，在``data_id``中收集的不同环境/情节在RNN中将具有不同的隐藏状态。

参数:

data_id (-) – 数据的id，用于重置由 data_id 指定的状态变量。

备注

此方法不是必须实现的。子类可以根据需要重写此方法。

_reset_eval(data_id: List[int] | None = None) → None

概述：

在必要时重置一些有状态变量以适应评估模式，例如RNN的隐藏状态或某些特殊算法的内存库。如果 data_id 为None，则意味着重置所有有状态变量。否则，它将根据 data_id 重置有状态变量。例如，在 data_id 中的不同环境/情节在评估中将具有RNN中的不同隐藏状态。

参数:

data_id (-) – 数据的id，用于重置由 data_id 指定的状态变量。

备注

此方法不是必须实现的。子类可以根据需要重写此方法。

_reset_learn(data_id: List[int] | None = None) → None

概述：

在必要时重置学习模式下的一些有状态变量，例如RNN的隐藏状态或某些特殊算法的状态库。如果 data_id 为None，则意味着重置所有有状态变量。否则，它将根据 data_id 重置有状态变量。例如， data_id 中的不同轨迹在RNN中将具有不同的隐藏状态。

参数:

data_id (-) – 数据的id，用于重置由 data_id 指定的状态变量。

备注

此方法不是必须实现的。子类可以根据需要重写此方法。

_set_attribute(name: str, value: Any) → None

概述：

为了控制策略属性的访问，我们对外暴露不同的模式，而不是直接使用策略实例。我们还提供了一种方法，用于在不同模式下设置策略的属性。新属性将被命名为 _{name}。

参数:

• name (-) – 属性的名称。

• value (-) – 属性的值。

_state_dict_collect() → Dict[str, Any]

概述：

返回收集模式的 state_dict，仅包括通常情况下的模型，这对于分布式训练场景中自动恢复收集器是必要的。

返回:

当前策略收集状态的字典，用于保存和恢复。

返回类型:

• state_dict (Dict[str, Any])

小技巧

并非所有场景都需要自动恢复收集器，有时，我们可以直接关闭崩溃的收集器并重新启动一个新的。

_state_dict_eval() → Dict[str, Any]

概述：

返回评估模式下的 state_dict，通常只包括模型，这对于分布式训练场景是必要的，以便自动恢复评估器。

返回:

当前策略评估状态的字典，用于保存和恢复。

返回类型:

• state_dict (Dict[str, Any])

小技巧

并非所有场景都需要自动恢复评估器，有时，我们可以直接关闭崩溃的评估器并重新启动一个新的评估器。

_state_dict_learn() → Dict[str, Any]

概述：

返回学习模式的 state_dict，通常包括模型和优化器。

返回:

当前策略学习状态的字典，用于保存和恢复。

返回类型:

• state_dict (Dict[str, Any])

property cfg: EasyDict

class collect_function(forward, process_transition, get_train_sample, reset, get_attribute, set_attribute, state_dict, load_state_dict)

基类：tuple

_asdict()

返回一个新字典，该字典将字段名称映射到它们的值。

_field_defaults = {}

_fields = ('forward', 'process_transition', 'get_train_sample', 'reset', 'get_attribute', 'set_attribute', 'state_dict', 'load_state_dict')

classmethod _make(iterable)

从序列或可迭代对象创建一个新的 collect_function 对象

_replace(**kwds)

返回一个新的 collect_function 对象，用新值替换指定字段

count(value, /)

返回值出现的次数。

forward

字段编号 0 的别名

get_attribute

字段编号4的别名

get_train_sample

字段编号2的别名

index(value, start=0, stop=9223372036854775807, /)

返回值的第一个索引。

如果值不存在，则引发 ValueError。

load_state_dict

字段编号 7 的别名

process_transition

字段编号1的别名

reset

字段编号 3 的别名

set_attribute

字段编号 5 的别名

state_dict

字段编号 6 的别名

property collect_mode: collect_function

概述：

返回策略的收集模式的接口，该接口用于训练模型。这里我们使用 namedtuple 来定义不可变的接口，并限制策略在不同模式下的使用。此外，派生子类可以重写这些接口以自定义其自己的收集模式。

返回:

策略的收集模式的接口，它是一个 namedtuple，其不同字段的值是不同的内部方法。

返回类型:

• interfaces (Policy.collect_function)

示例

>>> policy = Policy(cfg, model)
>>> policy_collect = policy.collect_mode
>>> obs = env_manager.ready_obs
>>> inference_output = policy_collect.forward(obs)
>>> next_obs, rew, done, info = env_manager.step(inference_output.action)

config = {'batch_size': 256, 'collector_env_num': 8, 'cuda': False, 'evaluator_env_num': 3, 'fixed_temperature_value': 0.25, 'grad_clip_value': 10, 'learning_rate': 0.2, 'manual_temperature_decay': False, 'mcts': {'action_space_size': 9, 'continuous_action_space': False, 'legal_actions': None, 'max_moves': 512, 'num_of_sampled_actions': 2, 'num_simulations': 50, 'pb_c_base': 19652, 'pb_c_init': 1.25, 'root_dirichlet_alpha': 0.3, 'root_noise_weight': 0.25}, 'mcts_ctree': True, 'model': {'num_channels': 32, 'num_res_blocks': 1, 'observation_shape': (3, 6, 6)}, 'momentum': 0.9, 'normalize_prob_of_sampled_actions': False, 'optim_type': 'SGD', 'other': {'replay_buffer': {'replay_buffer_size': 1000000, 'save_episode': False}}, 'piecewise_decay_lr_scheduler': True, 'policy_loss_type': 'cross_entropy', 'replay_ratio': 0.25, 'sampled_algo': False, 'tensor_float_32': False, 'threshold_training_steps_for_final_lr': 500000, 'threshold_training_steps_for_final_temperature': 100000, 'torch_compile': False, 'type': 'alphazero', 'update_per_collect': None, 'value_weight': 1.0, 'weight_decay': 0.0001}

classmethod default_config() → EasyDict

概述：

获取策略的默认配置。此方法用于创建策略的默认配置。

返回:

对应策略的默认配置。对于派生策略类，它将递归合并基类的默认配置及其自身的默认配置。

返回类型:

• cfg (EasyDict)

小技巧

此方法将深度复制类的 config 属性并返回结果。因此，用户无需担心返回配置的修改。

default_model() → Tuple[str, List[str]]

概述：

返回此算法的默认模型设置以供演示。

返回:

The model type used in this algorithm, which is registered in ModelRegistry. - import_names (List[str]): The model class path list used in this algorithm.

返回类型:

• model_type (str)

class eval_function(forward, reset, get_attribute, set_attribute, state_dict, load_state_dict)

基类：tuple

_asdict()

返回一个新字典，该字典将字段名称映射到它们的值。

_field_defaults = {}

_fields = ('forward', 'reset', 'get_attribute', 'set_attribute', 'state_dict', 'load_state_dict')

classmethod _make(iterable)

从序列或可迭代对象创建一个新的 eval_function 对象

_replace(**kwds)

返回一个新的 eval_function 对象，用新值替换指定字段。

count(value, /)

返回值出现的次数。

forward

字段编号 0 的别名

get_attribute

字段编号2的别名

index(value, start=0, stop=9223372036854775807, /)

返回值的第一个索引。

如果值不存在，则引发 ValueError。

load_state_dict

字段编号 5 的别名

reset

字段编号1的别名

set_attribute

字段编号 3 的别名

state_dict

字段编号4的别名

property eval_mode: eval_function

概述：

返回策略的评估模式的接口，该接口用于训练模型。这里我们使用 namedtuple 来定义不可变的接口，并限制策略在不同模式下的使用。此外，派生子类可以重写这些接口以自定义其自己的评估模式。

返回:

策略的eval模式的接口，它是一个namedtuple，其不同字段的值是不同的内部方法。

返回类型:

• interfaces (Policy.eval_function)

示例

>>> policy = Policy(cfg, model)
>>> policy_eval = policy.eval_mode
>>> obs = env_manager.ready_obs
>>> inference_output = policy_eval.forward(obs)
>>> next_obs, rew, done, info = env_manager.step(inference_output.action)

class learn_function(forward, reset, info, monitor_vars, get_attribute, set_attribute, state_dict, load_state_dict)

基类：tuple

_asdict()

返回一个新字典，该字典将字段名称映射到它们的值。

_field_defaults = {}

_fields = ('forward', 'reset', 'info', 'monitor_vars', 'get_attribute', 'set_attribute', 'state_dict', 'load_state_dict')

classmethod _make(iterable)

从序列或可迭代对象创建一个新的 learn_function 对象

_replace(**kwds)

返回一个新的 learn_function 对象，用新值替换指定的字段

count(value, /)

返回值出现的次数。

forward

字段编号 0 的别名

get_attribute

字段编号4的别名

index(value, start=0, stop=9223372036854775807, /)

返回值的第一个索引。

如果值不存在，则引发 ValueError。

info

字段编号2的别名

load_state_dict

字段编号 7 的别名

monitor_vars

字段编号 3 的别名

reset

字段编号1的别名

set_attribute

字段编号 5 的别名

state_dict

字段编号 6 的别名

property learn_mode: learn_function

概述：

返回策略学习模式的接口，该接口用于训练模型。这里我们使用 namedtuple 来定义不可变的接口，并限制策略在不同模式下的使用。此外，派生子类可以重写这些接口以自定义其自己的学习模式。

返回:

策略的学习模式的接口，它是一个 namedtuple，其不同字段的值是不同的内部方法。

返回类型:

• interfaces (Policy.learn_function)

示例

>>> policy = Policy(cfg, model)
>>> policy_learn = policy.learn_mode
>>> train_output = policy_learn.forward(data)
>>> state_dict = policy_learn.state_dict()

sync_gradients(model: Module) → None

概述：

在数据并行的多GPU训练中，同步（allreduce）模型参数的梯度。

参数:

model (-) – 用于同步梯度的模型。

备注

此方法仅在多GPU训练中使用，并且应在 backward 方法之后和 step 方法之前调用。用户还可以使用 bp_update_sync 配置来控制是否同步梯度allreduce和优化器更新。

total_field = {'collect', 'eval', 'learn'}

Sampled MuZeroPolicy

class lzero.policy.sampled_muzero.SampledMuZeroPolicy(cfg: EasyDict, model: Module | None = None, enable_field: List[str] | None = None)

基类：MuZeroPolicy

概述：

论文 https://arxiv.org/abs/2104.06303 中提出的 Sampled MuZero 的策略类。

__init__(cfg: EasyDict, model: Module | None = None, enable_field: List[str] | None = None) → None

概述：

根据输入配置和模型初始化策略实例。此方法将初始化策略中的不同字段，包括 learn、collect、eval。learn 字段用于训练策略，collect 字段用于收集训练数据，eval 字段用于评估策略。enable_field 用于指定要初始化的字段，如果为 None，则将初始化所有字段。

参数:

• cfg (-) – 用于初始化策略的最终合并配置。对于默认配置，请参阅策略类的 config 属性及其注释。

• model (-) – 用于初始化策略的神经网络模型。如果它是 None，则模型将根据 default_model 方法和 cfg.model 字段创建。否则，模型将被设置为由外部调用者创建的 model 实例。

• enable_field (-) – 要初始化的字段列表。如果为 None，则所有字段都将被初始化。否则，只有 enable_field 中的字段将被初始化，这有助于节省资源。

备注

对于派生的策略类，它应该实现 _init_learn、_init_collect、_init_eval 方法来初始化相应的字段。

_abc_impl = <_abc._abc_data object>

_calculate_policy_loss_cont(policy_loss: Tensor, policy_logits: Tensor, target_policy: Tensor, mask_batch: Tensor, child_sampled_actions_batch: Tensor, unroll_step: int) → Tuple[Tensor]

概述：

计算连续动作空间的策略损失。

参数:

• policy_loss (-) – 策略损失张量。

• policy_logits (-) – 策略对数张量。

• target_policy (-) – 目标策略张量。

• mask_batch (-) – 掩码张量。

• child_sampled_actions_batch (-) – 子样本动作张量。

• unroll_step (-) – 展开步骤。

返回:

The policy loss tensor. - policy_entropy (torch.Tensor): The policy entropy tensor. - policy_entropy_loss (torch.Tensor): The policy entropy loss tensor. - target_policy_entropy (torch.Tensor): The target policy entropy tensor. - target_sampled_actions (torch.Tensor): The target sampled actions tensor. - mu (torch.Tensor): The mu tensor. - sigma (torch.Tensor): The sigma tensor.

返回类型:

• policy_loss (torch.Tensor)

_calculate_policy_loss_disc(policy_loss: Tensor, policy_logits: Tensor, target_policy: Tensor, mask_batch: Tensor, child_sampled_actions_batch: Tensor, unroll_step: int) → Tuple[Tensor]

概述：

计算离散动作空间的策略损失。

参数:

• policy_loss (-) – 策略损失张量。

• policy_logits (-) – 策略对数张量。

• target_policy (-) – 目标策略张量。

• mask_batch (-) – 掩码张量。

• child_sampled_actions_batch (-) – 子样本动作张量。

• unroll_step (-) – 展开步骤。

返回:

The policy loss tensor. - policy_entropy (torch.Tensor): The policy entropy tensor. - policy_entropy_loss (torch.Tensor): The policy entropy loss tensor. - target_policy_entropy (torch.Tensor): The target policy entropy tensor. - target_sampled_actions (torch.Tensor): The target sampled actions tensor.

返回类型:

• policy_loss (torch.Tensor)

_create_model(cfg: EasyDict, model: Module | None = None) → Module

概述：

根据输入配置和模型创建或验证神经网络模型。如果输入模型为 None，则将根据 default_model 方法和 cfg.model 字段创建模型。否则，模型将被验证为 torch.nn.Module 的实例，并设置为由外部调用者创建的 model 实例。

参数:

• cfg (-) – 用于初始化策略的最终合并配置。

• model (-) – 用于初始化策略的神经网络模型。用户可以参考相应策略中定义的默认模型来自定义自己的模型。

返回:

创建的神经网络模型。策略的不同模式将向模型添加不同的包装器和插件，这些包装器和插件用于训练、收集和评估。

返回类型:

• model (torch.nn.Module)

抛出:

- RuntimeError – 如果输入模型不是 None 且不是 torch.nn.Module 的实例。

_forward_collect(data: Tensor, action_mask: list = None, temperature: ndarray = 1, to_play=-1, epsilon: float = 0.25, ready_env_id: array = None)

概述：

在收集模式下收集数据的正向函数。使用模型执行MCTS搜索。在收集模式期间通过采样选择动作。

参数:

• data (-) – 输入数据，即观察结果。

• action_mask (-) – 动作掩码，即不能被选择的动作。

• temperature (-) – 政策的温度。

• to_play (-) – 要播放的玩家。

• ready_env_id (-) – 准备好收集的环境的id。

形状：

• data (torch.Tensor):

  • 对于Atari，\((N, C*S, H, W)\)，其中N是collect_env的数量，C是通道的数量，S是堆叠帧的数量，H是图像的高度，W是图像的宽度。

  • 对于 lunarlander，\((N, O)\)，其中 N 是 collect_env 的数量，O 是观察空间的大小。

• action_mask: \((N, action_space_size)\)，其中 N 是 collect_env 的数量。

• 温度: \((1, )\).

• to_play: \((N, 1)\)，其中 N 是 collect_env 的数量。

• ready_env_id: None

返回:

字典类型数据，包含的键有 action、distributions、visit_count_distribution_entropy、value、pred_value、policy_logits。

返回类型:

• output (Dict[int, Any])

_forward_eval(data: Tensor, action_mask: list, to_play: -1, ready_env_id: array = None)

概述：

用于评估当前策略在评估模式下的前向函数。使用模型执行MCTS搜索。在评估模式下选择值最高的动作（argmax）而不是采样。

参数:

• data (-) – 输入数据，即观察结果。

• action_mask (-) – 动作掩码，即不能被选择的动作。

• to_play (-) – 要播放的玩家。

• ready_env_id (-) – 准备好收集的环境的id。

形状：

• data (torch.Tensor):

  • 对于Atari，\((N, C*S, H, W)\)，其中N是collect_env的数量，C是通道的数量，S是堆叠帧的数量，H是图像的高度，W是图像的宽度。

  • 对于 lunarlander，\((N, O)\)，其中 N 是 collect_env 的数量，O 是观察空间的大小。

• action_mask: \((N, action_space_size)\)，其中 N 是 collect_env 的数量。

• to_play: \((N, 1)\)，其中 N 是 collect_env 的数量。

• ready_env_id: None

返回:

字典类型数据，包含的键有 action、distributions、visit_count_distribution_entropy、value、pred_value、policy_logits。

返回类型:

• output (Dict[int, Any])

_forward_learn(data: Tensor) → Dict[str, float | int]

概述：

学习模式下用于学习策略的前向函数，这是学习过程的核心。数据从回放缓冲区中采样。通过损失函数计算损失，并将损失反向传播以更新模型。

参数:

data (-) – 从回放缓冲区中采样的数据，这是一个张量元组。第一个张量是当前批次，第二个张量是目标批次。

返回:

要记录的信息字典，其中包含当前的学习损失和学习统计数据。

返回类型:

• info_dict (Dict[str, Union[float, int]])

_get_attribute(name: str) → Any

概述：

为了控制策略属性的访问，我们对外暴露不同的模式，而不是直接使用策略实例。我们还提供了一种方法来获取不同模式下的策略属性。

参数:

name (-) – 属性的名称。

返回:

属性的值。

返回类型:

• value (Any)

备注

DI-engine 的策略将首先尝试访问 _get_{name} 方法，然后尝试访问 _{name} 属性。如果两者都未找到，它将引发一个 NotImplementedError。

_get_batch_size() → int | Dict[str, int]

_get_n_episode() → int | None

_get_n_sample() → int | None

_get_target_obs_index_in_step_k(step)

概述：

获取步骤 k 中目标 obs 的起始索引和结束索引。

参数:

step (-) – 当前步骤 k。

返回:

The begin index of the target obs in step k. - end_index (int): The end index of the target obs in step k.

返回类型:

• beg_index (int)

示例

>>> self._cfg.model.model_type = 'conv'
>>> self._cfg.model.image_channel = 3
>>> self._cfg.model.frame_stack_num = 4
>>> self._get_target_obs_index_in_step_k(0)
>>> (0, 12)

_get_train_sample(data)

概述：

对于给定的轨迹（过渡，过渡列表）数据，将其处理为可以直接用于训练的样本列表。一个训练样本可以是一个经过处理的过渡（DQN 使用 n步 TD）或者一些多时间步的过渡（DRQN）。这个方法通常在收集器中使用，以在训练前执行必要的强化学习数据预处理，这可以帮助学习者分摊相关的时间消耗。此外，你也可以将这个方法实现为一个恒等函数，并在 self._forward_learn 方法中进行数据处理。

参数:

transitions (-) – 轨迹数据（一个过渡列表），每个元素的格式与 self._process_transition 方法的返回值相同。

返回:

处理后的训练样本，每个元素的格式与输入转换相似，但可能包含更多用于训练的数据，如n步奖励、优势等。

返回类型:

• samples (List[Dict[str, Any]])

备注

我们将在下一个发布版本中向量化 process_transition 和 get_train_sample 方法。用户可以通过重写这两个方法和收集器本身来自定义此数据处理过程。

_init_collect() → None

概述：

收集模式初始化方法。由 self.__init__ 调用。初始化收集模型和MCTS工具。

_init_eval() → None

概述：

评估模式初始化方法。由 self.__init__ 调用。初始化评估模型和MCTS工具。

_init_learn() → None

概述：

学习模式初始化方法。由 self.__init__ 调用。初始化学习模型、优化器和MCTS工具。

_init_multi_gpu_setting(model: Module, bp_update_sync: bool) → None

概述：

初始化多GPU数据并行训练设置，包括在训练开始时广播模型参数，并为所有模型参数梯度准备hook函数以进行allreduce操作。

参数:

• model (-) – 要训练的神经网络模型。

• bp_update_sync (-) – 是否在所有reduce模型参数的梯度后同步更新模型参数。异步更新可以在不同的网络层中并行进行，如流水线，从而节省时间。

_load_state_dict_collect(state_dict: Dict[str, Any]) → None

概述：

将 state_dict 变量加载到策略收集模式中，例如加载预训练的 state_dict、自动恢复检查点，或在分布式训练场景中从学习者复制模型。

参数:

state_dict (-) – 之前保存的策略收集状态的字典。

小技巧

如果你想只加载模型的一部分，你可以简单地将 load_state_dict 中的 strict 参数设置为 False，或者参考 ding.torch_utils.checkpoint_helper 以进行更复杂的操作。

_load_state_dict_eval(state_dict: Dict[str, Any]) → None

概述：

将 state_dict 变量加载到策略评估模式中，例如加载自动恢复检查点，或在分布式训练场景中从学习者加载模型副本。

参数:

state_dict (-) – 保存之前的策略评估状态的字典。

小技巧

如果你想只加载模型的一部分，你可以简单地将 load_state_dict 中的 strict 参数设置为 False，或者参考 ding.torch_utils.checkpoint_helper 以进行更复杂的操作。

_load_state_dict_learn(state_dict: Dict[str, Any]) → None

概述：

将 state_dict 变量加载到策略学习模式中。

参数:

state_dict (-) – 之前保存的策略学习状态的字典。

_monitor_vars_learn() → List[str]

概述：

在学习模式中注册要监控的变量。根据返回值 _forward_learn，注册的变量将被记录在 tensorboard 中。

_process_transition(obs, policy_output, timestep)

概述：

将一个时间步的过渡数据处理并打包成一个字典，例如 <s, a, r, s’, done>。一些策略需要进行一些特殊处理并打包其自身的必要属性（例如隐藏状态和logit），因此此方法留给子类实现。

参数:

• obs (-) – 当前时间步的观察。

• policy_output (-) – 以观察为输入的策略网络的输出。通常，它包含动作和动作的对数概率。

• timestep (-) – 环境步骤方法返回的名为namedtuple的执行结果，除了所有元素都已转换为张量数据。通常，它包含下一个观察值、奖励、完成状态、信息等。

返回:

当前时间步的已处理过渡数据。

返回类型:

• transition (Dict[str, torch.Tensor])

_reset_collect(data_id: List[int] | None = None) → None

概述：

重置收集器环境的观察和动作。

参数:

data_id (-) – 要重置的数据ID列表（在此实现中未使用）。

_reset_eval(data_id: List[int] | None = None) → None

概述：

重置评估环境的观察和动作。

参数:

data_id (-) – 要重置的数据ID列表（在此实现中未使用）。

_reset_learn(data_id: List[int] | None = None) → None

概述：

在必要时重置学习模式下的一些有状态变量，例如RNN的隐藏状态或某些特殊算法的状态库。如果 data_id 为None，则意味着重置所有有状态变量。否则，它将根据 data_id 重置有状态变量。例如， data_id 中的不同轨迹在RNN中将具有不同的隐藏状态。

参数:

data_id (-) – 数据的id，用于重置由 data_id 指定的状态变量。

备注

此方法不是必须实现的。子类可以根据需要重写此方法。

_set_attribute(name: str, value: Any) → None

概述：

为了控制策略属性的访问，我们对外暴露不同的模式，而不是直接使用策略实例。我们还提供了一种方法，用于在不同模式下设置策略的属性。新属性将被命名为 _{name}。

参数:

• name (-) – 属性的名称。

• value (-) – 属性的值。

_state_dict_collect() → Dict[str, Any]

概述：

返回收集模式的 state_dict，仅包括通常情况下的模型，这对于分布式训练场景中自动恢复收集器是必要的。

返回:

当前策略收集状态的字典，用于保存和恢复。

返回类型:

• state_dict (Dict[str, Any])

小技巧

并非所有场景都需要自动恢复收集器，有时，我们可以直接关闭崩溃的收集器并重新启动一个新的。

_state_dict_eval() → Dict[str, Any]

概述：

返回评估模式下的 state_dict，通常只包括模型，这对于分布式训练场景是必要的，以便自动恢复评估器。

返回:

当前策略评估状态的字典，用于保存和恢复。

返回类型:

• state_dict (Dict[str, Any])

小技巧

并非所有场景都需要自动恢复评估器，有时，我们可以直接关闭崩溃的评估器并重新启动一个新的评估器。

_state_dict_learn() → Dict[str, Any]

概述：

返回学习模式的 state_dict，通常包括模型、目标模型和优化器。

返回:

当前策略学习状态的字典，用于保存和恢复。

返回类型:

• state_dict (Dict[str, Any])

property cfg: EasyDict

class collect_function(forward, process_transition, get_train_sample, reset, get_attribute, set_attribute, state_dict, load_state_dict)

基类：tuple

_asdict()

返回一个新字典，该字典将字段名称映射到它们的值。

_field_defaults = {}

_fields = ('forward', 'process_transition', 'get_train_sample', 'reset', 'get_attribute', 'set_attribute', 'state_dict', 'load_state_dict')

classmethod _make(iterable)

从序列或可迭代对象创建一个新的 collect_function 对象

_replace(**kwds)

返回一个新的 collect_function 对象，用新值替换指定字段

count(value, /)

返回值出现的次数。

forward

字段编号 0 的别名

get_attribute

字段编号4的别名

get_train_sample

字段编号2的别名

index(value, start=0, stop=9223372036854775807, /)

返回值的第一个索引。

如果值不存在，则引发 ValueError。

load_state_dict

字段编号 7 的别名

process_transition

字段编号1的别名

reset

字段编号 3 的别名

set_attribute

字段编号 5 的别名

state_dict

字段编号 6 的别名

property collect_mode: collect_function

概述：

返回策略的收集模式的接口，该接口用于训练模型。这里我们使用 namedtuple 来定义不可变的接口，并限制策略在不同模式下的使用。此外，派生子类可以重写这些接口以自定义其自己的收集模式。

返回:

策略的收集模式的接口，它是一个 namedtuple，其不同字段的值是不同的内部方法。

返回类型:

• interfaces (Policy.collect_function)

示例

>>> policy = Policy(cfg, model)
>>> policy_collect = policy.collect_mode
>>> obs = env_manager.ready_obs
>>> inference_output = policy_collect.forward(obs)
>>> next_obs, rew, done, info = env_manager.step(inference_output.action)

config = {'action_type': 'fixed_action_space', 'augmentation': ['shift', 'intensity'], 'batch_size': 256, 'battle_mode': 'play_with_bot_mode', 'collector_env_num': 8, 'cos_lr_scheduler': False, 'cuda': True, 'discount_factor': 0.997, 'env_type': 'not_board_games', 'eps': {'decay': 100000, 'end': 0.05, 'eps_greedy_exploration_in_collect': False, 'start': 1.0, 'type': 'linear'}, 'eval_offline': False, 'evaluator_env_num': 3, 'fixed_temperature_value': 0.25, 'game_segment_length': 200, 'grad_clip_value': 10, 'gray_scale': False, 'gumbel_algo': False, 'ignore_done': False, 'init_w': 0.003, 'learning_rate': 0.0001, 'lstm_horizon_len': 5, 'manual_temperature_decay': False, 'mcts_ctree': True, 'model': {'action_space_size': 6, 'bias': True, 'categorical_distribution': True, 'continuous_action_space': False, 'discrete_action_encoding_type': 'one_hot', 'fixed_sigma_value': 0.3, 'frame_stack_num': 1, 'image_channel': 1, 'lstm_hidden_size': 512, 'model_type': 'conv', 'norm_type': 'LN', 'num_res_blocks': 1, 'observation_shape': (4, 96, 96), 'res_connection_in_dynamics': True, 'self_supervised_learning_loss': True, 'sigma_type': 'conditioned', 'support_scale': 300}, 'momentum': 0.9, 'monitor_extra_statistics': True, 'multi_gpu': False, 'n_episode': 8, 'normalize_prob_of_sampled_actions': False, 'num_simulations': 50, 'num_unroll_steps': 5, 'optim_type': 'AdamW', 'piecewise_decay_lr_scheduler': False, 'policy_entropy_weight': 0.005, 'policy_loss_type': 'cross_entropy', 'policy_loss_weight': 1, 'priority_prob_alpha': 0.6, 'priority_prob_beta': 0.4, 'random_collect_episode_num': 0, 'reanalyze_noise': True, 'replay_ratio': 0.25, 'reward_loss_weight': 1, 'root_dirichlet_alpha': 0.3, 'root_noise_weight': 0.25, 'sampled_algo': True, 'ssl_loss_weight': 2, 'target_update_freq': 100, 'td_steps': 5, 'threshold_training_steps_for_final_lr': 50000, 'threshold_training_steps_for_final_temperature': 100000, 'transform2string': False, 'update_per_collect': None, 'use_augmentation': False, 'use_priority': False, 'use_ture_chance_label_in_chance_encoder': False, 'value_loss_weight': 0.25, 'weight_decay': 0.0001}

classmethod default_config() → EasyDict

概述：

获取策略的默认配置。此方法用于创建策略的默认配置。

返回:

对应策略的默认配置。对于派生策略类，它将递归合并基类的默认配置及其自身的默认配置。

返回类型:

• cfg (EasyDict)

小技巧

此方法将深度复制类的 config 属性并返回结果。因此，用户无需担心返回配置的修改。

default_model() → Tuple[str, List[str]]

概述：

返回此算法的默认模型设置。

返回:

模型名称和模型导入名称。

• model_type (str): The model type used in this algorithm, which is registered in ModelRegistry.

• import_names (List[str]): The model class path list used in this algorithm.

返回类型:

• model_info (Tuple[str, List[str]])

备注

用户可以定义和使用自定义的网络模型，但必须遵守由 import_names 路径指示的相同接口定义。对于 Sampled MuZero，使用 lzero.model.sampled_muzero_model.SampledMuZeroModel。

class eval_function(forward, reset, get_attribute, set_attribute, state_dict, load_state_dict)

基类：tuple

_asdict()

返回一个新字典，该字典将字段名称映射到它们的值。

_field_defaults = {}

_fields = ('forward', 'reset', 'get_attribute', 'set_attribute', 'state_dict', 'load_state_dict')

classmethod _make(iterable)

从序列或可迭代对象创建一个新的 eval_function 对象

_replace(**kwds)

返回一个新的 eval_function 对象，用新值替换指定字段。

count(value, /)

返回值出现的次数。

forward

字段编号 0 的别名

get_attribute

字段编号2的别名

index(value, start=0, stop=9223372036854775807, /)

返回值的第一个索引。

如果值不存在，则引发 ValueError。

load_state_dict

字段编号 5 的别名

reset

字段编号1的别名

set_attribute

字段编号 3 的别名

state_dict

字段编号4的别名

property eval_mode: eval_function

概述：

返回策略的评估模式的接口，该接口用于训练模型。这里我们使用 namedtuple 来定义不可变的接口，并限制策略在不同模式下的使用。此外，派生子类可以重写这些接口以自定义其自己的评估模式。

返回:

策略的eval模式的接口，它是一个namedtuple，其不同字段的值是不同的内部方法。

返回类型:

• interfaces (Policy.eval_function)

示例

>>> policy = Policy(cfg, model)
>>> policy_eval = policy.eval_mode
>>> obs = env_manager.ready_obs
>>> inference_output = policy_eval.forward(obs)
>>> next_obs, rew, done, info = env_manager.step(inference_output.action)

class learn_function(forward, reset, info, monitor_vars, get_attribute, set_attribute, state_dict, load_state_dict)

基类：tuple

_asdict()

返回一个新字典，该字典将字段名称映射到它们的值。

_field_defaults = {}

_fields = ('forward', 'reset', 'info', 'monitor_vars', 'get_attribute', 'set_attribute', 'state_dict', 'load_state_dict')

classmethod _make(iterable)

从序列或可迭代对象创建一个新的 learn_function 对象

_replace(**kwds)

返回一个新的 learn_function 对象，用新值替换指定的字段

count(value, /)

返回值出现的次数。

forward

字段编号 0 的别名

get_attribute

字段编号4的别名

index(value, start=0, stop=9223372036854775807, /)

返回值的第一个索引。

如果值不存在，则引发 ValueError。

info

字段编号2的别名

load_state_dict

字段编号 7 的别名

monitor_vars

字段编号 3 的别名

reset

字段编号1的别名

set_attribute

字段编号 5 的别名

state_dict

字段编号 6 的别名

property learn_mode: learn_function

概述：

返回策略学习模式的接口，该接口用于训练模型。这里我们使用 namedtuple 来定义不可变的接口，并限制策略在不同模式下的使用。此外，派生子类可以重写这些接口以自定义其自己的学习模式。

返回:

策略的学习模式的接口，它是一个 namedtuple，其不同字段的值是不同的内部方法。

返回类型:

• interfaces (Policy.learn_function)

示例

>>> policy = Policy(cfg, model)
>>> policy_learn = policy.learn_mode
>>> train_output = policy_learn.forward(data)
>>> state_dict = policy_learn.state_dict()

set_train_iter_env_step(train_iter, env_step) → None

概述：

设置策略的 train_iter 和 env_step。

参数:

• train_iter (-) – 策略的 train_iter。

• env_step (-) – 策略的 env_step。

sync_gradients(model: Module) → None

概述：

在数据并行的多GPU训练中，同步（allreduce）模型参数的梯度。

参数:

model (-) – 用于同步梯度的模型。

备注

此方法仅在多GPU训练中使用，并且应在 backward 方法之后和 step 方法之前调用。用户还可以使用 bp_update_sync 配置来控制是否同步梯度allreduce和优化器更新。

total_field = {'collect', 'eval', 'learn'}

Sampled EfficientZeroPolicy

class lzero.policy.sampled_efficientzero.SampledEfficientZeroPolicy(cfg: EasyDict, model: Module | None = None, enable_field: List[str] | None = None)

基类：MuZeroPolicy

概述：

Sampled EfficientZero 论文中提出的策略类，论文地址为 https://arxiv.org/abs/2104.06303。

__init__(cfg: EasyDict, model: Module | None = None, enable_field: List[str] | None = None) → None

概述：

根据输入配置和模型初始化策略实例。此方法将初始化策略中的不同字段，包括 learn、collect、eval。learn 字段用于训练策略，collect 字段用于收集训练数据，eval 字段用于评估策略。enable_field 用于指定要初始化的字段，如果为 None，则将初始化所有字段。

参数:

• cfg (-) – 用于初始化策略的最终合并配置。对于默认配置，请参阅策略类的 config 属性及其注释。

• model (-) – 用于初始化策略的神经网络模型。如果它是 None，则模型将根据 default_model 方法和 cfg.model 字段创建。否则，模型将被设置为由外部调用者创建的 model 实例。

• enable_field (-) – 要初始化的字段列表。如果为 None，则所有字段都将被初始化。否则，只有 enable_field 中的字段将被初始化，这有助于节省资源。

备注

对于派生的策略类，它应该实现 _init_learn、_init_collect、_init_eval 方法来初始化相应的字段。

_abc_impl = <_abc._abc_data object>

_calculate_policy_loss_cont(policy_loss: Tensor, policy_logits: Tensor, target_policy: Tensor, mask_batch: Tensor, child_sampled_actions_batch: Tensor, unroll_step: int) → Tuple[Tensor]

概述：

计算连续动作空间的策略损失。

参数:

• policy_loss (-) – 策略损失张量。

• policy_logits (-) – 策略对数张量。

• target_policy (-) – 目标策略张量。

• mask_batch (-) – 掩码张量。

• child_sampled_actions_batch (-) – 子样本动作张量。

• unroll_step (-) – 展开步骤。

返回:

The policy loss tensor. - policy_entropy (torch.Tensor): The policy entropy tensor. - policy_entropy_loss (torch.Tensor): The policy entropy loss tensor. - target_policy_entropy (torch.Tensor): The target policy entropy tensor. - target_sampled_actions (torch.Tensor): The target sampled actions tensor. - mu (torch.Tensor): The mu tensor. - sigma (torch.Tensor): The sigma tensor.

返回类型:

• policy_loss (torch.Tensor)

_calculate_policy_loss_disc(policy_loss: Tensor, policy_logits: Tensor, target_policy: Tensor, mask_batch: Tensor, child_sampled_actions_batch: Tensor, unroll_step: int) → Tuple[Tensor]

概述：

计算离散动作空间的策略损失。

参数:

• policy_loss (-) – 策略损失张量。

• policy_logits (-) – 策略对数张量。

• target_policy (-) – 目标策略张量。

• mask_batch (-) – 掩码张量。

• child_sampled_actions_batch (-) – 子样本动作张量。

• unroll_step (-) – 展开步骤。

返回:

The policy loss tensor. - policy_entropy (torch.Tensor): The policy entropy tensor. - policy_entropy_loss (torch.Tensor): The policy entropy loss tensor. - target_policy_entropy (torch.Tensor): The target policy entropy tensor. - target_sampled_actions (torch.Tensor): The target sampled actions tensor.

返回类型:

• policy_loss (torch.Tensor)

_create_model(cfg: EasyDict, model: Module | None = None) → Module

概述：

根据输入配置和模型创建或验证神经网络模型。如果输入模型为 None，则将根据 default_model 方法和 cfg.model 字段创建模型。否则，模型将被验证为 torch.nn.Module 的实例，并设置为由外部调用者创建的 model 实例。

参数:

• cfg (-) – 用于初始化策略的最终合并配置。

• model (-) – 用于初始化策略的神经网络模型。用户可以参考相应策略中定义的默认模型来自定义自己的模型。

返回:

创建的神经网络模型。策略的不同模式将向模型添加不同的包装器和插件，这些包装器和插件用于训练、收集和评估。

返回类型:

• model (torch.nn.Module)

抛出:

- RuntimeError – 如果输入模型不是 None 且不是 torch.nn.Module 的实例。

_forward_collect(data: Tensor, action_mask: list = None, temperature: ndarray = 1, to_play=-1, epsilon: float = 0.25, ready_env_id: array = None)

概述：

在收集模式下收集数据的正向函数。使用模型执行MCTS搜索。在收集模式期间通过采样选择动作。

参数:

• data (-) – 输入数据，即观察结果。

• action_mask (-) – 动作掩码，即不能被选择的动作。

• temperature (-) – 政策的温度。

• to_play (-) – 要播放的玩家。

• ready_env_id (-) – 准备好收集的环境的id。

形状：

• data (torch.Tensor):

  • 对于Atari，\((N, C*S, H, W)\)，其中N是collect_env的数量，C是通道的数量，S是堆叠帧的数量，H是图像的高度，W是图像的宽度。

  • 对于 lunarlander，\((N, O)\)，其中 N 是 collect_env 的数量，O 是观察空间的大小。

• action_mask: \((N, action_space_size)\)，其中 N 是 collect_env 的数量。

• 温度: \((1, )\).

• to_play: \((N, 1)\)，其中 N 是 collect_env 的数量。

• ready_env_id: None

返回:

字典类型数据，包含的键有 action、distributions、visit_count_distribution_entropy、value、pred_value、policy_logits。

返回类型:

• output (Dict[int, Any])

_forward_eval(data: Tensor, action_mask: list, to_play: -1, ready_env_id: array = None)

概述：

用于评估当前策略在评估模式下的前向函数。使用模型执行MCTS搜索。在评估模式下选择值最高的动作（argmax）而不是采样。

参数:

• data (-) – 输入数据，即观察结果。

• action_mask (-) – 动作掩码，即不能被选择的动作。

• to_play (-) – 要播放的玩家。

• ready_env_id (-) – 准备好收集的环境的id。

形状：

• data (torch.Tensor):

  • 对于Atari，\((N, C*S, H, W)\)，其中N是collect_env的数量，C是通道的数量，S是堆叠帧的数量，H是图像的高度，W是图像的宽度。

  • 对于 lunarlander，\((N, O)\)，其中 N 是 collect_env 的数量，O 是观察空间的大小。

• action_mask: \((N, action_space_size)\)，其中 N 是 collect_env 的数量。

• to_play: \((N, 1)\)，其中 N 是 collect_env 的数量。

• ready_env_id: None

返回:

字典类型数据，包含的键有 action、distributions、visit_count_distribution_entropy、value、pred_value、policy_logits。

返回类型:

• output (Dict[int, Any])

_forward_learn(data: Tensor) → Dict[str, float | int]

概述：

学习模式下用于学习策略的前向函数，这是学习过程的核心。数据从回放缓冲区中采样。通过损失函数计算损失，并将损失反向传播以更新模型。

参数:

data (-) – 从回放缓冲区中采样的数据，这是一个张量元组。第一个张量是当前批次，第二个张量是目标批次。

返回:

要记录的信息字典，其中包含当前的学习损失和学习统计数据。

返回类型:

• info_dict (Dict[str, Union[float, int]])

_get_attribute(name: str) → Any

概述：

为了控制策略属性的访问，我们对外暴露不同的模式，而不是直接使用策略实例。我们还提供了一种方法来获取不同模式下的策略属性。

参数:

name (-) – 属性的名称。

返回:

属性的值。

返回类型:

• value (Any)

备注

DI-engine 的策略将首先尝试访问 _get_{name} 方法，然后尝试访问 _{name} 属性。如果两者都未找到，它将引发一个 NotImplementedError。

_get_batch_size() → int | Dict[str, int]

_get_n_episode() → int | None

_get_n_sample() → int | None

_get_target_obs_index_in_step_k(step)

概述：

获取步骤 k 中目标 obs 的起始索引和结束索引。

参数:

step (-) – 当前步骤 k。

返回:

The begin index of the target obs in step k. - end_index (int): The end index of the target obs in step k.

返回类型:

• beg_index (int)

示例

>>> self._cfg.model.model_type = 'conv'
>>> self._cfg.model.image_channel = 3
>>> self._cfg.model.frame_stack_num = 4
>>> self._get_target_obs_index_in_step_k(0)
>>> (0, 12)

_get_train_sample(data)

概述：

对于给定的轨迹（过渡，过渡列表）数据，将其处理为可以直接用于训练的样本列表。一个训练样本可以是一个经过处理的过渡（DQN 使用 n步 TD）或者一些多时间步的过渡（DRQN）。这个方法通常在收集器中使用，以在训练前执行必要的强化学习数据预处理，这可以帮助学习者分摊相关的时间消耗。此外，你也可以将这个方法实现为一个恒等函数，并在 self._forward_learn 方法中进行数据处理。

参数:

transitions (-) – 轨迹数据（一个过渡列表），每个元素的格式与 self._process_transition 方法的返回值相同。

返回:

处理后的训练样本，每个元素的格式与输入转换相似，但可能包含更多用于训练的数据，如n步奖励、优势等。

返回类型:

• samples (List[Dict[str, Any]])

备注

我们将在下一个发布版本中向量化 process_transition 和 get_train_sample 方法。用户可以通过重写这两个方法和收集器本身来自定义此数据处理过程。

_init_collect() → None

概述：

收集模式初始化方法。由 self.__init__ 调用。初始化收集模型和MCTS工具。

_init_eval() → None

概述：

评估模式初始化方法。由 self.__init__ 调用。初始化评估模型和MCTS工具。

_init_learn() → None

概述：

学习模式初始化方法。由 self.__init__ 调用。初始化学习模型、优化器和MCTS工具。

_init_multi_gpu_setting(model: Module, bp_update_sync: bool) → None

概述：

初始化多GPU数据并行训练设置，包括在训练开始时广播模型参数，并为所有模型参数梯度准备hook函数以进行allreduce操作。

参数:

• model (-) – 要训练的神经网络模型。

• bp_update_sync (-) – 是否在所有reduce模型参数的梯度后同步更新模型参数。异步更新可以在不同的网络层中并行进行，如流水线，从而节省时间。

_load_state_dict_collect(state_dict: Dict[str, Any]) → None

概述：

将 state_dict 变量加载到策略收集模式中，例如加载预训练的 state_dict、自动恢复检查点，或在分布式训练场景中从学习者复制模型。

参数:

state_dict (-) – 之前保存的策略收集状态的字典。

小技巧

如果你想只加载模型的一部分，你可以简单地将 load_state_dict 中的 strict 参数设置为 False，或者参考 ding.torch_utils.checkpoint_helper 以进行更复杂的操作。

_load_state_dict_eval(state_dict: Dict[str, Any]) → None

概述：

将 state_dict 变量加载到策略评估模式中，例如加载自动恢复检查点，或在分布式训练场景中从学习者加载模型副本。

参数:

state_dict (-) – 保存之前的策略评估状态的字典。

小技巧

如果你想只加载模型的一部分，你可以简单地将 load_state_dict 中的 strict 参数设置为 False，或者参考 ding.torch_utils.checkpoint_helper 以进行更复杂的操作。

_load_state_dict_learn(state_dict: Dict[str, Any]) → None

概述：

将 state_dict 变量加载到策略学习模式中。

参数:

state_dict (-) – 之前保存的策略学习状态的字典。

_monitor_vars_learn() → List[str]

概述：

在学习模式中注册要监控的变量。根据返回值 _forward_learn，注册的变量将被记录在 tensorboard 中。

_process_transition(obs, policy_output, timestep)

概述：

将一个时间步的过渡数据处理并打包成一个字典，例如 <s, a, r, s’, done>。一些策略需要进行一些特殊处理并打包其自身的必要属性（例如隐藏状态和logit），因此此方法留给子类实现。

参数:

• obs (-) – 当前时间步的观察。

• policy_output (-) – 以观察为输入的策略网络的输出。通常，它包含动作和动作的对数概率。

• timestep (-) – 环境步骤方法返回的名为namedtuple的执行结果，除了所有元素都已转换为张量数据。通常，它包含下一个观察值、奖励、完成状态、信息等。

返回:

当前时间步的已处理过渡数据。

返回类型:

• transition (Dict[str, torch.Tensor])

_reset_collect(data_id: List[int] | None = None) → None

概述：

重置收集器环境的观察和动作。

参数:

data_id (-) – 要重置的数据ID列表（在此实现中未使用）。

_reset_eval(data_id: List[int] | None = None) → None

概述：

重置评估环境的观察和动作。

参数:

data_id (-) – 要重置的数据ID列表（在此实现中未使用）。

_reset_learn(data_id: List[int] | None = None) → None

概述：

在必要时重置学习模式下的一些有状态变量，例如RNN的隐藏状态或某些特殊算法的状态库。如果 data_id 为None，则意味着重置所有有状态变量。否则，它将根据 data_id 重置有状态变量。例如， data_id 中的不同轨迹在RNN中将具有不同的隐藏状态。

参数:

data_id (-) – 数据的id，用于重置由 data_id 指定的状态变量。

备注

此方法不是必须实现的。子类可以根据需要重写此方法。

_set_attribute(name: str, value: Any) → None

概述：

为了控制策略属性的访问，我们对外暴露不同的模式，而不是直接使用策略实例。我们还提供了一种方法，用于在不同模式下设置策略的属性。新属性将被命名为 _{name}。

参数:

• name (-) – 属性的名称。

• value (-) – 属性的值。

_state_dict_collect() → Dict[str, Any]

概述：

返回收集模式的 state_dict，仅包括通常情况下的模型，这对于分布式训练场景中自动恢复收集器是必要的。

返回:

当前策略收集状态的字典，用于保存和恢复。

返回类型:

• state_dict (Dict[str, Any])

小技巧

并非所有场景都需要自动恢复收集器，有时，我们可以直接关闭崩溃的收集器并重新启动一个新的。

_state_dict_eval() → Dict[str, Any]

概述：

返回评估模式下的 state_dict，通常只包括模型，这对于分布式训练场景是必要的，以便自动恢复评估器。

返回:

当前策略评估状态的字典，用于保存和恢复。

返回类型:

• state_dict (Dict[str, Any])

小技巧

并非所有场景都需要自动恢复评估器，有时，我们可以直接关闭崩溃的评估器并重新启动一个新的评估器。

_state_dict_learn() → Dict[str, Any]

概述：

返回学习模式的 state_dict，通常包括模型和优化器。

返回:

当前策略学习状态的字典，用于保存和恢复。

返回类型:

• state_dict (Dict[str, Any])

property cfg: EasyDict

class collect_function(forward, process_transition, get_train_sample, reset, get_attribute, set_attribute, state_dict, load_state_dict)

基类：tuple

_asdict()

返回一个新字典，该字典将字段名称映射到它们的值。

_field_defaults = {}

_fields = ('forward', 'process_transition', 'get_train_sample', 'reset', 'get_attribute', 'set_attribute', 'state_dict', 'load_state_dict')

classmethod _make(iterable)

从序列或可迭代对象创建一个新的 collect_function 对象

_replace(**kwds)

返回一个新的 collect_function 对象，用新值替换指定字段

count(value, /)

返回值出现的次数。

forward

字段编号 0 的别名

get_attribute

字段编号4的别名

get_train_sample

字段编号2的别名

index(value, start=0, stop=9223372036854775807, /)

返回值的第一个索引。

如果值不存在，则引发 ValueError。

load_state_dict

字段编号 7 的别名

process_transition

字段编号1的别名

reset

字段编号 3 的别名

set_attribute

字段编号 5 的别名

state_dict

字段编号 6 的别名

property collect_mode: collect_function

概述：

返回策略的收集模式的接口，该接口用于训练模型。这里我们使用 namedtuple 来定义不可变的接口，并限制策略在不同模式下的使用。此外，派生子类可以重写这些接口以自定义其自己的收集模式。

返回:

策略的收集模式的接口，它是一个 namedtuple，其不同字段的值是不同的内部方法。

返回类型:

• interfaces (Policy.collect_function)

示例

>>> policy = Policy(cfg, model)
>>> policy_collect = policy.collect_mode
>>> obs = env_manager.ready_obs
>>> inference_output = policy_collect.forward(obs)
>>> next_obs, rew, done, info = env_manager.step(inference_output.action)

config = {'action_type': 'fixed_action_space', 'augmentation': ['shift', 'intensity'], 'batch_size': 256, 'battle_mode': 'play_with_bot_mode', 'collector_env_num': 8, 'cos_lr_scheduler': False, 'cuda': True, 'discount_factor': 0.997, 'env_type': 'not_board_games', 'eps': {'decay': 100000, 'end': 0.05, 'eps_greedy_exploration_in_collect': False, 'start': 1.0, 'type': 'linear'}, 'eval_offline': False, 'evaluator_env_num': 3, 'fixed_temperature_value': 0.25, 'game_segment_length': 200, 'grad_clip_value': 10, 'gray_scale': False, 'gumbel_algo': False, 'ignore_done': False, 'init_w': 0.003, 'learning_rate': 0.0001, 'lstm_horizon_len': 5, 'manual_temperature_decay': False, 'mcts_ctree': True, 'model': {'action_space_size': 6, 'bias': True, 'categorical_distribution': True, 'continuous_action_space': False, 'discrete_action_encoding_type': 'one_hot', 'fixed_sigma_value': 0.3, 'frame_stack_num': 1, 'image_channel': 1, 'lstm_hidden_size': 512, 'model_type': 'conv', 'norm_type': 'LN', 'num_res_blocks': 1, 'observation_shape': (4, 96, 96), 'res_connection_in_dynamics': True, 'self_supervised_learning_loss': True, 'sigma_type': 'conditioned', 'support_scale': 300}, 'momentum': 0.9, 'monitor_extra_statistics': True, 'multi_gpu': False, 'n_episode': 8, 'normalize_prob_of_sampled_actions': False, 'num_simulations': 50, 'num_unroll_steps': 5, 'optim_type': 'AdamW', 'piecewise_decay_lr_scheduler': False, 'policy_entropy_weight': 0.005, 'policy_loss_type': 'cross_entropy', 'policy_loss_weight': 1, 'priority_prob_alpha': 0.6, 'priority_prob_beta': 0.4, 'random_collect_episode_num': 0, 'reanalyze_noise': True, 'replay_ratio': 0.25, 'reward_loss_weight': 1, 'root_dirichlet_alpha': 0.3, 'root_noise_weight': 0.25, 'sampled_algo': True, 'ssl_loss_weight': 2, 'target_update_freq': 100, 'td_steps': 5, 'threshold_training_steps_for_final_lr': 50000, 'threshold_training_steps_for_final_temperature': 100000, 'transform2string': False, 'update_per_collect': None, 'use_augmentation': False, 'use_priority': True, 'use_ture_chance_label_in_chance_encoder': False, 'value_loss_weight': 0.25, 'weight_decay': 0.0001}

classmethod default_config() → EasyDict

概述：

获取策略的默认配置。此方法用于创建策略的默认配置。

返回:

对应策略的默认配置。对于派生策略类，它将递归合并基类的默认配置及其自身的默认配置。

返回类型:

• cfg (EasyDict)

小技巧

此方法将深度复制类的 config 属性并返回结果。因此，用户无需担心返回配置的修改。

default_model() → Tuple[str, List[str]]

概述：

返回此算法的默认模型设置。

返回:

模型名称和模型导入名称。

• model_type (str): The model type used in this algorithm, which is registered in ModelRegistry.

• import_names (List[str]): The model class path list used in this algorithm.

返回类型:

• model_info (Tuple[str, List[str]])

备注

用户可以定义和使用自定义的网络模型，但必须遵守由 import_names 路径指示的相同接口定义。对于 Sampled EfficientZero，使用 lzero.model.sampled_efficientzero_model.SampledEfficientZeroModel。

class eval_function(forward, reset, get_attribute, set_attribute, state_dict, load_state_dict)

基类：tuple

_asdict()

返回一个新字典，该字典将字段名称映射到它们的值。

_field_defaults = {}

_fields = ('forward', 'reset', 'get_attribute', 'set_attribute', 'state_dict', 'load_state_dict')

classmethod _make(iterable)

从序列或可迭代对象创建一个新的 eval_function 对象

_replace(**kwds)

返回一个新的 eval_function 对象，用新值替换指定字段。

count(value, /)

返回值出现的次数。

forward

字段编号 0 的别名

get_attribute

字段编号2的别名

index(value, start=0, stop=9223372036854775807, /)

返回值的第一个索引。

如果值不存在，则引发 ValueError。

load_state_dict

字段编号 5 的别名

reset

字段编号1的别名

set_attribute

字段编号 3 的别名

state_dict

字段编号4的别名

property eval_mode: eval_function

概述：

返回策略的评估模式的接口，该接口用于训练模型。这里我们使用 namedtuple 来定义不可变的接口，并限制策略在不同模式下的使用。此外，派生子类可以重写这些接口以自定义其自己的评估模式。

返回:

策略的eval模式的接口，它是一个namedtuple，其不同字段的值是不同的内部方法。

返回类型:

• interfaces (Policy.eval_function)

示例

>>> policy = Policy(cfg, model)
>>> policy_eval = policy.eval_mode
>>> obs = env_manager.ready_obs
>>> inference_output = policy_eval.forward(obs)
>>> next_obs, rew, done, info = env_manager.step(inference_output.action)

class learn_function(forward, reset, info, monitor_vars, get_attribute, set_attribute, state_dict, load_state_dict)

基类：tuple

_asdict()

返回一个新字典，该字典将字段名称映射到它们的值。

_field_defaults = {}

_fields = ('forward', 'reset', 'info', 'monitor_vars', 'get_attribute', 'set_attribute', 'state_dict', 'load_state_dict')

classmethod _make(iterable)

从序列或可迭代对象创建一个新的 learn_function 对象

_replace(**kwds)

返回一个新的 learn_function 对象，用新值替换指定的字段

count(value, /)

返回值出现的次数。

forward

字段编号 0 的别名

get_attribute

字段编号4的别名

index(value, start=0, stop=9223372036854775807, /)

返回值的第一个索引。

如果值不存在，则引发 ValueError。

info

字段编号2的别名

load_state_dict

字段编号 7 的别名

monitor_vars

字段编号 3 的别名

reset

字段编号1的别名

set_attribute

字段编号 5 的别名

state_dict

字段编号 6 的别名

property learn_mode: learn_function

概述：

返回策略学习模式的接口，该接口用于训练模型。这里我们使用 namedtuple 来定义不可变的接口，并限制策略在不同模式下的使用。此外，派生子类可以重写这些接口以自定义其自己的学习模式。

返回:

策略的学习模式的接口，它是一个 namedtuple，其不同字段的值是不同的内部方法。

返回类型:

• interfaces (Policy.learn_function)

示例

>>> policy = Policy(cfg, model)
>>> policy_learn = policy.learn_mode
>>> train_output = policy_learn.forward(data)
>>> state_dict = policy_learn.state_dict()

set_train_iter_env_step(train_iter, env_step) → None

概述：

设置策略的 train_iter 和 env_step。

参数:

• train_iter (-) – 策略的 train_iter。

• env_step (-) – 策略的 env_step。

sync_gradients(model: Module) → None

概述：

在数据并行的多GPU训练中，同步（allreduce）模型参数的梯度。

参数:

model (-) – 用于同步梯度的模型。

备注

此方法仅在多GPU训练中使用，并且应在 backward 方法之后和 step 方法之前调用。用户还可以使用 bp_update_sync 配置来控制是否同步梯度allreduce和优化器更新。

total_field = {'collect', 'eval', 'learn'}

Stochastic MuZeroPolicy

class lzero.policy.stochastic_muzero.StochasticMuZeroPolicy(cfg: EasyDict, model: Module | None = None, enable_field: List[str] | None = None)

基类：MuZeroPolicy

概述：

论文 https://openreview.net/pdf?id=X6D9bAHhBQ1 中提出的 Stochastic MuZero 的策略类。

__init__(cfg: EasyDict, model: Module | None = None, enable_field: List[str] | None = None) → None

概述：

根据输入配置和模型初始化策略实例。此方法将初始化策略中的不同字段，包括 learn、collect、eval。learn 字段用于训练策略，collect 字段用于收集训练数据，eval 字段用于评估策略。enable_field 用于指定要初始化的字段，如果为 None，则将初始化所有字段。

参数:

• cfg (-) – 用于初始化策略的最终合并配置。对于默认配置，请参阅策略类的 config 属性及其注释。

• model (-) – 用于初始化策略的神经网络模型。如果它是 None，则模型将根据 default_model 方法和 cfg.model 字段创建。否则，模型将被设置为由外部调用者创建的 model 实例。

• enable_field (-) – 要初始化的字段列表。如果为 None，则所有字段都将被初始化。否则，只有 enable_field 中的字段将被初始化，这有助于节省资源。

备注

对于派生的策略类，它应该实现 _init_learn、_init_collect、_init_eval 方法来初始化相应的字段。

_abc_impl = <_abc._abc_data object>

_create_model(cfg: EasyDict, model: Module | None = None) → Module

概述：

根据输入配置和模型创建或验证神经网络模型。如果输入模型为 None，则将根据 default_model 方法和 cfg.model 字段创建模型。否则，模型将被验证为 torch.nn.Module 的实例，并设置为由外部调用者创建的 model 实例。

参数:

• cfg (-) – 用于初始化策略的最终合并配置。

• model (-) – 用于初始化策略的神经网络模型。用户可以参考相应策略中定义的默认模型来自定义自己的模型。

返回:

创建的神经网络模型。策略的不同模式将向模型添加不同的包装器和插件，这些包装器和插件用于训练、收集和评估。

返回类型:

• model (torch.nn.Module)

抛出:

- RuntimeError – 如果输入模型不是 None 且不是 torch.nn.Module 的实例。

_forward_collect(data: Tensor, action_mask: list = None, temperature: float = 1, to_play: List = [-1], epsilon: float = 0.25, ready_env_id: array = None) → Dict

概述：

在收集模式下收集数据的前向函数。使用模型执行MCTS搜索。在收集模式期间通过采样选择动作。

参数:

• data (-) – 输入数据，即观察结果。

• action_mask (-) – 动作掩码，即不能被选择的动作。

• temperature (-) – 政策的温度。

• to_play (-) – 要播放的玩家。

• ready_env_id (-) – 准备好收集的环境的id。

形状：

• data (torch.Tensor):

  • 对于Atari，其形状为 \((N, C*S, H, W)\)，其中N是collect_env的数量，C是通道的数量，S是堆叠帧的数量，H是图像的高度，W是图像的宽度。

  • 对于 lunarlander，其形状为 \((N, O)\)，其中 N 是 collect_env 的数量，O 是观测空间的大小。

• action_mask: \((N, action_space_size)\)，其中 N 是 collect_env 的数量。

• 温度: \((1, )\).

• to_play: \((N, 1)\)，其中 N 是 collect_env 的数量。

• ready_env_id: None

返回:

字典类型数据，包含的键有 action、distributions、visit_count_distribution_entropy、value、pred_value、policy_logits。

返回类型:

• output (Dict[int, Any])

_forward_eval(data: Tensor, action_mask: list, to_play: int = -1, ready_env_id: array = None) → Dict

概述：

用于在评估模式下评估当前策略的前向函数。使用模型执行MCTS搜索。在评估模式下选择值最高的动作（argmax），而不是进行采样。

参数:

• data (-) – 输入数据，即观察结果。

• action_mask (-) – 动作掩码，即不能被选择的动作。

• to_play (-) – 要播放的玩家。

• ready_env_id (-) – 准备好收集的环境的id。

形状：

• data (torch.Tensor):

  • 对于Atari，\((N, C*S, H, W)\)，其中N是collect_env的数量，C是通道的数量，S是堆叠帧的数量，H是图像的高度，W是图像的宽度。

  • 对于 lunarlander，\((N, O)\)，其中 N 是 collect_env 的数量，O 是观察空间的大小。

• action_mask: \((N, action_space_size)\)，其中 N 是 collect_env 的数量。

• to_play: \((N, 1)\)，其中 N 是 collect_env 的数量。

• ready_env_id: None

返回:

字典类型数据，包含的键有 action、distributions、visit_count_distribution_entropy、value、pred_value、policy_logits。

返回类型:

• output (Dict[int, Any])

_forward_learn(data: Tuple[Tensor]) → Dict[str, float | int]

概述：

学习模式下用于学习策略的前向函数，这是学习过程的核心。数据从回放缓冲区中采样。通过损失函数计算损失，并将损失反向传播以更新模型。

参数:

data (-) – 从回放缓冲区中采样的数据，这是一个张量元组。第一个张量是 current_batch，第二个张量是 target_batch。

返回:

要记录的信息字典，其中包含当前的学习损失和学习统计数据。

返回类型:

• info_dict (Dict[str, Union[float, int]])

_get_attribute(name: str) → Any

概述：

为了控制策略属性的访问，我们对外暴露不同的模式，而不是直接使用策略实例。我们还提供了一种方法来获取不同模式下的策略属性。

参数:

name (-) – 属性的名称。

返回:

属性的值。

返回类型:

• value (Any)

备注

DI-engine 的策略将首先尝试访问 _get_{name} 方法，然后尝试访问 _{name} 属性。如果两者都未找到，它将引发一个 NotImplementedError。

_get_batch_size() → int | Dict[str, int]

_get_n_episode() → int | None

_get_n_sample() → int | None

_get_target_obs_index_in_step_k(step)

概述：

获取步骤 k 中目标 obs 的起始索引和结束索引。

参数:

step (-) – 当前步骤 k。

返回:

The begin index of the target obs in step k. - end_index (int): The end index of the target obs in step k.

返回类型:

• beg_index (int)

示例

>>> self._cfg.model.model_type = 'conv'
>>> self._cfg.model.image_channel = 3
>>> self._cfg.model.frame_stack_num = 4
>>> self._get_target_obs_index_in_step_k(0)
>>> (0, 12)

_get_train_sample(data)

概述：

对于给定的轨迹（过渡，过渡列表）数据，将其处理为可以直接用于训练的样本列表。一个训练样本可以是一个经过处理的过渡（DQN 使用 n步 TD）或者一些多时间步的过渡（DRQN）。这个方法通常在收集器中使用，以在训练前执行必要的强化学习数据预处理，这可以帮助学习者分摊相关的时间消耗。此外，你也可以将这个方法实现为一个恒等函数，并在 self._forward_learn 方法中进行数据处理。

参数:

transitions (-) – 轨迹数据（一个过渡列表），每个元素的格式与 self._process_transition 方法的返回值相同。

返回:

处理后的训练样本，每个元素的格式与输入转换相似，但可能包含更多用于训练的数据，如n步奖励、优势等。

返回类型:

• samples (List[Dict[str, Any]])

备注

我们将在下一个发布版本中向量化 process_transition 和 get_train_sample 方法。用户可以通过重写这两个方法和收集器本身来自定义此数据处理过程。

_init_collect() → None

概述：

收集模式初始化方法。由 self.__init__ 调用。初始化收集模型和MCTS工具。

_init_eval() → None

概述：

评估模式初始化方法。由 self.__init__ 调用。初始化评估模型和MCTS工具。

_init_learn() → None

概述：

学习模式初始化方法。由 self.__init__ 调用。初始化学习模型、优化器和MCTS工具。

_init_multi_gpu_setting(model: Module, bp_update_sync: bool) → None

概述：

初始化多GPU数据并行训练设置，包括在训练开始时广播模型参数，并为所有模型参数梯度准备hook函数以进行allreduce操作。

参数:

• model (-) – 要训练的神经网络模型。

• bp_update_sync (-) – 是否在所有reduce模型参数的梯度后同步更新模型参数。异步更新可以在不同的网络层中并行进行，如流水线，从而节省时间。

_load_state_dict_collect(state_dict: Dict[str, Any]) → None

概述：

将 state_dict 变量加载到策略收集模式中，例如加载预训练的 state_dict、自动恢复检查点，或在分布式训练场景中从学习者复制模型。

参数:

state_dict (-) – 之前保存的策略收集状态的字典。

小技巧

如果你想只加载模型的一部分，你可以简单地将 load_state_dict 中的 strict 参数设置为 False，或者参考 ding.torch_utils.checkpoint_helper 以进行更复杂的操作。

_load_state_dict_eval(state_dict: Dict[str, Any]) → None

概述：

将 state_dict 变量加载到策略评估模式中，例如加载自动恢复检查点，或在分布式训练场景中从学习者加载模型副本。

参数:

state_dict (-) – 保存之前的策略评估状态的字典。

小技巧

如果你想只加载模型的一部分，你可以简单地将 load_state_dict 中的 strict 参数设置为 False，或者参考 ding.torch_utils.checkpoint_helper 以进行更复杂的操作。

_load_state_dict_learn(state_dict: Dict[str, Any]) → None

概述：

将 state_dict 变量加载到策略学习模式中。

参数:

state_dict (-) – 之前保存的策略学习状态的字典。

_monitor_vars_learn() → List[str]

概述：

在学习模式中注册要监控的变量。根据返回值 _forward_learn ，注册的变量将被记录在 tensorboard 中。

_process_transition(obs, policy_output, timestep)

概述：

将一个时间步的过渡数据处理并打包成一个字典，例如 <s, a, r, s’, done>。一些策略需要进行一些特殊处理并打包其自身的必要属性（例如隐藏状态和logit），因此此方法留给子类实现。

参数:

• obs (-) – 当前时间步的观察。

• policy_output (-) – 以观察为输入的策略网络的输出。通常，它包含动作和动作的对数概率。

• timestep (-) – 环境步骤方法返回的名为namedtuple的执行结果，除了所有元素都已转换为张量数据。通常，它包含下一个观察值、奖励、完成状态、信息等。

返回:

当前时间步的已处理过渡数据。

返回类型:

• transition (Dict[str, torch.Tensor])

_reset_collect(data_id: List[int] | None = None) → None

概述：

重置收集器环境的观察和动作。

参数:

data_id (-) – 要重置的数据ID列表（在此实现中未使用）。

_reset_eval(data_id: List[int] | None = None) → None

概述：

重置评估环境的观察和动作。

参数:

data_id (-) – 要重置的数据ID列表（在此实现中未使用）。

_reset_learn(data_id: List[int] | None = None) → None

概述：

在必要时重置学习模式下的一些有状态变量，例如RNN的隐藏状态或某些特殊算法的状态库。如果 data_id 为None，则意味着重置所有有状态变量。否则，它将根据 data_id 重置有状态变量。例如， data_id 中的不同轨迹在RNN中将具有不同的隐藏状态。

参数:

data_id (-) – 数据的id，用于重置由 data_id 指定的状态变量。

备注

此方法不是必须实现的。子类可以根据需要重写此方法。

_set_attribute(name: str, value: Any) → None

概述：

为了控制策略属性的访问，我们对外暴露不同的模式，而不是直接使用策略实例。我们还提供了一种方法，用于在不同模式下设置策略的属性。新属性将被命名为 _{name}。

参数:

• name (-) – 属性的名称。

• value (-) – 属性的值。

_state_dict_collect() → Dict[str, Any]

概述：

返回收集模式的 state_dict，仅包括通常情况下的模型，这对于分布式训练场景中自动恢复收集器是必要的。

返回:

当前策略收集状态的字典，用于保存和恢复。

返回类型:

• state_dict (Dict[str, Any])

小技巧

并非所有场景都需要自动恢复收集器，有时，我们可以直接关闭崩溃的收集器并重新启动一个新的。

_state_dict_eval() → Dict[str, Any]

概述：

返回评估模式下的 state_dict，通常只包括模型，这对于分布式训练场景是必要的，以便自动恢复评估器。

返回:

当前策略评估状态的字典，用于保存和恢复。

返回类型:

• state_dict (Dict[str, Any])

小技巧

并非所有场景都需要自动恢复评估器，有时，我们可以直接关闭崩溃的评估器并重新启动一个新的评估器。

_state_dict_learn() → Dict[str, Any]

概述：

返回学习模式的 state_dict，通常包括模型、目标模型和优化器。

返回:

当前策略学习状态的字典，用于保存和恢复。

返回类型:

• state_dict (Dict[str, Any])

property cfg: EasyDict

class collect_function(forward, process_transition, get_train_sample, reset, get_attribute, set_attribute, state_dict, load_state_dict)

基类：tuple

_asdict()

返回一个新字典，该字典将字段名称映射到它们的值。

_field_defaults = {}

_fields = ('forward', 'process_transition', 'get_train_sample', 'reset', 'get_attribute', 'set_attribute', 'state_dict', 'load_state_dict')

classmethod _make(iterable)

从序列或可迭代对象创建一个新的 collect_function 对象

_replace(**kwds)

返回一个新的 collect_function 对象，用新值替换指定字段

count(value, /)

返回值出现的次数。

forward

字段编号 0 的别名

get_attribute

字段编号4的别名

get_train_sample

字段编号2的别名

index(value, start=0, stop=9223372036854775807, /)

返回值的第一个索引。

如果值不存在，则引发 ValueError。

load_state_dict

字段编号 7 的别名

process_transition

字段编号1的别名

reset

字段编号 3 的别名

set_attribute

字段编号 5 的别名

state_dict

字段编号 6 的别名

property collect_mode: collect_function

概述：

返回策略的收集模式的接口，该接口用于训练模型。这里我们使用 namedtuple 来定义不可变的接口，并限制策略在不同模式下的使用。此外，派生子类可以重写这些接口以自定义其自己的收集模式。

返回:

策略的收集模式的接口，它是一个 namedtuple，其不同字段的值是不同的内部方法。

返回类型:

• interfaces (Policy.collect_function)

示例

>>> policy = Policy(cfg, model)
>>> policy_collect = policy.collect_mode
>>> obs = env_manager.ready_obs
>>> inference_output = policy_collect.forward(obs)
>>> next_obs, rew, done, info = env_manager.step(inference_output.action)

config = {'action_type': 'fixed_action_space', 'afterstate_policy_loss_weight': 1, 'afterstate_value_loss_weight': 0.25, 'analyze_chance_distribution': False, 'augmentation': ['shift', 'intensity'], 'batch_size': 256, 'battle_mode': 'play_with_bot_mode', 'collector_env_num': 8, 'commitment_loss_weight': 1.0, 'cuda': True, 'discount_factor': 0.997, 'env_type': 'not_board_games', 'eps': {'decay': 100000, 'end': 0.05, 'eps_greedy_exploration_in_collect': False, 'start': 1.0, 'type': 'linear'}, 'eval_offline': False, 'evaluator_env_num': 3, 'fixed_temperature_value': 0.25, 'game_segment_length': 200, 'grad_clip_value': 10, 'gray_scale': False, 'gumbel_algo': False, 'ignore_done': False, 'learning_rate': 0, 'manual_temperature_decay': False, 'mcts_ctree': True, 'model': {'bias': True, 'categorical_distribution': True, 'chance_space_size': 2, 'continuous_action_space': False, 'frame_stack_num': 1, 'image_channel': 1, 'model_type': 'conv', 'num_channels': 64, 'num_res_blocks': 1, 'observation_shape': (4, 96, 96), 'self_supervised_learning_loss': False, 'support_scale': 300}, 'momentum': 0.9, 'monitor_extra_statistics': True, 'n_episode': 8, 'num_simulations': 50, 'num_unroll_steps': 5, 'optim_type': 'Adam', 'piecewise_decay_lr_scheduler': False, 'policy_loss_weight': 1, 'priority_prob_alpha': 0.6, 'priority_prob_beta': 0.4, 'random_collect_episode_num': 0, 'reanalyze_noise': True, 'replay_ratio': 0.25, 'reward_loss_weight': 1, 'root_dirichlet_alpha': 0.3, 'root_noise_weight': 0.25, 'sampled_algo': False, 'ssl_loss_weight': 0, 'target_update_freq': 100, 'td_steps': 5, 'threshold_training_steps_for_final_lr': 50000, 'threshold_training_steps_for_final_temperature': 100000, 'transform2string': False, 'update_per_collect': 100, 'use_augmentation': False, 'use_max_priority_for_new_data': True, 'use_priority': True, 'use_ture_chance_label_in_chance_encoder': False, 'value_loss_weight': 0.25, 'weight_decay': 0.0001}

classmethod default_config() → EasyDict

概述：

获取策略的默认配置。此方法用于创建策略的默认配置。

返回:

对应策略的默认配置。对于派生策略类，它将递归合并基类的默认配置及其自身的默认配置。

返回类型:

• cfg (EasyDict)

小技巧

此方法将深度复制类的 config 属性并返回结果。因此，用户无需担心返回配置的修改。

default_model() → Tuple[str, List[str]]

概述：

返回此算法的默认模型设置。

返回:

模型名称和模型导入名称。

• model_type (str): The model type used in this algorithm, which is registered in ModelRegistry.

• import_names (List[str]): The model class path list used in this algorithm.

返回类型:

• model_info (Tuple[str, List[str]])

备注

用户可以定义和使用自定义的网络模型，但必须遵守由 import_names 路径指示的相同接口定义。对于 MuZero，使用 lzero.model.muzero_model.MuZeroModel。

class eval_function(forward, reset, get_attribute, set_attribute, state_dict, load_state_dict)

基类：tuple

_asdict()

返回一个新字典，该字典将字段名称映射到它们的值。

_field_defaults = {}

_fields = ('forward', 'reset', 'get_attribute', 'set_attribute', 'state_dict', 'load_state_dict')

classmethod _make(iterable)

从序列或可迭代对象创建一个新的 eval_function 对象

_replace(**kwds)

返回一个新的 eval_function 对象，用新值替换指定字段。

count(value, /)

返回值出现的次数。

forward

字段编号 0 的别名

get_attribute

字段编号2的别名

index(value, start=0, stop=9223372036854775807, /)

返回值的第一个索引。

如果值不存在，则引发 ValueError。

load_state_dict

字段编号 5 的别名

reset

字段编号1的别名

set_attribute

字段编号 3 的别名

state_dict

字段编号4的别名

property eval_mode: eval_function

概述：

返回策略的评估模式的接口，该接口用于训练模型。这里我们使用 namedtuple 来定义不可变的接口，并限制策略在不同模式下的使用。此外，派生子类可以重写这些接口以自定义其自己的评估模式。

返回:

策略的eval模式的接口，它是一个namedtuple，其不同字段的值是不同的内部方法。

返回类型:

• interfaces (Policy.eval_function)

示例

>>> policy = Policy(cfg, model)
>>> policy_eval = policy.eval_mode
>>> obs = env_manager.ready_obs
>>> inference_output = policy_eval.forward(obs)
>>> next_obs, rew, done, info = env_manager.step(inference_output.action)

class learn_function(forward, reset, info, monitor_vars, get_attribute, set_attribute, state_dict, load_state_dict)

基类：tuple

_asdict()

返回一个新字典，该字典将字段名称映射到它们的值。

_field_defaults = {}

_fields = ('forward', 'reset', 'info', 'monitor_vars', 'get_attribute', 'set_attribute', 'state_dict', 'load_state_dict')

classmethod _make(iterable)

从序列或可迭代对象创建一个新的 learn_function 对象

_replace(**kwds)

返回一个新的 learn_function 对象，用新值替换指定的字段

count(value, /)

返回值出现的次数。

forward

字段编号 0 的别名

get_attribute

字段编号4的别名

index(value, start=0, stop=9223372036854775807, /)

返回值的第一个索引。

如果值不存在，则引发 ValueError。

info

字段编号2的别名

load_state_dict

字段编号 7 的别名

monitor_vars

字段编号 3 的别名

reset

字段编号1的别名

set_attribute

字段编号 5 的别名

state_dict

字段编号 6 的别名

property learn_mode: learn_function

概述：

返回策略学习模式的接口，该接口用于训练模型。这里我们使用 namedtuple 来定义不可变的接口，并限制策略在不同模式下的使用。此外，派生子类可以重写这些接口以自定义其自己的学习模式。

返回:

策略的学习模式的接口，它是一个 namedtuple，其不同字段的值是不同的内部方法。

返回类型:

• interfaces (Policy.learn_function)

示例

>>> policy = Policy(cfg, model)
>>> policy_learn = policy.learn_mode
>>> train_output = policy_learn.forward(data)
>>> state_dict = policy_learn.state_dict()

set_train_iter_env_step(train_iter, env_step) → None

概述：

设置策略的 train_iter 和 env_step。

参数:

• train_iter (-) – 策略的 train_iter。

• env_step (-) – 策略的 env_step。

sync_gradients(model: Module) → None

概述：

在数据并行的多GPU训练中，同步（allreduce）模型参数的梯度。

参数:

model (-) – 用于同步梯度的模型。

备注

此方法仅在多GPU训练中使用，并且应在 backward 方法之后和 step 方法之前调用。用户还可以使用 bp_update_sync 配置来控制是否同步梯度allreduce和优化器更新。

total_field = {'collect', 'eval', 'learn'}

UniZeroPolicy

class lzero.policy.unizero.UniZeroPolicy(cfg: EasyDict, model: Module | None = None, enable_field: List[str] | None = None)

基类：MuZeroPolicy

概述：

UniZero 的策略类，是论文 UniZero: Generalized and Efficient Planning with Scalable LatentWorld Models 的官方实现。UniZero 旨在通过解决 MuZero 风格算法中的局限性来增强强化学习代理的规划能力，特别是在需要捕捉长期依赖关系的环境中。更多详情请参见 https://arxiv.org/abs/2406.10667。

__init__(cfg: EasyDict, model: Module | None = None, enable_field: List[str] | None = None) → None

概述：

根据输入配置和模型初始化策略实例。此方法将初始化策略中的不同字段，包括 learn、collect、eval。learn 字段用于训练策略，collect 字段用于收集训练数据，eval 字段用于评估策略。enable_field 用于指定要初始化的字段，如果为 None，则将初始化所有字段。

参数:

• cfg (-) – 用于初始化策略的最终合并配置。对于默认配置，请参阅策略类的 config 属性及其注释。

• model (-) – 用于初始化策略的神经网络模型。如果它是 None，则模型将根据 default_model 方法和 cfg.model 字段创建。否则，模型将被设置为由外部调用者创建的 model 实例。

• enable_field (-) – 要初始化的字段列表。如果为 None，则所有字段都将被初始化。否则，只有 enable_field 中的字段将被初始化，这有助于节省资源。

备注

对于派生的策略类，它应该实现 _init_learn、_init_collect、_init_eval 方法来初始化相应的字段。

_abc_impl = <_abc._abc_data object>

_create_model(cfg: EasyDict, model: Module | None = None) → Module

概述：

根据输入配置和模型创建或验证神经网络模型。如果输入模型为 None，则将根据 default_model 方法和 cfg.model 字段创建模型。否则，模型将被验证为 torch.nn.Module 的实例，并设置为由外部调用者创建的 model 实例。

参数:

• cfg (-) – 用于初始化策略的最终合并配置。

• model (-) – 用于初始化策略的神经网络模型。用户可以参考相应策略中定义的默认模型来自定义自己的模型。

返回:

创建的神经网络模型。策略的不同模式将向模型添加不同的包装器和插件，这些包装器和插件用于训练、收集和评估。

返回类型:

• model (torch.nn.Module)

抛出:

- RuntimeError – 如果输入模型不是 None 且不是 torch.nn.Module 的实例。

_forward_collect(data: Tensor, action_mask: list = None, temperature: float = 1, to_play: List = [-1], epsilon: float = 0.25, ready_env_id: array = None) → Dict

概述：

在收集模式下收集数据的正向函数。使用模型执行MCTS搜索。在收集模式期间通过采样选择动作。

参数:

• data (-) – 输入数据，即观察结果。

• action_mask (-) – 动作掩码，即不能被选择的动作。

• temperature (-) – 政策的温度。

• to_play (-) – 要播放的玩家。

• ready_env_id (-) – 准备好收集的环境的id。

形状：

• data (torch.Tensor):

  • 对于Atari，\((N, C*S, H, W)\)，其中N是collect_env的数量，C是通道的数量，S是堆叠帧的数量，H是图像的高度，W是图像的宽度。

  • 对于 lunarlander，\((N, O)\)，其中 N 是 collect_env 的数量，O 是观察空间的大小。

• action_mask: \((N, action_space_size)\)，其中 N 是 collect_env 的数量。

• 温度: \((1, )\).

• to_play: \((N, 1)\)，其中 N 是 collect_env 的数量。

• ready_env_id: None

返回:

字典类型数据，包含的键有 action、distributions、visit_count_distribution_entropy、value、pred_value、policy_logits。

返回类型:

• output (Dict[int, Any])

_forward_eval(data: Tensor, action_mask: list, to_play: int = -1, ready_env_id: array = None) → Dict

概述：

用于评估当前策略在评估模式下的前向函数。使用模型执行MCTS搜索。在评估模式下选择值最高的动作（argmax）而不是采样。

参数:

• data (-) – 输入数据，即观察结果。

• action_mask (-) – 动作掩码，即不能被选择的动作。

• to_play (-) – 要播放的玩家。

• ready_env_id (-) – 准备好收集的环境的id。

形状：

• data (torch.Tensor):

  • 对于Atari，\((N, C*S, H, W)\)，其中N是collect_env的数量，C是通道的数量，S是堆叠帧的数量，H是图像的高度，W是图像的宽度。

  • 对于 lunarlander，\((N, O)\)，其中 N 是 collect_env 的数量，O 是观察空间的大小。

• action_mask: \((N, action_space_size)\)，其中 N 是 collect_env 的数量。

• to_play: \((N, 1)\)，其中 N 是 collect_env 的数量。

• ready_env_id: None

返回:

字典类型数据，包含的键有 action、distributions、visit_count_distribution_entropy、value、pred_value、policy_logits。

返回类型:

• output (Dict[int, Any])

_forward_learn(data: Tuple[Tensor]) → Dict[str, float | int]

概述：

学习模式下用于学习策略的前向函数，这是学习过程的核心。数据从回放缓冲区中采样。通过损失函数计算损失，并将损失反向传播以更新模型。

参数:

data (-) – 从回放缓冲区中采样的数据，这是一个张量元组。第一个张量是当前批次，第二个张量是目标批次。

返回:

要记录的信息字典，其中包含当前的学习损失和学习统计数据。

返回类型:

• info_dict (Dict[str, Union[float, int]])

_get_attribute(name: str) → Any

概述：

为了控制策略属性的访问，我们对外暴露不同的模式，而不是直接使用策略实例。我们还提供了一种方法来获取不同模式下的策略属性。

参数:

name (-) – 属性的名称。

返回:

属性的值。

返回类型:

• value (Any)

备注

DI-engine 的策略将首先尝试访问 _get_{name} 方法，然后尝试访问 _{name} 属性。如果两者都未找到，它将引发一个 NotImplementedError。

_get_batch_size() → int | Dict[str, int]

_get_n_episode() → int | None

_get_n_sample() → int | None

_get_target_obs_index_in_step_k(step)

概述：

获取步骤 k 中目标 obs 的起始索引和结束索引。

参数:

step (-) – 当前步骤 k。

返回:

The begin index of the target obs in step k. - end_index (int): The end index of the target obs in step k.

返回类型:

• beg_index (int)

示例

>>> self._cfg.model.model_type = 'conv'
>>> self._cfg.model.image_channel = 3
>>> self._cfg.model.frame_stack_num = 4
>>> self._get_target_obs_index_in_step_k(0)
>>> (0, 12)

_get_train_sample(data)

概述：

对于给定的轨迹（过渡，过渡列表）数据，将其处理为可以直接用于训练的样本列表。一个训练样本可以是一个经过处理的过渡（DQN 使用 n步 TD）或者一些多时间步的过渡（DRQN）。这个方法通常在收集器中使用，以在训练前执行必要的强化学习数据预处理，这可以帮助学习者分摊相关的时间消耗。此外，你也可以将这个方法实现为一个恒等函数，并在 self._forward_learn 方法中进行数据处理。

参数:

transitions (-) – 轨迹数据（一个过渡列表），每个元素的格式与 self._process_transition 方法的返回值相同。

返回:

处理后的训练样本，每个元素的格式与输入转换相似，但可能包含更多用于训练的数据，如n步奖励、优势等。

返回类型:

• samples (List[Dict[str, Any]])

备注

我们将在下一个发布版本中向量化 process_transition 和 get_train_sample 方法。用户可以通过重写这两个方法和收集器本身来自定义此数据处理过程。

_init_collect() → None

概述：

收集模式初始化方法。由 self.__init__ 调用。初始化收集模型和MCTS工具。

_init_eval() → None

概述：

评估模式初始化方法。由 self.__init__ 调用。初始化评估模型和MCTS工具。

_init_learn() → None

概述：

学习模式初始化方法。由 self.__init__ 调用。初始化学习模型、优化器和MCTS工具。

_init_multi_gpu_setting(model: Module, bp_update_sync: bool) → None

概述：

初始化多GPU数据并行训练设置，包括在训练开始时广播模型参数，并为所有模型参数梯度准备hook函数以进行allreduce操作。

参数:

• model (-) – 要训练的神经网络模型。

• bp_update_sync (-) – 是否在所有reduce模型参数的梯度后同步更新模型参数。异步更新可以在不同的网络层中并行进行，如流水线，从而节省时间。

_load_state_dict_collect(state_dict: Dict[str, Any]) → None

概述：

将 state_dict 变量加载到策略收集模式中，例如加载预训练的 state_dict、自动恢复检查点，或在分布式训练场景中从学习者复制模型。

参数:

state_dict (-) – 之前保存的策略收集状态的字典。

小技巧

如果你想只加载模型的一部分，你可以简单地将 load_state_dict 中的 strict 参数设置为 False，或者参考 ding.torch_utils.checkpoint_helper 以进行更复杂的操作。

_load_state_dict_eval(state_dict: Dict[str, Any]) → None

概述：

将 state_dict 变量加载到策略评估模式中，例如加载自动恢复检查点，或在分布式训练场景中从学习者加载模型副本。

参数:

state_dict (-) – 保存之前的策略评估状态的字典。

小技巧

如果你想只加载模型的一部分，你可以简单地将 load_state_dict 中的 strict 参数设置为 False，或者参考 ding.torch_utils.checkpoint_helper 以进行更复杂的操作。

_load_state_dict_learn(state_dict: Dict[str, Any]) → None

概述：

将 state_dict 变量加载到策略学习模式中。

参数:

state_dict (-) – 之前保存的策略学习状态的字典。

_monitor_vars_learn() → List[str]

概述：

在学习模式中注册要监控的变量。根据返回值 _forward_learn，注册的变量将被记录在 tensorboard 中。

_process_transition(obs, policy_output, timestep)

概述：

将一个时间步的过渡数据处理并打包成一个字典，例如 <s, a, r, s’, done>。一些策略需要进行一些特殊处理并打包其自身的必要属性（例如隐藏状态和logit），因此此方法留给子类实现。

参数:

• obs (-) – 当前时间步的观察。

• policy_output (-) – 以观察为输入的策略网络的输出。通常，它包含动作和动作的对数概率。

• timestep (-) – 环境步骤方法返回的名为namedtuple的执行结果，除了所有元素都已转换为张量数据。通常，它包含下一个观察值、奖励、完成状态、信息等。

返回:

当前时间步的已处理过渡数据。

返回类型:

• transition (Dict[str, torch.Tensor])

_reset_collect(env_id: int = None, current_steps: int = None, reset_init_data: bool = True) → None

概述：

此方法重置特定环境的收集过程。当满足某些条件时，它会清除缓存和内存，以确保最佳性能。如果 reset_init_data 为 True，则初始数据将被重置。

参数:

• env_id (-) – 要重置的环境的ID。如果为None或列表，函数将立即返回。

• current_steps (-) – 当前环境中的步骤计数。用于确定是否清除缓存。

• reset_init_data (-) – 是否重置初始数据。如果为 True，初始数据将被重置。

_reset_eval(env_id: int = None, current_steps: int = None, reset_init_data: bool = True) → None

概述：

此方法重置特定环境的评估过程。当满足某些条件时，它会清除缓存和内存，以确保最佳性能。如果 reset_init_data 为 True，初始数据将被重置。

参数:

• env_id (-) – 要重置的环境的ID。如果为None或列表，函数将立即返回。

• current_steps (-) – 当前环境中的步骤计数。用于确定是否清除缓存。

• reset_init_data (-) – 是否重置初始数据。如果为 True，初始数据将被重置。

_reset_learn(data_id: List[int] | None = None) → None

概述：

在必要时重置学习模式下的一些有状态变量，例如RNN的隐藏状态或某些特殊算法的状态库。如果 data_id 为None，则意味着重置所有有状态变量。否则，它将根据 data_id 重置有状态变量。例如， data_id 中的不同轨迹在RNN中将具有不同的隐藏状态。

参数:

data_id (-) – 数据的id，用于重置由 data_id 指定的状态变量。

备注

此方法不是必须实现的。子类可以根据需要重写此方法。

_set_attribute(name: str, value: Any) → None

概述：

为了控制策略属性的访问，我们对外暴露不同的模式，而不是直接使用策略实例。我们还提供了一种方法，用于在不同模式下设置策略的属性。新属性将被命名为 _{name}。

参数:

• name (-) – 属性的名称。

• value (-) – 属性的值。

_state_dict_collect() → Dict[str, Any]

概述：

返回收集模式的 state_dict，仅包括通常情况下的模型，这对于分布式训练场景中自动恢复收集器是必要的。

返回:

当前策略收集状态的字典，用于保存和恢复。

返回类型:

• state_dict (Dict[str, Any])

小技巧

并非所有场景都需要自动恢复收集器，有时，我们可以直接关闭崩溃的收集器并重新启动一个新的。

_state_dict_eval() → Dict[str, Any]

概述：

返回评估模式下的 state_dict，通常只包括模型，这对于分布式训练场景是必要的，以便自动恢复评估器。

返回:

当前策略评估状态的字典，用于保存和恢复。

返回类型:

• state_dict (Dict[str, Any])

小技巧

并非所有场景都需要自动恢复评估器，有时，我们可以直接关闭崩溃的评估器并重新启动一个新的评估器。

_state_dict_learn() → Dict[str, Any]

概述：

返回学习模式的 state_dict，通常包括模型、目标模型和优化器。

返回:

当前策略学习状态的字典，用于保存和恢复。

返回类型:

• state_dict (Dict[str, Any])

property cfg: EasyDict

class collect_function(forward, process_transition, get_train_sample, reset, get_attribute, set_attribute, state_dict, load_state_dict)

基类：tuple

_asdict()

返回一个新字典，该字典将字段名称映射到它们的值。

_field_defaults = {}

_fields = ('forward', 'process_transition', 'get_train_sample', 'reset', 'get_attribute', 'set_attribute', 'state_dict', 'load_state_dict')

classmethod _make(iterable)

从序列或可迭代对象创建一个新的 collect_function 对象

_replace(**kwds)

返回一个新的 collect_function 对象，用新值替换指定字段

count(value, /)

返回值出现的次数。

forward

字段编号 0 的别名

get_attribute

字段编号4的别名

get_train_sample

字段编号2的别名

index(value, start=0, stop=9223372036854775807, /)

返回值的第一个索引。

如果值不存在，则引发 ValueError。

load_state_dict

字段编号 7 的别名

process_transition

字段编号1的别名

reset

字段编号 3 的别名

set_attribute

字段编号 5 的别名

state_dict

字段编号 6 的别名

property collect_mode: collect_function

概述：

返回策略的收集模式的接口，该接口用于训练模型。这里我们使用 namedtuple 来定义不可变的接口，并限制策略在不同模式下的使用。此外，派生子类可以重写这些接口以自定义其自己的收集模式。

返回:

策略的收集模式的接口，它是一个 namedtuple，其不同字段的值是不同的内部方法。

返回类型:

• interfaces (Policy.collect_function)

示例

>>> policy = Policy(cfg, model)
>>> policy_collect = policy.collect_mode
>>> obs = env_manager.ready_obs
>>> inference_output = policy_collect.forward(obs)
>>> next_obs, rew, done, info = env_manager.step(inference_output.action)

config = {'action_type': 'fixed_action_space', 'analysis_sim_norm': False, 'augmentation': ['shift', 'intensity'], 'batch_size': 256, 'battle_mode': 'play_with_bot_mode', 'collect_with_pure_policy': False, 'collector_env_num': 8, 'cuda': True, 'discount_factor': 0.997, 'env_type': 'not_board_games', 'eps': {'decay': 100000, 'end': 0.05, 'eps_greedy_exploration_in_collect': False, 'start': 1.0, 'type': 'linear'}, 'eval_freq': 2000, 'evaluator_env_num': 3, 'fixed_temperature_value': 0.25, 'game_segment_length': 400, 'grad_clip_value': 20, 'gray_scale': False, 'gumbel_algo': False, 'ignore_done': False, 'learning_rate': 0.0001, 'manual_temperature_decay': False, 'mcts_ctree': True, 'model': {'analysis_sim_norm': False, 'bias': True, 'categorical_distribution': True, 'continuous_action_space': False, 'frame_stack_num': 1, 'image_channel': 3, 'learn': {'learner': {'hook': {'save_ckpt_after_iter': 10000}}}, 'model_type': 'conv', 'norm_type': 'BN', 'num_channels': 64, 'num_res_blocks': 1, 'observation_shape': (3, 64, 64), 'res_connection_in_dynamics': True, 'self_supervised_learning_loss': True, 'support_scale': 50, 'world_model_cfg': {'action_space_size': 6, 'analysis_dormant_ratio': False, 'analysis_sim_norm': False, 'attention': 'causal', 'attn_pdrop': 0.1, 'context_length': 8, 'continuous_action_space': False, 'device': 'cpu', 'dormant_threshold': 0.025, 'embed_dim': 768, 'embed_pdrop': 0.1, 'env_num': 8, 'gamma': 1, 'group_size': 8, 'gru_gating': False, 'latent_recon_loss_weight': 0.0, 'max_blocks': 10, 'max_cache_size': 5000, 'max_tokens': 20, 'num_heads': 8, 'num_layers': 2, 'obs_type': 'image', 'perceptual_loss_weight': 0.0, 'policy_entropy_weight': 0, 'predict_latent_loss_type': 'group_kl', 'resid_pdrop': 0.1, 'support_size': 101, 'tokens_per_block': 2}}, 'momentum': 0.9, 'monitor_extra_statistics': True, 'multi_gpu': False, 'n_episode': 8, 'num_segments': 8, 'num_simulations': 50, 'num_unroll_steps': 10, 'optim_type': 'AdamW', 'piecewise_decay_lr_scheduler': False, 'policy_loss_weight': 1, 'priority_prob_alpha': 0.6, 'priority_prob_beta': 0.4, 'random_collect_episode_num': 0, 'replay_ratio': 0.25, 'reward_loss_weight': 1, 'root_dirichlet_alpha': 0.3, 'root_noise_weight': 0.25, 'sample_type': 'transition', 'sampled_algo': False, 'ssl_loss_weight': 0, 'target_update_freq': 100, 'target_update_freq_for_intrinsic_reward': 1000, 'target_update_theta': 0.05, 'td_steps': 5, 'threshold_training_steps_for_final_lr': 50000, 'threshold_training_steps_for_final_temperature': 50000, 'train_start_after_envsteps': 0, 'transform2string': False, 'type': 'unizero', 'update_per_collect': None, 'use_augmentation': False, 'use_priority': False, 'use_rnd_model': False, 'use_ture_chance_label_in_chance_encoder': False, 'value_loss_weight': 0.25, 'weight_decay': 0.0001}

classmethod default_config() → EasyDict

概述：

获取策略的默认配置。此方法用于创建策略的默认配置。

返回:

对应策略的默认配置。对于派生策略类，它将递归合并基类的默认配置及其自身的默认配置。

返回类型:

• cfg (EasyDict)

小技巧

此方法将深度复制类的 config 属性并返回结果。因此，用户无需担心返回配置的修改。

default_model() → Tuple[str, List[str]]

概述：

返回此算法的默认模型设置以供演示。

返回:

模型名称和模型导入名称。

• model_type (str): The model type used in this algorithm, which is registered in ModelRegistry.

• import_names (List[str]): The model class path list used in this algorithm.

返回类型:

• model_info (Tuple[str, List[str]])

备注

用户可以定义和使用自定义的网络模型，但必须遵守由 import_names 路径指示的相同接口定义。对于 MuZero，使用 lzero.model.unizero_model.MuZeroModel。

class eval_function(forward, reset, get_attribute, set_attribute, state_dict, load_state_dict)

基类：tuple

_asdict()

返回一个新字典，该字典将字段名称映射到它们的值。

_field_defaults = {}

_fields = ('forward', 'reset', 'get_attribute', 'set_attribute', 'state_dict', 'load_state_dict')

classmethod _make(iterable)

从序列或可迭代对象创建一个新的 eval_function 对象

_replace(**kwds)

返回一个新的 eval_function 对象，用新值替换指定字段。

count(value, /)

返回值出现的次数。

forward

字段编号 0 的别名

get_attribute

字段编号2的别名

index(value, start=0, stop=9223372036854775807, /)

返回值的第一个索引。

如果值不存在，则引发 ValueError。

load_state_dict

字段编号 5 的别名

reset

字段编号1的别名

set_attribute

字段编号 3 的别名

state_dict

字段编号4的别名

property eval_mode: eval_function

概述：

返回策略的评估模式的接口，该接口用于训练模型。这里我们使用 namedtuple 来定义不可变的接口，并限制策略在不同模式下的使用。此外，派生子类可以重写这些接口以自定义其自己的评估模式。

返回:

策略的eval模式的接口，它是一个namedtuple，其不同字段的值是不同的内部方法。

返回类型:

• interfaces (Policy.eval_function)

示例

>>> policy = Policy(cfg, model)
>>> policy_eval = policy.eval_mode
>>> obs = env_manager.ready_obs
>>> inference_output = policy_eval.forward(obs)
>>> next_obs, rew, done, info = env_manager.step(inference_output.action)

class learn_function(forward, reset, info, monitor_vars, get_attribute, set_attribute, state_dict, load_state_dict)

基类：tuple

_asdict()

返回一个新字典，该字典将字段名称映射到它们的值。

_field_defaults = {}

_fields = ('forward', 'reset', 'info', 'monitor_vars', 'get_attribute', 'set_attribute', 'state_dict', 'load_state_dict')

classmethod _make(iterable)

从序列或可迭代对象创建一个新的 learn_function 对象

_replace(**kwds)

返回一个新的 learn_function 对象，用新值替换指定的字段

count(value, /)

返回值出现的次数。

forward

字段编号 0 的别名

get_attribute

字段编号4的别名

index(value, start=0, stop=9223372036854775807, /)

返回值的第一个索引。

如果值不存在，则引发 ValueError。

info

字段编号2的别名

load_state_dict

字段编号 7 的别名

monitor_vars

字段编号 3 的别名

reset

字段编号1的别名

set_attribute

字段编号 5 的别名

state_dict

字段编号 6 的别名

property learn_mode: learn_function

概述：

返回策略学习模式的接口，该接口用于训练模型。这里我们使用 namedtuple 来定义不可变的接口，并限制策略在不同模式下的使用。此外，派生子类可以重写这些接口以自定义其自己的学习模式。

返回:

策略的学习模式的接口，它是一个 namedtuple，其不同字段的值是不同的内部方法。

返回类型:

• interfaces (Policy.learn_function)

示例

>>> policy = Policy(cfg, model)
>>> policy_learn = policy.learn_mode
>>> train_output = policy_learn.forward(data)
>>> state_dict = policy_learn.state_dict()

monitor_weights_and_grads(model)

recompute_pos_emb_diff_and_clear_cache() → None

概述：

清除缓存并在模型中预计算位置嵌入矩阵。

set_train_iter_env_step(train_iter, env_step) → None

概述：

设置策略的 train_iter 和 env_step。

参数:

• train_iter (-) – 策略的 train_iter。

• env_step (-) – 策略的 env_step。

sync_gradients(model: Module) → None

概述：

在数据并行的多GPU训练中，同步（allreduce）模型参数的梯度。

参数:

model (-) – 用于同步梯度的模型。

备注

此方法仅在多GPU训练中使用，并且应在 backward 方法之后和 step 方法之前调用。用户还可以使用 bp_update_sync 配置来控制是否同步梯度allreduce和优化器更新。

total_field = {'collect', 'eval', 'learn'}