memory_tools

class xuance.common.memory_tools.Buffer(observation_space: gymnasium.Space, action_space: gymnasium.Space, auxiliary_info_shape: dict | None, num_envs: int, buffer_size: int)

Bases: ABC

Basic buffer single-agent DRL algorithms.

Parameters:

• observation_space – the space for observation data.

• action_space – the space for action data.

• auxiliary_info_shape – the shape for auxiliary data if needed.

abstractmethod clear(*args)

finish_path(*args)

property full

abstractmethod sample(*args)

abstractmethod store(*args)

class xuance.common.memory_tools.DummyOffPolicyBuffer(observation_space: gymnasium.Space, action_space: gymnasium.Space, auxiliary_shape: dict | None, n_envs: int, buffer_size: int, batch_size: int)

Bases: Buffer

Replay buffer for off-policy DRL algorithms.

Parameters:

• observation_space – the observation space of the environment.

• action_space – the action space of the environment.

• auxiliary_shape – data shape of auxiliary information (if exists).

• n_envs – number of parallel environments.

• buffer_size – the total size of the replay buffer.

• batch_size – size of transition data for a batch of sample.

clear()

sample(batch_size=None)

store(obs, acts, rews, terminals, next_obs)

class xuance.common.memory_tools.DummyOffPolicyBuffer_Atari(observation_space: gymnasium.Space, action_space: gymnasium.Space, auxiliary_shape: dict | None, n_envs: int, buffer_size: int, batch_size: int)

Bases: DummyOffPolicyBuffer

Replay buffer for off-policy DRL algorithms and Atari tasks.

Parameters:

• observation_space – the observation space of the environment.

• action_space – the action space of the environment.

• auxiliary_shape – data shape of auxiliary information (if exists).

• n_envs – number of parallel environments.

• buffer_size – the total size of the replay buffer.

• batch_size – batch size of transition data for a sample.

clear()

class xuance.common.memory_tools.DummyOnPolicyBuffer(observation_space: gymnasium.Space, action_space: gymnasium.Space, auxiliary_shape: dict | None, n_envs: int, horizon_size: int, use_gae: bool = True, use_advnorm: bool = True, gamma: float = 0.99, gae_lam: float = 0.95)

Bases: Buffer

Replay buffer for on-policy DRL algorithms.

Parameters:

• observation_space – the observation space of the environment.

• action_space – the action space of the environment.

• auxiliary_shape – data shape of auxiliary information (if exists).

• n_envs – number of parallel environments.

• horizon_size – max length of steps to store for one environment.

• use_gae – if use GAE trick.

• use_advnorm – if use Advantage normalization trick.

• gamma – discount factor.

• gae_lam – gae lambda.

clear()

finish_path(val, i)

property full

sample(indexes)

store(obs, acts, rews, value, terminals, aux_info=None)

class xuance.common.memory_tools.DummyOnPolicyBuffer_Atari(observation_space: gymnasium.Space, action_space: gymnasium.Space, auxiliary_shape: dict | None, n_envs: int, horizon_size: int, use_gae: bool = True, use_advnorm: bool = True, gamma: float = 0.99, gae_lam: float = 0.95)

Bases: DummyOnPolicyBuffer

Replay buffer for on-policy DRL algorithms and Atari tasks.

Parameters:

• observation_space – the observation space of the environment.

• action_space – the action space of the environment.

• auxiliary_shape – data shape of auxiliary information (if exists).

• n_envs – number of parallel environments.

• horizon_size – max length of steps to store for one environment.

• use_gae – if use GAE trick.

• use_advnorm – if use Advantage normalization trick.

• gamma – discount factor.

• gae_lam – gae lambda.

clear()

class xuance.common.memory_tools.EpisodeBuffer

Bases: object

Episode buffer for DRQN agent.

put(transition)

sample(lookup_step=None, idx=None) → Dict[str, numpy.ndarray]

class xuance.common.memory_tools.PerOffPolicyBuffer(observation_space: gymnasium.Space, action_space: gymnasium.Space, auxiliary_shape: dict | None, n_envs: int, buffer_size: int, batch_size: int, alpha: float = 0.6)

Bases: Buffer

Prioritized Replay Buffer.

Parameters:

• observation_space – the observation space of the environment.

• action_space – the action space of the environment.

• auxiliary_shape – data shape of auxiliary information (if exists).

• n_envs – number of parallel environments.

• buffer_size – the total size of the replay buffer.

• batch_size – batch size of transition data for a sample.

• alpha – prioritized factor.

clear()

sample(beta)

store(obs, acts, rews, terminals, next_obs)

update_priorities(idxes, priorities)

class xuance.common.memory_tools.RecurrentOffPolicyBuffer(observation_space: gymnasium.Space, action_space: gymnasium.Space, auxiliary_shape: dict | None, n_envs: int, buffer_size: int, batch_size: int, episode_length: int, lookup_length: int)

Bases: Buffer

Replay buffer for DRQN-based algorithms.

Parameters:

• observation_space – the observation space of the environment.

• action_space – the action space of the environment.

• auxiliary_shape – data shape of auxiliary information (if exists).

• n_envs – number of parallel environments.

• buffer_size – the size of replay buffer that stores episodes of data.

• batch_size – batch size of transition data for a sample.

• episode_length – data length for an episode.

• lookup_length – the length of history data.

can_sample()

clear(*args)

property full

sample()

store(episode)

class xuance.common.memory_tools.SequentialReplayBuffer(observation_space: gymnasium.Space, action_space: gymnasium.Space, auxiliary_shape: dict | None, n_envs: int, buffer_size: int, batch_size: int)

Bases: Buffer

Sequential Replay buffer for Dreamerv3

Parameters:

• observation_space – the observation space of the environment.

• action_space – the action space of the environment.

• auxiliary_shape – data shape of auxiliary information (if exists).

• n_envs – number of parallel environments.

• buffer_size – the total size of the replay buffer.

• batch_size – size of transition data for a batch of sample.

clear()

sample(seq_len: int)

Sample elements from the replay buffer in a sequential manner, without considering the episode boundaries. :param seq_len: :type seq_len: int

Returns:

the sampled dictionary with a shape of [envs, sequence_length, batch_size, …].

Return type:

Dict[str, np.ndarray]

store(obs, acts, rews, terms, truncs, is_first)

Parameters:

• arrays (all arguments are numpy) – [envs, ~] if ~ != 1 else [envs, ]

• shape – [envs, ~] if ~ != 1 else [envs, ]

Returns:

xuance.common.memory_tools.create_memory(shape: tuple | dict | None, n_envs: int, n_size: int, dtype: type = numpy.float32)

Create a numpy array for memory data.

Parameters:

• shape – data shape.

• n_envs – number of parallel environments.

• n_size – length of data sequence for each environment.

• dtype – numpy data type.

Returns:

numpy.zeros())

Return type:

An empty memory space to store data. (initial

xuance.common.memory_tools.sample_batch(memory: numpy.ndarray | dict | None, index: numpy.ndarray | tuple | None)

Sample a batch of data from the selected memory.

Parameters:

• memory – memory that contains experience data.

• index – pointer to the location for the selected data.

Returns:

A batch of data.

xuance.common.memory_tools.store_element(data: numpy.ndarray | dict | float | None, memory: dict | numpy.ndarray, ptr: int)

Insert a step of data into current memory.

Parameters:

• data – target data that to be stored.

• memory – the memory where data will be stored.

• ptr – pointer to the location for the data.