&`i&lddlZddlZddlZddlZddlZddlZddlZddlmZddl m Z ddl m Z m Z mZmZmZmZmZmZddlZddlZddlmZddlmZddlmZddlmZ ddl!m"Z"dd l#m$Z$dd l%m&Z&m'Z'dd l(m)Z)dd l*m+Z+dd l,m-Z-ddl.m/Z/m0Z0m1Z1ddl2m3Z3m4Z4ddl5m6Z6ddl7m8Z8ddl9m:Z:m;Z;mZ>ddl?m@Z@dZAedegdffZBejCeDZEdZFde3ZGde3ZHe 0`, or `max_failures = -1` for unlimited retries.N)__name__ __module__ __qualname____doc__ _RESTORE_MSG_FAILURE_CONFIG_MSGj/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/train/base_trainer.pyr'r'Gs*77 < !r/r'config trainer_cls BaseTrainermetadatacB| J||t_|di|}tj}|r||_d|jD|_|| dS)a This is the function that defines the logic of the Ray Train coordinator. This is responsible for setting up a remote instance of the `trainer_cls` (a different instance than the one calling `trainer.fit` on the driver!) and running the training loop. NcLi|]!\}}|t|r |n|"Sr.)callable).0kds r0 z)_train_coordinator_fn..vs@)-A(1++ $113331r/r.) rr4raytuneget_checkpointstarting_checkpointdatasetsitemssetup training_loop)r1r2r4trainer checkpoints r0_train_coordinator_fnrF\s      %KMMk##F##G((**J1'1# 181A1G1G1I1IG MMOOO r/cLeZdZUdZdgZeeed<dZe ed<dZ e ed<gZ ddddddd e e d e ed e eeefd e eeefd e ef dZeee d)deddeeejfde ejjd e eeefd e e ddf dZeeee d*deddeeejfde ejjde fdZ dZ!fdZ"dZ#dZ$ed e de fdZ%d+dZ&d+dZ'e(j)d+dZ*e+d de,fd!Z-d"ejjd#efd$Z.defd%Z/ded&fd'Z0ded&fd(Z1xZ2S),r3aDefines interface for distributed training on Ray. Note: The base ``BaseTrainer`` class cannot be instantiated directly. Only one of its subclasses can be used. Note to developers: If a new trainer is added, please update `air/_internal/usage.py`. **How does a trainer work?** - First, initialize the Trainer. The initialization runs locally, so heavyweight setup should not be done in ``__init__``. - Then, when you call ``trainer.fit()``, the Trainer is serialized and copied to a remote Ray actor. The following methods are then called in sequence on the remote actor. - ``trainer.setup()``: Any heavyweight Trainer setup should be specified here. - ``trainer.training_loop()``: Executes the main training logic. - Calling ``trainer.fit()`` will return a ``ray.result.Result`` object where you can access metrics from your training run, as well as any checkpoints that may have been saved. **How do I create a new Trainer?** Subclass ``ray.train.trainer.BaseTrainer``, and override the ``training_loop`` method, and optionally ``setup``. .. testcode:: :skipif: True import torch from ray.train.trainer import BaseTrainer from ray import train, tune class MyPytorchTrainer(BaseTrainer): def setup(self): self.model = torch.nn.Linear(1, 1) self.optimizer = torch.optim.SGD( self.model.parameters(), lr=0.1) def training_loop(self): # You can access any Trainer attributes directly in this method. # self.datasets["train"] has already been dataset = self.datasets["train"] torch_ds = dataset.iter_torch_batches(dtypes=torch.float) loss_fn = torch.nn.MSELoss() for epoch_idx in range(10): loss = 0 num_batches = 0 torch_ds = dataset.iter_torch_batches( dtypes=torch.float, batch_size=2 ) for batch in torch_ds: X = torch.unsqueeze(batch["x"], 1) y = torch.unsqueeze(batch["y"], 1) # Compute prediction error pred = self.model(X) batch_loss = loss_fn(pred, y) # Backpropagation self.optimizer.zero_grad() batch_loss.backward() self.optimizer.step() loss += batch_loss.item() num_batches += 1 loss /= num_batches # Use Tune functions to report intermediate # results. train.report({"loss": loss, "epoch": epoch_idx}) # Initialize the Trainer, and call Trainer.fit() import ray train_dataset = ray.data.from_items( [{"x": i, "y": i} for i in range(10)]) my_trainer = MyPytorchTrainer(datasets={"train": train_dataset}) result = my_trainer.fit() Args: scaling_config: Configuration for how to scale training. run_config: Configuration for the execution of the training run. datasets: Any Datasets to use for training. Use the key "train" to denote which dataset is the training dataset. 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. trainer_resources_scaling_config_allowed_keysF_handles_checkpoint_freq_handles_checkpoint_at_endN)scaling_config run_configr@r4resume_from_checkpointrLrMr@r4rNc||n t|_|tj|n t|_||_||ni|_||_tr,|tt|ttd|_ d|_ |tjdt#j|dS)Ntrain)rrLcopyrrMr4r@r?rrr+_RESUME_FROM_CHECKPOINT_DEPRECATION_WARNING _restore_path_restore_storage_filesystem_validate_attributesr record_library_usage air_usagetag_air_trainer)selfrLrMr@r4rNs r0__init__zBaseTrainer.__init__s-8NNmoo &0%;DIj ! ! ! ! $,$8b #9 ) + + V#()JKKK%1()TUUU"+/( !!###&w///!$'''''r/)messageclspathstorage_filesystemreturnc trtt|||st d|dt ||\}}t |t}| |5} tj | \} } dddn #1swxYwY| |ur(tjd| d|jd| jd| di} | r<|s:t d t#| d |jd |pi}t'| t'|krSt d t#| d t#||| d<|r|| d<|D] \} }||| | < |di| }n+#t*$r}t d|jd|d}~wwxYw||_||_|S)aRestores a Train experiment from a previously interrupted/failed run. Restore should be used for experiment-level fault tolerance in the event that the head node crashes (e.g., OOM or some other runtime error) or the entire cluster goes down (e.g., network error affecting all nodes). A run that has already completed successfully will not be resumed from this API. To continue training from a successful run, launch a new run with the ``Trainer(resume_from_checkpoint)`` API instead, passing in a checkpoint from the previous run to start with. .. note:: Restoring an experiment from a path that's pointing to a *different* location than the original experiment path is supported. However, Ray Train assumes that the full experiment directory is available (including checkpoints) so that it's possible to resume trials from their latest state. For example, if the original experiment path was run locally, then the results are uploaded to cloud storage, Ray Train expects the full contents to be available in cloud storage if attempting to resume via ``Trainer.restore("s3://...")``. The restored run will continue writing results to the same cloud storage location. The following example can be paired with implementing job retry using :ref:`Ray Jobs ` to produce a Train experiment that will attempt to resume on both experiment-level and trial-level failures: .. testcode:: :skipif: True import os import ray from ray import train from ray.train.trainer import BaseTrainer experiment_name = "unique_experiment_name" storage_path = os.path.expanduser("~/ray_results") experiment_dir = os.path.join(storage_path, experiment_name) # Define some dummy inputs for demonstration purposes datasets = {"train": ray.data.from_items([{"a": i} for i in range(10)])} class CustomTrainer(BaseTrainer): def training_loop(self): pass if CustomTrainer.can_restore(experiment_dir): trainer = CustomTrainer.restore( experiment_dir, datasets=datasets ) else: trainer = CustomTrainer( datasets=datasets, run_config=train.RunConfig( name=experiment_name, storage_path=storage_path, # Tip: You can also enable retries on failure for # worker-level fault tolerance failure_config=train.FailureConfig(max_failures=3), ), ) result = trainer.fit() Args: path: The path to the experiment directory of the training run to restore. This can be a local path or a remote URI if the experiment was uploaded to the cloud. storage_filesystem: Custom ``pyarrow.fs.FileSystem`` corresponding to the ``path``. This may be necessary if the original experiment passed in a custom filesystem. datasets: Re-specified datasets used in the original training run. This must include all the datasets that were passed in the original trainer constructor. scaling_config: Optionally re-specified scaling config. This can be modified to be different from the original spec. **kwargs: Other optionally re-specified arguments, passed in by subclasses. Raises: ValueError: If all datasets were not re-supplied on restore. Returns: BaseTrainer: A restored instance of the class that is calling this method. zInvalid restore path: zn. Make sure that this path exists and is the experiment directory that results from a call to `trainer.fit()`.NzFInvalid trainer type. You are attempting to restore a trainer of type z with `z-.restore`, which will most likely fail. Use `z.restore` instead.r@z=The following datasets need to be provided again on restore: z Use z^.restore(..., datasets=datasets) with the datasets that were provided to the original trainer.z_The provided datasets don't match the original dataset keys. Expected datasets for the keys: z Actual datasets provided: rLzxTrainer restoration failed (see above for the stack trace). Make sure that you use the right trainer class to restore: `z .restore` r.)rr$_TRAINER_RESTORE_DEPRECATION_WARNING can_restore ValueErrorrr _TRAINER_PKLas_posixopen_input_filepickleloadsreadallwarningswarnr(poplistkeyssetrA ExceptionrSrT)r\r]r^r@rLkwargsfsfs_pathtrainer_pkl_pathfr2 param_dictoriginal_datasets param_namevalrDes r0restorezBaseTrainer.restore sL@ * + + K $%I J J Jt%788 ####  &d,>?? G66??AA    0 1 1 @Q&,l199;;&?&? #K @ @ @ @ @ @ @ @ @ @ @ @ @ @ @ c ! ! MAAA(+ AA$,AAA   'NN:r::  X P)..0011PP|PPP  >r  ! !S]] 2 2G59:K:P:P:R:R5S5SGG/3HMMOO/D/DGG  "* :  :+9J' (%||~~ - -OJ), :& c''J''GG   .L...    !%.@+s**CCCH H2H--H2)r[warningctrttt||\}}t |t }t||S)aChecks whether a given directory contains a restorable Train experiment. Args: path: The path to the experiment directory of the Train experiment. This can be either a local directory (e.g., ~/ray_results/exp_name) or a remote URI (e.g., s3://bucket/exp_name). Returns: bool: Whether this path exists and contains the trainer state to resume from )rrrarrrdrer)r\r]r^rrrsrts r0rbzBaseTrainer.can_restoresc( * + + K $%I J J J%d,>?? G66??AA!"&6777r/cHttidd}g}|D]5\}}t||}||kr||d|6|r&d|jjdd|dSd|jjdS)N)rLrMr@r?=< >)rrrAgetattrappend __class__r(join)rYdefault_valuesnon_default_arguments parameter default_valuevalues r0__repr__zBaseTrainer.__repr__s,oo#++#' * * !#(6(<(<(>(> G G $I}D),,E %%%,, -E-EE-E-EFFF  TSt~.SS:O1P1PSSS S-4>*----r/c:tt||}tj|jj}t|}|dd}tt||}i|||_ |S)N) superr3__new__inspect signaturerZ parametersrmrndictzip _param_dict)r\argsrqrDrarg_dictrs r0rzBaseTrainer.__new__s S))11#66&s|44? *//++,, ^ J--..44V4r/ct|jts-tdt |jd|jdt|jt s-tdt |jd|jdt|jts-tdt |jd|jd|j D]I\}}t|tj j s%t|std|d|dJ|jpi|_t|jts%td t |jd  t!jt!j|j|_n,#t&$r}td |jd |d }~wwxYw|jGt|jt*s-tdt |jd|jd|d S)z4Called on __init()__ to validate trainer attributes.zC`run_config` should be an instance of `ray.train.RunConfig`, found z with value `z`.zA`scaling_config` should be an instance of `ScalingConfig`, found zW`datasets` should be a dict mapping from a string to `ray.data.Dataset` objects, found zThe Dataset under 'z,' key is not a `ray.data.Dataset`. Received z instead.z*The provided metadata must be a dict, was .z1The provided metadata must be JSON-serializable: z: NzP`resume_from_checkpoint` should be an instance of `ray.train.Checkpoint`, found ) isinstancerMrrctyperLrr@rrAr<datar!r7r4 TypeErrorjsonrhdumpsrpr?r_log_v2_deprecation_warnings)rYkeydatasetrzs r0rUz BaseTrainer._validate_attributess$/955 Qdo..QQ=A_QQQ  $-}== Yd122YYAEATYYY  $-.. Mdm,,MM;?=MMM  !% 3 3 5 5   W!'38+;<<XgEVEV$7c77$+777  + $-.. ST$-=P=PSSS   Jtz$-'@'@AADMM   (=(($%((    # /  $j9 9 /<15d6N1O1O<<#7<<<  ))+++++s0F88 G!GG!cddlm}m}|r td|jjd|dt sdSddlm}m }m }m }m }m }m} m} |jjt%| |jjjrt%||jjt%| |jjrt%||jjt%||jjt%||jjt%||jjt8jkrt%|dSdS)aLogs deprecation warnings for v2 migration. Log them here in the Ray Train case rather than in the configuration constructors to avoid logging incorrect deprecation warnings when `ray.train.RunConfig` is passed to Ray Tune. r)V2_ENABLED_ENV_VAR is_v2_enabledz2Detected use of a deprecated Trainer import from `a `. This Trainer class is not compatible with Ray Train V2. To fix this: - Update to use the new import path. For example, `from ray.train.torch.torch_trainer import TorchTrainer` -> `from ray.train.torch import TorchTrainer` - Or, explicitly disable V2 by setting: zS=0 See this issue for more context: https://github.com/ray-project/ray/issues/49454N)CALLBACKS_DEPRECATION_MESSAGEFAIL_FAST_DEPRECATION_MESSAGELOG_TO_FILE_DEPRECATION_MESSAGE%PROGRESS_REPORTER_DEPRECATION_MESSAGESTOP_DEPRECATION_MESSAGESYNC_CONFIG_DEPRECATION_MESSAGE%TRAINER_RESOURCES_DEPRECATION_MESSAGEVERBOSE_DEPRECATION_MESSAGE) ray.train.v2._internal.constantsrrDeprecationWarningrr)r&ray.train.v2._internal.migration_utilsrrrrrrrrrLrHrrMfailure_config fail_fast_verbose log_to_filestop callbacksprogress_reporter sync_configr<rP SyncConfig) rYrrrrrrrrrrs r0rz(BaseTrainer._log_v2_deprecation_warningss WVVVVVVV =?? $BT^E^BB >P BBB   .//  F    0 < $%J K K K ? ) 3 D $%B C C C ? # / $%@ A A A ? & F $%D E E E ?  + $%= > > > ? $ 0 $%B C C C ? , 8 $%J K K K ? &#)*>*>*@*@ @ @ $%D E E E E E A @r/c2t||j|S)z?Returns scaling config dataclass after validating updated keys.) dataclass allowed_keys)rrIr\rLs r0_validate_scaling_configz$BaseTrainer._validate_scaling_configVs, 3$9    r/cdS)aoCalled during fit() to perform initial setup on the Trainer. .. note:: This method is run on a remote process. This method will not be called on the driver, so any expensive setup operations should be placed here and not in ``__init__``. This method is called prior to ``preprocess_datasets`` and ``training_loop``. Nr.rYs r0rBzBaseTrainer.setup_s  r/c0tdt)z Deprecated.za`preprocess_datasets` is no longer used, since preprocessors are no longer accepted by Trainers. )r PREPROCESSOR_DEPRECATION_MESSAGErs r0preprocess_datasetszBaseTrainer.preprocess_datasetsls)  W4T W W   r/ct)aLoop called by fit() to run training and report results to Tune. .. note:: This method runs on a remote process. ``self.datasets`` have already been evaluated if they were wrapped in a factory. You can use the :ref:`Ray Train utilities ` (:func:`train.report() ` and :func:`train.get_checkpoint() `) inside this training loop. Example: .. testcode:: from ray.train.trainer import BaseTrainer from ray import train class MyTrainer(BaseTrainer): def training_loop(self): for epoch_idx in range(5): ... train.report({"epoch": epoch_idx}) )NotImplementedErrorrs r0rCzBaseTrainer.training_loopss 6"!r/r#r$c ddlm}m}ddlm}|}|}|jjptj ||j_t|jj |jj|jj }|j rO||j ||||jj|jj|jj|j}n||||jt$j}||j |jt,j|jjt7|j} |} n&#|$r} | jp| } t-|| d } ~ wwxYwt=| d ksJ| d} | jr4t-d |t,j!g| j| S) a"Runs training. Returns: A Result object containing the training result. Raises: ray.train.base_trainer.TrainingFailedError: If any failures during the execution of ``self.as_trainable()``, or during the Tune execution loop. r) ResumeConfig TuneError)Tuner) storage_pathexperiment_dir_namer^)finished unfinishederrored)r] trainable param_space_resume_configr^)rrrM _entrypoint)trainer_cls_namer]Nr )"ray.tunerrray.tune.tunerr as_trainable%_extract_fields_for_tuner_param_spacerMnamerget_experiment_dir_namerr^rSr{ ResumeTypeRESUMErTrTRAINER_saveexperiment_fs_pathr'r,formatrr(strfit __cause__lenerrorrr-) rYrrrrrstoragetuner restore_msg result_gridrz parent_errorresults r0rzBaseTrainer.fitsB 54444444((((((%%'' @@BB  O UN$J9$U$U  !5 $ 4#A      MM'#'+|)4;+6=(3:    $(#C"  EEE#'?)1 E 7-w/IJJJ)6==!^4W/00>   E))++KK E E E;+!L&k22 D  E;1$$$$Q < & ;(;(OPQQ<  s#E88F=FFrrexperiment_pathc|j}|di}d|rfd|D|d<|j|f}|||t |t5}| tj |ddddS#1swxYwYdS)aSaves the current trainer's class along with the `param_dict` of parameters passed to this trainer's constructor. This is used to recreate the trainer on restore. Unless a parameter is re-specified during restoration (only a subset of parameters can be passed in again), that parameter will be loaded from the saved copy. Datasets should not be saved as part of the state. Instead, we save the keys and replace the dataset values with dummy functions that will raise an error if invoked. The error only serves as a guardrail for misuse (e.g., manually unpickling and constructing the Trainer again) and is not typically surfaced, since datasets must be re-specified upon restoration. r@ctN) RuntimeErrorr.r/r0raise_fnz#BaseTrainer._save..raise_fns r/ci|]}|Sr.r.)r8 dataset_namers r0r;z%BaseTrainer._save..s(&&&+7 h&&&r/N) rrQrlr create_diropen_output_streamrrdrewritergr)rYrrrrvr@cls_and_param_dictrurs @r0rzBaseTrainer._saves< %**,, >>*b11     &&&&;C&&&Jz "#nj9 o&&&  " "4#F#F#O#O#Q#Q R R 6VW GGFL!344 5 5 5 6 6 6 6 6 6 6 6 6 6 6 6 6 6 6 6 6 6s(CCCci}|jD]?}||jvr"tj|j|||<@|S)asExtracts fields to be included in `Tuner.param_space`. This is needed to leverage the full logging/integration offerings from Tune. For example, `param_space` is logged automatically to wandb integration. Currently only done for `train_loop_config`. Returns: A dictionary that should be passed to Tuner.param_space. )_fields_for_tuner_param_spacerrnrQdeepcopy)rYrrs r0rz1BaseTrainer._extract_fields_for_tuner_param_spacesX5 C CCd&++----"mD,.TrainTrainablez(Adds default resources to the Trainable.r_cS)z2Whether a dataset is provided through the Trainer.r.)r\has_base_datasets r0rzLBaseTrainer._generate_trainable_cls..TrainTrainable.has_base_dataset4s ('r/cS)zBReturns the unchanged scaling config provided through the Trainer.r.rs r0base_scaling_configzOBaseTrainer._generate_trainable_cls..TrainTrainable.base_scaling_config9s &%r/ct|}|dd}t||jd|_||jd<|jdt }t|tr t di|}|||jd<| r t| |dS)NrMT)new_keys_allowedrLr.) rrlrr1_merged_configgetrr._reconcile_scaling_config_with_trial_resourcesr _set_currentrrB) rYr1rq base_configrMmerged_scaling_configrTrainTrainablerdataset_contexts r0rBzABaseTrainer._generate_trainable_cls..TrainTrainable.setup>s"6ll )__\4@@ &1t'''#5?#L1(,(;(?(?$moo))%3T::S,9,R,R ,,_===nd++11&99999r/rLc|j}t|s|S|j|kr|S|t j|S)z ResourceChangingScheduler workaround. Ensures that the scaling config matches trial resources. This should be replaced with RCS returning a ScalingConfig in the future. )trial_resourcesras_placement_group_factoryrrfrom_placement_group_factory)rYrLrrr2s r0r zjBaseTrainer._generate_trainable_cls..TrainTrainable._reconcile_scaling_config_with_trial_resources\sm#'"6!/3HII*))'>+T+T+V+VVV))44^DDD$A/RRRr/cTt|jdSr)r_trainable_funcr)rYr1rs r0rzKBaseTrainer._generate_trainable_cls..TrainTrainable._trainable_funcvs&''(;<<<<.TrainTrainable.default_resource_request{sp*04Dn)U)U&4d;;U-:-T-T=S-T-T*+6+O+O*,,(0JJLLLr/)r(r)r*r+rJrK classmethodboolrrrrBr rr __classcell__)rrrrrrrLr2s@r0rr.sI : :'2'K $)4)O &  ( ( ( ( ( ([ ( &M & & & & &[ & : : : : : : : :< S&3 S S S S S S S S4 = = = = =  M M M M M[ M M M M Mr/r)#ray.tune.execution.placement_groupsrray.tune.trainablerrrLr4rrFr(rr@ray.data.contextr get_current) rYrr4train_coordinator_fn trainable_clsrrrrrrLr2s @@@@@@@r0_generate_trainable_clsz#BaseTrainer._generate_trainable_clss= NMMMMM444444n ,=& !{X    )4(<%% &:;;  ..  # 4 4 4 4 4 4)5577OO"O[ M[ M[ M[ M[ M[ M[ M[ M[ M[ M[ M[ M[ M][ M[ M[ Mzr/c`ddlm}|j}|}|j|fi|S)z,Converts self to a ``tune.Trainable`` class.r)r=)r<r=rr#with_parameters)rYr=r r"s r0rzBaseTrainer.as_trainablesJ& 4466 $t#MAA[AAAr/)NNNr)r_N)3r(r)r*r+rIr r__annotations__rJrrKrr rrr GenDatasetrrrZrrrar r osPathLikepyarrowrr FileSystemr{rrbrrrUrrrBrabcabstractmethodrCr rrrrr#rr)rs@r0r3r3~s]]@ / $s)&+d***',,,,%'! 37*.48-17; ( ( (!/ (Y' ( 4Z01 ( 4S>* (!) 4 ( ( ( (DZ<===?C4826 WW - WC$%W%WZ%:;W4Z01 W !/ W WWW>=[WrZ4..00?C88 - 8C$%8%WZ%:;8  888 [ 8,...(     4,4,4,l>F>F>F@m [          """"8Y   JVJJJ! JX6 -66666Bt"|k):|||||Bd;/BBBBBBBBr/)Or,rQrrloggingr(rj functoolsrpathlibrtypingrrrrr r r r pyarrow.fsr*r<ray.cloudpickle cloudpicklergray._common.usager ray._private.dictrray.air._internalrrWray.air._internal.configrray.air._internal.usagerray.air.configrrray.air.resultr ray.trainrray.train._internal.sessionrray.train._internal.storagerrrray.train.constantsrrray.train.contextrray.train.utilsrray.util.annotationsrrr ray.datar!rr"rdr' getLoggerr(loggerrrarRrr'rrFABCr3r.r/r0rGsy   RRRRRRRRRRRRRRRRRRRR ''''''))))))000000OOOOOO11111133333333!!!!!! 333333 @?????444444DDDDDDDDDD#      """""" 9hr9}55 6   8 $ $u!I,FII% -*--,  V,( #M2>BDV BV BV BV BV B#'V BV BV BV BV Br/