csle_agents.agents.random_search package

Submodules

csle_agents.agents.random_search.random_search_agent module

class csle_agents.agents.random_search.random_search_agent.RandomSearchAgent(simulation_env_config: csle_common.dao.simulation_config.simulation_env_config.SimulationEnvConfig, emulation_env_config: Union[None, csle_common.dao.emulation_config.emulation_env_config.EmulationEnvConfig], experiment_config: csle_common.dao.training.experiment_config.ExperimentConfig, env: Optional[csle_common.dao.simulation_config.base_env.BaseEnv] = None, training_job: Optional[csle_common.dao.jobs.training_job_config.TrainingJobConfig] = None, save_to_metastore: bool = True)

Bases: csle_agents.agents.base.base_agent.BaseAgent

Random Search Agent

static compute_avg_metrics(metrics: Dict[str, List[Union[float, int]]]) → Dict[str, Union[float, int]]

Computes the average metrics of a dict with aggregated metrics

Parameters

metrics – the dict with the aggregated metrics

Returns

the average metrics

eval_theta(policy: Union[csle_common.dao.training.multi_threshold_stopping_policy.MultiThresholdStoppingPolicy, csle_common.dao.training.linear_threshold_stopping_policy.LinearThresholdStoppingPolicy], max_steps: int = 200) → Dict[str, Union[float, int]]

Evaluates a given threshold policy by running monte-carlo simulations

Parameters

policy – the policy to evaluate

Returns

the average metrics of the evaluation

get_policy(theta: List[float], L: int) → Union[csle_common.dao.training.multi_threshold_stopping_policy.MultiThresholdStoppingPolicy, csle_common.dao.training.linear_threshold_stopping_policy.LinearThresholdStoppingPolicy]

Gets the policy of a given parameter vector

Parameters

• theta – the parameter vector

• L – the number of parameters

Returns

the policy

hparam_names() → List[str]

Returns

a list with the hyperparameter names

static initial_theta(L: int) → numpy.ndarray[Any, numpy.dtype[Any]]

Initializes theta randomly

Parameters

L – the dimension of theta

Returns

the initialized theta vector

random_perturbation(delta: float, theta: numpy.ndarray[Any, numpy.dtype[Any]]) → numpy.ndarray[Any, numpy.dtype[Any]]

Performs a random perturbation to the theta vector

Parameters

• delta – the step size for the perturbation

• theta – the current theta vector

Returns

the perturbed theta vector

random_search(exp_result: csle_common.dao.training.experiment_result.ExperimentResult, seed: int, training_job: csle_common.dao.jobs.training_job_config.TrainingJobConfig, random_seeds: List[int]) → csle_common.dao.training.experiment_result.ExperimentResult

Runs the random search algorithm

Parameters

• exp_result – the experiment result object to store the result

• seed – the seed

• training_job – the training job config

• random_seeds – list of seeds

Returns

the updated experiment result and the trained policy

static round_vec(vec) → List[float]

Rounds a vector to 3 decimals

Parameters

vec – the vector to round

Returns

the rounded vector

train() → csle_common.dao.training.experiment_execution.ExperimentExecution

Performs the policy training for the given random seeds using random search

Returns

the training metrics and the trained policies

static update_metrics(metrics: Dict[str, List[Union[float, int]]], info: Dict[str, Union[float, int]]) → Dict[str, List[Union[float, int]]]

Update a dict with aggregated metrics using new information from the environment

Parameters

• metrics – the dict with the aggregated metrics

• info – the new information

Returns

the updated dict

Module contents