multi_agent_rl

coma_agents

class xuance.mindspore.agents.multi_agent_rl.coma_agents.COMA_Agents(config: Namespace, envs: DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv | None = None, num_agents: int | None = None, agent_keys: List[str] | None = None, state_space: gymnasium.spaces.Space | None = None, observation_space: gymnasium.spaces.Space | None = None, action_space: gymnasium.spaces.Space | None = None, callback: MultiAgentBaseCallback | None = None)

Bases: OnPolicyMARLAgents

get_actions(obs_dict: List[dict], state: numpy.ndarray | None = None, avail_actions_dict: List[dict] | None = None, rnn_hidden_actor: dict | None = None, rnn_hidden_critic: dict | None = None, test_mode: bool | None = False, **kwargs)

Compute actions (and optional value/log-prob outputs) for multi-agent execution.

This method performs a forward pass through the current multi-agent actor-critic policy to produce actions for each agent in each parallel environment. When RNN-based representations are enabled, the method consumes and returns recurrent hidden states for both the actor and the critic. During training (test_mode=False), this method also computes critic values and action log-probabilities needed for on-policy updates. During evaluation (test_mode=True), critic values and log-probabilities are not computed to reduce overhead.

Parameters:

• obs_dict (List[dict]) – Observations for each parallel environment. Each element is a dict keyed by self.agent_keys.

• state (Optional[np.ndarray]) – Global state array used by centralized critics when use_global_state=True. The expected shape depends on the environment wrapper.

• avail_actions_dict (Optional[List[dict]]) – Available-action masks for each parallel environment when use_actions_mask=True. Each element is a dict keyed by self.agent_keys. Can be None when action masking is disabled.

• rnn_hidden_actor (Optional[dict]) – Current actor RNN hidden states keyed by self.model_keys. Required when self.use_rnn is True.

• rnn_hidden_critic (Optional[dict]) – Current critic RNN hidden states keyed by self.model_keys. Required when self.use_rnn is True and values are requested.

• test_mode (bool) – Whether to run in evaluation mode. When True, only actions are produced and training-specific outputs (values/log_pi) are omitted.

Returns:

A dictionary containing:

• rnn_hidden_actor (Optional[dict]): Updated actor RNN hidden states when self.use_rnn is True;

  otherwise the value returned by the policy (typically None).

• rnn_hidden_critic (Optional[dict]): Updated critic RNN hidden states when computed;

  otherwise an empty dict.

• actions (List[dict]): Actions for each parallel environment. Each element is a dict keyed by

  self.agent_keys.

• log_pi (dict): Log-probabilities of sampled actions for each agent when test_mode=False;

  otherwise an empty dict.

• values (dict): Critic value estimates for each agent when test_mode=False; otherwise an empty dict.

Return type:

dict

run_episodes(n_episodes: int = 1, run_envs: DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv | None = None, test_mode: bool = False, close_envs: bool = True) → list

Run vectorized multi-agent episodes for rollout collection or evaluation.

This method steps a vectorized multi-agent environment using the current actor-critic policy until n_episodes episodes have completed. When test_mode is False, collected transitions are stored into the on-policy trajectory buffer and episode boundaries are tracked for bootstrapping and advantage computation (GAE). When test_mode is True, training-time outputs (values/log-probabilities) are skipped, exploration schedules are disabled by default, and episode scores are returned; optional RGB-array frames can be recorded and logged as a video.

Parameters:

• n_episodes (int) – Number of completed episodes to run across all parallel environments.

• run_envs (Optional[DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv]) – Vectorized environments to run. If None, self.train_envs is used.

• test_mode (bool) – Whether to run in evaluation mode. When True, the trajectory buffer is not written and only episode scores are collected.

• close_envs (bool) – Whether to close run_envs before returning when test_mode is True. Set this to False if the caller manages the environment lifecycle externally.

Returns:

Episode scores (mean reward across agents) for each completed episode.

Return type:

list

store_experience(obs_dict, avail_actions, actions_dict, log_pi_a, rewards_dict, values_dict, terminals_dict, info, **kwargs)

Store a batch of multi-agent transitions into the on-policy buffer.

This method converts per-environment dictionaries (one dict per vector environment) into per-agent batched arrays and writes them into the on-policy trajectory buffer. It also stores auxiliary fields such as agent masks and (optionally) global state and action masks. For RNN-based policies, episode-step indices are recorded to support episode-aware bookkeeping.

Parameters:

• obs_dict (List[dict]) – Observations for each parallel environment. Each element is a dict keyed by self.agent_keys.

• avail_actions (Optional[List[dict]]) – Available-action masks for each parallel environment when use_actions_mask=True. Each element is a dict keyed by self.agent_keys. Can be None when action masking is disabled.

• actions_dict (List[dict]) – Actions executed by each agent for each parallel environment. Each element is a dict keyed by self.agent_keys.

• log_pi_a (dict) – Log-probabilities of the actions under the current policy (typically computed during rollout collection).

• rewards_dict (List[dict]) – Rewards for each agent for each parallel environment. Each element is a dict keyed by self.agent_keys.

• values_dict (dict) – Value estimates produced by the critic for each agent (used for advantage/return computation).

• terminals_dict (List[dict]) – Termination flags for each agent for each parallel environment. Each element is a dict keyed by self.agent_keys.

• info (List[dict]) – Environment info for each parallel environment at the current step. Must contain agent_mask for each agent key.

• **kwargs – Optional extra fields. When use_global_state=True, this method expects state to be provided.

train(train_steps: int) → dict

Run the main multi-agent on-policy training loop.

This method interacts with the training environments to collect fresh rollouts from the current policy, stores transitions in the on-policy trajectory buffer, and triggers policy/value updates when the buffer is full. Training advances in vectorized increments (one iteration corresponds to stepping all parallel environments once).

Parameters:

train_steps (int) – Number of rollout collection iterations to run. Each iteration steps all parallel environments once, so the total number of environment steps is approximately train_steps * self.n_envs.

Returns:

A dictionary containing aggregated training information and logged metrics collected during

training (e.g., policy loss, value loss, entropy, KL divergence, and episode statistics).

Return type:

dict

Notes

• This method assumes that training environments (self.train_envs) and the trajectory buffer self.memory

  have already been initialized.

• When the buffer becomes full, the agent finalizes trajectories by computing bootstrapped terminal values

  via values_next and calling finish_path, then performs n_epochs optimization passes over mini-batches using train_epochs.

• Episode termination and reset logic are handled per environment,

  and episode-level statistics are reported via callbacks.

train_epochs(n_epochs: int = 1) → dict

Update policies for multiple epochs using mini-batches from the trajectory buffer.

This method performs n_epochs optimization passes over the rollout data stored in self.memory. For each epoch, it shuffles transition indices and iterates over mini-batches to compute gradient updates via the learner. When RNN-based policies are enabled, the RNN-specific update method is used.

Parameters:

n_epochs (int) – Number of optimization epochs to perform over the current trajectory buffer.

Returns:

A dictionary of training metrics returned by the learner from the last mini-batch update (e.g., policy

loss, value loss, entropy, KL divergence). Implementations may include additional diagnostics depending on the algorithm.

Return type:

dict

values_next(i_env: int, obs_dict: dict, state: numpy.ndarray | None = None, actions_n: numpy.ndarray | None = None, rnn_hidden_critic: dict | None = None)

Compute bootstrapped critic values for an environment that reached a boundary.

This method evaluates the critic on the terminal/next observations of a specific vectorized environment (i_env) and returns per-agent value estimates used for bootstrapping when finalizing trajectories (e.g., for GAE/return computation).

Parameters:

• i_env (int) – Index of the vectorized environment that is finishing an episode or trajectory segment.

• obs_dict (dict) – Per-agent observations for the selected environment. This dict is keyed by self.agent_keys.

• state (Optional[np.ndarray]) – Global state for the selected environment when use_global_state=True. If provided, it should correspond to the same i_env instance.

• rnn_hidden_critic (Optional[dict]) – Current critic RNN hidden states keyed by self.model_keys. Required when self.use_rnn is True.

Returns:

A tuple of (rnn_hidden_critic_new, values_dict):

• rnn_hidden_critic_new (Optional[dict]): Updated critic hidden states for the selected environment

  when self.use_rnn is True; otherwise the value returned by the critic (typically None).

• values_dict (dict): Per-agent critic value estimates keyed by self.agent_keys.

Return type:

Tuple[Optional[dict], dict]

commnet_agents

class xuance.mindspore.agents.multi_agent_rl.commnet_agents.CommNet_Agents(config: Namespace, envs: DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv | None = None, num_agents: int | None = None, agent_keys: List[str] | None = None, state_space: gymnasium.Space | None = None, observation_space: gymnasium.Space | None = None, action_space: gymnasium.Space | None = None, callback: MultiAgentBaseCallback | None = None)

Bases: MAPPO_Agents

get_actions(obs_dict: List[dict], state: numpy.ndarray | None = None, avail_actions_dict: List[dict] | None = None, rnn_hidden_actor: dict | None = None, rnn_hidden_critic: dict | None = None, test_mode: bool | None = False, info: dict = None, **kwargs)

Returns actions for agents.

Parameters:

• obs_dict (dict) – Observations for each agent in self.agent_keys.

• state (Optional[np.ndarray]) – The global state.

• avail_actions_dict (Optional[List[dict]]) – Actions mask values, default is None.

• rnn_hidden_actor (Optional[dict]) – The RNN hidden states of actor representation.

• rnn_hidden_critic (Optional[dict]) – The RNN hidden states of critic representation.

• test_mode (Optional[bool]) – True for testing without noises.

• deterministic (bool) – True for deterministic policy and False for stochastic policy.

Returns:

The new RNN hidden states of actor representation (if self.use_rnn=True). rnn_hidden_critic_new (dict): The new RNN hidden states of critic representation (if self.use_rnn=True). actions_dict (dict): The output actions. log_pi_a (dict): The log of pi. values_dict (dict): The evaluated critic values (when test_mode is False).

Return type:

rnn_hidden_actor_new (dict)

static pad_observation(obs_dict)

run_episodes(n_episodes: int = 1, run_envs: DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv | None = None, test_mode: bool = False, close_envs: bool = True) → list

Run vectorized multi-agent episodes for rollout collection or evaluation.

This method steps a vectorized multi-agent environment using the current actor-critic policy until n_episodes episodes have completed. When test_mode is False, collected transitions are stored into the on-policy trajectory buffer and episode boundaries are tracked for bootstrapping and advantage computation (GAE). When test_mode is True, training-time outputs (values/log-probabilities) are skipped, exploration schedules are disabled by default, and episode scores are returned; optional RGB-array frames can be recorded and logged as a video.

Parameters:

• n_episodes (int) – Number of completed episodes to run across all parallel environments.

• deterministic_policy (bool) – True for evaluating the deterministic policy, and False for evaluating the stochastic policy.

• run_envs (Optional[DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv]) – Vectorized environments to run. If None, self.train_envs is used.

• test_mode (bool) – Whether to run in evaluation mode. When True, the trajectory buffer is not written and only episode scores are collected.

• close_envs (bool) – Whether to close run_envs before returning when test_mode is True. Set this to False if the caller manages the environment lifecycle externally.

Returns:

Episode scores (mean reward across agents) for each completed episode.

Return type:

list

dcg_agents

class xuance.mindspore.agents.multi_agent_rl.dcg_agents.DCG_Agents(config: Namespace, envs: DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv, device: str = 'cpu:0')

Bases: MARLAgents

act(obs_n, *rnn_hidden, avail_actions=None, test_mode=False)

train(i_step, n_epochs=1)

iac_agents

class xuance.mindspore.agents.multi_agent_rl.iac_agents.IAC_Agents(config: Namespace, envs: DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv | None = None, num_agents: int | None = None, agent_keys: List[str] | None = None, state_space: gymnasium.spaces.Space | None = None, observation_space: gymnasium.spaces.Space | None = None, action_space: gymnasium.spaces.Space | None = None, callback: MultiAgentBaseCallback | None = None)

Bases: OnPolicyMARLAgents

The implementation of IAC agents.

Parameters:

• config – the Namespace variable that provides hyperparameters and other settings.

• envs – the vectorized environments.

• callback – A user-defined callback function object to inject custom logic during training.

get_actions(obs_dict: List[dict], state: numpy.ndarray | None = None, avail_actions_dict: List[dict] | None = None, rnn_hidden_actor: dict | None = None, rnn_hidden_critic: dict | None = None, test_mode: bool | None = False, **kwargs)

Returns actions for agents.

Parameters:

• obs_dict (dict) – Observations for each agent in self.agent_keys.

• state (Optional[np.ndarray]) – The global state.

• avail_actions_dict (Optional[List[dict]]) – Actions mask values, default is None.

• rnn_hidden_actor (Optional[dict]) – The RNN hidden states of actor representation.

• rnn_hidden_critic (Optional[dict]) – The RNN hidden states of critic representation.

• test_mode (Optional[bool]) – True for testing without noises.

Returns:

The new RNN hidden states of actor representation (if self.use_rnn=True). rnn_hidden_critic_new (dict): The new RNN hidden states of critic representation (if self.use_rnn=True). actions_dict (dict): The output actions. log_pi_a (dict): The log of pi. values_dict (dict): The evaluated critic values (when test_mode is False).

Return type:

rnn_hidden_actor_new (dict)

store_experience(obs_dict, avail_actions, actions_dict, log_pi_a, rewards_dict, values_dict, terminals_dict, info, **kwargs)

Store experience data into replay buffer.

Parameters:

• obs_dict (List[dict]) – Observations for each agent in self.agent_keys.

• avail_actions (List[dict]) – Actions mask values for each agent in self.agent_keys.

• actions_dict (List[dict]) – Actions for each agent in self.agent_keys.

• log_pi_a (dict) – The log of pi.

• rewards_dict (List[dict]) – Rewards for each agent in self.agent_keys.

• values_dict (dict) – Critic values for each agent in self.agent_keys.

• terminals_dict (List[dict]) – Terminated values for each agent in self.agent_keys.

• info (List[dict]) – Other information for the environment at current step.

• **kwargs – Other inputs.

values_next(i_env: int, obs_dict: dict, state: numpy.ndarray | None = None, rnn_hidden_critic: dict | None = None)

Returns critic values of one environment that finished an episode.

Parameters:

• i_env (int) – The index of environment.

• obs_dict (dict) – Observations for each agent in self.agent_keys.

• state (Optional[np.ndarray]) – The global state.

• rnn_hidden_critic (Optional[dict]) – The RNN hidden states of critic representation.

Returns:

The new RNN hidden states of critic representation (if self.use_rnn=True). values_dict: The critic values.

Return type:

rnn_hidden_critic_new (dict)

ic3net_agents

class xuance.mindspore.agents.multi_agent_rl.ic3net_agents.IC3Net_Agents(config: Namespace, envs: DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv | None = None, num_agents: int | None = None, agent_keys: List[str] | None = None, state_space: gymnasium.Space | None = None, observation_space: gymnasium.Space | None = None, action_space: gymnasium.Space | None = None, callback: MultiAgentBaseCallback | None = None)

Bases: CommNet_Agents

get_actions(obs_dict: List[dict], state: numpy.ndarray | None = None, avail_actions_dict: List[dict] | None = None, rnn_hidden_actor: dict | None = None, rnn_hidden_critic: dict | None = None, test_mode: bool | None = False, info: dict = None, **kwargs)

Returns actions for agents.

Parameters:

• obs_dict (dict) – Observations for each agent in self.agent_keys.

• state (Optional[np.ndarray]) – The global state.

• avail_actions_dict (Optional[List[dict]]) – Actions mask values, default is None.

• rnn_hidden_actor (Optional[dict]) – The RNN hidden states of actor representation.

• rnn_hidden_critic (Optional[dict]) – The RNN hidden states of critic representation.

• test_mode (Optional[bool]) – True for testing without noises.

• deterministic (bool) – True for deterministic policy and False for stochastic policy.

Returns:

The new RNN hidden states of actor representation (if self.use_rnn=True). rnn_hidden_critic_new (dict): The new RNN hidden states of critic representation (if self.use_rnn=True). actions_dict (dict): The output actions. log_pi_a (dict): The log of pi. values_dict (dict): The evaluated critic values (when test_mode is False).

Return type:

rnn_hidden_actor_new (dict)

run_episodes(n_episodes: int = 1, run_envs: DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv | None = None, test_mode: bool = False, close_envs: bool = True) → list

Run vectorized multi-agent episodes for rollout collection or evaluation.

This method steps a vectorized multi-agent environment using the current actor-critic policy until n_episodes episodes have completed. When test_mode is False, collected transitions are stored into the on-policy trajectory buffer and episode boundaries are tracked for bootstrapping and advantage computation (GAE). When test_mode is True, training-time outputs (values/log-probabilities) are skipped, exploration schedules are disabled by default, and episode scores are returned; optional RGB-array frames can be recorded and logged as a video.

Parameters:

• n_episodes (int) – Number of completed episodes to run across all parallel environments.

• deterministic_policy (bool) – True for evaluating the deterministic policy, and False for evaluating the stochastic policy.

• run_envs (Optional[DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv]) – Vectorized environments to run. If None, self.train_envs is used.

• test_mode (bool) – Whether to run in evaluation mode. When True, the trajectory buffer is not written and only episode scores are collected.

• close_envs (bool) – Whether to close run_envs before returning when test_mode is True. Set this to False if the caller manages the environment lifecycle externally.

Returns:

Episode scores (mean reward across agents) for each completed episode.

Return type:

list

store_experience(obs_dict, avail_actions, actions_dict, log_pi_a, rewards_dict, values_dict, terminals_dict, info, **kwargs)

Store a batch of multi-agent transitions into the on-policy buffer.

This method converts per-environment dictionaries (one dict per vector environment) into per-agent batched arrays and writes them into the on-policy trajectory buffer. It also stores auxiliary fields such as agent masks and (optionally) global state and action masks. For RNN-based policies, episode-step indices are recorded to support episode-aware bookkeeping.

Parameters:

• obs_dict (List[dict]) – Observations for each parallel environment. Each element is a dict keyed by self.agent_keys.

• avail_actions (Optional[List[dict]]) – Available-action masks for each parallel environment when use_actions_mask=True. Each element is a dict keyed by self.agent_keys. Can be None when action masking is disabled.

• actions_dict (List[dict]) – Actions executed by each agent for each parallel environment. Each element is a dict keyed by self.agent_keys.

• log_pi_a (dict) – Log-probabilities of the actions under the current policy (typically computed during rollout collection).

• rewards_dict (List[dict]) – Rewards for each agent for each parallel environment. Each element is a dict keyed by self.agent_keys.

• values_dict (dict) – Value estimates produced by the critic for each agent (used for advantage/return computation).

• terminals_dict (List[dict]) – Termination flags for each agent for each parallel environment. Each element is a dict keyed by self.agent_keys.

• info (List[dict]) – Environment info for each parallel environment at the current step. Must contain agent_mask for each agent key.

• **kwargs – Optional extra fields. When use_global_state=True, this method expects state to be provided.

iddpg_agents

class xuance.mindspore.agents.multi_agent_rl.iddpg_agents.IDDPG_Agents(config: Namespace, envs: DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv | None = None, num_agents: int | None = None, agent_keys: List[str] | None = None, state_space: gymnasium.spaces.Space | None = None, observation_space: gymnasium.spaces.Space | None = None, action_space: gymnasium.spaces.Space | None = None, callback: MultiAgentBaseCallback | None = None)

Bases: OffPolicyMARLAgents

The implementation of Independent DDPG agents.

Parameters:

• config – The Namespace variable that provides hyperparameters and other settings.

• envs – the vectorized environments.

• callback – A user-defined callback function object to inject custom logic during training.

get_actions(obs_dict: List[dict], avail_actions_dict: List[dict] | None = None, rnn_hidden: dict | None = None, test_mode: bool | None = False)

Returns actions for agents.

Parameters:

• obs_dict (List[dict]) – Observations for each agent in self.agent_keys.

• avail_actions_dict (Optional[List[dict]]) – Actions mask values, default is None.

• rnn_hidden (Optional[dict]) – The hidden variables of the RNN.

• test_mode (Optional[bool]) – True for testing without noises.

Returns:

The new hidden states for RNN (if self.use_rnn=True). actions_dict (dict): The output actions.

Return type:

rnn_hidden_state (dict)

init_hidden_item(i_env: int, rnn_hidden: dict | None = None)

Returns initialized hidden states of RNN for i-th environment.

Parameters:

• i_env (int) – The index of environment that to be selected.

• rnn_hidden (Optional[dict]) – The RNN hidden states of actor representation.

init_rnn_hidden(n_envs)

Returns initialized hidden states of RNN if use RNN-based representations.

Parameters:

n_envs (int) – The number of parallel environments.

Returns:

The hidden states for RNN.

Return type:

rnn_hidden_states

ippo_agents

class xuance.mindspore.agents.multi_agent_rl.ippo_agents.IPPO_Agents(config: Namespace, envs: DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv | None = None, num_agents: int | None = None, agent_keys: List[str] | None = None, state_space: gymnasium.spaces.Space | None = None, observation_space: gymnasium.spaces.Space | None = None, action_space: gymnasium.spaces.Space | None = None, callback: MultiAgentBaseCallback | None = None)

Bases: OnPolicyMARLAgents

The implementation of Independent PPO agents.

Parameters:

• config – the Namespace variable that provides hyperparameters and other settings.

• envs – the vectorized environments.

• callback – A user-defined callback function object to inject custom logic during training.

init_hidden_item(i_env: int, rnn_hidden_actor: dict | None = None, rnn_hidden_critic: dict | None = None)

Returns initialized hidden states of RNN for i-th environment.

Parameters:

• i_env (int) – The index of environment that to be selected.

• rnn_hidden_actor (Optional[dict]) – The RNN hidden states of actor representation.

• rnn_hidden_critic (Optional[dict]) – The RNN hidden states of critic representation.

init_rnn_hidden(n_envs)

Returns initialized hidden states of RNN if use RNN-based representations.

Parameters:

n_envs (int) – The number of parallel environments.

iql_agents

class xuance.mindspore.agents.multi_agent_rl.iql_agents.IQL_Agents(config: Namespace, envs: DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv | None = None, num_agents: int | None = None, agent_keys: List[str] | None = None, state_space: gymnasium.spaces.Space | None = None, observation_space: gymnasium.spaces.Space | None = None, action_space: gymnasium.spaces.Space | None = None, callback: MultiAgentBaseCallback | None = None)

Bases: OffPolicyMARLAgents

The implementation of Independent Q-Learning agents.

Parameters:

• config – the Namespace variable that provides hyperparameters and other settings.

• envs – the vectorized environments.

• callback – A user-defined callback function object to inject custom logic during training.

isac_agents

class xuance.mindspore.agents.multi_agent_rl.isac_agents.ISAC_Agents(config: Namespace, envs: DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv | None = None, num_agents: int | None = None, agent_keys: List[str] | None = None, state_space: gymnasium.spaces.Space | None = None, observation_space: gymnasium.spaces.Space | None = None, action_space: gymnasium.spaces.Space | None = None, callback: MultiAgentBaseCallback | None = None)

Bases: OffPolicyMARLAgents

The implementation of Independent SAC agents.

Parameters:

• config – the Namespace variable that provides hyperparameters and other settings.

• envs – the vectorized environments.

• callback – A user-defined callback function object to inject custom logic during training.

get_actions(obs_dict: List[dict], avail_actions_dict: List[dict] | None = None, rnn_hidden: dict | None = None, test_mode: bool | None = False, **kwargs)

Returns actions for agents.

Parameters:

• obs_dict (List[dict]) – Observations for each agent in self.agent_keys.

• avail_actions_dict (Optional[List[dict]]) – Actions mask values, default is None.

• rnn_hidden (Optional[dict]) – The hidden variables of the RNN.

• test_mode (Optional[bool]) – True for testing without noises.

Returns:

The new hidden states for RNN (if self.use_rnn=True). actions_dict (dict): The output actions.

Return type:

rnn_hidden_state (dict)

maddpg_agents

class xuance.mindspore.agents.multi_agent_rl.maddpg_agents.MADDPG_Agents(config: Namespace, envs: DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv | None = None, num_agents: int | None = None, agent_keys: List[str] | None = None, state_space: gymnasium.spaces.Space | None = None, observation_space: gymnasium.spaces.Space | None = None, action_space: gymnasium.spaces.Space | None = None, callback: MultiAgentBaseCallback | None = None)

Bases: IDDPG_Agents

The implementation of MADDPG agents.

Parameters:

• config – The Namespace variable that provides hyperparameters and other settings.

• envs – the vectorized environments.

• callback – A user-defined callback function object to inject custom logic during training.

mappo_agents

class xuance.mindspore.agents.multi_agent_rl.mappo_agents.MAPPO_Agents(config: Namespace, envs: DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv | None = None, num_agents: int | None = None, agent_keys: List[str] | None = None, state_space: gymnasium.spaces.Space | None = None, observation_space: gymnasium.spaces.Space | None = None, action_space: gymnasium.spaces.Space | None = None, callback: MultiAgentBaseCallback | None = None)

Bases: IPPO_Agents

The implementation of MAPPO agents.

Parameters:

• config – the Namespace variable that provides hyperparameters and other settings.

• envs – the vectorized environments.

• callback – A user-defined callback function object to inject custom logic during training.

get_actions(obs_dict: List[dict], state: numpy.ndarray | None = None, avail_actions_dict: List[dict] | None = None, rnn_hidden_actor: dict | None = None, rnn_hidden_critic: dict | None = None, test_mode: bool | None = False)

Returns actions for agents.

Parameters:

• obs_dict (dict) – Observations for each agent in self.agent_keys.

• state (Optional[np.ndarray]) – The global state.

• avail_actions_dict (Optional[List[dict]]) – Actions mask values, default is None.

• rnn_hidden_actor (Optional[dict]) – The RNN hidden states of actor representation.

• rnn_hidden_critic (Optional[dict]) – The RNN hidden states of critic representation.

• test_mode (Optional[bool]) – True for testing without noises.

Returns:

The new RNN hidden states of actor representation (if self.use_rnn=True). rnn_hidden_critic_new (dict): The new RNN hidden states of critic representation (if self.use_rnn=True). actions_dict (dict): The output actions. log_pi_a (dict): The log of pi. values_dict (dict): The evaluated critic values (when test_mode is False).

Return type:

rnn_hidden_actor_new (dict)

values_next(i_env: int, obs_dict: dict, state: numpy.ndarray | None = None, rnn_hidden_critic: dict | None = None)

Returns critic values of one environment that finished an episode.

Parameters:

• i_env (int) – The index of environment.

• obs_dict (dict) – Observations for each agent in self.agent_keys.

• state (Optional[np.ndarray]) – The global state.

• rnn_hidden_critic (Optional[dict]) – The RNN hidden states of critic representation.

Returns:

The new RNN hidden states of critic representation (if self.use_rnn=True). values_dict: The critic values.

Return type:

rnn_hidden_critic_new (dict)

masac_agents

class xuance.mindspore.agents.multi_agent_rl.masac_agents.MASAC_Agents(config: Namespace, envs: DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv | None = None, num_agents: int | None = None, agent_keys: List[str] | None = None, state_space: gymnasium.spaces.Space | None = None, observation_space: gymnasium.spaces.Space | None = None, action_space: gymnasium.spaces.Space | None = None, callback: MultiAgentBaseCallback | None = None)

Bases: ISAC_Agents

The implementation of MASAC agents.

Parameters:

• config – the Namespace variable that provides hyperparameters and other settings.

• envs – the vectorized environments.

• callback – A user-defined callback function object to inject custom logic during training.

init_hidden_item(i_env: int, rnn_hidden: dict | None = None)

Returns initialized hidden states of RNN for i-th environment.

Parameters:

• i_env (int) – The index of environment that to be selected.

• rnn_hidden (Optional[dict]) – The RNN hidden states of actor representation.

init_rnn_hidden(n_envs)

Returns initialized hidden states of RNN if use RNN-based representations.

Parameters:

n_envs (int) – The number of parallel environments.

Returns:

The hidden states for RNN.

Return type:

rnn_hidden_states

matd3_agents

class xuance.mindspore.agents.multi_agent_rl.matd3_agents.MATD3_Agents(config: Namespace, envs: DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv | None = None, num_agents: int | None = None, agent_keys: List[str] | None = None, state_space: gymnasium.spaces.Space | None = None, observation_space: gymnasium.spaces.Space | None = None, action_space: gymnasium.spaces.Space | None = None, callback: MultiAgentBaseCallback | None = None)

Bases: IDDPG_Agents

The implementation of MATD3 agents.

Parameters:

• config – The Namespace variable that provides hyperparameters and other settings.

• envs – the vectorized environments.

• callback – A user-defined callback function object to inject custom logic during training.

mfac_agents

class xuance.mindspore.agents.multi_agent_rl.mfac_agents.MFAC_Agents(config: Namespace, envs: DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv | None = None, num_agents: int | None = None, agent_keys: List[str] | None = None, state_space: gymnasium.spaces.Space | None = None, observation_space: gymnasium.spaces.Space | None = None, action_space: gymnasium.spaces.Space | None = None, callback: MultiAgentBaseCallback | None = None)

Bases: OnPolicyMARLAgents

get_actions(obs_dict: List[dict], agent_mask: List[dict] | None = None, act_mean_dict: List[dict] | None = None, state: numpy.ndarray | None = None, avail_actions_dict: List[dict] | None = None, rnn_hidden_actor: dict | None = None, rnn_hidden_critic: dict | None = None, test_mode: bool | None = False, **kwargs)

Returns actions for agents.

Parameters:

• obs_dict (dict) – Observations for each agent in self.agent_keys.

• agent_mask (Optional[List[dict]]) – Mask the agents that are alive.

• state (Optional[np.ndarray]) – The global state.

• act_mean_dict (Optional[List[dict]]) – Mean actions of each agent’s neighbors.

• avail_actions_dict (Optional[List[dict]]) – Actions mask values, default is None.

• rnn_hidden_actor (Optional[dict]) – The RNN hidden states of actor representation.

• rnn_hidden_critic (Optional[dict]) – The RNN hidden states of critic representation.

• test_mode (Optional[bool]) – True for testing without noises.

Returns:

The new RNN hidden states of actor representation (if self.use_rnn=True). rnn_hidden_critic_new (dict): The new RNN hidden states of critic representation (if self.use_rnn=True). actions_dict (dict): The output actions. log_pi_a (dict): The log of pi. values_dict (dict): The evaluated critic values (when test_mode is False).

Return type:

rnn_hidden_actor_new (dict)

run_episodes(n_episodes: int = 1, run_envs: DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv | None = None, test_mode: bool = False, close_envs: bool = True) → list

Run vectorized multi-agent episodes for rollout collection or evaluation.

This method steps a vectorized multi-agent environment using the current actor-critic policy until n_episodes episodes have completed. When test_mode is False, collected transitions are stored into the on-policy trajectory buffer and episode boundaries are tracked for bootstrapping and advantage computation (GAE). When test_mode is True, training-time outputs (values/log-probabilities) are skipped, exploration schedules are disabled by default, and episode scores are returned; optional RGB-array frames can be recorded and logged as a video.

Parameters:

• n_episodes (int) – Number of completed episodes to run across all parallel environments.

• run_envs (Optional[DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv]) – Vectorized environments to run. If None, self.train_envs is used.

• test_mode (bool) – Whether to run in evaluation mode. When True, the trajectory buffer is not written and only episode scores are collected.

• close_envs (bool) – Whether to close run_envs before returning when test_mode is True. Set this to False if the caller manages the environment lifecycle externally.

Returns:

Episode scores (mean reward across agents) for each completed episode.

Return type:

list

store_experience(obs_dict, avail_actions, actions_dict, log_pi_a, rewards_dict, values_dict, terminals_dict, info, **kwargs)

Store experience data into replay buffer.

Parameters:

• obs_dict (List[dict]) – Observations for each agent in self.agent_keys.

• avail_actions (List[dict]) – Actions mask values for each agent in self.agent_keys.

• actions_dict (List[dict]) – Actions for each agent in self.agent_keys.

• log_pi_a (dict) – The log of pi.

• rewards_dict (List[dict]) – Rewards for each agent in self.agent_keys.

• values_dict (dict) – Critic values for each agent in self.agent_keys.

• terminals_dict (List[dict]) – Terminated values for each agent in self.agent_keys.

• info (List[dict]) – Other information for the environment at current step.

• **kwargs – Other inputs.

train(train_steps: int) → dict

Run the main multi-agent on-policy training loop.

This method interacts with the training environments to collect fresh rollouts from the current policy, stores transitions in the on-policy trajectory buffer, and triggers policy/value updates when the buffer is full. Training advances in vectorized increments (one iteration corresponds to stepping all parallel environments once).

Parameters:

train_steps (int) – Number of rollout collection iterations to run. Each iteration steps all parallel environments once, so the total number of environment steps is approximately train_steps * self.n_envs.

Returns:

A dictionary containing aggregated training information and logged metrics collected during

training (e.g., policy loss, value loss, entropy, KL divergence, and episode statistics).

Return type:

dict

Notes

• This method assumes that training environments (self.train_envs) and the trajectory buffer self.memory

  have already been initialized.

• When the buffer becomes full, the agent finalizes trajectories by computing bootstrapped terminal values

  via values_next and calling finish_path, then performs n_epochs optimization passes over mini-batches using train_epochs.

• Episode termination and reset logic are handled per environment,

  and episode-level statistics are reported via callbacks.

values_next(i_env: int, obs_dict: dict, state: numpy.ndarray | None = None, agent_mask: dict = None, act_mean_dict: dict = None, rnn_hidden_critic: dict | None = None)

Returns critic values of one environment that finished an episode.

Parameters:

• i_env (int) – The index of environment.

• obs_dict (dict) – Observations for each agent in self.agent_keys.

• state (Optional[np.ndarray]) – The global state.

• agent_mask (dict) – Mask the agents that are alive.

• act_mean_dict (dict) – The mean actions of each agent’s neighbors.

• rnn_hidden_critic (Optional[dict]) – The RNN hidden states of critic representation.

Returns:

The new RNN hidden states of critic representation (if self.use_rnn=True). values_dict: The critic values.

Return type:

rnn_hidden_critic_new (dict)

mfq_agents

class xuance.mindspore.agents.multi_agent_rl.mfq_agents.MFQ_Agents(config: Namespace, envs: DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv | None = None, num_agents: int | None = None, agent_keys: List[str] | None = None, state_space: gymnasium.spaces.Space | None = None, observation_space: gymnasium.spaces.Space | None = None, action_space: gymnasium.spaces.Space | None = None, callback: MultiAgentBaseCallback | None = None)

Bases: OffPolicyMARLAgents

get_actions(obs_dict: List[dict], agent_mask: List[dict] | None = None, act_mean_dict: List[dict] | None = None, avail_actions_dict: List[dict] | None = None, rnn_hidden: dict | None = None, test_mode: bool | None = False)

Compute actions for all agents given vectorized observations.

This method performs a forward pass through the current multi-agent policy to obtain actions for each agent in each parallel environment. When RNN-based representations are enabled, it also consumes and returns recurrent hidden states. During training (test_mode=False), this method applies the configured exploration strategy (epsilon-greedy or additive noise); during evaluation (test_mode=True), exploration is disabled.

Parameters:

• obs_dict (List[dict]) – Observations for each parallel environment. Each element is a dict keyed by self.agent_keys.

• avail_actions_dict (Optional[List[dict]]) – Available-action masks for each parallel environment when use_actions_mask=True. Each element is a dict keyed by self.agent_keys. Can be None when action masking is disabled.

• rnn_hidden (Optional[dict]) – Current RNN hidden states keyed by self.model_keys. Required when self.use_rnn is True.

• test_mode (bool) – Whether to run in evaluation mode. When True, exploration is disabled and actions are produced deterministically (or without training-time noise).

Returns:

A dictionary containing:

• hidden_state (Optional[dict]): Updated RNN hidden states when self.use_rnn is True;

  otherwise the value returned by the policy (typically None).

• actions (List[dict]): Actions for each parallel environment.

  Each element is a dict keyed by self.agent_keys.

Return type:

dict

run_episodes(n_episodes: int = 1, run_envs: DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv | None = None, test_mode: bool = False, close_envs: bool = True) → list

Run vectorized multi-agent episodes for rollout collection or evaluation.

This method steps a vectorized multi-agent environment using the current policy until n_episodes episodes have completed. When test_mode is False, collected transitions are stored into the replay buffer (and episode boundaries are tracked for RNN-aware buffers). When test_mode is True, exploration is disabled and episode scores are returned; optional RGB-array frames can be recorded and logged as a video.

Parameters:

• n_episodes (int) – Number of completed episodes to run across all parallel environments.

• run_envs (Optional[DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv]) – Vectorized environments to run. If None, self.train_envs is used.

• test_mode (bool) – Whether to run in evaluation mode. When True, exploration is disabled and the replay buffer is not written.

• close_envs (bool) – Whether to close run_envs before returning when test_mode is True. Set this to False if the caller manages the environment lifecycle externally.

Returns:

Episode scores (mean reward across agents) for each completed episode.

Return type:

list

store_experience(obs_dict, avail_actions, actions_dict, obs_next_dict, avail_actions_next, rewards_dict, terminals_dict, info, **kwargs)

Store experience data into replay buffer.

Parameters:

• obs_dict (List[dict]) – Observations for each agent in self.agent_keys.

• avail_actions (List[dict]) – Actions mask values for each agent in self.agent_keys.

• actions_dict (List[dict]) – Actions for each agent in self.agent_keys.

• obs_next_dict (List[dict]) – Next observations for each agent in self.agent_keys.

• avail_actions_next (List[dict]) – The next actions mask values for each agent in self.agent_keys.

• rewards_dict (List[dict]) – Rewards for each agent in self.agent_keys.

• terminals_dict (List[dict]) – Terminated values for each agent in self.agent_keys.

• info (List[dict]) – Other information for the environment at current step.

train(train_steps: int) → dict

Run the main multi-agent off-policy training loop.

This method interacts with the training environments to collect multi-agent transitions, stores them in the replay buffer, and performs periodic policy updates by sampling mini-batches from the buffer. The training loop is step-based and advances in vectorized increments (one iteration corresponds to self.n_envs environment steps).

Parameters:

• train_steps (int) – Number of training iterations to run. Each iteration steps all parallel environments

• once

• self.n_envs. (so the total number of environment steps is approximately train_steps *)

Returns:

A dictionary containing aggregated training information and logged metrics collected during

training (e.g., losses, episode statistics, exploration factors).

Return type:

dict

Notes

• This method assumes that training environments (self.train_envs) and the replay buffer (self.memory)

  have already been initialized.

• When self.use_rnn is enabled, rollout collection and buffer bookkeeping are handled in run_episodes(),

  and updates are performed once enough experience is available.

• Policy updates are triggered after self.start_training steps and then periodically according to

  self.training_frequency.

qmix_agents

class xuance.mindspore.agents.multi_agent_rl.qmix_agents.QMIX_Agents(config: Namespace, envs: DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv | None = None, num_agents: int | None = None, agent_keys: List[str] | None = None, state_space: gymnasium.spaces.Space | None = None, observation_space: gymnasium.spaces.Space | None = None, action_space: gymnasium.spaces.Space | None = None, callback: MultiAgentBaseCallback | None = None)

Bases: OffPolicyMARLAgents

The implementation of QMIX agents.

Parameters:

• config – the Namespace variable that provides hyperparameters and other settings.

• envs – the vectorized environments.

• callback – A user-defined callback function object to inject custom logic during training.

qtran_agents

class xuance.mindspore.agents.multi_agent_rl.qtran_agents.QTRAN_Agents(config: Namespace, envs: DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv | None = None, num_agents: int | None = None, agent_keys: List[str] | None = None, state_space: gymnasium.spaces.Space | None = None, observation_space: gymnasium.spaces.Space | None = None, action_space: gymnasium.spaces.Space | None = None, callback: MultiAgentBaseCallback | None = None)

Bases: OffPolicyMARLAgents

The implementation of QTRAN agents.

Parameters:

• config – the Namespace variable that provides hyperparameters and other settings.

• envs – the vectorized environments.

• callback – A user-defined callback function object to inject custom logic during training.

get_actions(obs_dict: List[dict], avail_actions_dict: List[dict] | None = None, rnn_hidden: dict | None = None, test_mode: bool | None = False, **kwargs)

Returns actions for agents.

Parameters:

• obs_dict (List[dict]) – Observations for each agent in self.agent_keys.

• avail_actions_dict (Optional[List[dict]]) – Actions mask values, default is None.

• rnn_hidden (Optional[dict]) – The hidden variables of the RNN.

• test_mode (Optional[bool]) – True for testing without noises.

Returns:

The new hidden states for RNN (if self.use_rnn=True). actions_dict (dict): The output actions.

Return type:

rnn_hidden_state (dict)

tarmac_agents

class xuance.mindspore.agents.multi_agent_rl.tarmac_agents.TarMAC_Agents(config: Namespace, envs: DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv | None = None, num_agents: int | None = None, agent_keys: List[str] | None = None, state_space: gymnasium.Space | None = None, observation_space: gymnasium.Space | None = None, action_space: gymnasium.Space | None = None, callback: MultiAgentBaseCallback | None = None)

Bases: IC3Net_Agents

vdac_agents

class xuance.mindspore.agents.multi_agent_rl.vdac_agents.VDAC_Agents(config: Namespace, envs: DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv | None = None, num_agents: int | None = None, agent_keys: List[str] | None = None, state_space: gymnasium.spaces.Space | None = None, observation_space: gymnasium.spaces.Space | None = None, action_space: gymnasium.spaces.Space | None = None, callback: MultiAgentBaseCallback | None = None)

Bases: OnPolicyMARLAgents

get_actions(obs_dict: List[dict], state: numpy.ndarray | None = None, avail_actions_dict: List[dict] | None = None, rnn_hidden_actor: dict | None = None, rnn_hidden_critic: dict | None = None, test_mode: bool | None = False, **kwargs)

Returns actions for agents.

Parameters:

• obs_dict (dict) – Observations for each agent in self.agent_keys.

• state (Optional[np.ndarray]) – The global state.

• avail_actions_dict (Optional[List[dict]]) – Actions mask values, default is None.

• rnn_hidden_actor (Optional[dict]) – The RNN hidden states of actor representation.

• rnn_hidden_critic (Optional[dict]) – The RNN hidden states of critic representation.

• test_mode (Optional[bool]) – True for testing without noises.

Returns:

The new RNN hidden states of actor representation (if self.use_rnn=True). rnn_hidden_critic_new (dict): The new RNN hidden states of critic representation (if self.use_rnn=True). actions_dict (dict): The output actions. log_pi_a (dict): The log of pi. values_dict (dict): The evaluated critic values (when test_mode is False).

Return type:

rnn_hidden_actor_new (dict)

store_experience(obs_dict, avail_actions, actions_dict, log_pi_a, rewards_dict, values_dict, terminals_dict, info, **kwargs)

Store experience data into replay buffer.

Parameters:

• obs_dict (List[dict]) – Observations for each agent in self.agent_keys.

• avail_actions (List[dict]) – Actions mask values for each agent in self.agent_keys.

• actions_dict (List[dict]) – Actions for each agent in self.agent_keys.

• log_pi_a (dict) – The log of pi.

• rewards_dict (List[dict]) – Rewards for each agent in self.agent_keys.

• values_dict (dict) – Critic values for each agent in self.agent_keys.

• terminals_dict (List[dict]) – Terminated values for each agent in self.agent_keys.

• info (List[dict]) – Other information for the environment at current step.

• **kwargs – Other inputs.

values_next(i_env: int, obs_dict: dict, state: numpy.ndarray | None = None, rnn_hidden_critic: dict | None = None)

Returns critic values of one environment that finished an episode.

Parameters:

• i_env (int) – The index of environment.

• obs_dict (dict) – Observations for each agent in self.agent_keys.

• state (Optional[np.ndarray]) – The global state.

• rnn_hidden_critic (Optional[dict]) – The RNN hidden states of critic representation.

Returns:

The new RNN hidden states of critic representation (if self.use_rnn=True). values_dict: The critic values.

Return type:

rnn_hidden_critic_new (dict)

vdn_agents

class xuance.mindspore.agents.multi_agent_rl.vdn_agents.VDN_Agents(config: Namespace, envs: DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv | None = None, num_agents: int | None = None, agent_keys: List[str] | None = None, state_space: gymnasium.spaces.Space | None = None, observation_space: gymnasium.spaces.Space | None = None, action_space: gymnasium.spaces.Space | None = None, callback: MultiAgentBaseCallback | None = None)

Bases: OffPolicyMARLAgents

The implementation of Value Decomposition Networks (VDN) agents.

Parameters:

• config – the Namespace variable that provides hyperparameters and other settings.

• envs – the vectorized environments.

• callback – A user-defined callback function object to inject custom logic during training.

wqmix_agents

class xuance.mindspore.agents.multi_agent_rl.wqmix_agents.WQMIX_Agents(config: Namespace, envs: DummyVecMultiAgentEnv | SubprocVecMultiAgentEnv | None = None, num_agents: int | None = None, agent_keys: List[str] | None = None, state_space: gymnasium.spaces.Space | None = None, observation_space: gymnasium.spaces.Space | None = None, action_space: gymnasium.spaces.Space | None = None, callback: MultiAgentBaseCallback | None = None)

Bases: QMIX_Agents

The implementation of Weighted QMIX agents.

Parameters:

• config – the Namespace variable that provides hyperparameters and other settings.

• envs – the vectorized environments.

• callback – A user-defined callback function object to inject custom logic during training.