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

Full Quickstart Migration (Step-by-Step)

Step 1: Import Modules

Before (Forge SDK):

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

After (SageMaker SDK V3):

from sagemaker.train import SFTTrainer, CPTTrainer, DPOTrainer
from sagemaker.train.evaluate import BenchMarkEvaluator, get_benchmarks
from sagemaker.train.data_mixing_config import DataMixingConfig
from sagemaker.core.training.configs import HyperPodCompute, 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):

from sagemaker.train.data_mixing_config import DataMixingConfig

data_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):

from sagemaker.train.evaluate import BenchMarkEvaluator, get_benchmarks

Benchmark = 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:

from sagemaker.train.evaluate import CustomScorerEvaluator

evaluator = 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:

from sagemaker.train.evaluate import InspectAIEvaluator

evaluator = 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):

# Deploy
deployer = 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",
)

# Inference
inference = 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:

import json
from sagemaker.serve import ModelBuilder

# Deploy
builder = ModelBuilder(
    model=trainer,
    role_arn="arn:aws:iam::123456789012:role/SageMakerRole",
    instance_type="ml.p4d.24xlarge",
)
builder.accept_eula = True
builder.build(region="us-east-1")
endpoint = builder.deploy(
    endpoint_name="my-endpoint",
    instance_type="ml.p4d.24xlarge",
)

# Inference
response = 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:

import json
import boto3
from sagemaker.serve.bedrock_model_builder import BedrockModelBuilder

# Deploy
builder = 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"]

# Inference
bedrock_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:

from sagemaker.train import RLVRTrainer

trainer = 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