Training job hyperparameter name configurable via pipeline parameter.

### Product Version

- [ ] Amazon SageMaker Studio Classic
- [ ] Amazon SageMaker Studio
- [x] It is not related to SageMaker Studio

### Product Category

Pipelines

### Description

I'd like to be able to configure the training job hyperparameter name via a pipeline variable. The following code illustrates what I'd like to achieve:
```python
from sagemaker.pytorch import PyTorch
from sagemaker.workflow.parameters import ParameterString
from sagemaker.workflow.steps import TrainingStep
from sagemaker.workflow.pipeline import Pipeline
from sagemaker.workflow.pipeline_context import (
PipelineSession
)

if __name__ == "__main__":
pipeline_session = PipelineSession()

hparam_name = ParameterString("HParamName")
hparam_value = ParameterString("HParamValue")
estimator = PyTorch(
sagemaker_session=pipeline_session,
instance_type='ml.m5.large',
instance_count=1,
framework_version="2.3",
py_version="py311",
source_dir='source',
entry_point='main.py',
hyperparameters={
hparam_name: hparam_value
},
metric_definitions=[
{'Name': 'valid:loss', 'Regex': 'valid_loss=([0-9]+\\.?[0-9]*)'}
]
)

training_step = TrainingStep(
name="Training",
step_args=estimator.fit(),
)

pipeline = Pipeline(
name="TestTrainingPipeline",
parameters=[hparam_name, hparam_value],
steps=[training_step],
sagemaker_session=pipeline_session
)

pipeline.upsert()
pipeline.start(
execution_display_name="TrainingTestParam1",
parameters={
"HParamName": "hparam1",
"HParamValue": 1.0
}
)
pipeline.start(
execution_display_name="TrainingTestParam2",
parameters={
"HParamName": "hparam2",
"HParamValue": 1.0
}
)
```
I would expect this code to create a pipeline with one training step and run it the first time using the hyperparameter hparam1 and the second time using the hyperparameter hparam2.

The above code ends up with exception: `TypeError: Pipeline variables do not support __str__ operation. Please use `.to_string()` to convert it to string type in execution time or use `.expr` to translate it to Json for display purpose in Python SDK.`
Traceback:
```
╭─────────────────────────────── Traceback (most recent call last) ────────────────────────────────╮
│ HIDDEN_PATH/test.py:44 in │
│ │
│ 41 │ │ sagemaker_session=pipeline_session │
│ 42 │ ) │
│ 43 │ │
│ ❱ 44 │ pipeline.upsert() │
│ 45 │ pipeline.start( │
│ 46 │ │ execution_display_name="TrainingTestParam1", │
│ 47 │ │ parameters={"HParamName": "hparam1"} │
│ │
│ HIDDEN_PATH/.venv/lib/python3.12/site-packages/sagemaker/workflow/pipeline.py:292 in upsert │
│ │
│ 289 │ │ │ # after fetching the config. │
│ 290 │ │ │ raise ValueError("An AWS IAM role is required to create or update a Pipeline │
│ 291 │ │ try: │
│ ❱ 292 │ │ │ response = self.create(role_arn, description, tags, parallelism_config) │
│ 293 │ │ except ClientError as ce: │
│ 294 │ │ │ error_code = ce.response["Error"]["Code"] │
│ 295 │ │ │ error_message = ce.response["Error"]["Message"] │
│ │
│ HIDDEN_PATH/.venv/lib/python3.12/site-packages/sagemaker/workflow/pipeline.py:164 in create │
│ │
│ 161 │ │ tags = format_tags(tags) │
│ 162 │ │ tags = _append_project_tags(tags) │
│ 163 │ │ tags = self.sagemaker_session._append_sagemaker_config_tags(tags, PIPELINE_TAGS_ │
│ ❱ 164 │ │ kwargs = self._create_args(role_arn, description, parallelism_config) │
│ 165 │ │ update_args( │
│ 166 │ │ │ kwargs, │
│ 167 │ │ │ Tags=tags, │
│ │
│ HIDDEN_PATH/.venv/lib/python3.12/site-packages/sagemaker/workflow/pipeline.py:186 │
│ in _create_args │
│ │
│ 183 │ │ Returns: │
│ 184 │ │ │ A keyword argument dict for calling create_pipeline. │
│ 185 │ │ """ │
│ ❱ 186 │ │ pipeline_definition = self.definition() │
│ 187 │ │ kwargs = dict( │
│ 188 │ │ │ PipelineName=self.name, │
│ 189 │ │ │ RoleArn=role_arn, │
│ │
│ HIDDEN_PATH/.venv/lib/python3.12/site-packages/sagemaker/workflow/pipeline.py:392 in definition │
│ │
│ 389 │ │ │ sagemaker_session=self.sagemaker_session, │
│ 390 │ │ │ steps=self.steps, │
│ 391 │ │ │ pipeline_definition_config=self.pipeline_definition_config, │
│ ❱ 392 │ │ ).build() │
│ 393 │ │ │
│ 394 │ │ request_dict = { │
│ 395 │ │ │ "Version": self._version, │
│ │
│ HIDDEN_PATH/.venv/lib/python3.12/site-packages/sagemaker/workflow/_steps_compiler.py:406 │
│ in build │
│ │
│ 403 │ │ if self._build_count > 1: │
│ 404 │ │ │ raise RuntimeError("Cannot build a pipeline more than once with the same com │
│ 405 │ │ │
│ ❱ 406 │ │ return self._initialize_queue_and_build(self._input_steps) │
│ 407 │
│ │
│ HIDDEN_PATH/.venv/lib/python3.12/site-packages/sagemaker/workflow/_steps_compiler.py:390 │
│ in _initialize_queue_and_build │
│ │
│ 387 │ │ │ if isinstance(step, ConditionStep): │
│ 388 │ │ │ │ compiled_steps.append(self._build_condition_step(step)) │
│ 389 │ │ │ else: │
│ ❱ 390 │ │ │ │ compiled_steps.append(self._build_step(step)) │
│ 391 │ │ │
│ 392 │ │ self._set_serialize_output_to_json_flag(compiled_steps) │
│ 393 │ │ return compiled_steps │
│ │
│ HIDDEN_PATH/.venv/lib/python3.12/site-packages/sagemaker/workflow/_steps_compiler.py:331 │
│ in _build_step │
│ │
│ 328 │ │ │ pipeline_build_time=self.pipeline_build_time, │
│ 329 │ │ │ function_step_secret_token=self._function_step_secret_token, │
│ 330 │ │ ) as context: │
│ ❱ 331 │ │ │ request_dict = step.to_request() │
│ 332 │ │ │ │
│ 333 │ │ │ self.upload_runtime_scripts = context.upload_runtime_scripts │
│ 334 │ │ │ self.upload_workspace = context.upload_workspace │
│ │
│ HIDDEN_PATH/.venv/lib/python3.12/site-packages/sagemaker/workflow/steps.py:551 in to_request │
│ │
│ 548 │ │
│ 549 │ def to_request(self) -> RequestType: │
│ 550 │ │ """Updates the request dictionary with cache configuration.""" │
│ ❱ 551 │ │ request_dict = super().to_request() │
│ 552 │ │ if self.cache_config: │
│ 553 │ │ │ request_dict.update(self.cache_config.config) │
│ 554 │
│ │
│ HIDDEN_PATH/.venv/lib/python3.12/site-packages/sagemaker/workflow/steps.py:390 in to_request │
│ │
│ 387 │ │
│ 388 │ def to_request(self) -> RequestType: │
│ 389 │ │ """Gets the request structure for `ConfigurableRetryStep`.""" │
│ ❱ 390 │ │ step_dict = super().to_request() │
│ 391 │ │ if self.retry_policies: │
│ 392 │ │ │ step_dict["RetryPolicies"] = self._resolve_retry_policy(self.retry_policies) │
│ 393 │ │ return step_dict │
│ │
│ HIDDEN_PATH/.venv/lib/python3.12/site-packages/sagemaker/workflow/steps.py:147 in to_request │
│ │
│ 144 │ │ request_dict = { │
│ 145 │ │ │ "Name": self.name, │
│ 146 │ │ │ "Type": self.step_type.value, │
│ ❱ 147 │ │ │ "Arguments": self.arguments, │
│ 148 │ │ } │
│ 149 │ │ if self.depends_on: │
│ 150 │ │ │ request_dict["DependsOn"] = list(self.depends_on) │
│ │
│ HIDDEN_PATH/.venv/lib/python3.12/site-packages/sagemaker/workflow/steps.py:522 in arguments │
│ │
│ 519 │ │ if self.step_args: │
│ 520 │ │ │ # execute fit function with saved parameters, │
│ 521 │ │ │ # and store args in PipelineSession's _context │
│ ❱ 522 │ │ │ execute_job_functions(self.step_args) │
│ 523 │ │ │ │
│ 524 │ │ │ # populate request dict with args │
│ 525 │ │ │ estimator = self.step_args.func_args[0] │
│ │
│ HIDDEN_PATH/.venv/lib/python3.12/site-packages/sagemaker/workflow/utilities.py:444 │
│ in execute_job_functions │
│ │
│ 441 │ │ │ a pipeline step, contains the necessary function information │
│ 442 │ """ │
│ 443 │ │
│ ❱ 444 │ chained_args = step_args.func(*step_args.func_args, **step_args.func_kwargs) │
│ 445 │ if isinstance(chained_args, _StepArguments): │
│ 446 │ │ execute_job_functions(chained_args) │
│ 447 │
│ │
│ HIDDEN_PATH/.venv/lib/python3.12/site-packages/sagemaker/estimator.py:1373 in fit │
│ │
│ 1370 │ │ self._prepare_for_training(job_name=job_name) │
│ 1371 │ │ │
│ 1372 │ │ experiment_config = check_and_get_run_experiment_config(experiment_config) │
│ ❱ 1373 │ │ self.latest_training_job = _TrainingJob.start_new(self, inputs, experiment_confi │
│ 1374 │ │ self.jobs.append(self.latest_training_job) │
│ 1375 │ │ forward_to_mlflow_tracking_server = False │
│ 1376 │ │ if os.environ.get("MLFLOW_TRACKING_URI") and self.enable_network_isolation(): │
│ │
│ HIDDEN_PATH/.venv/lib/python3.12/site-packages/sagemaker/estimator.py:2511 in start_new │
│ │
│ 2508 │ │ │ sagemaker.estimator._TrainingJob: Constructed object that captures │
│ 2509 │ │ │ all information about the started training job. │
│ 2510 │ │ """ │
│ ❱ 2511 │ │ train_args = cls._get_train_args(estimator, inputs, experiment_config) │
│ 2512 │ │ │
│ 2513 │ │ logger.debug("Train args after processing defaults: %s", train_args) │
│ 2514 │ │ estimator.sagemaker_session.train(**train_args) │
│ │
│ HIDDEN_PATH/.venv/lib/python3.12/site-packages/sagemaker/estimator.py:2556 in _get_train_args │
│ │
│ 2553 │ │ │
│ 2554 │ │ config = _Job._load_config(inputs, estimator) │
│ 2555 │ │ │
│ ❱ 2556 │ │ current_hyperparameters = estimator.hyperparameters() │
│ 2557 │ │ if current_hyperparameters is not None: │
│ 2558 │ │ │ hyperparameters = {str(k): to_string(v) for (k, v) in current_hyperparameter │
│ 2559 │
│ │
│ HIDDEN_PATH/.venv/lib/python3.12/site-packages/sagemaker/pytorch/estimator.py:487 │
│ in hyperparameters │
│ │
│ 484 │ │
│ 485 │ def hyperparameters(self): │
│ 486 │ │ """Return hyperparameters used by your custom PyTorch code during model training │
│ ❱ 487 │ │ hyperparameters = super(PyTorch, self).hyperparameters() │
│ 488 │ │ additional_hyperparameters = self._pytorch_distribution_configuration( │
│ 489 │ │ │ distribution=self.distribution │
│ 490 │ │ ) │
│ │
│ HIDDEN_PATH/.venv/lib/python3.12/site-packages/sagemaker/estimator.py:3725 in hyperparameters │
│ │
│ 3722 │ │ Returns: │
│ 3723 │ │ │ dict[str, str]: The hyperparameters. │
│ 3724 │ │ """ │
│ ❱ 3725 │ │ return EstimatorBase._json_encode_hyperparameters(self._hyperparameters) │
│ 3726 │ │
│ 3727 │ @classmethod │
│ 3728 │ def _prepare_init_params_from_job_description(cls, job_details, model_channel_name=N │
│ │
│ HIDDEN_PATH/.venv/lib/python3.12/site-packages/sagemaker/estimator.py:900 │
│ in _json_encode_hyperparameters │
│ │
│ 897 │ │ current_hyperparameters = hyperparameters │
│ 898 │ │ if current_hyperparameters is not None: │
│ 899 │ │ │ hyperparameters = { │
│ ❱ 900 │ │ │ │ str(k): (v.to_string() if is_pipeline_variable(v) else json.dumps(v)) │
│ 901 │ │ │ │ for (k, v) in current_hyperparameters.items() │
│ 902 │ │ │ } │
│ 903 │ │ return hyperparameters │
│ │
│ HIDDEN_PATH/.venv/lib/python3.12/site-packages/sagemaker/workflow/entities.py:90 in __str__ │ │
│ │
│ 87 │ │
│ 88 │ def __str__(self): │
│ 89 │ │ """Override built-in String function for PipelineVariable""" │
│ ❱ 90 │ │ raise TypeError( │
│ 91 │ │ │ "Pipeline variables do not support __str__ operation. " │
│ 92 │ │ │ "Please use `.to_string()` to convert it to string type in execution time " │
│ 93 │ │ │ "or use `.expr` to translate it to Json for display purpose in Python SDK." │
╰──────────────────────────────────────────────────────────────────────────────────────────────────╯
TypeError: Pipeline variables do not support __str__ operation. Please use `.to_string()` to convert it to string type
in execution time or use `.expr` to translate it to Json for display purpose in Python SDK.
```
But it seems that it is not only a problem with the request serialization on the SDK side, but in general CreateTrainingJob request does not support hyperparameters names being pipeline variable expressions.

Why I need that?
I want to train and evaluate several different ML models for the same task within a single pipeline. These models differ in some hyperparameters. Currently, I'm working around this problem by passing the hyperparameters via a file that is one of the inputs to the training step.

### Other Details

_No response_

Reproduce the example with the SageMaker Python SDK, then trace pipeline.py, _steps_compiler.py, and steps.py from Pipeline.upsert() through step serialization. Determine whether a ParameterString can supply a training hyperparameter name, and verify that the pipeline can be created and run with hparam1 and hparam2 across separate executions.

Written by the indexing model from the issue text.