DEPRECATED — This package (amzn-nova-forge) is deprecated and will no longer receive feature updates. Please use the SageMaker Python SDK V3 (pip install "sagemaker>=3.19.0") for Amazon Nova model customization. 📓 SageMaker SDK sample notebook: Nova Serverless End-to-End Example on GitHub

Amazon Nova Forge SDK

A comprehensive Python SDK for fine-tuning and customizing Amazon Nova models. This SDK provides a unified interface for training, evaluation, deployment, and monitoring of Nova models across both SageMaker Training Jobs and SageMaker HyperPod.

---

Migrating from Nova Forge SDK to SageMaker Python SDK V3

Why Migrate

The amzn-nova-forge package is deprecated. Amazon Nova model customization functionality is available in the SageMaker Python SDK V3.

What's Different (Summary)

• Compute is a config object (HyperPodCompute, TrainingJobCompute), not a runtime manager
• Model is a string identifier (e.g. "nova-textgeneration-lite-v2"), not an enum; also accepts S3 checkpoint paths for iterative training
• Deployment uses ModelBuilder/BedrockModelBuilder pattern instead of ForgeDeployer
• Overrides use full recipe paths (e.g. "recipes.training_config.trainer.lr"); use trainer.get_resolved_recipe() to inspect the final merged recipe
• No ForgeConfig object — shared settings are passed directly to trainer constructors
• Job notifications currently support SMTJ only — pass a notifications dict with SNS topic and EventBridge event bus ARNs

Installation

pip install "sagemaker>=3.19.0"

Requires Python 3.10 or later.

Concept Mapping

Forge SDK Concept	SageMaker SDK V3 Equivalent
ForgeTrainer (SFT)	sagemaker.train.sft_trainer.SFTTrainer
ForgeTrainer (CPT)	sagemaker.train.cpt_trainer.CPTTrainer
ForgeTrainer (DPO)	sagemaker.train.dpo_trainer.DPOTrainer
ForgeTrainer (RFT)	sagemaker.train.rlvr_trainer.RLVRTrainer
ForgeTrainer (MTRL)	sagemaker.train.multi_turn_rl_trainer.MultiTurnRLTrainer
ForgeEvaluator	BenchMarkEvaluator, LLMAsJudgeEvaluator, InspectAIEvaluator, CustomScorerEvaluator, MultiTurnRLEvaluator
SMHPRuntimeManager	sagemaker.core.training.configs.HyperPodCompute
SMTJRuntimeManager	TrainingJobCompute for serverful or omit for serverless
data_mixing_enabled	sagemaker.train.data_mixing_config.DataMixingConfig
NovaModelCustomizer	Individual trainer classes above
ForgeDeployer	BedrockModelBuilder or ModelBuilder
ForgeInference	SageMaker SDK Predictor / Bedrock InvokeModel

Full Quickstart Migration (Step-by-Step)

Step 1: Import Modules

Before (Forge SDK):

fromamzn_nova_forgeimport (
ForgeTrainer,
ForgeEvaluator,
ForgeDeployer,
ForgeInference,
ForgeConfig,
Model,
TrainingMethod,
DeployPlatform,
SMTJRuntimeManager,
SMHPRuntimeManager,
SMTJServerlessRuntimeManager,
BedrockRuntimeManager,
CloudWatchLogMonitor,
MLflowMonitor,
JSONLDatasetLoader,
TransformMethod,
ValidateMethod,
DataMixingConfig,
EvalTaskConfig,
EvaluationTask,
)

After (SageMaker SDK V3):

fromsagemaker.trainimportSFTTrainer, CPTTrainer, DPOTrainerfromsagemaker.train.evaluateimportBenchMarkEvaluator, get_benchmarksfromsagemaker.train.data_mixing_configimportDataMixingConfigfromsagemaker.core.training.configsimportHyperPodCompute, TrainingJobCompute

Step 2: Configure Compute

Before (Forge SDK) — SMTJ:

runtime=SMTJRuntimeManager(instance_type="ml.p5.48xlarge", instance_count=4)

After (SageMaker SDK V3) — SMTJ:

compute=TrainingJobCompute(instance_type="ml.p5.48xlarge", instance_count=4)

Before (Forge SDK) — SMHP:

runtime=SMHPRuntimeManager(
instance_type="ml.p5.48xlarge",
instance_count=4,
cluster_name="my-cluster",
namespace="default",
)

After (SageMaker SDK V3) — SMHP:

compute=HyperPodCompute(
cluster_name="my-cluster",
instance_type="ml.p5.48xlarge",
node_count=4,
)

Before (Forge SDK) — Serverless:

runtime=SMTJServerlessRuntimeManager(model_package_group_name="test-package")

After (SageMaker SDK V3) — Serverless:

Omit the compute parameter entirely. The trainer runs serverless by default.

Step 3: Training (SFT)

Before (Forge SDK):

trainer=ForgeTrainer(
model=Model.NOVA_LITE_2,
method=TrainingMethod.SFT_LORA,
infra=runtime,
training_data_s3_path="s3://bucket/train.jsonl",
config=ForgeConfig(output_s3_path="s3://bucket/output"),
)
result=trainer.train(job_name="my-sft-job", overrides={"lr": 5e-6, "warmup_steps": 100})

After (SageMaker SDK V3):

trainer=SFTTrainer(
model="nova-textgeneration-lite-v2",
compute=compute,
training_dataset="s3://bucket/train.jsonl",
s3_output_path="s3://bucket/output/",
overrides={
"recipes.training_config.trainer.lr": 5e-6,
"recipes.training_config.trainer.warmup_steps": 100,
},
)
job_name=trainer.train(wait=False)

Step 4: Data Mixing (Optional)

Before (Forge SDK):

trainer=ForgeTrainer(..., data_mixing_enabled=True)
trainer.data_mixing.set_config(
{
"customer_data_percent": 50,
"nova_code_percent": 30,
"nova_general_percent": 70,
}
)

After (SageMaker SDK V3):

fromsagemaker.train.data_mixing_configimportDataMixingConfigdata_mixing=DataMixingConfig(
customer_data_percent=50.0,
nova_data_percentages={"code": 30.0, "reasoning": 70.0},
)
trainer=SFTTrainer(..., data_mixing_config=data_mixing)

Step 5: Monitor, Notifications & Dry Run

Log Streaming

Before (Forge SDK):

trainer.get_logs(job_result=result, limit=50)
monitor=CloudWatchLogMonitor.from_job_id(job_id=result.job_id, platform=platform)
monitor.show_logs(limit=100)

After (SageMaker SDK V3):

# Stream logs (works on both trainer and evaluator)trainer.stream_logs()
trainer.stream_logs(tail_logs=50) # last 50 log entries

Metrics Visualization

Before (Forge SDK):

monitor.plot_metrics(training_method=TrainingMethod.SFT_LORA)

After (SageMaker SDK V3):

trainer.show_metrics()

Job Notifications (SMTJ only)

Before (Forge SDK):

result=trainer.train(job_name="my-job")
result.enable_job_notifications(emails=["user@example.com"])

After (SageMaker SDK V3):

trainer=SFTTrainer(
model="amazon.nova-lite-v2",
training_dataset="s3://bucket/train.jsonl",
notifications={
"sns_topic_arn": "arn:aws:sns:us-east-1:123456789012:my-topic",
"event_bus_arn": "arn:aws:events:us-east-1:123456789012:event-bus/my-bus",
"events": ["Completed", "Failed"],
"job_name_prefix": "my-team-",
},
)
trainer.train()

Requires a pre-created SNS topic. Notifications fire on job state changes (Completed, Failed, Stopped).

Dry Run Mode

Before (Forge SDK):

trainer.train(job_name="my-job", dry_run=True)

After (SageMaker SDK V3):

trainer.train(dry_run=True)

Runs all validations (IAM, compute, dataset) without submitting a job.

Step 6: Evaluate

Before (Forge SDK):

evaluator=ForgeEvaluator(
model=Model.NOVA_LITE_2,
infra=eval_infra,
data_s3_path="s3://bucket/eval-data.jsonl",
config=ForgeConfig(output_s3_path="s3://bucket/eval-output"),
)
mmlu_result=evaluator.evaluate(job_name="eval-mmlu", eval_task=EvaluationTask.MMLU)
byod_result=evaluator.evaluate(
job_name="eval-byod",
eval_task=EvaluationTask.GEN_QA,
task_config=EvalTaskConfig(override_data_s3_path="s3://bucket/custom-eval.jsonl"),
)

After (SageMaker SDK V3) — Benchmark (MMLU):

fromsagemaker.train.evaluateimportBenchMarkEvaluator, get_benchmarksBenchmark=get_benchmarks()
evaluator=BenchMarkEvaluator(
benchmark=Benchmark.MMLU,
model="nova-textgeneration-lite-v2",
s3_output_path="s3://bucket/eval-output/",
)
execution=evaluator.evaluate(checkpoint_path="s3://bucket/output/checkpoint/")

After (SageMaker SDK V3) — Custom Evaluator:

fromsagemaker.train.evaluateimportCustomScorerEvaluatorevaluator=CustomScorerEvaluator(
model="nova-textgeneration-lite-v2",
eval_dataset="s3://bucket/custom-eval.jsonl",
s3_output_path="s3://bucket/eval-output/",
)
execution=evaluator.evaluate(checkpoint_path="s3://bucket/output/checkpoint/")

After (SageMaker SDK V3) — InspectAI Evaluator:

fromsagemaker.train.evaluateimportInspectAIEvaluatorevaluator=InspectAIEvaluator(
model="nova-textgeneration-lite",
bedrock_model_id="us.amazon.nova-lite-v1:0",
benchmarks_path="s3://bucket/benchmarks/boolq/",
tasks=[{"name": "boolq_pt", "limit": 10}],
s3_output_path="s3://bucket/inspectai-eval-output/",
instance_type="ml.m5.large",
)
execution=evaluator.evaluate()
execution.wait(target_status="Succeeded")
execution.show_results()

Step 7: Deploy & Inference

Before (Forge SDK):

# Deploydeployer=ForgeDeployer(model=Model.NOVA_LITE_2)
result=deployer.deploy(
model_artifact_path=training_result.model_artifacts.checkpoint_s3_path,
deploy_platform=DeployPlatform.SAGEMAKER,
unit_count=1,
endpoint_name="my-endpoint",
)
# Inferenceinference=ForgeInference()
result=inference.invoke(
endpoint_arn=deployment_result.endpoint.endpoint_arn,
request_body={"messages": [{"role": "user", "content": "Hello!"}], "max_tokens": 100},
)
result.show()

After (SageMaker SDK V3) — SageMaker Endpoint:

importjsonfromsagemaker.serveimportModelBuilder# Deploybuilder=ModelBuilder(
model=trainer,
role_arn="arn:aws:iam::123456789012:role/SageMakerRole",
instance_type="ml.p4d.24xlarge",
)
builder.accept_eula=Truebuilder.build(region="us-east-1")
endpoint=builder.deploy(
endpoint_name="my-endpoint",
instance_type="ml.p4d.24xlarge",
)
# Inferenceresponse=endpoint.invoke(
body=json.dumps(
{"messages": [{"role": "user", "content": [{"type": "text", "text": "Hello!"}]}]}
),
content_type="application/json",
accept="application/json",
)
body=json.loads(response.body.read())

After (SageMaker SDK V3) — Bedrock:

importjsonimportboto3fromsagemaker.serve.bedrock_model_builderimportBedrockModelBuilder# Deploybuilder=BedrockModelBuilder(model="s3://bucket/output/checkpoint/")
result=builder.deploy(
custom_model_name="my-custom-model",
role_arn="arn:aws:iam::123456789012:role/SageMakerRole",
)
model_arn=result["modelArn"]
# Inferencebedrock_runtime=boto3.client("bedrock-runtime", region_name="us-east-1")
response=bedrock_runtime.invoke_model(
modelId=model_arn,
contentType="application/json",
accept="application/json",
body=json.dumps(
{"messages": [{"role": "user", "content": [{"type": "text", "text": "Hello!"}]}]}
),
)
body=json.loads(response["body"].read())

Additional Training Methods

CPT (Continued Pre-Training)

Before:

ForgeTrainer(model=Model.NOVA_LITE_2, method=TrainingMethod.CPT, infra=smhp_runtime, ...)

After:

CPTTrainer(model="nova-textgeneration-lite-v2", compute=HyperPodCompute(...), ...)

DPO (Direct Preference Optimization)

Before:

ForgeTrainer(model=Model.NOVA_MICRO, method=TrainingMethod.DPO_LORA, infra=runtime, ...)

After:

DPOTrainer(model="nova-textgeneration-micro", compute=compute, ...)

RLVR (Reinforcement Learning with Verifiable Rewards)

Before:

ForgeTrainer(model=Model.NOVA_LITE_2, method=TrainingMethod.RFT_LORA, infra=runtime, ...)

After:

fromsagemaker.trainimportRLVRTrainertrainer=RLVRTrainer(
model="nova-textgeneration-lite-v2",
compute=compute,
training_dataset="s3://bucket/rlvr-data.jsonl",
custom_reward_function="arn:aws:lambda:us-east-1:123456789012:function:my-reward",
s3_output_path="s3://bucket/output/",
)
trainer.train()

Iterative Training (Resume from Checkpoint)

Before:

trainer=ForgeTrainer(
model=Model.NOVA_LITE_2,
method=TrainingMethod.SFT_LORA,
infra=runtime,
training_data_s3_path="s3://bucket/stage2-data.jsonl",
model_s3_path="s3://bucket/stage1-output/checkpoint/",
config=ForgeConfig(output_s3_path="s3://bucket/stage2-output"),
)

After:

trainer=SFTTrainer(
model="s3://bucket/stage1-output/checkpoint/",
compute=compute,
training_dataset="s3://bucket/stage2-data.jsonl",
s3_output_path="s3://bucket/stage2-output/",
)
trainer.train()

Support

• SageMaker Python SDK docs: https://sagemaker.readthedocs.io/en/stable/
• SageMaker Python SDK GitHub: https://github.com/aws/sagemaker-python-sdk