lit_sb3.vec_env.async_eval

source module lit_sb3.vec_env.async_eval

Classes

• AsyncEval — Helper class to do asynchronous evaluation of different policies with multiple processes.

source class AsyncEval(envs_fn: list[Callable[[], VecEnv]], train_policy: BasePolicy, start_method: str | None = None, n_eval_episodes: int = 1)

Helper class to do asynchronous evaluation of different policies with multiple processes.

It is useful when implementing population based methods like Evolution Strategies (ES), Cross Entropy Method (CEM) or Augmented Random Search (ARS).

.. warning::

Only 'forkserver' and 'spawn' start methods are thread-safe,
which is important to avoid race conditions.
However, compared to
'fork' they incur a small start-up cost and have restrictions on
global variables. With those methods, users must wrap the code in an
``if __name__ == "__main__":`` block.
For more information, see the multiprocessing documentation.

Parameters

• envs_fn : list[Callable[[], VecEnv]] — Vectorized environments to run in subprocesses (callable)

• train_policy : BasePolicy — The policy object that will load the different candidate weights.

• start_method : str | None — method used to start the subprocesses. Must be one of the methods returned by multiprocessing.get_all_start_methods(). Defaults to 'forkserver' on available platforms, and 'spawn' otherwise.

• n_eval_episodes : int — The number of episodes to test each agent

Methods

• send_jobs — Send jobs to the workers to evaluate new candidates.

• seed — Seed the environments.

• set_options — Set environment options for all environments.

• get_results — Retreive episode rewards and lengths from each worker.

• get_obs_rms — Retrieve the observation filters (observation running mean std).

• sync_obs_rms — Synchronise (and update) the observation filters.

• close — Close the processes.

source method AsyncEval.send_jobs(candidate_weights: torch.Tensor, pop_size: int) → None

Send jobs to the workers to evaluate new candidates.

Parameters

• candidate_weights : torch.Tensor — The weights to be evaluated. :pop_size: The number of candidate (size of the population)

• pop_size : int — The pop size parameter.

source method AsyncEval.seed(seed: int | None = None) → list[None | int]

Seed the environments.

Parameters

• seed : int | None — The seed for the pseudo-random generators.

source method AsyncEval.set_options(options: list[dict] | dict | None = None) → list[None | int]

Set environment options for all environments.

If a dict is passed instead of a list, the same options will be used for all environments. WARNING: Those options will only be passed to the environment at the next reset.

Parameters

• options : list[dict] | dict | None — A dictionary of environment options to pass to each environment at the next reset.

source method AsyncEval.get_results() → list[tuple[int, tuple[np.ndarray, np.ndarray]]]

Retreive episode rewards and lengths from each worker.

for all candidates (there might be multiple candidates per worker).

Returns

• list[tuple[int, tuple[np.ndarray, np.ndarray]]] — A list of tuples containing each candidate index and its result (episodic reward and episode length)

source method AsyncEval.get_obs_rms() → list[RunningMeanStd]

Retrieve the observation filters (observation running mean std).

of each process, they will be combined in the main process. Synchronisation is done afterward using sync_obs_rms().

Returns

• list[RunningMeanStd] — A list of RunningMeanStd objects (one per process)

source method AsyncEval.sync_obs_rms(obs_rms: RunningMeanStd) → None

Synchronise (and update) the observation filters.

(observation running mean std).

Parameters

• obs_rms : RunningMeanStd — The updated RunningMeanStd to be used by workers for normalizing observations.

source method AsyncEval.close() → None

Close the processes.