Skip to content

Commit 3f8b299

Browse files
author
nayan3107
committed
fix(train): complete PipelineSession support for V3 trainers (SFT/DPO/RLAIF/RLVR)
Add @runnable_by_pipeline decorator and PascalCase request serialization so TrainingStep can consume step_args from V3 fine-tuning trainers. Changes: - Add @runnable_by_pipeline decorator on train() for all 4 trainers - Build PascalCase serialized request (remove session/region, pop job name) - Fix Tags to PascalCase (JumpStart returns lowercase key/value) - Add source_code=None to BaseTrainer (required by get_code_hash) - Add consumer tests (TrainingStep.arguments validation) - Add regular session regression tests Fixes: #6163
1 parent 0316c38 commit 3f8b299

10 files changed

Lines changed: 1022 additions & 32 deletions

File tree

‎sagemaker-mlops/tests/integ/workflow/test_v3_trainer_pipeline.py‎

Lines changed: 399 additions & 0 deletions
Large diffs are not rendered by default.

‎sagemaker-train/src/sagemaker/train/base_trainer.py‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -160,6 +160,7 @@ def __init__(
160160
self.base_model_name = base_model_name
161161
self.disable_output_compression = disable_output_compression
162162
self.notification_rule_arn = None
163+
self.source_code = None
163164

164165
# Set up notifications if configured
165166
if notifications:

‎sagemaker-train/src/sagemaker/train/dpo_trainer.py‎

Lines changed: 17 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,9 @@
55
from sagemaker.train.common import TrainingType, CustomizationTechnique, JOB_TYPE
66
from sagemaker.core.resources import TrainingJob, ModelPackageGroup, ModelPackage
77
from sagemaker.core.shapes import VpcConfig
8-
from sagemaker.core.workflow.pipeline_context import PipelineSession
8+
from sagemaker.core.workflow.pipeline_context import PipelineSession, runnable_by_pipeline
9+
from sagemaker.core.utils.utils import serialize
10+
from sagemaker.core.apiutils._boto_functions import to_pascal_case
911
from sagemaker.train.defaults import TrainDefaults
1012
from sagemaker.train.utils import _get_unique_name, _get_jumpstart_tags
1113
from sagemaker.train.configs import StoppingCondition
@@ -230,6 +232,7 @@ def _process_hyperparameters(self):
230232
("compute", TelemetryParamType.ATTR_TYPE),
231233
],
232234
)
235+
@runnable_by_pipeline
233236
def train(self,
234237
training_dataset: Optional[Union[str, DataSet]] = None,
235238
validation_dataset: Optional[Union[str, DataSet]] = None,
@@ -375,8 +378,19 @@ def train(self,
375378
# This must come before data path validation since in pipeline mode
376379
# the data path may be a pipeline parameter that doesn't exist yet.
377380
if isinstance(sagemaker_session, PipelineSession):
378-
sagemaker_session._intercept_create_request(create_args, None, "train")
379-
return sagemaker_session.context
381+
pipeline_args = {k: v for k, v in create_args.items()
382+
if k not in ("session", "region")}
383+
pipeline_args.pop("training_job_name", None)
384+
pipeline_request = {to_pascal_case(k): v for k, v in pipeline_args.items()}
385+
# Fix Tag dicts to PascalCase (JumpStart tags use lowercase key/value)
386+
if "Tags" in pipeline_request and pipeline_request["Tags"]:
387+
pipeline_request["Tags"] = [
388+
{"Key": t.get("key", t.get("Key")), "Value": t.get("value", t.get("Value"))}
389+
for t in pipeline_request["Tags"]
390+
]
391+
serialized_request = serialize(pipeline_request)
392+
sagemaker_session._intercept_create_request(serialized_request, None, "train")
393+
return
380394

381395
# Validate data paths exist before submission
382396
effective_training = training_dataset or self.training_dataset

‎sagemaker-train/src/sagemaker/train/rlaif_trainer.py‎

Lines changed: 17 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,9 @@
44
from sagemaker.train.common import TrainingType, CustomizationTechnique, JOB_TYPE
55
from sagemaker.core.resources import TrainingJob, ModelPackageGroup, MlflowTrackingServer, ModelPackage
66
from sagemaker.core.shapes import VpcConfig
7-
from sagemaker.core.workflow.pipeline_context import PipelineSession
7+
from sagemaker.core.workflow.pipeline_context import PipelineSession, runnable_by_pipeline
8+
from sagemaker.core.utils.utils import serialize
9+
from sagemaker.core.apiutils._boto_functions import to_pascal_case
810
from sagemaker.train.defaults import TrainDefaults
911
from sagemaker.train.utils import _get_unique_name, _get_jumpstart_tags
1012
from sagemaker.train.common_utils.recipe_utils import _get_hub_content_metadata
@@ -224,6 +226,7 @@ def _validate_reward_model_id(self, reward_model_id):
224226
("custom_reward_function", TelemetryParamType.ATTR_EXISTS),
225227
],
226228
)
229+
@runnable_by_pipeline
227230
def train(self, training_dataset: Optional[Union[str, DataSet]] = None, validation_dataset: Optional[Union[str, DataSet]] = None, wait: bool = True, wait_timeout: Optional[int] = None, poll: int = 5, dry_run: bool = False):
228231
"""Execute the RLAIF training job.
229232
@@ -341,8 +344,19 @@ def train(self, training_dataset: Optional[Union[str, DataSet]] = None, validati
341344
# This must come before data path validation since in pipeline mode
342345
# the data path may be a pipeline parameter that doesn't exist yet.
343346
if isinstance(sagemaker_session, PipelineSession):
344-
sagemaker_session._intercept_create_request(create_args, None, "train")
345-
return sagemaker_session.context
347+
pipeline_args = {k: v for k, v in create_args.items()
348+
if k not in ("session", "region")}
349+
pipeline_args.pop("training_job_name", None)
350+
pipeline_request = {to_pascal_case(k): v for k, v in pipeline_args.items()}
351+
# Fix Tag dicts to PascalCase (JumpStart tags use lowercase key/value)
352+
if "Tags" in pipeline_request and pipeline_request["Tags"]:
353+
pipeline_request["Tags"] = [
354+
{"Key": t.get("key", t.get("Key")), "Value": t.get("value", t.get("Value"))}
355+
for t in pipeline_request["Tags"]
356+
]
357+
serialized_request = serialize(pipeline_request)
358+
sagemaker_session._intercept_create_request(serialized_request, None, "train")
359+
return
346360

347361
# Validate data paths exist before submission
348362
effective_training = training_dataset or self.training_dataset

‎sagemaker-train/src/sagemaker/train/rlvr_trainer.py‎

Lines changed: 17 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,9 @@
66
from sagemaker.train.common import TrainingType, CustomizationTechnique, JOB_TYPE
77
from sagemaker.core.resources import TrainingJob, ModelPackageGroup, MlflowTrackingServer, ModelPackage
88
from sagemaker.core.shapes import VpcConfig
9-
from sagemaker.core.workflow.pipeline_context import PipelineSession
9+
from sagemaker.core.workflow.pipeline_context import PipelineSession, runnable_by_pipeline
10+
from sagemaker.core.utils.utils import serialize
11+
from sagemaker.core.apiutils._boto_functions import to_pascal_case
1012
from sagemaker.train.defaults import TrainDefaults
1113
from sagemaker.train.utils import _get_unique_name, _get_jumpstart_tags
1214
from sagemaker.ai_registry.dataset import DataSet
@@ -383,6 +385,7 @@ def _verify_reward_function(
383385
)
384386

385387
@_telemetry_emitter(feature=Feature.MODEL_CUSTOMIZATION, func_name="RLVRTrainer.train")
388+
@runnable_by_pipeline
386389
def train(self, training_dataset: Optional[Union[str, DataSet]] = None,
387390
validation_dataset: Optional[Union[str, DataSet]] = None, wait: bool = True, wait_timeout: Optional[int] = None, poll: int = 5, dry_run: bool = False):
388391
"""Execute the RLVR training job.
@@ -561,8 +564,19 @@ def train(self, training_dataset: Optional[Union[str, DataSet]] = None,
561564
# This must come before data path validation since in pipeline mode
562565
# the data path may be a pipeline parameter that doesn't exist yet.
563566
if isinstance(sagemaker_session, PipelineSession):
564-
sagemaker_session._intercept_create_request(create_args, None, "train")
565-
return sagemaker_session.context
567+
pipeline_args = {k: v for k, v in create_args.items()
568+
if k not in ("session", "region")}
569+
pipeline_args.pop("training_job_name", None)
570+
pipeline_request = {to_pascal_case(k): v for k, v in pipeline_args.items()}
571+
# Fix Tag dicts to PascalCase (JumpStart tags use lowercase key/value)
572+
if "Tags" in pipeline_request and pipeline_request["Tags"]:
573+
pipeline_request["Tags"] = [
574+
{"Key": t.get("key", t.get("Key")), "Value": t.get("value", t.get("Value"))}
575+
for t in pipeline_request["Tags"]
576+
]
577+
serialized_request = serialize(pipeline_request)
578+
sagemaker_session._intercept_create_request(serialized_request, None, "train")
579+
return
566580

567581
# Validate data paths exist before submission
568582
effective_training = training_dataset or self.training_dataset

‎sagemaker-train/src/sagemaker/train/sft_trainer.py‎

Lines changed: 19 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,9 @@
44
from sagemaker.train.common import TrainingType, CustomizationTechnique, JOB_TYPE
55
from sagemaker.core.resources import TrainingJob, ModelPackageGroup, ModelPackage
66
from sagemaker.core.shapes import VpcConfig
7-
from sagemaker.core.workflow.pipeline_context import PipelineSession
7+
from sagemaker.core.workflow.pipeline_context import PipelineSession, runnable_by_pipeline
8+
from sagemaker.core.utils.utils import serialize
9+
from sagemaker.core.apiutils._boto_functions import to_pascal_case
810
from sagemaker.train.defaults import TrainDefaults
911
from sagemaker.train.utils import _get_unique_name, _get_jumpstart_tags
1012
from sagemaker.ai_registry.dataset import DataSet
@@ -262,6 +264,7 @@ def _process_hyperparameters(self):
262264
("compute", TelemetryParamType.ATTR_TYPE),
263265
],
264266
)
267+
@runnable_by_pipeline
265268
def train(self, training_dataset: Optional[Union[str, DataSet]] = None, validation_dataset: Optional[Union[str, DataSet]] = None, wait: bool = True, wait_timeout: Optional[int] = None, poll: int = 5, dry_run: bool = False):
266269
"""Execute the SFT training job.
267270
@@ -443,8 +446,20 @@ def train(self, training_dataset: Optional[Union[str, DataSet]] = None, validati
443446
# This must come before data path validation since in pipeline mode
444447
# the data path may be a pipeline parameter that doesn't exist yet.
445448
if isinstance(sagemaker_session, PipelineSession):
446-
sagemaker_session._intercept_create_request(create_args, None, "train")
447-
return sagemaker_session.context
449+
# Build pipeline-compatible request: PascalCase, serialized, no session/region
450+
pipeline_args = {k: v for k, v in create_args.items()
451+
if k not in ("session", "region")}
452+
pipeline_args.pop("training_job_name", None)
453+
pipeline_request = {to_pascal_case(k): v for k, v in pipeline_args.items()}
454+
# Fix Tag dicts to PascalCase (JumpStart tags use lowercase key/value)
455+
if "Tags" in pipeline_request and pipeline_request["Tags"]:
456+
pipeline_request["Tags"] = [
457+
{"Key": t.get("key", t.get("Key")), "Value": t.get("value", t.get("Value"))}
458+
for t in pipeline_request["Tags"]
459+
]
460+
serialized_request = serialize(pipeline_request)
461+
sagemaker_session._intercept_create_request(serialized_request, None, "train")
462+
return
448463

449464
# Validate data paths exist before submission
450465
effective_training = training_dataset or self.training_dataset
@@ -481,4 +496,4 @@ def train(self, training_dataset: Optional[Union[str, DataSet]] = None, validati
481496
logger.error("Error: %s", e)
482497

483498
self._latest_training_job = training_job
484-
return training_job
499+
return training_job

‎sagemaker-train/tests/unit/train/test_dpo_trainer.py‎

Lines changed: 131 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -815,7 +815,135 @@ def test_train_with_pipeline_session_does_not_launch_job(
815815

816816
result = trainer.train()
817817

818+
from sagemaker.core.workflow.pipeline_context import _StepArguments
819+
# @runnable_by_pipeline intercepts and returns _StepArguments
820+
assert isinstance(result, _StepArguments)
821+
assert result.caller_name == "train"
822+
assert result.func is not None
823+
assert result.func_args[0] is trainer
818824
mock_training_job_create.assert_not_called()
819-
pipeline_session._intercept_create_request.assert_called_once()
820-
assert pipeline_session._intercept_create_request.call_args[0][2] == "train"
821-
assert result == step_args
825+
826+
@patch('sagemaker.train.dpo_trainer._create_model_package_config')
827+
@patch('sagemaker.train.dpo_trainer._create_mlflow_config')
828+
@patch('sagemaker.train.dpo_trainer._create_output_config')
829+
@patch('sagemaker.train.dpo_trainer._create_serverless_config')
830+
@patch('sagemaker.train.dpo_trainer._convert_input_data_to_channels')
831+
@patch('sagemaker.train.dpo_trainer._create_input_data_config')
832+
@patch('sagemaker.train.dpo_trainer._get_unique_name')
833+
@patch('sagemaker.train.dpo_trainer.TrainDefaults.get_role')
834+
@patch('sagemaker.train.dpo_trainer.TrainDefaults.get_sagemaker_session')
835+
@patch('sagemaker.train.dpo_trainer._validate_and_resolve_model_package_group')
836+
@patch('sagemaker.train.dpo_trainer._get_fine_tuning_options_and_model_arn')
837+
@patch('sagemaker.train.dpo_trainer._resolve_model_and_name')
838+
@patch('sagemaker.train.common_utils.finetune_utils._get_beta_session')
839+
@patch('sagemaker.core.resources.TrainingJob.create')
840+
def test_train_pipeline_session_produces_valid_step_arguments(
841+
self, mock_training_job_create, mock_beta_session, mock_resolve_model,
842+
mock_finetuning_options, mock_validate_group, mock_get_session, mock_get_role,
843+
mock_unique_name, mock_input_config, mock_convert_channels,
844+
mock_serverless_config, mock_output_config, mock_mlflow_config, mock_model_package_config,
845+
):
846+
"""TrainingStep.arguments produces valid PascalCase dict."""
847+
from sagemaker.train.dpo_trainer import DPOTrainer
848+
from sagemaker.core.workflow.pipeline_context import PipelineSession, _StepArguments
849+
from sagemaker.mlops.workflow.steps import TrainingStep
850+
851+
pipeline_session = PipelineSession.__new__(PipelineSession)
852+
pipeline_session._context = None
853+
pipeline_session.boto_session = Mock()
854+
pipeline_session.boto_session.region_name = "us-west-2"
855+
mock_get_session.return_value = pipeline_session
856+
857+
mock_resolve_model.return_value = ("test-model", "test")
858+
mock_hyperparams = Mock()
859+
mock_hyperparams.to_dict.return_value = {"lr": "0.001"}
860+
mock_hyperparams._specs = {"lr": {"type": "string"}}
861+
mock_hyperparams._user_set = set()
862+
mock_finetuning_options.return_value = (mock_hyperparams, "arn:model", False)
863+
mock_validate_group.return_value = "grp"
864+
mock_get_role.return_value = "arn:aws:iam::123:role/Role"
865+
mock_unique_name.return_value = "test-job"
866+
mock_input_config.return_value = [{"DataSource": {"S3DataSource": {"S3Uri": "s3://data"}}}]
867+
mock_convert_channels.return_value = [{"ChannelName": "train"}]
868+
mock_serverless_config.return_value = {"BaseModelArn": "arn:model", "JobType": "FineTuning"}
869+
mock_output_config.return_value = {"S3OutputPath": "s3://output"}
870+
mock_mlflow_config.return_value = None
871+
mock_model_package_config.return_value = None
872+
mock_beta_session.return_value = pipeline_session
873+
874+
trainer = DPOTrainer(model="test-model", training_dataset="s3://bucket/data", model_package_group="grp", sagemaker_session=pipeline_session)
875+
trainer._model_arn = "arn:model"
876+
trainer._model_name = "test-model"
877+
trainer.accept_eula = True
878+
trainer.hyperparameters = mock_hyperparams
879+
880+
result = trainer.train()
881+
step = TrainingStep(name="TestDPO", step_args=result)
882+
arguments = step.arguments
883+
884+
assert isinstance(arguments, dict)
885+
assert "session" not in arguments
886+
assert "region" not in arguments
887+
tags = arguments.get("Tags", [])
888+
for t in tags:
889+
assert "Key" in t and "Value" in t
890+
891+
@patch('sagemaker.train.dpo_trainer._create_model_package_config')
892+
@patch('sagemaker.train.dpo_trainer._create_mlflow_config')
893+
@patch('sagemaker.train.dpo_trainer._create_output_config')
894+
@patch('sagemaker.train.dpo_trainer._create_serverless_config')
895+
@patch('sagemaker.train.dpo_trainer._convert_input_data_to_channels')
896+
@patch('sagemaker.train.dpo_trainer._create_input_data_config')
897+
@patch('sagemaker.train.dpo_trainer._get_unique_name')
898+
@patch('sagemaker.train.dpo_trainer.TrainDefaults.get_role')
899+
@patch('sagemaker.train.dpo_trainer.TrainDefaults.get_sagemaker_session')
900+
@patch('sagemaker.train.dpo_trainer._validate_and_resolve_model_package_group')
901+
@patch('sagemaker.train.dpo_trainer._get_fine_tuning_options_and_model_arn')
902+
@patch('sagemaker.train.dpo_trainer._resolve_model_and_name')
903+
@patch('sagemaker.train.common_utils.finetune_utils._get_beta_session')
904+
@patch('sagemaker.train.common_utils.data_utils.validate_data_path_exists')
905+
@patch('sagemaker.core.resources.TrainingJob.create')
906+
def test_train_without_pipeline_session_launches_job(
907+
self, mock_training_job_create, mock_validate_path, mock_beta_session,
908+
mock_resolve_model, mock_finetuning_options, mock_validate_group,
909+
mock_get_session, mock_get_role, mock_unique_name, mock_input_config,
910+
mock_convert_channels, mock_serverless_config, mock_output_config,
911+
mock_mlflow_config, mock_model_package_config,
912+
):
913+
"""Regular Session launches job normally."""
914+
from sagemaker.train.dpo_trainer import DPOTrainer
915+
916+
regular_session = Mock()
917+
regular_session.boto_session = Mock()
918+
regular_session.boto_session.region_name = "us-west-2"
919+
regular_session.sagemaker_config = {}
920+
mock_get_session.return_value = regular_session
921+
922+
mock_resolve_model.return_value = ("test-model", "test")
923+
mock_hyperparams = Mock()
924+
mock_hyperparams.to_dict.return_value = {"lr": "0.001"}
925+
mock_hyperparams._specs = {"lr": {"type": "string"}}
926+
mock_hyperparams._user_set = set()
927+
mock_finetuning_options.return_value = (mock_hyperparams, "arn:model", False)
928+
mock_validate_group.return_value = "grp"
929+
mock_get_role.return_value = "arn:aws:iam::123:role/Role"
930+
mock_unique_name.return_value = "test-job"
931+
mock_input_config.return_value = {}
932+
mock_convert_channels.return_value = []
933+
mock_serverless_config.return_value = {"BaseModelArn": "arn:model"}
934+
mock_output_config.return_value = {"S3OutputPath": "s3://output"}
935+
mock_mlflow_config.return_value = None
936+
mock_model_package_config.return_value = None
937+
mock_beta_session.return_value = regular_session
938+
mock_training_job = Mock()
939+
mock_training_job_create.return_value = mock_training_job
940+
941+
trainer = DPOTrainer(model="test-model", training_dataset="s3://bucket/data", model_package_group="grp", sagemaker_session=regular_session)
942+
trainer._model_arn = "arn:model"
943+
trainer._model_name = "test-model"
944+
trainer.accept_eula = True
945+
trainer.hyperparameters = mock_hyperparams
946+
947+
result = trainer.train(wait=False)
948+
mock_training_job_create.assert_called_once()
949+
assert result == mock_training_job

0 commit comments

Comments
 (0)