colosseum.hyperopt.utils

 1from typing import List, Type, Dict, Tuple
 2
 3from colosseum.agent.agents.base import BaseAgent
 4from colosseum.agent.utils import sample_agent_gin_configs_file
 5from colosseum.benchmark.benchmark import ColosseumBenchmark
 6from colosseum.hyperopt import HyperOptConfig
 7from colosseum.utils import get_colosseum_mdp_classes
 8from colosseum.utils.miscellanea import sample_mdp_gin_configs_file
 9
10
11def sample_agent_configs_and_benchmarks_for_hyperopt(
12    agent_cls: List[Type[BaseAgent]], hpoc: HyperOptConfig
13) -> List[Tuple[Dict[Type[BaseAgent], str], ColosseumBenchmark]]:
14    """
15    samples the agent configurations from the agents parameters sample spaces and the MDP configuration from the
16    MDP sampling functions to be used in the parameters optimization procedures.
17
18    Parameters
19    ----------
20    agent_cls : Type["BaseAgent"]
21        The agent classes to be optimized.
22    hpoc : HyperOptConfig
23        The parameters optimization procedure configuration.
24    Returns
25    -------
26    List[Tuple[Dict[Type[BaseAgent], str], ColosseumBenchmark]]
27        The agents configurations and the benchmarks with the MDP configs for the parameters optimization procedure.
28    """
29
30    agents_and_benchmarks = []
31
32    episodic_benchmark_name = f"hyperopt_episodic"
33    episodic_agents_configs = dict()
34
35    continuous_benchmark_name = f"hyperopt_continuous"
36    continuous_agents_configs = dict()
37
38    for agent_cl in agent_cls:
39
40        # First sample the agent parameters
41        agent_samples = sample_agent_gin_configs_file(
42            agent_cl, hpoc.n_samples_agents, hpoc.seed
43        )
44
45        if agent_cl.is_episodic():
46            episodic_agents_configs[agent_cl] = agent_samples
47        else:
48            continuous_agents_configs[agent_cl] = agent_samples
49
50    # Sampling the episodic MDPs
51    if len(episodic_agents_configs) > 0:
52        episodic_mdps_configs = dict()
53        for cl in get_colosseum_mdp_classes(True):
54            episodic_mdps_configs[cl] = sample_mdp_gin_configs_file(
55                cl, hpoc.n_samples_mdps, hpoc.seed
56            )
57        episodic_benchmark = ColosseumBenchmark(
58            episodic_benchmark_name, episodic_mdps_configs, hpoc.experiment_config
59        )
60        agents_and_benchmarks.append((episodic_agents_configs, episodic_benchmark))
61
62    # Sampling the continuous MDPs
63    if len(continuous_agents_configs) > 0:
64        continuous_mdps_configs = dict()
65        for cl in get_colosseum_mdp_classes(False):
66            continuous_mdps_configs[cl] = sample_mdp_gin_configs_file(
67                cl, hpoc.n_samples_mdps, hpoc.seed
68            )
69        continuous_benchmark = ColosseumBenchmark(
70            continuous_benchmark_name, continuous_mdps_configs, hpoc.experiment_config
71        )
72        agents_and_benchmarks.append((continuous_agents_configs, continuous_benchmark))
73
74    return agents_and_benchmarks