mindspore_rl

MindSpore强化学习框架的组件。

mindspore_rl.agent

agent、actor、learner、trainer的组件。

class mindspore_rl.agent.Actor[源代码]

所有Actor的基类。Actor 是一个用来和环境交互并产生数据的类。

样例：

>>> from mindspore_rl.agent.actor import Actor
>>> from mindspore_rl.network import FullyConnectedNet
>>> from mindspore_rl.environment import GymEnvironment
>>> class MyActor(Actor):
...   def __init__(self):
...     super(MyActor, self).__init__()
...     self.argmax = P.Argmax()
...     self.actor_net = FullyConnectedNet(4, 10, 2)
...     self.env = GymEnvironment({'name': 'CartPole-v0'})
>>> my_actor = MyActor()
>>> print(my_actor)
MyActor<
(actor_net): FullyConnectedNet<
(linear1): Dense<input_channels=4, output_channels=10, has_bias=True>
(linear2): Dense<input_channels=10, output_channels=2, has_bias=True>
(relu): ReLU<>
>
(environment): GymEnvironment<>
act(phase, params)[源代码]

act 方法接收一个枚举值和观察数据或计算动作期间所需的数据。它将返回一组包含新观察数据或其他经验的输出。此接口将与环境交互。

参数：

  • phase (enum) - 一个int型的枚举值，用于初始化、收集、评估或其他用户定义的阶段。

  • params (tuple(Tensor)) - 作为输入的张量元组，用于计算动作。

返回：

  • observation (tuple(Tensor)) - 作为输出的张量元组，用于生成经验数据。

get_action(phase, params)[源代码]

get_action 是用来获得动作的方法。用户需要根据算法重载此函数。但该函数入参需为phase和params。此接口不会与环境交互。

参数：

  • phase (enum) - 一个int型的枚举值，用于初始化、收集、评估或者其他用户定义的阶段。

  • params (tuple(Tensor)) - 作为输入的张量元组，用于计算动作。

返回：

  • action (tuple(Tensor)) - 作为输出的张量元组，包含动作和其他所需数据的张量。

class mindspore_rl.agent.Learner[源代码]

Learner的基类。通过输入的经验数据，计算并更新自生的网络。

样例：

>>> from mindspore_rl.agent.learner import Learner
>>> from mindspore_rl.network import FullyConnectedNet
>>> class MyLearner(Learner):
...   def init(self):
...     super(MyLearner, self).init()
...     self.target_network = FullyConnectedNet(4, 10, 2)
>>> my_learner = MyLearner()
>>> print(my_learner)
MyLearner<
(target_network): FullyConnectedNet<
(linear1): Dense<input_channels=4, output_channels=10, has_bias=True>
(linear2): Dense<input_channels=10, output_channels=2, has_bias=True>
(relu): ReLU<>
>
learn(experience)[源代码]

learn 方法的接口。 learn 方法的行为取决于用户的实现。通常，它接受来自重放缓存中的 samples 或其他Tensors，并计算用于更新网络的损失。

参数：

  • experience (tuple(Tensor)) - 缓存中的经验数据。

返回：

  • results (tuple(Tensor)) - 更新权重后输出的结果。

class mindspore_rl.agent.Trainer(msrl)[源代码]

Trainer的基类。是一个流程类，提供训练的基本模式。

说明

引用 dqn_trainer.py。

参数：

  • msrl (MSRL) - 函数句柄。

evaluate()[源代码]

在训练中用于评估的评估方法。

load_and_eval(ckpt_path=None)[源代码]

离线评估的方法。必须提供一个checkpoint。

参数：

  • ckpt_path (string) - 需要加载到网络的checkpoint文件。默认值：None。

train(episodes, callbacks=None, ckpt_path=None)[源代码]

train 方法中提供一个标准的训练流程，包含整个循环和回调。用户可根据需要自行继承或覆写。

参数：

  • episodes (int) - 训练回合数。

  • callbacks (Optional[list[Callback]]) - 回调函数的列表。默认值：None。

  • ckpt_path (Optional[str]) - 要初始化或重加载的网络文件路径。默认值：None。

train_one_episode()[源代码]

在训练中，训练一个回合的接口。该函数的输出必须按顺序限制为 loss, rewards, steps, [Optional]others。

trainable_variables()[源代码]

用于保存至checkpoint的变量。

class mindspore_rl.agent.Agent(actors, learner)[源代码]

Agent的基类。作为智能体的定义，由Actor和Learner构成。具备基本的act和learn功能用于和环境交互和自我更新。

参数：

  • actors (Actor) - Actor 实例。

  • learner (Learner) - learner 实例。

样例：

>>> from mindspore_rl.agent.learner import Learner
>>> from mindspore_rl.agent.actor import Actor
>>> from mindspore_rl.agent.agent import Agent
>>> actors = Actor()
>>> learner = Learner()
>>> agent = Agent(actors, learner)
>>> print(agent)
Agent<
(_actors): Actor<>
(_learner): Learner<>
>
act(phase, params)[源代码]

act 方法接收一个枚举值和观察数据或计算动作期间所需的数据。它将返回一组包含新观察数据或其他经验的输出。此接口中，Agent将与环境交互。

参数：

  • phase (enum) - 一个int型的枚举值，用于初始化、收集或评估的阶段。

  • params (tuple(Tensor)) - 作为输入的张量元组，用于计算动作。

返回：

  • observation (tuple(Tensor)) - 作为输出的张量元组，用于生成经验数据。

get_action(phase, params)[源代码]

get_action 方法接收一个枚举值和观察数据或计算动作期间所需的数据。它将返回一组包含动作和其他数据的输出。此接口中，Agent不与环境交互。

参数：

  • phase (enum) - 一个int型的枚举值，用于初始化、收集、评估或者其他用户定义的阶段。

  • params (tuple(Tensor)) - 作为输入的张量元组，用于计算动作。

返回：

  • action (tuple(Tensor)) - 作为输出的张量元组，包含动作和其他所需数据的张量。

learn(experience)[源代码]

learn 方法接收一组经验数据作为输入，以计算损失并更新权重。

参数：

  • experience (tuple(Tensor)) - 经验的张量状态元组。

返回：

  • results (tuple(Tensor)) - 更新权重后输出的结果。

mindspore_rl.core

用于实现 RL 算法的Helper程序组件。

class mindspore_rl.core.MSRL(alg_config, deploy_config=None)[源代码]

MSRL提供了用于强化学习算法开发的方法和API。 它向用户公开以下方法。这些方法的输入和输出与用户定义的方法相同。

agent_act
agent_get_action
sample_buffer
agent_learn
replay_buffer_sample
replay_buffer_insert
replay_buffer_reset
参数：

  • alg_config (dict) - 提供算法配置。

  • deploy_config (dict) - 提供分布式配置。

    • 顶层 - 定义算法组件。

  • 关键字: actor， 值： actor的配置 (dict)。

  • 关键字: learner， 值： learner的配置 (dict)。

  • 关键字: policy_and_network， 值： actor和learner使用的策略和网络 (dict)。

  • 关键字: collect_environment， 值： 收集环境的配置 (dict)。

  • 关键字: eval_environment， 值： 评估环境的配置 (dict)。

  • 关键字: replay_buffer， 值： 重放缓存的配置 (dict)。

    • 第二层 - 每个算法组件的配置。

  • 关键字: number， 值： actor/learner的数量 (int)。

  • 关键字: type， 值： actor/learner/policy_and_network/environment (class)。

  • 关键字: params， 值： actor/learner/policy_and_network/environment的参数 (dict)。

  • 关键字: policies， 值： actor/learner使用的策略列表 (list)。

  • 关键字: networks， 值： actor/learner使用的网络列表 (list)。

  • 关键字: pass_environment， 值： 如果为True， 用户需要传递环境实例给actor， 为False则不需要 (bool)。

get_replay_buffer()[源代码]

返回重放缓存的实例。

返回：

  • buffers (object) - 重放缓存的实例。如果缓存为None， 返回也为None。

get_replay_buffer_elements(transpose=False, shape=None)[源代码]

返回重放缓存中的所有元素。

参数：

  • transpose (bool) - 输出元素是否需要转置，如果为True，则shape也需指定。默认值：False。

  • shape (tuple[int]) - 转置的shape。默认值：None。

返回：

  • elements (List[Tensor]) - 一组包含所有重放缓存中数据的张量。

init(config)[源代码]

MSRL 对象的初始化。该方法创建算法所需的所有数据/对象。它会初始化所有的方法。

参数：

  • config (dict) - 算法的配置文件。

class mindspore_rl.core.Session(alg_config, deploy_config=None, params=None, callbacks=None)[源代码]

Session是一个用于运行MindSpore RL算法的类。

参数：

  • alg_config (dict) - 算法的配置或算法的部署配置。

  • deploy_config (dict) - 分布式的部署配置，默认：None。更多算法配置的详细信息，请看 detail。

  • params (dict) - 算法特定的训练参数。默认值：None。

  • callbacks (list[Callback]) - 回调列表。默认值：None。

run(class_type=None, is_train=True, episode=0, duration=0)[源代码]

执行强化学习算法。

参数：

  • class_type (Trainer) - 算法的trainer类的类型。默认值：None。

  • is_train (bool) - 在训练或推理中执行算法，True为训练，False为推理。默认值：True。

  • episode (int) - 训练的回合数。默认值：0。

  • duration (int) - 每回合的步数。默认值：0。

class mindspore_rl.core.UniformReplayBuffer(batch_size, capacity, shapes, types)[源代码]

重放缓存类。重放缓存区中存放来自环境的经验数据。在该类中，每个元素都是一组Tensor。因此，ReplayBuffer类的构造函数将每个Tensor的形状和类型作为参数。

参数：

  • batch_size (int) - 从缓存区采样的batch大小。

  • capacity (int) - 缓存区的大小。

  • shapes (list[int]) - 缓存区中每个元素对应的Tensor shape列表。

  • types (list[mindspore.dtype]) - 缓存区中每个元素对应的Tensor dtype列表。

样例：

>>> import mindspore as ms
>>> from mindspore_rl.core.uniform_replay_buffer import UniformReplayBuffer
>>> batch_size = 10
>>> capacity = 10000
>>> shapes = [(4,), (1,), (1,), (4,)]
>>> types = [ms.float32, ms.int32, ms.float32, ms.float32]
>>> replaybuffer = UniformReplayBuffer(batch_size, capacity, shapes, types)
>>> print(replaybuffer)
UniformReplayBuffer<>
full()[源代码]

检查缓存区是否已满。

返回：

  • Full (bool) - 缓存区已满返回True，否则返回False。

get_item(index)[源代码]

从缓存区的指定位置取出元素。

参数：

  • index (int) - 元素的索引。

返回：

  • element (list[Tensor]) - 返回指定位置的元素。

insert(exp)[源代码]

将元素插入缓存区。如果缓存区已满，则将使用先进先出的策略替换缓存区的元素。

参数：

  • exp (list[Tensor]) - 插入的Tensor组，需要符合缓存初始化时的shape和type。

返回：

  • element (list[Tensor]) - 返回插入数据后的缓存区。

reset()[源代码]

重置缓存区，将count值置零。

返回：

  • success (bool) - 重置是否成功。

sample()[源代码]

缓存区采样，随机地选择一组元素并输出。

返回：

  • data (Tuple(Tensor)) - 一组从缓存区随机采样出的元素。

size()[源代码]

返回缓存区的大小。

返回：

  • size (int) - 缓存区的元素个数。

class mindspore_rl.core.PriorityReplayBuffer(alpha, capacity, sample_size, shapes, dtypes, seed0=0, seed1=0)[源代码]

优先级经验回放缓存，用于深度Q学习存储经验数据。 该算法在 Prioritized Experience Replay 中提出。 与普通的经验回放缓存相同，它允许强化学习智能体记住和重用过去的经验。此外，它更频繁的回放重要的transition，提高样本效率。

参数：

  • alpha (float) - 控制优先级程度的参数。0表示均匀采样，1表示优先级采样。

  • capacity (int) - 缓存的容量。

  • sample_size (int) - 从缓存采样的大小

  • shapes (list[int]) - 缓存区中张量维度列表。

  • types (list[mindspore.dtype]) - 缓存区张量数据类型列表。

  • seed0 (int) - 随机数种子0值。默认值：0。

  • seed1 (int) - 随机数种子1值。默认值：0。

样例：

>>> import mindspore as ms
>>> from mindspore import Tensor
>>> from mindspore_rl.core.priority_replay_buffer import PriorityReplayBuffer
>>> capacity = 10000
>>> batch_size = 10
>>> shapes = [(4,), (1,), (1,), (4,)]
>>> dtypes = [ms.float32, ms.int32, ms.float32, ms.float32]
>>> replaybuffer = PriorityReplayBuffer(alpha, capacity, batch_size, shapes, dtypes)
>>> print(replaybuffer)
PriorityReplayBuffer<>
destroy()[源代码]

销毁经验回放缓存。

返回：

  • handle (Tensor) - 优先级经验回放缓存句柄，数据和shape分别是int64和 \((1,)\)。

push(*transition)[源代码]

将transition推送到缓存区。如果缓存区已满，则覆盖最早的数据。

参数：

  • transition (List[Tensor]) - 与初始化的shapes和dtypes匹配的张量列表。

返回：

  • handle (Tensor) - 优先级经验回放缓存句柄，数据和shape分别是int64和 \((1,)\)。

sample(beta)[源代码]

从缓存区中采样一批transition。

参数：

  • beta (float) - 控制采样校正程度的参数。0表示不校正，1表示完全校正。

返回：

  • indices (Tensor) - transition在缓存区中的索引。

  • weights (Tensor) - 用于校正采样偏差的权重。

  • transition - 采样得到的transition。

update_priorities(indices, priorities)[源代码]

更新transition的优先级。

参数：

  • indices (Tensor) - transition在缓存区中的索引。

  • priorities (Tensor) - transition优先级。

返回：

  • handle (Tensor) - 优先级经验回放缓存句柄，数据和shape分别是int64和 \((1,)\)。

mindspore_rl.environment

用于实现自定义环境的组件。

class mindspore_rl.environment.GymEnvironment(params, env_id=0)[源代码]

GymEnvironment将 Gym 封装成一个类来提供在MindSpore图模式下也能和Gym环境交互的能力。

参数：

  • params (dict) - 字典包含GymEnvironment类中所需要的所有参数。

配置参数

备注

name

Gym内游戏的名字

seed

Gym内使用的随机种子

  • env_id (int，可选) - 环境id，用于设置环境内种子，默认为第0个环境。默认：0。

支持平台：

Ascend GPU CPU

样例：

>>> env_params = {'name': 'CartPole-v0'}
>>> environment = GymEnvironment(env_params, 0)
>>> print(environment)
GymEnvironment<>
property action_space

获取环境的动作空间。

返回：

Space，环境的动作空间。

close()[源代码]

关闭环境以释放环境资源

返回：

  • Success (np.bool_) - 是否成功释放资源。

property config

获取环境的配置信息。

返回：

dict，一个包含环境信息的字典。

property done_space

获取环境的终止空间。

返回：

Space，环境的终止空间。

property observation_space

获取环境的状态空间。

返回：

Space，环境的状态空间。

render()[源代码]

渲染环境，仅支持PyNative模式。

reset()[源代码]

将环境重置为初始状态。reset方法一般在每一局游戏开始时使用，并返回环境的初始状态值。

返回：

Tensor，表示环境初始状态。

property reward_space

获取环境的状态空间。

返回：

Space，环境的奖励空间。

step(action)[源代码]

执行环境Step函数来和环境交互一回合。

参数：

  • action (Tensor) - 包含动作信息的Tensor。

返回：

  • state (Tensor) - 输入动作后的环境返回的新状态。

  • reward (Tensor) - 输入动作后环境返回的奖励。

  • done (Tensor) - 输入动作后环境是否终止。

class mindspore_rl.environment.MultiEnvironmentWrapper(env_instance, num_proc=1)[源代码]

MultiEnvironmentWrapper是平行环境场景下的包装器。用户实现自己的单环境类，并在配置文件中设置环境数量大于1时，框架将自动调用此类创建并行环境。

参数：

  • env_instance (list[Environment]) - 包含环境实例（继承Environment类）的List。

  • num_proc (int，可选) - 在和环境交互时使用的进程数量。默认值： 1。

支持平台：

Ascend GPU CPU

样例：

>>> env_params = {'name': 'CartPole-v0'}
>>> multi_env = [GymEnvironment(env_params), GymEnvironment(env_params)]
>>> wrapper = MultiEnvironmentWrapper(multi_env)
>>> print(wrapper)
MultiEnvironmentWrapper<>
property action_space

获取环境的动作空间。

返回：

Space，环境的动作空间。

close()[源代码]

关闭环境以释放环境资源

返回：

  • Success (np.bool_) - 是否成功释放资源。

property config

获取环境的配置信息。

返回：

dict，一个包含环境信息的字典。

property done_space

获取环境的终止空间。

返回：

Space，环境的终止空间。

property observation_space

获取环境的状态空间。

返回：

Space，返回环境的状态空间。

render()[源代码]

渲染环境，仅支持PyNative模式。

reset()[源代码]

将环境重置为初始状态。reset方法一般在每一局游戏开始时使用，并返回环境的初始状态值。

返回：

表示环境初始状态的Tensor List。

property reward_space

获取环境的状态空间。

返回：

Space，环境的奖励空间。

step(action)[源代码]

执行环境Step函数来和环境交互一回合。

参数：

  • action (Tensor) - 包含动作信息的Tensor。

返回：

  • state (list(Tensor)) - 输入动作后的环境返回的新状态List。

  • reward (list(Tensor)) - 输入动作后环境返回的奖励List。

  • done (list(Tensor)) - 输入动作后环境是否终止的List。

class mindspore_rl.environment.Environment(env_name=None, env=None, config=None)[源代码]

环境的虚基类。每一个子类环境都需要继承这个基类，并且需要在子类中实现_reset，_get_action，_step，_get_min_max_action和_get_min_max_observation。基类提供了自动将python实现的reset和step方法用mindspore算子（PyFunc）抱起来的能力，并且也提供了自动生成环境Space的能力。

参数：

  • env_name (str) - 子类环境的名字。Default：None

  • env (Environment) - 子类环境的实例。Default：None

  • config (dict) - 环境的配置信息，可以通过调用环境的config属性来获得。Default: None

property action_space

获取环境的动作空间。

返回：

  • action_space (Space) - 返回环境的动作空间。

close()[源代码]

关闭环境以释放环境资源

返回：

  • Success (np.bool_) - 是否成功释放资源。

property config

获取环境的配置信息。

返回：

  • config (dict) - 一个包含环境信息的字典。

property done_space

获取环境的终止空间。

返回：

  • done_space (Space) - 返回环境的终止空间。

property observation_space

获取环境的状态空间。

返回：

  • observation_space (Space) - 返回环境的状态空间。

reset()[源代码]

将环境重置为初始状态。reset方法一般在每一局游戏开始时使用，并返回环境的初始状态值以及其reset方法初始信息。

返回：

  • state (Tensor) - 一个表示环境初始状态的Tensor。

  • other (Tensor) - _reset方法中除了state以外的其他输出。

property reward_space

获取环境的状态空间。

返回：

  • reward_space (Space) - 返回环境的奖励空间。

step(action: Union[Tensor, np.ndarray])[源代码]

执行环境Step函数来和环境交互一回合。

参数：

  • action (Union[Tensor, np.ndarray]) - 包含动作信息的Tensor。

返回：

  • state (Tensor) - 输入动作后的环境返回的新状态。

  • reward (Tensor) - 输入动作后环境返回的奖励。

  • done (Tensor) - 输入动作后环境是否终止。

  • other (Tensor) - _step方法中剩下的返回值。

class mindspore_rl.environment.Space(feature_shape, dtype, low=None, high=None, batch_shape=None)[源代码]

包含环境动作/状态空间的类。

参数：

  • feature_shape (Union[list(int), tuple(int), int]) - 批处理前的动作/状态的Shape。

  • dtype (np.dtype) - 动作/状态空间的数据类型。

  • low (Union[int, float]，可选) - 动作/状态空间的下边界。默认：None。

  • high (Union[int, float]，可选) - 动作/状态空间的上边界。默认：None。

  • batch_shape (Union[list(int), tuple(int), int]，可选) - 矢量化的批量Shape。通常用于多环境和多智能体的场景。默认：None。

样例：

>>> action_space = Space(feature_shape=(6,), dtype=np.int32)
>>> print(action_space.ms_dtype)
Int32
property boundary

当前Space的边界。

返回：

当前空间的上下边界。

property is_discrete

当前Space是否为离散。

返回：

是否为离散空间。

property ms_dtype

当前Space的MindSpore数据类型。

返回：

当前空间的MindSpore的数据类型。

property np_dtype

当前Space的numpy数据类型。

返回：

当前空间的Numpy的数据类型。

property num_values

当前Space的可用动作数量。

返回：

当前空间可选动作的数量。

sample()[源代码]

从当前Space里随机采样一个合法动作。

返回：

  • action (Tensor) - 一个合法动作的Tensor。

property shape

批处理后的shape。

返回：

批处理后的Space的Shape。

class mindspore_rl.environment.MsEnvironment(kwargs=None)[源代码]

封装了内置环境（c++实现的环境）的类。

参数：

  • kwargs (dict) - 和环境相关的特定配置信息。详细信息请参见下表：

    环境名称

    配置参数

    默认值

    备注

    Tag环境

    seed

    42

    随机种子

    environment_num

    2

    环境数量

    predator_num

    10

    Predator的数量

    max_timestep

    100

    每一局游戏的最大步长

    map_length

    100

    地图的长

    map_width

    100

    地图的宽

    wall_hit_penalty

    0.1

    智能体撞击到墙的惩罚

    catch_reward

    10

    Predator抓捕到目标的奖励

    caught_penalty

    5

    Prey被捕捉到的惩罚

    step_cost

    0.01

    单个Step的基础成本
支持平台：

GPU

样例：

>>> config = {'name': 'Tag', 'predator_num': 4}
>>> env = MsEnvironment(config)
>>> observation = env.reset()
>>> action = Tensor(env.action_space.sample())
>>> observation, reward, done = env.step(action)
>>> print(observation.shape)
(2, 5, 21)
property action_space

获取环境的动作空间。

返回：

Space，环境的动作空间。

property config

获取环境的配置信息。

返回：

dict，一个包含环境信息的字典。

property done_space

获取环境的终止空间。

返回：

Space，环境的终止空间。

property observation_space

获取环境的状态空间。

返回：

Space，环境的状态空间。

reset()[源代码]

将环境重置为初始状态，并返回环境的初始状态值。

输入：

没有输入。

返回：

Tensor，表示环境初始状态。

支持平台：

GPU

样例：

>>> config = {'name': 'Tag', 'predator_num': 4}
>>> env = MsEnvironment(config)
>>> observation = env.reset()
>>> print(observation.shape)
(2, 5, 21)
property reward_space

获取环境的状态空间。

返回：

Space，环境的奖励空间。

step(action)[源代码]

执行环境Step函数来和环境交互一回合。

参数：

  • action (Tensor) - 由所有智能体提供的动作。

返回：

3 个张量的元组，状态、奖励和终止。

  • observation (Tensor) - 输入动作后的环境返回的所有智能体的新状态。

  • reward (Tensor) - 输入动作后环境返回的奖励。

  • done (Tensor) - 输入动作后环境是否终止。

支持平台：

GPU

样例：

>>> config = {'name': 'Tag', 'predator_num': 4}
>>> env = MsEnvironment(config)
>>> observation = env.reset()
>>> action = Tensor(env.action_space.sample())
>>> observation, reward, done = env.step(action)
>>> print(observation.shape)
(2, 5, 21)
class mindspore_rl.environment.EnvironmentProcess(proc_no, env_num, envs, actions, observations, initial_states)[源代码]

负责创建一个独立进程用作与一个或多个环境交互。

参数：

  • proc_no (int) - 被分配的进程号。

  • env_num (int) - 传入此进程的环境数量。

  • envs (list(Environment)) - 包含环境实例（继承Environment类）的List。

  • actions (Queue) - 用于将动作传递给环境进程的队列。

  • observations (Queue) - 用于将状态传递给环境进程的队列。

  • initial_states (Queue) - 用于将初始状态传递给环境进程的队列。

样例：

>>> from multiprocessing import Queue
>>> actions = Queue()
>>> observations = Queue()
>>> initial_states = Queue()
>>> proc_no = 1
>>> env_num = 2
>>> env_params = {'name': 'CartPole-v0'}
>>> multi_env = [GymEnvironment(env_params), GymEnvironment(env_params)]
>>> env_proc = EnvironmentProcess(proc_no, env_num, multi_env, actions, observations, initial_states)
>>> env_proc.start()
run()[源代码]

在子进程中运行的方法，可以在子类中重写。

class mindspore_rl.environment.StarCraft2Environment(params, env_id=0)[源代码]

StarCraft2Environment是一个SMAC的包装器。SMAC是WhiRL的一个基于暴雪星际争霸2开发的用于多智能体合作场景的强化学习环境。

SMAC通过调用暴雪星际争霸2的机器学习API和DeepMind的PySC2提供的API，方便算法中的智能体与星际争霸2交互来获得环境的状态和合法的动作。更多的信息请查阅官方的SMAC官方的GitHub： <https://github.com/oxwhirl/smac>。

参数：

  • params (dict) - 字典包含StarCraft2Environment类中所需要的所有参数。

    配置参数

    备注

    sc2_args

    一个用于创建SMAC实例的字典包含一些SMAC需要的key值 如map_name. 详细配置信息请查看官方GitHub。

  • env_id (int，可选) - 环境id，用于设置环境内种子，默认为第0个环境。默认值：0

支持平台：

Ascend GPU CPU

样例：

>>> env_params = {'sc2_args': {'map_name': '2s3z'}}
>>> environment = StarCraft2Environment(env_params, 0)
>>> print(environment)
property action_space

获取环境的动作空间。

返回：

Space，返回环境的动作空间。

close()[源代码]

关闭环境以释放环境资源

返回：

  • Success (np.bool_) - 是否成功释放资源。

property config

获取环境的配置信息。

返回：

dict，返回一个包含环境信息的字典。

property done_space

获取环境的终止空间。

返回：

Space，返回环境的终止空间。

get_step_info()[源代码]

在与环境交互后，获得环境的信息。

返回：

  • battle_won (Tensor) - 是否这局游戏取得胜利。

  • dead_allies (Tensor) - 己方单位阵亡数量。

  • dead_enemies (Tensor) - 敌方单位阵亡数量。

property observation_space

获取环境的状态空间。

返回：

返回环境的状态空间。

reset()[源代码]

将环境重置为初始状态。reset方法一般在每一局游戏开始时使用，并返回环境的初始状态值，全局状态以及新的合法动作。

返回：

tuple，包含了环境的初始状态值，全局状态以及新的合法动作这几个Tensor。

property reward_space

获取环境的奖励空间。

返回：

Space，返回环境的奖励空间。

step(action)[源代码]

执行环境Step函数来和环境交互一回合。

参数：

  • action (Tensor) - 包含动作信息的Tensor。

返回：

  • state (Tensor) - 输入动作后的环境返回的新状态。

  • reward (Tensor) - 输入动作后环境返回的奖励。

  • done (Tensor) - 输入动作后环境是否终止。

  • global_obs (Tensor) - 输入动作后环境返回的新的全局状态。

  • avail_actions (Tensor) - 输入动作后环境返回的新的合法动作。

class mindspore_rl.environment.TicTacToeEnvironment(params, env_id=0)[源代码]

井字棋是一款有名的纸笔游戏<https://en.wikipedia.org/wiki/Tic-tac-toe>。这个游戏的规则是两个玩家在一个3X3的格子上交互的画O和X。当三个相同的标记在水平，垂直或者对角线连成一条线时，对应的玩家将获得胜利。下图就是一个井字棋游戏的例子。

o

x

x

o

x

o
参数：

  • params (dict) - 字典包含TicTacToeEnvironment类中所需要的所有参数。

  • env_id (int，可选) - 环境id，用于设置环境内种子，默认为第0个环境。默认：0。

支持平台：

Ascend GPU CPU

样例：

>>> from mindspore_rl.environment import TicTacToeEnvironment
>>> env_params = {}
>>> environment = TicTacToeEnvironment(env_params, 0)
>>> print(environment)
TicTacToeEnvironment<>
property action_space

获取环境的动作空间。

返回：

Space，环境的动作空间。

calculate_rewards()[源代码]

返回当前状态的收益。

返回：

Tensor，表示当前状态收益。

property config

获取环境的配置信息。

返回：

dict，一个包含环境信息的字典。

current_player()[源代码]

返回当前状态下，轮到哪个玩家。

返回：

Tensor，表示当前玩家。

property done_space

获取环境的终止空间。

返回：

Space，环境的终止空间。

is_terminal()[源代码]

返回当前状态下，游戏是否已经终止。

返回：

当前状态下，游戏是否已经终止。

legal_action()[源代码]

返回当前状态的合法动作

返回：

Tensor，表示合法动作。

load(state)[源代码]

加载输入的状态。环境会根据输入的状态，更新当前的状态，合法动作和是否结束。

参数：

  • state (Tensor) - 输入的环境状态。

返回：

  • state (Tensor) - 存档点的状态。

  • reward (Tensor) - 存档点的收益。

  • done (Tensor) - 是否在输入存档点时，游戏已经结束。

max_utility()[源代码]

返回井字棋游戏的最大收益。

返回：

Tensor，表示最大收益。

property observation_space

获取环境的状态空间。

返回：

Space，环境的状态空间。

reset()[源代码]

将环境重置为初始状态。reset方法一般在每一局游戏开始时使用，并返回环境的初始状态值。

返回：

Tensor，表示环境初始状态。

property reward_space

获取环境的状态空间。

返回：

Space，环境的奖励空间。

save()[源代码]

返回一个环境的副本。在井字棋游戏中不需要返回环境的副本，因此他会返回当前状态。

返回：

一个代表当前状态的Tensor。

step(action)[源代码]

执行环境Step函数来和环境交互一回合。

参数：

  • action (Tensor) - 包含动作信息的Tensor。

返回：

  • state (Tensor) - 输入动作后的环境返回的新状态。

  • reward (Tensor) - 输入动作后环境返回的奖励。

  • done (Tensor) - 输入动作后环境是否终止。

total_num_player()[源代码]

返回总玩家数量。

返回：

Tensor，表示总玩家数量。

class mindspore_rl.environment.DeepMindControlEnvironment(params, env_id=0)[源代码]

DeepMindControlEnvironment将DeepMind Control Suite(DMC)通过MindSpore算子再次封装。它用于基于物理的模拟和强化学习环境，使用MUJOCO。

参数：

  • params (dict) - 字典包含DeepMindControlEnvironment类中所需要的所有参数。

配置参数

备注

env_name

DMC内游戏的名字

seed

DMC内使用的随机种子

camera

在渲染中使用的camera位置

action_repeat

同一个动作和环境交互几次

normalize_action

是否需要归一化输入动作

img_size

渲染图像的大小

  • env_id (int，可选) - 环境id，用于设置环境内种子，默认为第0个环境。默认：0。

样例：

>>> env_params = {'env_name': 'walker_walk', 'img_size': (64, 64),
'action_repeat': 2, 'normalize_action': True, 'seed': 1,
'episode_limits': 1000, 'prefill_value': 5000}
>>> environment = DeepMindControlEnvironment(env_params, 0)
>>> print(environment)
DeepMindControlEnvironment<>
property action_space

获取环境的动作空间。

返回：

Space，环境的动作空间。

close()[源代码]

关闭环境以释放环境资源

返回：

  • Success (np.bool_) - 是否成功释放资源。

property config

获取环境的配置信息。

返回：

dict，一个包含环境信息的字典。

property done_space

获取环境的终止空间。

返回：

Space，环境的终止空间。

property observation_space

获取环境的状态空间。

返回：

Space，环境的状态空间。

reset()[源代码]

将环境重置为初始状态。reset方法一般在每一局游戏开始时使用，并返回环境的初始状态值。

返回：

Tensor，表示环境初始状态。

property reward_space

获取环境的奖励空间。

返回：

Space，环境的奖励空间。

step(action)[源代码]

执行环境Step函数来和环境交互一回合。

参数：

  • action (Tensor) - 包含动作信息的Tensor。

返回：

  • state (Tensor) - 输入动作后的环境返回的新状态。

  • reward (Tensor) - 输入动作后环境返回的奖励。

  • done (Tensor) - 输入动作后环境是否终止。

  • discount (Tensor) - 环境对于当前状态返回的折扣

class mindspore_rl.environment.PettingZooMPEEnvironment(params, env_id=0)[源代码]

The PettingZooMPEEnvironment class is a wrapper that encapsulates PettingZoo to provide the ability to interact with PettingZoo environments in MindSpore Graph Mode.

参数

  • params (dict) –

    A dictionary contains all the parameters which are used in this class.

    Configuration Parameters

    Notices

    scenario_name

    the name of game

    num

    Number of Environment

    continuous_actions

    type of actions space

  • env_id (int, optional) – A integer which is used to set the seed of this environment, default value means the 0th environment. Default: 0.

Supported Platforms:

Ascend GPU CPU

样例

>>> env_params = {'name': 'simple_spread', 'num': 3, 'continuous_actions': False}
>>> environment = PettingZooMPEEnvironment(env_params)
>>> print(environment)
PettingZooMPEEnvironment<>
property action_space

Get the action space of the environment.

返回

The action space of environment.

close()[源代码]

Close the environment to release the resource.

返回

Success(np.bool_), Whether shutdown the process or threading successfully.

property config

Get the config of environment.

返回

A dictionary which contains environment’s info.

property done_space

Get the done space of the environment.

返回

The done space of environment.

property observation_space

Get the state space of the environment.

返回

The state space of environment.

render()[源代码]

Render the game. Only support on PyNative mode.

reset()[源代码]

Reset the environment to the initial state. It is always used at the beginning of each episode. It will return the value of initial state.

返回

A tensor which states for the initial state of environment.

property reward_space

Get the reward space of the environment.

返回

The reward space of environment.

step(action)[源代码]

Execute the environment step, which means that interact with environment once.

参数

action (Tensor) – A tensor that contains the action information.

返回

  • state (Tensor), the environment state after performing the action.

  • reward (Tensor), the reward after performing the action.

  • done (Tensor), whether the simulation finishes or not.

mindspore_rl.network

用于实现策略的网络组件。

class mindspore_rl.network.FullyConnectedNet(input_size, hidden_size, output_size, compute_type=mstype.float32)[源代码]

一个基本的全连接神经网络。

参数：

  • input_size (int) - 输入的数量。

  • hidden_size (int) - 隐藏层的数量。

  • output_size (int) - 输出大小的数量。

  • compute_type (mindspore.dtype) - 用于全连接层的数据类型。默认值： mindspore.float32。

样例：

>>> from mindspore import Tensor
>>> from mindspore_rl.network.fully_connected_net import FullyConnectedNet
>>> input = Tensor(np.ones([2, 4]).astype(np.float32))
>>> net = FullyConnectedNet(4, 10, 2)
>>> output = net(input)
>>> print(output.shape)
(2, 2)
construct(x)[源代码]

返回网络的输出。

参数：

  • x (Tensor) - 网络的输入张量。

返回：

网络的输出。

class mindspore_rl.network.FullyConnectedLayers(fc_layer_params, dropout_layer_params=None, activation_fn=nn.ReLU(), weight_init='normal', bias_init='zeros')[源代码]

这是一个全连接层的模块。用户可以输入任意数量的fc_layer_params，然后该模块可以创建相应数量的全链接层。

参数：

  • fc_layer_params (list[int]) - 全连接层输入和输出大小的值列表。例如，输入列表为[10，20，30]，模块将创建两个全连接层， 其输入和输出大小分别为(10, 20)和(20,30)。fc_layer_params的长度应大于等于3。

  • dropout_layer_params (list[float]) - 丢弃率的列表。如果输入为[0.5, 0.3]，则在每个全连接层之后将创建两个丢弃层。 dropout_layer_params的长度应小于fc_layer_params。 dropout_layer_params是个可选值。默认值： None。

  • activation_fn (Union[str, Cell, Primitive]) - 激活函数的实例。默认值： nn.ReLU()。

  • weight_init (Union[Tensor, str, Initializer, numbers.Number]) - 可训练的初始化权重参数。类型与 x 相同。str的值代表 Initializer 函数，如normal、uniform。默认值： ‘normal’。

  • bias_init (Union[Tensor, str, Initializer, numbers.Number]) - 可训练的初始化偏置参数。类型与 x 相同。str的值代表 Initializer 函数，如normal、uniform。默认值： ‘zeros’。

输入：

  • x (Tensor) - Tensor的shape为 \((*, fc\_layers\_params[0])\)。

输出：

Tensor的shape为 \((*, fc\_layers\_params[-1])\)。

样例：

>>> import numpy as np
>>> from mindspore import Tensor
>>> from mindspore_rl.network.fully_connected_net import FullyConnectedLayers
>>> input = Tensor(np.ones([2, 4]).astype(np.float32))
>>> net = FullyConnectedLayers(fc_layer_params=[4, 10, 2])
>>> output = net(input)
>>> print(output.shape)
(2, 2)
construct(x)[源代码]

返回网络的输出。

参数：

  • x (Tensor) - Tensor的shape为 \((*, fc\_layers\_params[0])\)。

返回：

Tensor的shape为 \((*, fc\_layers\_params[-1])\)。

class mindspore_rl.network.GruNet(input_size, hidden_size, weight_init='normal', num_layers=1, has_bias=True, batch_first=False, dropout=0.0, bidirectional=False, enable_fusion=True)[源代码]

GRU (门控递归单元)层。 将GRU层应用于输入。 有关详细信息，请参见： mindspore.nn.GRU 。

参数：

  • input_size (int) - 输入的特征数。

  • hidden_size (int) - 隐藏层的特征数量。

  • weight_init (str or Initializer) - 初始化方法，如normal、uniform。默认值： ‘normal’。

  • num_layers (int) - GRU层的数量。默认值： 1。

  • has_bias (bool) - cell中是否有偏置。默认值： True。

  • batch_first (bool) - 指定输入 x 的第一个维度是否为批处理大小。默认值： False。

  • dropout (float) - 如果不是0.0, 则在除最后一层外的每个GRU层的输出上附加 Dropout 层。默认值： 0.0。取值范围 [0.0, 1.0)。

  • bidirectional (bool) - 指定它是否为双向GRU，如果bidirectional=True则为双向，否则为单向。默认值： False。

  • enable_fusion (bool) - 是否需要使用GRU的融合算子。默认值：True。

输入：

  • x_in (Tensor) - 数据类型为mindspore.float32和shape为 \((seq\_len, batch\_size, input\_size)\) 或 \((batch\_size, seq\_len, input\_size)\) 的Tensor。

  • h_in (Tensor) - 数据类型为mindspore.float32和shape为 \((num\_directions * num\_layers, batch\_size, hidden\_size)\) 的Tensor。h_in 的数据类型必须和 x_in 一致。

输出：

元组，包含(x_out, h_out)。

  • x_out (Tensor) - shape为 \((seq\_len, batch\_size, num\_directions * hidden\_size)\) 的Tensor。

  • h_out (Tensor) - shape为 \((num\_directions * num\_layers, batch\_size, hidden\_size)\) 的Tensor。

样例：

>>> net = GruNet(10, 16, 2, has_bias=True, bidirectional=False)
>>> x_in = Tensor(np.ones([3, 5, 10]).astype(np.float32))
>>> h_in = Tensor(np.ones([1, 5, 16]).astype(np.float32))
>>> x_out, h_out = net(x_in, h_in)
>>> print(x_out.shape)
(3, 5, 16)
construct(x_in, h_in)[源代码]

gru网络的正向输出。

参数：

  • x_in (Tensor) - 数据类型为mindspore.float32和shape为 \((seq\_len, batch\_size, input\_size)\) 或 \((batch\_size, seq\_len, input\_size)\) 的Tensor。

  • h_in (Tensor) - 数据类型为mindspore.float32和shape为 \((num\_directions * num\_layers, batch\_size, hidden\_size)\) 的Tensor。h_in 的数据类型必须和 x_in 一致。

返回：

  • x_out (Tensor) - shape为 \((seq\_len, batch\_size, num\_directions * hidden\_size)\) 的Tensor。

  • h_out (Tensor) - shape为 \((num\_directions * num\_layers, batch\_size, hidden\_size)\) 的Tensor。

mindspore_rl.policy

RL 算法中使用的策略。

class mindspore_rl.policy.Policy[源代码]

策略的虚基类。在调用模型之前，应该重写此类。

construct(*inputs, **kwargs)[源代码]

构造函数接口。由用户继承使用，参数可参考 EpsilonGreedyPolicy， RandomPolicy 等。

参数：

  • inputs - 取决于用户的定义。

  • kwargs - 取决于用户的定义。

返回：

取决于用户的定义。通常返回一个动作值或者动作的概率分布。

class mindspore_rl.policy.RandomPolicy(action_space_dim)[源代码]

在[0, action_space_dim)之间产生随机动作。

参数：

  • action_space_dim (int) - 动作空间的维度。

样例：

>>> action_space_dim = 2
>>> policy = RandomPolicy(action_space_dim)
>>> output = policy()
>>> print(output.shape)
(1,)
construct()[源代码]

返回[0, action_space_dim)之间的随机数。

返回：

[0, action_space_dim)之间的随机数。

class mindspore_rl.policy.GreedyPolicy(input_network)[源代码]

基于给定的贪婪策略生成采样动作。

参数：

  • input_network (Cell) - 用于按输入状态产生动作的网络。

样例：

>>> state_dim, hidden_dim, action_dim = 4, 10, 2
>>> input_net = FullyConnectedNet(state_dim, hidden_dim, action_dim)
>>> policy = GreedyPolicy(input_net)
>>> state = Tensor(np.ones([2, 4]).astype(np.float32))
>>> output = policy(state)
>>> print(output.shape)
(2,)
construct(state)[源代码]

返回最佳动作。

参数：

  • state (Tensor) - 网络的输入状态Tensor。

返回：

action_max，输出最佳动作。

class mindspore_rl.policy.EpsilonGreedyPolicy(input_network, size, epsi_high, epsi_low, decay, action_space_dim)[源代码]

基于给定的epsilon-greedy策略生成采样动作。

参数：

  • input_network (Cell) - 返回策略动作的输入网络。

  • size (int) - epsilon的shape。

  • epsi_high (float) - 探索的上限epsilon值，介于[0, 1]。

  • epsi_low (float) - 探索的下限epsilon值，介于[0, epsi_high]。

  • decay (float) - epsilon的衰减系数。

  • action_space_dim (int) - 动作空间的维度。

样例：

>>> state_dim, hidden_dim, action_dim = (4, 10, 2)
>>> input_net = FullyConnectedNet(state_dim, hidden_dim, action_dim)
>>> policy = EpsilonGreedyPolicy(input_net, 1, 0.1, 0.1, 100, action_dim)
>>> state = Tensor(np.ones([1, state_dim]).astype(np.float32))
>>> step =  Tensor(np.array([10,]).astype(np.float32))
>>> output = policy(state, step)
>>> print(output.shape)
(1,)
construct(state, step)[源代码]

构造函数接口。

参数：

  • state (Tensor) - 网络的输入Tensor。

  • step (Tensor) - 当前step, 影响epsilon的衰减。

返回：

输出动作。

mindspore_rl.utils

RL 算法中工具组件。

class mindspore_rl.utils.DiscountedReturn(gamma, need_bprop=False)[源代码]

计算折扣回报。

设折扣回报为 \(G\)，折扣系数为 \(\gamma\)，奖励为 \(R\)，时间步 \(t\)，最大时间步 \(N\)。 则 \(G_{t} = \Sigma_{t=0}^N{\gamma^tR_{t+1}}\)。

对于奖励序列包含多个episode的情况， \(done\) 用来标识episode边界， \(last\_state\_value\) 表示最后一个epsode的最后一个step的价值。

参数：

  • gamma (float) - 折扣系数。

  • need_bprop (bool) - 是否需要计算discounted return的反向，默认值：False。

输入：

  • reward (Tensor) - 包含多个episode的奖励序列。 张量的维度 \((Timestep, Batch, ...)\)。

  • done (Tensor) - Episode结束标识。 张量维度 \((Timestep, Batch)\)。

  • last_state_value (Tensor) - 表示最后一个epsode的最后一个step的价值， 张量的维度 \((Batch, ...)\)。

返回：

折扣回报。

样例：

>>> net = DiscountedReturn(gamma=0.99)
>>> reward = Tensor([[1, 1, 1, 1]], dtype=mindspore.float32)
>>> done = Tensor([[False, False, True, False]])
>>> last_state_value = Tensor([2.], dtype=mindspore.float32)
>>> ret = net(reward, done, last_state_value)
>>> print(output.shape)
(2, 2)
class mindspore_rl.utils.OUNoise(stddev, damping, action_shape)[源代码]

在action上加入Ornstein-Uhlenbeck (OU)噪声。

设均值为0的正态分布为 \(N(0, stddev)\)， 则下一个时序值是 \(x\_next = (1 - damping) * x - N(0, stddev)\)， 加入OU噪声的action是 \(action += x\_next\)。

参数：

  • stddev (float) - Ornstein-Uhlenbeck (OU) 噪声标准差。

  • damping (float) - Ornstein-Uhlenbeck (OU) 噪声阻尼。

  • action_shape (tuple) - 动作的维度。

输入：

  • actions (Tensor) - 添加OU噪声之前的动作。

输出：

  • actions (Tensor) - 添加OU噪声之后的动作。

样例：

>>> import numpy as np
>>> from mindspore import Tensor
>>> from mindspore_rl.utils import OUNoise
>>> action_shape = (6,)
>>> actions = Tensor(np.ones(action_shape))
>>> net = OUNoise(stddev=0.2, damping=0.15, action_shape=action_shape)
>>> actions = net(actions)
>>> print(actions.shape)
(6,)
class mindspore_rl.utils.SoftUpdate(factor, update_interval, behavior_params, target_params)[源代码]

采用滑动平均方式更新目标网络的参数。

设目标网络参数为 \(target\_param\)，行为网络参数为 \(behavior\_param\)， 滑动平均系数为 \(factor\)。 则 \(target\_param = (1. - factor) * behavior\_param + factor * target\_param\)。

参数：

  • factor (float) - 滑动平均系数，范围[0, 1]。

  • update_interval (int) - 目标网络参数更新间隔。

  • behavior_params (list(Parameter)) - 行为网络参数列表。

  • target_params (list(Parameter)) - 目标网络参数列表。

样例：

>>> import numpy as np
>>> import mindspore.nn as nn
>>> from mindspore.common.parameter import ParameterTuple
>>> from mindspore_rl.utils import SoftUpdate
>>> class Net(nn.Cell):
>>>     def __init__(self):
>>>         super(Net, self).__init__()
>>>         self.behavior_params = ParameterTuple(nn.Dense(10, 20).trainable_params())
>>>         self.target_params = ParameterTuple(nn.Dense(10, 20).trainable_params())
>>>         self.updater = SoftUpdate(0.9, 2, self.behavior_params, self.target_params)
>>>     def construct(self):
>>>         return self.updater()
>>> net = Net()
>>> for _ in range(10):
>>>     net()
>>> np.allclose(net.behavior_params[0].asnumpy(), net.target_params[0].asnumpy(), atol=1e-5)
True
class mindspore_rl.utils.CallbackParam[源代码]

包含回调函数执行时需要的参数。

class mindspore_rl.utils.CallbackManager(callbacks)[源代码]

依次执行回调函数。

参数：

  • callbacks (list[Callback]) - 一个包含回调函数的list。

begin(params)[源代码]

在训练执行开始调用，仅执行一次。

参数：

  • params (CallbackParam) - begin执行用的参数。

end(params)[源代码]

在训练执行结束调用，仅执行一次。

参数：

  • params (CallbackParam) - end执行用的参数。

episode_begin(params)[源代码]

在每个episode执行前调用。

参数：

  • params (CallbackParam) - episode_begin执行用的参数。

episode_end(params)[源代码]

在每个episode执行后调用。

参数：

  • params (CallbackParam) - episode_end执行用的参数。

class mindspore_rl.utils.LossCallback(print_rate=1)[源代码]

在每个episode结束时打印loss值。

参数：

  • print_rate (int, 可选) - 打印loss的频率。默认值：1。

样例：

>>> from mindspore_rl.utils.callback import LossCallback
>>> from mindspore_rl.core import Session
>>> from mindspore_rl.algorithm.dqn import config
>>> loss_cb = LossCallback()
>>> cbs = [loss_cb]
>>> session = Session(config.algorithm_config, None, None, cbs)
episode_end(params)[源代码]

在每个episode执行后调用，打印loss值。

参数：

  • params (CallbackParam) - 训练参数，用于获取结果。

class mindspore_rl.utils.TimeCallback(print_rate=1, fixed_steps_in_episode=None)[源代码]

在每个episode结束时打印耗时。

参数：

  • print_rate (int, 可选) - 打印耗时的频率, 默认值：1。

  • fixed_steps_in_episode (int, 可选) - 如果每个episode的steps是固定的，则提供一个固定steps值。如果时None, params中需要提供实际steps。默认值：None。

样例：

>>> from mindspore_rl.utils.callback import TimeCallback
>>> from mindspore_rl.core import Session
>>> from mindspore_rl.algorithm.dqn import config
>>> time_cb = TimeCallback()
>>> cbs = [time_cb]
>>> session = Session(config.algorithm_config, None, None, cbs)
episode_begin(params)[源代码]

在每个episode执行前调用，打印耗时。

参数：

  • params (CallbackParam) - 训练参数，用于获取结果。

episode_end(params)[源代码]

在每个episode执行后记录时间。

参数：

  • params (CallbackParam) - 训练参数，用于获取结果。

class mindspore_rl.utils.CheckpointCallback(save_per_episode=0, directory=None, max_ckpt_nums=5)[源代码]

保存模型的checkpoint文件，保留最新的 max_ckpt_nums 个。

参数：

  • save_per_episode (int, 可选) - 保存ckpt文件的频率。默认值：0（不保存）。

  • directory (str, 可选) - 保存ckpt文件的路径。默认：’./’。

  • max_ckpt_nums (int, 可选) - 最大保留ckpt的个数。默认值：5。

样例：

>>> from mindspore_rl.utils.callback import CheckpointCallback
>>> from mindspore_rl.core import Session
>>> from mindspore_rl.algorithm.dqn import config
>>> ckpt_cb = CheckpointCallback()
>>> cbs = [ckpt_cb]
>>> session = Session(config.algorithm_config, None, None, cbs)
episode_end(params)[源代码]

在每个episode执行后调用，保存ckpt文件。

参数：

  • params (CallbackParam) - 训练参数，用于获取结果。

class mindspore_rl.utils.EvaluateCallback(eval_rate=0)[源代码]

推理回调。

参数：

  • eval_rate (int, 可选) - 推理的频率。默认值：0（不推理）。

样例：

>>> from mindspore_rl.utils.callback import EvaluateCallback
>>> from mindspore_rl.core import Session
>>> from mindspore_rl.algorithm.dqn import config
>>> eval_cb = EvaluateCallback()
>>> cbs = [eval_cb]
>>> session = Session(config.algorithm_config, None, None, cbs)
begin(params)[源代码]

在训练开始前保存推理频率。

参数：

  • params (CallbackParam) - episode开始时用的参数。

episode_end(params)[源代码]

在每个episode执行后调用，推理并打印结果。

参数：

  • params (CallbackParam) - episode结束后用的参数。

mindspore_rl.utils.update_config(config, env_yaml, algo_yaml)[源代码]

通过传入的yaml文件更新config。参考 mindspore_rl/algorithm/dqn/config.py, mindspore_rl/example/env_yaml/ and mindspore_rl/example/algo_yaml/ 获取用法。

参数：

  • config (dict) - 待更新的配置。

  • env_yaml (str) - 环境yaml文件。

  • algo_yaml (str) - 算法yaml文件。

class mindspore_rl.utils.MCTS(env, tree_type, node_type, root_player, customized_func, device, args, has_init_reward=False, max_action=- 1.0, max_iteration=1000)[源代码]

蒙特卡洛树搜索（MCTS）是一种通用搜索决策算法，在棋类游戏（如围棋，国际象棋）中效果尤为显著。MCTS在2006年被首次提出。一个通用的MCTS会有以下四个阶段：

  1. 选择（Selection） - 根据选择策略（如UCT, RAVE, AMAF等）选择下一个节点。

  2. 扩展（Expansion） - 除非搜索达到了终止节点，新的子节点都会被添加到选择阶段达到的叶节点。

  3. 模拟（Simulation） - 使用一个算法（随机，神经网络或者其他算法）去获得当前状态的回报。

  4. 反向传播（Backpropagation） - 把模拟计算出的回报传播给所有经过的节点。

随着时间的推移，MCTS中的四步都更新迭代。AlphaGo中就在MCTS中引入了神经网络，使得MCTS更加强大。

本MCTS类由MindSpore算子组成。用户可以直接使用提供的MCTS算法（如CPUCommon，CPUVanilla），或者通过继承C++中的MonteCarloTreeNode去开发自己的MCTS算法。

参数：

  • env (Environment) - 必须是Environment的子类。

  • tree_type (str) - 树类型的名字。

  • node_type (str) - 节点类型的名字。

  • root_player (float) - 根节点的玩家，数值需要小于总玩家数。

  • customized_func (AlgorithmFunc) - 算法相关的类。更多信息请参考AlgorithmFunc的文档。

  • device (str) - 运行MCTS的设备[‘CPU’, ‘GPU’]，Ascend当前不支持。

  • args (Tensor) - 在MctsCreation中传入的常量值。请参考以下表格根据算法传入输入值。这里传入的值不会在 restore_tree_data 方法中被重置。

  • has_init_reward (bool，可选) - 是否把奖励在初始化时传给节点。默认：False。

  • max_action (float，可选) - 环境的最大动作。当max_action是-1.0时，环境的step函数只会获得最后一个动作，否则环境的step函数会获得所有动作。默认：-1.0.

  • max_iteration (int，可选) - 最多的训练迭代次数。默认：1000.

MCTS树类型

MCTS节点类型

配置参数

备注

CPUCommon

CPUVanilla

UCT常量

UCT常量被使用在Selection 阶段，去计算UCT值。

GPUCommon

GPUVanilla

UCT常量

样例：

>>> from mindspore import Tensor
>>> import mindspore as ms
>>> from mindspore_rl.environment import TicTacToeEnvironment
>>> from mindspore_rl.utils import VanillaFunc
>>> from mindspore_rl.utils import MCTS
>>> env = TicTacToeEnvironment(None)
>>> vanilla_func = VanillaFunc(env)
>>> uct = (Tensor(uct, ms.float32),)
>>> root_player = 0.0
>>> mcts = MCTS(env, "CPUCommon", "CPUVanilla", root_player, vanilla_func, device, args=uct)
>>> action, handle = mcts.mcts_search()
>>> print(action)
destroy(handle)[源代码]

销毁当前这棵树。请在算法结束或不再需要这棵树时调用。

参数：

  • handle (mindspore.int64) - 唯一的蒙特卡洛树句柄。

返回：

  • action (mindspore.bool_) - 是否成功重置。

mcts_search是MCTS中的主要方法。调用此方法会返回当前状态下的最优动作。

参数：

  • *args (Tensor) - 在迭代中会更新的变量，并且在调用 restore_tree_data 时会重置。输入值需要和传入的MCTS树以及节点对应。

返回：

  • action (mindspore.int32) - 蒙特卡洛树搜索返回的动作。

  • handle (mindspore.int64) - 唯一的蒙特卡洛树句柄。

restore_tree_data(handle)[源代码]

restore_tree_data会重置树中的所有信息，回到只有根节点的状态。

参数：

  • handle (mindspore.int64) - 唯一的蒙特卡洛树句柄。

返回：

  • action (mindspore.bool_) - 是否成功重置。

class mindspore_rl.utils.VanillaFunc(env)[源代码]

这是Vanilla MCTS的自定义算法。每个动作的先验概率是一个均匀分布。simulation中会进行随机选择动作从而获得结果。

参数：

  • env (Environment) - 传入的环境。

样例：

>>> env = TicTacToeEnvironment(None)
>>> vanilla_func = VanillaFunc(env)
>>> legal_action = env.legal_action()
>>> prior = vanilla_func.calculate_prior(legal_action, legal_action)
>>> print(prior)
calculate_prior(new_state, legal_action)[源代码]

计算输入合法动作的先验概率。

参数：

  • new_state (mindspore.float32) - 环境的状态。

  • legal_action (mindspore.int32) - 环境输出的合法动作。

返回：

  • prior (mindspore.float32) - 每个动作的先验概率。

simulation(new_state)[源代码]

计算输入状态的奖励（评估价值）。

参数：

  • new_state (mindspore.float32) - 环境的状态。

返回：

  • rewards (mindspore.float32) - simulation的结果。

class mindspore_rl.utils.AlgorithmFunc[源代码]

这是MCTS中用户用来自定义传入算法的基类。用户需要继承这个基类并且按照提供的输入输出实现对应的方法。

calculate_prior(new_state, legal_action)[源代码]

计算输入合法动作的先验概率。

参数：

  • new_state (mindspore.float32) - 环境的状态。

  • legal_action (mindspore.int32) - 环境输出的合法动作。

返回：

  • prior (mindspore.float32) - 每个动作的先验概率。

simulation(new_state)[源代码]

计算输入状态的奖励（评估价值）。

参数：

  • new_state (mindspore.float32) - 环境的状态。

返回：

  • rewards (mindspore.float32) - simulation的结果。

class mindspore_rl.utils.BatchWrite[源代码]

写一个list的参数覆盖到目标值。

警告

  • 这是一个实验特性，未来有可能被修改或删除。

支持平台：

GPU CPU

样例：

>>> import mindspore
>>> from mindspore import nn
>>> from mindspore.common.parameter import Parameter, ParameterTuple
>>> from mindspore_rl.utils import BatchWrite
>>> class SourceNet(nn.Cell):
...   def __init__(self):
...     super(SourceNet, self).__init__()
...     self.a = Parameter(Tensor(0.5, mstype.float32), name="a")
...     self.dense = nn.Dense(in_channels=16, out_channels=1, weight_init=0)
>>> class DstNet(nn.Cell):
...   def __init__(self):
...     super(DstNet, self).__init__()
...     self.a = Parameter(Tensor(0.1, mstype.float32), name="a")
...     self.dense = nn.Dense(in_channels=16, out_channels=1)
>>> class Write(nn.Cell):
...   def __init__(self, dst, src):
...     super(Write, self).__init__()
...     self.w = BatchWrite()
...     self.dst = ParameterTuple(dst.trainable_params())
...     self.src = ParameterTuple(src.trainable_params())
...   def construct(self):
...     success = self.w(self.dst, self.src)
...     return success
>>> dst_net = DstNet()
>>> source_net = SourceNet()
>>> nets = nn.CellList()
>>> nets.append(dst_net)
>>> nets.append(source_net)
>>> success = Write(nets[0], nets[1])()
construct(dst, src)[源代码]

将 src 中的参数覆盖到 dst 。

参数：

  • dst (tuple(Parameters)) - 目标位置的参数列表。

  • src (tuple(Parameters)) - 源位置的参数列表。

返回：

True。

class mindspore_rl.utils.BatchRead[源代码]

读一个list的参数覆盖到目标值。

警告

  • 这是一个实验特性，未来有可能被修改或删除。

支持平台：

GPU CPU

样例：

>>> import mindspore
>>> from mindspore import nn
>>> from mindspore.common.parameter import Parameter, ParameterTuple
>>> from mindspore_rl.utils import BatchRead
>>> class SNet(nn.Cell):
...   def __init__(self):
...     super(SNet, self).__init__()
...     self.a = Parameter(Tensor(0.5, mstype.float32), name="a")
...     self.dense = nn.Dense(in_channels=16, out_channels=1, weight_init=0)
>>> class DNet(nn.Cell):
...   def __init__(self):
...     super(DNet, self).__init__()
...     self.a = Parameter(Tensor(0.1, mstype.float32), name="a")
...     self.dense = nn.Dense(in_channels=16, out_channels=1)
>>> class Read(nn.Cell):
...   def __init__(self, dst, src):
...     super(Read, self).__init__()
...     self.read = BatchRead()
...     self.dst = ParameterTuple(dst.trainable_params())
...     self.src = ParameterTuple(src.trainable_params())
...   def construct(self):
...     success = self.read(self.dst, self.src)
...     return success
>>> dst_net = DNet()
>>> source_net = SNet()
>>> nets = nn.CellList()
>>> nets.append(dst_net)
>>> nets.append(source_net)
>>> success = Read(nets[0], nets[1])()
construct(dst, src)[源代码]

读取 src 中的参数覆盖到 dst 。

参数：

  • dst (tuple(Parameters)) - 目标位置的参数列表。

  • src (tuple(Parameters)) - 源位置的参数列表。

返回：

True。

class mindspore_rl.utils.TensorArray(dtype, element_shape, dynamic_size=True, size=0, name='TA')[源代码]

用来存Tensor的TensorArray。

警告

  • 这是一个实验特性，未来有可能被修改或删除。

参数：

  • dtype (mindspore.dtype) - TensorArray的数据类型。

  • element_shape (tuple(int)) - TensorArray中每个Tensor的shape。

  • dynamic_size (bool，可选) - 如果是True，则该数组可以动态增长，否则为固定大小。默认：True。

  • size (int，可选) - 如果 dynamic_size=False , 则 size 表示该数组的最大容量。

  • name (str，可选) - TensorArray的名字，任意str。默认：”TA”。

支持平台：

GPU CPU

样例：

>>> import mindspore
>>> from mindspore_rl.utils import TensorArray
>>> ta = TensorArray(mindspore.int64, ())
>>> ta.write(0, 1)
>>> ta.write(1, 2)
>>> ans = ta.read(1)
>>> print(ans)
2
>>> s = ta.stack()
>>> print(s)
[1 2]
>>> ta.clear()
>>> ta.write(0, 3)
>>> ans = ta.read(0)
>>> print(ans)
3
>>> ta.close()
clear()[源代码]

清理创建的TensorArray。仅重置该数组，清理数据和重置大小，保留数组实例。

返回：

True。

close()[源代码]

关闭TensorArray。

警告

  • 一旦关闭了TensorArray，每个属于该TensorArray的方法都将失效。所有该数组中的资源也将被清除。如果该数组还将在别的地方使用，如下一个循环，请用 clear 代替。

返回：

True。

read(index)[源代码]

从TensorArray的指定位置读Tensor。

参数：

  • index ([int, mindspore.int64]) - 读取的位置。

返回：

Tensor, 指定位置的值。

size()[源代码]

TensorArray的逻辑大小。

返回：

Tensor, TensorArray大小。

stack()[源代码]

堆叠TensorArray中的Tensor为一个整体。

返回：

Tensor, TensorArray中的所有Tensor将堆叠成一个整体。

write(index, value)[源代码]

向TensorArray的指定位置写入值（Tensor）。

参数：

  • index ([int, mindspore.int64]) - 写入的位置。

  • value (Tensor) - 写入的Tensor。

返回：

True。

class mindspore_rl.utils.TensorsQueue(dtype, shapes, size=0, name='TQ')[源代码]

用来存TensorsQueue的队列。

警告

  • 这是一个实验特性，未来有可能被修改或删除。

参数：

  • dtype (mindspore.dtype) - TensorsQueue的数据类型。每个Tensor需要相同的类型。

  • shapes (tuple[int64]) - TensorsQueue中每个Tensor的shape。

  • size (int，可选) - TensorsQueue的大小。默认：0。

  • name (str，可选) - TensorsQueue的名字。默认：”TQ”。

异常：

  • TypeError - dtype 不是 MindSpore 数字类型.

  • ValueError - size 小于0.

  • ValueError - shapes 的长度小于1.

支持平台：

GPU CPU

样例：

>>> import mindspore as ms
>>> from mindspore import Tensor
>>> from mindspore_rl.utils import TensorsQueue
>>> data1 = Tensor([[0, 1], [1, 2]], dtype=ms.float32)
>>> data2 = Tensor([1], dtype=ms.float32)
>>> tq = TensorsQueue(dtype=ms.float32, shapes=((2, 2), (1,)), size=5)
>>> tq.put((data1, data2))
>>> ans = tq.pop()
clear()[源代码]

清理创建的TensorsQueue。仅重置该队列，清理数据和重置大小，保留队列实例。

返回：

True。

close()[源代码]

关闭TensorsQueue。

警告

  • 一旦关闭了TensorsQueue，每个属于该TensorsQueue的方法都将失效。所有该队列中的资源也将被清除。如果该队列还将在别的地方使用，如下一个循环，请用 clear 代替。

返回：

True。

get()[源代码]

从TensorsQueue的头部取出一个元素。

返回：

tuple(Tensor)，一个元素。

pop()[源代码]

从TensorsQueue的头部取出一个元素并删除。

返回：

tuple(Tensor)，一个元素。

put(element)[源代码]

向TensorsQueue的底部放入元素（tuple(Tensors)）。

参数：

  • element (tuple(Tensor) 或 list[tensor]) - 写入的元素。

返回：

True。

size()[源代码]

TensorsQueue的已使用大小。

返回：

Tensor(mindspore.int64)，TensorsQueue的已使用大小。