Ray Tune: set epochs equal to max_t and disable early stopping

ludwig/utils/defaults.py

@@ -182,10 +182,45 @@ def _set_proc_column(config: dict) -> None:
             feature[PROC_COLUMN] = compute_feature_hash(feature)
 
 
+def _merge_hyperopt_with_training(config: dict) -> None:
+    if 'hyperopt' not in config or TRAINING not in config:
+        return
+
+    scheduler = config['hyperopt'].get('sampler', {}).get('scheduler')
+    if not scheduler:
+        return
+
+    # Disable early stopping when using a scheduler. We achieve this by setting the parameter
+    # to -1, which ensures the condition to apply early stopping is never met.
+    training = config[TRAINING]
+    early_stop = training.get('early_stop')
+    if early_stop is not None and early_stop != -1:
+        raise ValueError(
+            'Cannot set training parameter `early_stop` when using a hyperopt scheduler. '
+            'Unset this parameter in your config.'
+        )
+    training['early_stop'] = -1
+
+    # At most one of max_t and epochs may be specified by the user, and we set them to be equal to
+    # ensure that Ludwig does not stop training before the scheduler has finished the trial.
+    max_t = scheduler.get('max_t')
+    epochs = training.get('epochs')
+    if max_t is not None and epochs is not None and max_t != epochs:
+        raise ValueError(
+            'Cannot set training parameter `epochs` when using a hyperopt scheduler with `max_t`. '
+            'Unset one of these parameters in your config.'
+        )
+    elif max_t is not None:
+        training['epochs'] = max_t
+    elif epochs is not None:
+        scheduler['max_t'] = epochs
+
+
 def merge_with_defaults(config):
     _perform_sanity_checks(config)
     _set_feature_column(config)
     _set_proc_column(config)
+    _merge_hyperopt_with_training(config)
 
     # ===== Preprocessing =====
     config['preprocessing'] = merge_dict(

tests/integration_tests/test_hyperopt_ray.py

@@ -79,7 +79,6 @@
         "scheduler": {
             "type": "hb_bohb",
             "time_attr": "training_iteration",
-            "max_t": 100,
             "reduction_factor": 4,
         },
         "num_samples": 3
@@ -116,12 +115,17 @@ def run_hyperopt_executor(sampler, executor, csv_filename,
         "input_features": input_features,
         "output_features": output_features,
         "combiner": {"type": "concat", "num_fc_layers": 2},
-        "training": {"epochs": 2, "learning_rate": 0.001}
+        "training": {"epochs": 2, "learning_rate": 0.001},
+        "hyperopt": {
+            **HYPEROPT_CONFIG,
+            "executor": executor,
+            "sampler": sampler,
+        },
     }
 
     config = merge_with_defaults(config)
 
-    hyperopt_config = HYPEROPT_CONFIG.copy()
+    hyperopt_config = config["hyperopt"]
 
     if validate_output_feature:
         hyperopt_config['output_feature'] = output_features[0]['name']