ding.entry.serial_entry_preference_based_irl

ding.entry.serial_entry_preference_based_irl

serial_pipeline_preference_based_irl(input_cfg, seed=0, env_setting=None, model=None, max_train_iter=int(10000000000.0), max_env_step=int(10000000000.0))

Overview

serial_pipeline_preference_based_irl.

Arguments: - input_cfg (:obj:Union[str, Tuple[dict, dict]]): Config in dict type. str type means config file path. Tuple[dict, dict] type means [user_config, create_cfg]. - seed (:obj:int): Random seed. - env_setting (:obj:Optional[List[Any]]): A list with 3 elements: BaseEnv subclass, collector env config, and evaluator env config. - model (:obj:Optional[torch.nn.Module]): Instance of torch.nn.Module. - max_iterations (:obj:Optional[torch.nn.Module]): Learner's max iteration. Pipeline will stop when reaching this iteration. Returns: - policy (:obj:Policy): Converged policy.

Full Source Code

../ding/entry/serial_entry_preference_based_irl.py

1import copy 2from typing import Union, Optional, List, Any, Tuple 3import os 4import torch 5from ditk import logging 6from functools import partial 7from tensorboardX import SummaryWriter 8from copy import deepcopy 9 10from ding.envs import get_vec_env_setting, create_env_manager 11from ding.worker import BaseLearner, InteractionSerialEvaluator, BaseSerialCommander, create_buffer, \ 12 create_serial_collector 13from ding.config import read_config, compile_config 14from ding.policy import create_policy, PolicyFactory 15from ding.reward_model import create_reward_model 16from ding.utils import set_pkg_seed 17 18 19def serial_pipeline_preference_based_irl( 20 input_cfg: Union[str, Tuple[dict, dict]], 21 seed: int = 0, 22 env_setting: Optional[List[Any]] = None, 23 model: Optional[torch.nn.Module] = None, 24 max_train_iter: Optional[int] = int(1e10), 25 max_env_step: Optional[int] = int(1e10), 26) -> 'Policy': # noqa 27 """ 28 Overview: 29 serial_pipeline_preference_based_irl. 30 Arguments: 31 - input_cfg (:obj:`Union[str, Tuple[dict, dict]]`): Config in dict type. \ 32 ``str`` type means config file path. \ 33 ``Tuple[dict, dict]`` type means [user_config, create_cfg]. 34 - seed (:obj:`int`): Random seed. 35 - env_setting (:obj:`Optional[List[Any]]`): A list with 3 elements: \ 36 ``BaseEnv`` subclass, collector env config, and evaluator env config. 37 - model (:obj:`Optional[torch.nn.Module]`): Instance of torch.nn.Module. 38 - max_iterations (:obj:`Optional[torch.nn.Module]`): Learner's max iteration. Pipeline will stop \ 39 when reaching this iteration. 40 Returns: 41 - policy (:obj:`Policy`): Converged policy. 42 """ 43 if isinstance(input_cfg, str): 44 cfg, create_cfg = read_config(input_cfg) 45 else: 46 cfg, create_cfg = deepcopy(input_cfg) 47 create_cfg.policy.type = create_cfg.policy.type + '_command' 48 create_cfg.reward_model = dict(type=cfg.reward_model.type) 49 env_fn = None if env_setting is None else env_setting[0] 50 cfg = compile_config(cfg, seed=seed, env=env_fn, auto=True, create_cfg=create_cfg, save_cfg=True, renew_dir=False) 51 cfg_bak = copy.deepcopy(cfg) 52 # Create main components: env, policy 53 if env_setting is None: 54 env_fn, collector_env_cfg, evaluator_env_cfg = get_vec_env_setting(cfg.env) 55 else: 56 env_fn, collector_env_cfg, evaluator_env_cfg = env_setting 57 collector_env = create_env_manager(cfg.env.manager, [partial(env_fn, cfg=c) for c in collector_env_cfg]) 58 evaluator_env = create_env_manager(cfg.env.manager, [partial(env_fn, cfg=c) for c in evaluator_env_cfg]) 59 collector_env.seed(cfg.seed) 60 evaluator_env.seed(cfg.seed, dynamic_seed=False) 61 set_pkg_seed(cfg.seed, use_cuda=cfg.policy.cuda) 62 policy = create_policy(cfg.policy, model=model, enable_field=['learn', 'collect', 'eval', 'command']) 63 64 # Create worker components: learner, collector, evaluator, replay buffer, commander. 65 tb_logger = SummaryWriter(os.path.join('./{}/log/'.format(cfg.exp_name), 'serial')) 66 learner = BaseLearner(cfg.policy.learn.learner, policy.learn_mode, tb_logger, exp_name=cfg.exp_name) 67 collector = create_serial_collector( 68 cfg.policy.collect.collector, 69 env=collector_env, 70 policy=policy.collect_mode, 71 tb_logger=tb_logger, 72 exp_name=cfg.exp_name 73 ) 74 evaluator = InteractionSerialEvaluator( 75 cfg.policy.eval.evaluator, evaluator_env, policy.eval_mode, tb_logger, exp_name=cfg.exp_name 76 ) 77 replay_buffer = create_buffer(cfg.policy.other.replay_buffer, tb_logger=tb_logger, exp_name=cfg.exp_name) 78 commander = BaseSerialCommander( 79 cfg.policy.other.commander, learner, collector, evaluator, replay_buffer, policy.command_mode 80 ) 81 82 reward_model = create_reward_model(cfg_bak, policy.collect_mode.get_attribute('device'), tb_logger) 83 reward_model.train() 84 # ========== 85 # Main loop 86 # ========== 87 # Learner's before_run hook. 88 learner.call_hook('before_run') 89 90 # Accumulate plenty of data at the beginning of training. 91 if cfg.policy.get('random_collect_size', 0) > 0: 92 if cfg.policy.get('transition_with_policy_data', False): 93 collector.reset_policy(policy.collect_mode) 94 else: 95 action_space = collector_env.env_info().act_space 96 random_policy = PolicyFactory.get_random_policy(policy.collect_mode, action_space=action_space) 97 collector.reset_policy(random_policy) 98 collect_kwargs = commander.step() 99 new_data = collector.collect(n_sample=cfg.policy.random_collect_size, policy_kwargs=collect_kwargs) 100 replay_buffer.push(new_data, cur_collector_envstep=0) 101 collector.reset_policy(policy.collect_mode) 102 while True: 103 collect_kwargs = commander.step() 104 # Evaluate policy performance 105 if evaluator.should_eval(learner.train_iter): 106 stop, reward = evaluator.eval(learner.save_checkpoint, learner.train_iter, collector.envstep) 107 if stop: 108 break 109 # Collect data by default config n_sample/n_episode 110 new_data = collector.collect(train_iter=learner.train_iter, policy_kwargs=collect_kwargs) 111 replay_buffer.push(new_data, cur_collector_envstep=collector.envstep) 112 # Learn policy from collected data 113 for i in range(cfg.policy.learn.update_per_collect): 114 # Learner will train ``update_per_collect`` times in one iteration. 115 train_data = replay_buffer.sample(learner.policy.get_attribute('batch_size'), learner.train_iter) 116 if train_data is None: 117 # It is possible that replay buffer's data count is too few to train ``update_per_collect`` times 118 logging.warning( 119 "Replay buffer's data can only train for {} steps. ".format(i) + 120 "You can modify data collect config, e.g. increasing n_sample, n_episode." 121 ) 122 break 123 # update train_data reward using the augmented reward 124 train_data_augmented = reward_model.estimate(train_data) 125 learner.train(train_data_augmented, collector.envstep) 126 if learner.policy.get_attribute('priority'): 127 replay_buffer.update(learner.priority_info) 128 if collector.envstep >= max_env_step or learner.train_iter >= max_train_iter: 129 break 130 131 # Learner's after_run hook. 132 learner.call_hook('after_run') 133 return policy