&`iXbddlZddlZddlmZmZmZmZmZmZm Z ddl Z ddl m Z ddl mZddlmZmZddlmZmZddlmZddlmZmZdd lmZdd lmZmZdd lm Z m!Z!dd l"m#Z#dd l$m%Z%ddl&m'Z'm(Z(m)Z)ddl*m+Z+m,Z,ddl-m.Z.ddl/m0Z0ej1e2Z3e,Gdde'Z4dS)N)AnyCallableDictListOptionalTypeUnion) env_integer)tabulate) RunConfig ScalingConfig) BackendConfig Checkpoint)session)BackendExecutor TrialInfo) DataConfig)_TrainingResult get_session)construct_train_funccount_required_parameters)$_TRAINER_RESTORE_DEPRECATION_WARNING)RAY_TRAIN_ENABLE_STATE_TRACKING) BaseTrainer GenDatasetTrainingIterator) Deprecated DeveloperAPI)Template)repr_with_fallbackceZdZUdZeZeeed<eZ eeed<e j gdzZ dZ e j dgzZ ddddddddddeegdfeegdffdeed eed eed eed eed eeeefdeeeefdeeffdZeee d%dedeeegdfeegdffdeeffd ZfdZdededdfdZ ed edeffd Z!deddfdZ"de#e$fdZ%d&dZ&defdZ'e(dd gd!Z)defd"Z*defd#Z+defd$Z,xZ-S)'DataParallelTraineraA Trainer for data parallel training. You should subclass this Trainer if your Trainer follows SPMD (single program, multiple data) programming paradigm - you want multiple processes to run the same function, but on different data. This Trainer runs the function ``train_loop_per_worker`` on multiple Ray Actors. The ``train_loop_per_worker`` function is expected to take in either 0 or 1 arguments: .. testcode:: def train_loop_per_worker(): ... .. testcode:: def train_loop_per_worker(config: Dict): ... If ``train_loop_per_worker`` accepts an argument, then ``train_loop_config`` will be passed in as the argument. This is useful if you want to tune the values in ``train_loop_config`` as hyperparameters. If the ``datasets`` dict contains a training dataset (denoted by the "train" key), then it will be split into multiple dataset shards that can then be accessed by ``train.get_dataset_shard("train")`` inside ``train_loop_per_worker``. All the other datasets will not be split and ``train.get_dataset_shard(...)`` will return the entire Dataset. Inside the ``train_loop_per_worker`` function, you can use any of the :ref:`Ray Train loop methods `. .. testcode:: from ray import train def train_loop_per_worker(): # Report intermediate results for callbacks or logging and # checkpoint data. train.report(...) # Returns dict of last saved checkpoint. train.get_checkpoint() # Returns the Dataset shard for the given key. train.get_dataset_shard("my_dataset") # Returns the total number of workers executing training. train.get_context().get_world_size() # Returns the rank of this worker. train.get_context().get_world_rank() # Returns the rank of the worker on the current node. train.get_context().get_local_rank() Any returns from the ``train_loop_per_worker`` will be discarded and not used or persisted anywhere. **How do I use DataParallelTrainer or any of its subclasses?** Example: .. testcode:: :skipif: True import ray from ray import train from ray.train import ScalingConfig from ray.train.data_parallel_trainer import DataParallelTrainer def train_loop_for_worker(): dataset_shard_for_this_worker = train.get_dataset_shard("train") # 3 items for 3 workers, each worker gets 1 item batches = list(dataset_shard_for_this_worker.iter_batches(batch_size=1)) assert len(batches) == 1 train_dataset = ray.data.from_items([1, 2, 3]) assert train_dataset.count() == 3 trainer = DataParallelTrainer( train_loop_for_worker, scaling_config=ScalingConfig(num_workers=3), datasets={"train": train_dataset}, ) result = trainer.fit() **How do I develop on top of DataParallelTrainer?** In many cases, using DataParallelTrainer directly is sufficient to execute functions on multiple actors. However, you may want to subclass ``DataParallelTrainer`` and create a custom Trainer for the following 2 use cases: - **Use Case 1:** You want to do data parallel training, but want to have a predefined ``training_loop_per_worker``. - **Use Case 2:** You want to implement a custom :py:class:`~ray.train.backend.Backend` that automatically handles additional setup or teardown logic on each actor, so that the users of this new trainer do not have to implement this logic. For example, a ``TensorflowTrainer`` can be built on top of ``DataParallelTrainer`` that automatically handles setting the proper environment variables for distributed Tensorflow on each actor. For 1, you can set a predefined training loop in __init__ .. testcode:: from ray.train.data_parallel_trainer import DataParallelTrainer class MyDataParallelTrainer(DataParallelTrainer): def __init__(self, *args, **kwargs): predefined_train_loop_per_worker = lambda: 1 super().__init__(predefined_train_loop_per_worker, *args, **kwargs) For 2, you can implement the ``ray.train.Backend`` and ``ray.train.BackendConfig`` interfaces. .. testcode:: from dataclasses import dataclass from ray.train.backend import Backend, BackendConfig class MyBackend(Backend): def on_start(self, worker_group, backend_config): def set_env_var(env_var_value): import os os.environ["MY_ENV_VAR"] = env_var_value worker_group.execute(set_env_var, backend_config.env_var) @dataclass class MyBackendConfig(BackendConfig): env_var: str = "default_value" def backend_cls(self): return MyBackend class MyTrainer(DataParallelTrainer): def __init__(self, train_loop_per_worker, my_backend_config: MyBackendConfig, **kwargs): super().__init__( train_loop_per_worker, backend_config=my_backend_config, **kwargs) Args: train_loop_per_worker: The training function to execute. This can either take in no arguments or a ``config`` dict. train_loop_config: Configurations to pass into ``train_loop_per_worker`` if it accepts an argument. backend_config: Configuration for setting up a Backend (e.g. Torch, Tensorflow, Horovod) on each worker to enable distributed communication. If no Backend should be set up, then set this to None. scaling_config: Configuration for how to scale data parallel training. dataset_config: Configuration for dataset ingest. This is merged with the default dataset config for the given trainer (`cls._dataset_config`). run_config: Configuration for the execution of the training run. datasets: Ray Datasets to use for training and evaluation. This is a dict where the key is the name of the dataset, which can be accessed from within the ``train_loop_per_worker`` by calling ``train.get_dataset_shard(dataset_key)``. By default, all datasets are sharded equally across workers. This can be configured via ``dataset_config``. metadata: Dict that should be made available via `train.get_context().get_metadata()` and in `checkpoint.get_metadata()` for checkpoints saved from this Trainer. Must be JSON-serializable. resume_from_checkpoint: A checkpoint to resume training from. _backend_executor_cls_training_iterator_cls) num_workersresources_per_workeruse_gpuplacement_strategyaccelerator_typeNtrain_loop_config)r*backend_configscaling_configdataset_config run_configdatasetsmetadataresume_from_checkpointtrain_loop_per_workerr+r,r-r.r/r0r1c||_||_|t}t|tst d|||_||n t }||_tt| ||||| |j j } |j | dd| ddtt drddlm} | dSdS)NzC`dataset_config` must be an instance of ray.train.DataConfig, was: )r,r.r/r0r1CPUrGPU)get_or_create_state_actor)_train_loop_per_worker_train_loop_configr isinstance ValueError _data_configr_backend_configsuperr"__init__r,total_resourcesset_train_total_resourcesgetr r%ray.train._internal.state.state_actorr6) selfr2r*r+r,r-r.r/r0r1train_total_resourcesr6 __class__s s/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/train/data_parallel_trainer.pyr>zDataParallelTrainer.__init__sQ'<#"3  !'\\N.*55 )&)) +-8NNmoo  . !4((11)!#9 2   !% 3 C 33 ! % %eQ / / ! % %eQ / /   6 : : ( W W W W W W % % ' ' ' ' ' ( ()messagepathc Ltt|jd|||d|S)aRestores a DataParallelTrainer from a previously interrupted/failed run. Args: train_loop_per_worker: Optionally re-specified train loop function. This should be used to re-specify a function that is not restorable in a new Ray cluster (e.g., it holds onto outdated object references). This should be the same training loop that was passed to the original trainer constructor. train_loop_config: Optionally re-specified train config. This should similarly be used if the original `train_loop_config` contained outdated object references, and it should not be modified from what was originally passed in. See :meth:`BaseTrainer.restore() ` for descriptions of the other arguments. Returns a restored instance of the `DataParallelTrainer`. )rIr2r*)r=r"restore)clsrIr2r*kwargsrEs rFrLzDataParallelTrainer.restoresD:7u(#..6 "7/      rGc~t||jddS)Nr2)r=_validate_attributes_validate_train_loop_per_workerr7)rCrEs rFrPz(DataParallelTrainer._validate_attributes5sD $$&&& ,,  ')@     rGfn_namereturncZt|}|dkrt|d|ddS)Nz1 should take in 0 or 1 arguments, but it accepts z arguments instead.)rr:)rCr2rRnum_required_paramss rFrQz3DataParallelTrainer._validate_train_loop_per_worker<s\88MNN  " "KK"5KKK  # "rGcFtt||}|js/dt jvrt d|jtd|j d|jdkrtd|j|S)Nr5zGPUs are detected in your Ray cluster, but GPU training is not enabled for this trainer. To enable GPU training, make sure to set `use_gpu` to True in your scaling config.zQYou must specify the 'num_workers' in `scaling_config` as either an argument of `zR` or through the `param_space` of a `Tuner` (if performing hyperparameter tuning).rzG'num_workers' in `scaling_config` must be a positive integer. Received ) r=r"_validate_scaling_configr'rayavailable_resourcesloggerinfor%r:__name__)rMr,rEs rFrXz,DataParallelTrainer._validate_scaling_configFs2C88QQ    % %33J3L3L*L*L KK*     % -A # AAA   % * *B%3%?BB  rGtraining_iteratorc:|D]}||dS)aThis method loops over the `TrainingIterator`: The actual iteration (for ... in ...) waits for the training function on each worker to report a result and supplies it as a list of results. Afterwards (in the body of the loop), it will report the result to the Tune session. The iterator ends after the training function on each worker has finished. N)_propagate_results)rCr^training_resultss rF _run_trainingz!DataParallelTrainer._run_traininges6!2 6 6   # #$4 5 5 5 5 6 6rGrac |d}td|DsJtd|D}t|dk}|rj|jtfd|DsJ|r%t jjjjnd}t||j}t d|jd|j  |dS) Nrc3@K|]}t|tVdSN)r9r.0results rF z9DataParallelTrainer._propagate_results..ss,VV6:fo66VVVVVVrGc*g|]}|j |jSre) checkpointrfs rF z:DataParallelTrainer._propagate_results..~s.    ,  ,,,rGc3BK|]}|jjjkVdSre)rIstoragecheckpoint_fs_path)rgrk tune_sessions rFriz9DataParallelTrainer._propagate_results..sD   O|3F F      rG) filesystemrI)rkmetricsz A A    ,,-DEEEEErGc B||j}t|j|j|jjdd}ttj tj tj tj tjtjtjt'jj}||j||j|jd}||||j||j|j|j|j}||| dS)Nr2T)train_func_context fn_arg_namediscard_returns)nameid resourceslogdir driver_ipdriver_node_idexperiment_namerun_idr)r+ trial_infor%r& max_retries)backend_executorr+ train_funcr/r0 data_configrk)!rXr,rr7r8r<r~rrget_trial_name get_trial_idget_trial_resources get_trial_dirrYutilget_node_ip_addressget_runtime_context get_node_idget_experiment_nameuuiduuid4hexr#r%_resources_per_worker_not_nonestartr$r/r0r;starting_checkpointrbshutdown)rCr,r2rrr^s rF training_loopz!DataParallelTrainer.training_loops66t7JKK 4  '  ##3F/ ! ! ! '))#%%133(**h2244244@@BB#799:<<#     55/!&2!/!N 6       77-/,]])/8   ,--- !!#####rGc|jS)zReturns a copy of this Trainer's final dataset configs. Returns: The merged default + user-supplied dataset config. )r;)rCs rFget_dataset_configz&DataParallelTrainer.get_dataset_configs   rG ipywidgets8c ddlm}m}m}m}|d|jjd}g}g}|jr|| |d||| |d|j rE||| |d|j rJ|||j |d|jrJ|||j|d |jrJ|||j|d ||| } ||| g|d } | jdi|} | dt'|i| S)a,Returns a mimebundle with an ipywidget repr and a simple text repr. Depending on the frontend where the data is being displayed, different mimetypes will be used from this bundle. See https://ipython.readthedocs.io/en/stable/config/integrating.html for information about this method, and https://ipywidgets.readthedocs.io/en/latest/embedding.html for more information about the jupyter widget mimetype. Returns: A mimebundle containing an ipywidget repr and a simple text repr. r)HTMLLayoutTabVBoxz

z

Datasetsz Data ConfigTrain Loop ConfigzScaling Configz Run ConfigzBackend Config)titles100%widthlayoutz text/plainrK)rrrrrrEr]r/append_datasets_repr__data_config_repr_html_r8_train_loop_config_repr_html_r, _repr_html_r.r<_repr_mimebundle_updaterepr) rCrNrrrrtitlechildrenrtabwidgetbundles rFrz%DataParallelTrainer._repr_mimebundle_sb 766666666666:DN3:::;; = ) OOD0022 3 3 3 MM* % % % OODD!=!=!?!?@@ A A A MM- ( ( (  " / OODD!C!C!E!EFF G G G MM- . . .   , OODD!4!@!@!B!BCC D D D MM* + + + ? ( OODD!!>?? @ @ @ MM, ' ' '   , OODD!5!A!A!C!CDD E E E MM* + + +c(6***ucl66+?+?+?@@@))33F33 d4jj     rGc |jri}|jD]{\}}t|ts!t|r|||<At |dr|||<it|||<|tddtdt|ddgdd d  Sd S)Nrtitle_data.html.j2rzscrollableTable.html.j2SettingValueF unsafehtml)headers showindextablefmtnone)table max_heightrdata) r8itemsr9str isnumerichasattrrrrenderr )rC table_datakvs rFrz1DataParallelTrainer._train_loop_config_repr_html_s&  " J/5577 + +1a%%+Q)9)9););+$%JqMMQ ..+$%MMOOJqMM$'FFJqMM01188)788??""((**!*G 4"'!-  &@9   2rGcrt|jg}td|S)Nzrendered_html_common.html.j2)content)rr;rr)rCrs rFrz+DataParallelTrainer._data_config_repr_html_0s5t())*677>>w>OOOrGc ddlm}m}m}g}|jr|jD]\}}|}|rk||td d|dd|||||d S) Nr)rrrrzDataset - zrrrr) rrrrr/r _tab_repr_rrr)rCrrrrrconfigrs rFrz#DataParallelTrainer._datasets_repr_5s1111111111 = 8 $ 3 3 5 5 8 8 f''))8NN$%9::AA&F&F&F&FTBNN6#4#4#6#6777tGFF$8$8$89999rG)NN)rSN).r] __module__ __qualname____doc__rr#r__annotations__rr$r_scaling_config_allowed_keys_dataset_config_fields_for_tuner_param_spacer rrrrr rr rrrrr> classmethodrrrLrPrQrXrbrrr`rrr rrrr __classcell__)rEs@rFr"r"snnd4C40BBB5ED!12EEE#.#KOOO$ O$/$MQ%!-12626/3*.48-17;0(0(0($Xb$h%74&$,9O%OP0($D> 0( !/ 0( !/ 0(!,0(Y'0(4Z010(4S>*0(!) 40(0(0(0(0(0(dZ<=== ,0      ( (2t8$hvt|&<< =   $D>      >=[  D     %-8; m [< 6/? 6D 6 6 6 64F43H4F4F4F4Fl.$.$.$.$`!J!!!!s+,,33-,3js4PPPPP :::::::::rGr")5loggingrtypingrrrrrrr rYray._private.ray_constantsr )ray._private.thirdparty.tabulate.tabulater ray.air.configr r ray.trainrrray.train._internalr$ray.train._internal.backend_executorrrray.train._internal.data_configrray.train._internal.sessionrrray.train._internal.utilsrrray.train.base_trainerrray.train.constantsrray.train.trainerrrrray.util.annotationsrr ray.widgetsrray.widgets.utilr getLoggerr]r[r"rKrGrFrs CCCCCCCCCCCCCCCCCC 222222>>>>>>33333333////////''''''KKKKKKKK666666DDDDDDDDUUUUUUUUGGGGGG??????GGGGGGGGGG99999999 //////  8 $ $l:l:l:l:l:+l:l:l:l:l:rG