`ding.entry.serial_entry_mbrl`¶

`ding.entry.serial_entry_mbrl` ¶

`serial_pipeline_dyna(input_cfg, seed=0, env_setting=None, model=None, max_train_iter=int(10000000000.0), max_env_step=int(10000000000.0))` ¶

Overview

Serial pipeline entry for dyna-style model-based RL.

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_train_iter (:obj:Optional[int]): Maximum policy update iterations in training. - max_env_step (:obj:Optional[int]): Maximum collected environment interaction steps. Returns: - policy (:obj:Policy): Converged policy.

`serial_pipeline_dream(input_cfg, seed=0, env_setting=None, model=None, max_train_iter=int(10000000000.0), max_env_step=int(10000000000.0))` ¶

Overview

Serial pipeline entry for dreamer-style model-based RL.

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_train_iter (:obj:Optional[int]): Maximum policy update iterations in training. - max_env_step (:obj:Optional[int]): Maximum collected environment interaction steps. Returns: - policy (:obj:Policy): Converged policy.

`serial_pipeline_dreamer(input_cfg, seed=0, env_setting=None, model=None, max_train_iter=int(10000000000.0), max_env_step=int(10000000000.0))` ¶

Overview

Serial pipeline entry for dreamerv3.

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_train_iter (:obj:Optional[int]): Maximum policy update iterations in training. - max_env_step (:obj:Optional[int]): Maximum collected environment interaction steps. Returns: - policy (:obj:Policy): Converged policy.

Full Source Code

../ding/entry/serial_entry_mbrl.py

from typing import Union, Optional, List, Any, Tupleimport torchimport osfrom functools import partialfrom tensorboardX import SummaryWriterfrom copy import deepcopyfrom ding.worker import BaseLearner, InteractionSerialEvaluator, BaseSerialCommander, create_buffer, \    get_buffer_cls, create_serial_collectorfrom ding.world_model import WorldModelfrom ding.worker import IBufferfrom ding.envs import get_vec_env_setting, create_env_managerfrom ding.config import read_config, compile_configfrom ding.utils import set_pkg_seed, deep_merge_dictsfrom ding.policy import create_policyfrom ding.world_model import create_world_modelfrom ding.entry.utils import random_collectdef mbrl_entry_setup(        input_cfg: Union[str, Tuple[dict, dict]],        seed: int = 0,        env_setting: Optional[List[Any]] = None,        model: Optional[torch.nn.Module] = None,) -> Tuple:    if isinstance(input_cfg, str):        cfg, create_cfg = read_config(input_cfg)    else:        cfg, create_cfg = deepcopy(input_cfg)    create_cfg.policy.type = create_cfg.policy.type + '_command'    env_fn = None if env_setting is None else env_setting[0]    cfg = compile_config(        cfg,        seed=seed,        env=env_fn,        auto=True,        create_cfg=create_cfg,        save_cfg=True,        renew_dir=not cfg.policy.learn.get('resume_training', False)    )    if env_setting is None:        env_fn, collector_env_cfg, evaluator_env_cfg = get_vec_env_setting(cfg.env)    else:        env_fn, collector_env_cfg, evaluator_env_cfg = env_setting    collector_env = create_env_manager(cfg.env.manager, [partial(env_fn, cfg=c) for c in collector_env_cfg])    evaluator_env = create_env_manager(cfg.env.manager, [partial(env_fn, cfg=c) for c in evaluator_env_cfg])    collector_env.seed(cfg.seed)    evaluator_env.seed(cfg.seed, dynamic_seed=False)    set_pkg_seed(cfg.seed, use_cuda=cfg.policy.cuda)    # create logger    tb_logger = SummaryWriter(os.path.join('./{}/log/'.format(cfg.exp_name), 'serial'))    # create world model    world_model = create_world_model(cfg.world_model, env_fn(cfg.env), tb_logger)    # create policy    policy = create_policy(cfg.policy, model=model, enable_field=['learn', 'collect', 'eval', 'command'])    # create worker    learner = BaseLearner(cfg.policy.learn.learner, policy.learn_mode, tb_logger, exp_name=cfg.exp_name)    collector = create_serial_collector(        cfg.policy.collect.collector,        env=collector_env,        policy=policy.collect_mode,        tb_logger=tb_logger,        exp_name=cfg.exp_name    )    evaluator = InteractionSerialEvaluator(        cfg.policy.eval.evaluator, evaluator_env, policy.eval_mode, tb_logger, exp_name=cfg.exp_name    )    env_buffer = create_buffer(cfg.policy.other.replay_buffer, tb_logger=tb_logger, exp_name=cfg.exp_name)    commander = BaseSerialCommander(        cfg.policy.other.commander, learner, collector, evaluator, env_buffer, policy.command_mode    )    return (cfg, policy, world_model, env_buffer, learner, collector, collector_env, evaluator, commander, tb_logger)def create_img_buffer(        cfg: dict, input_cfg: Union[str, Tuple[dict, dict]], world_model: WorldModel, tb_logger: 'SummaryWriter') -> IBuffer:  # noqa    if isinstance(input_cfg, str):        _, create_cfg = read_config(input_cfg)    else:        _, create_cfg = input_cfg    img_buffer_cfg = cfg.world_model.other.imagination_buffer    img_buffer_cfg.update(create_cfg.imagination_buffer)    buffer_cls = get_buffer_cls(img_buffer_cfg)    cfg.world_model.other.imagination_buffer.update(deep_merge_dicts(buffer_cls.default_config(), img_buffer_cfg))    if img_buffer_cfg.type == 'elastic':        img_buffer_cfg.set_buffer_size = world_model.buffer_size_scheduler    img_buffer = create_buffer(cfg.world_model.other.imagination_buffer, tb_logger=tb_logger, exp_name=cfg.exp_name)    return img_bufferdef serial_pipeline_dyna(        input_cfg: Union[str, Tuple[dict, dict]],        seed: int = 0,        env_setting: Optional[List[Any]] = None,        model: Optional[torch.nn.Module] = None,        max_train_iter: Optional[int] = int(1e10),        max_env_step: Optional[int] = int(1e10),) -> 'Policy':  # noqa    """    Overview:        Serial pipeline entry for dyna-style model-based RL.    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_train_iter (:obj:`Optional[int]`): Maximum policy update iterations in training.        - max_env_step (:obj:`Optional[int]`): Maximum collected environment interaction steps.    Returns:        - policy (:obj:`Policy`): Converged policy.    """    cfg, policy, world_model, env_buffer, learner, collector, collector_env, evaluator, commander, tb_logger = \        mbrl_entry_setup(input_cfg, seed, env_setting, model)    img_buffer = create_img_buffer(cfg, input_cfg, world_model, tb_logger)    learner.call_hook('before_run')    if cfg.policy.learn.get('resume_training', False):        collector.envstep = learner.collector_envstep    if cfg.policy.get('random_collect_size', 0) > 0:        random_collect(cfg.policy, policy, collector, collector_env, commander, env_buffer)    while True:        collect_kwargs = commander.step()        # eval the policy        if evaluator.should_eval(collector.envstep):            stop, reward = evaluator.eval(learner.save_checkpoint, learner.train_iter, collector.envstep)            if stop:                break        # fill environment buffer        data = collector.collect(train_iter=learner.train_iter, policy_kwargs=collect_kwargs)        env_buffer.push(data, cur_collector_envstep=collector.envstep)        # eval&train world model and fill imagination buffer        if world_model.should_eval(collector.envstep):            world_model.eval(env_buffer, collector.envstep, learner.train_iter)        if world_model.should_train(collector.envstep):            world_model.train(env_buffer, collector.envstep, learner.train_iter)            world_model.fill_img_buffer(                policy.collect_mode, env_buffer, img_buffer, collector.envstep, learner.train_iter            )        for i in range(cfg.policy.learn.update_per_collect):            batch_size = learner.policy.get_attribute('batch_size')            train_data = world_model.sample(env_buffer, img_buffer, batch_size, learner.train_iter)            learner.train(train_data, collector.envstep)        if cfg.policy.on_policy:            # On-policy algorithm must clear the replay buffer.            env_buffer.clear()            img_buffer.clear()        if collector.envstep >= max_env_step or learner.train_iter >= max_train_iter:            break    learner.call_hook('after_run')    return policydef serial_pipeline_dream(        input_cfg: Union[str, Tuple[dict, dict]],        seed: int = 0,        env_setting: Optional[List[Any]] = None,        model: Optional[torch.nn.Module] = None,        max_train_iter: Optional[int] = int(1e10),        max_env_step: Optional[int] = int(1e10),) -> 'Policy':  # noqa    """    Overview:        Serial pipeline entry for dreamer-style model-based RL.    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_train_iter (:obj:`Optional[int]`): Maximum policy update iterations in training.        - max_env_step (:obj:`Optional[int]`): Maximum collected environment interaction steps.    Returns:        - policy (:obj:`Policy`): Converged policy.    """    cfg, policy, world_model, env_buffer, learner, collector, collector_env, evaluator, commander, tb_logger = \        mbrl_entry_setup(input_cfg, seed, env_setting, model)    learner.call_hook('before_run')    if cfg.policy.learn.get('resume_training', False):        collector.envstep = learner.collector_envstep    if cfg.policy.get('random_collect_size', 0) > 0:        random_collect(cfg.policy, policy, collector, collector_env, commander, env_buffer)    while True:        collect_kwargs = commander.step()        # eval the policy        if evaluator.should_eval(collector.envstep):            stop, reward = evaluator.eval(learner.save_checkpoint, learner.train_iter, collector.envstep)            if stop:                break        # fill environment buffer        data = collector.collect(train_iter=learner.train_iter, policy_kwargs=collect_kwargs)        env_buffer.push(data, cur_collector_envstep=collector.envstep)        # eval&train world model and fill imagination buffer        if world_model.should_eval(collector.envstep):            world_model.eval(env_buffer, collector.envstep, learner.train_iter)        if world_model.should_train(collector.envstep):            world_model.train(env_buffer, collector.envstep, learner.train_iter)        update_per_collect = cfg.policy.learn.update_per_collect // world_model.rollout_length_scheduler(            collector.envstep        )        update_per_collect = max(1, update_per_collect)        for i in range(update_per_collect):            batch_size = learner.policy.get_attribute('batch_size')            train_data = env_buffer.sample(batch_size, learner.train_iter)            # dreamer-style: use pure on-policy imagined rollout to train policy,            # which depends on the current envstep to decide the rollout length            learner.train(                train_data, collector.envstep, policy_kwargs=dict(world_model=world_model, envstep=collector.envstep)            )        if collector.envstep >= max_env_step or learner.train_iter >= max_train_iter:            break    learner.call_hook('after_run')    return policydef serial_pipeline_dreamer(        input_cfg: Union[str, Tuple[dict, dict]],        seed: int = 0,        env_setting: Optional[List[Any]] = None,        model: Optional[torch.nn.Module] = None,        max_train_iter: Optional[int] = int(1e10),        max_env_step: Optional[int] = int(1e10),) -> 'Policy':  # noqa    """    Overview:        Serial pipeline entry for dreamerv3.    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_train_iter (:obj:`Optional[int]`): Maximum policy update iterations in training.        - max_env_step (:obj:`Optional[int]`): Maximum collected environment interaction steps.    Returns:        - policy (:obj:`Policy`): Converged policy.    """    cfg, policy, world_model, env_buffer, learner, collector, collector_env, evaluator, commander, tb_logger = \        mbrl_entry_setup(input_cfg, seed, env_setting, model)    learner.call_hook('before_run')    # prefill environment buffer    if cfg.policy.get('random_collect_size', 0) > 0:        cfg.policy.random_collect_size = cfg.policy.random_collect_size // cfg.policy.collect.unroll_len        random_collect(cfg.policy, policy, collector, collector_env, commander, env_buffer)    while True:        collect_kwargs = commander.step()        # eval the policy        if evaluator.should_eval(collector.envstep):            stop, reward = evaluator.eval(                learner.save_checkpoint,                learner.train_iter,                collector.envstep,                policy_kwargs=dict(world_model=world_model)            )            if stop:                break        # train world model and fill imagination buffer        steps = (            cfg.world_model.pretrain            if world_model.should_pretrain() else int(world_model.should_train(collector.envstep))        )        for _ in range(steps):            batch_size = learner.policy.get_attribute('batch_size')            batch_length = cfg.policy.learn.batch_length            post, context = world_model.train(                env_buffer, collector.envstep, learner.train_iter, batch_size, batch_length            )            start = post            learner.train(                start, collector.envstep, policy_kwargs=dict(world_model=world_model, envstep=collector.envstep)            )        # fill environment buffer        data = collector.collect(            train_iter=learner.train_iter,            policy_kwargs=dict(world_model=world_model, envstep=collector.envstep, **collect_kwargs)        )        env_buffer.push(data, cur_collector_envstep=collector.envstep)        if collector.envstep >= max_env_step or learner.train_iter >= max_train_iter:            break    learner.call_hook('after_run')    return policy

ding.entry.serial_entry_mbrl¶