Nova Forge SDKを用いたNovaカスタマイズ実験の開始

# Validate data format
loader.validate(method=TrainingMethod.EVALUATION, model=MODEL)

# Save to S3
evap_s3_uri = loader.save_data(
    f"s3://{S3_BUCKET}/{S3_PREFIX}/data/eval.jsonl"
)

# Configure runtime infrastructure
runtime_manager = SMTJRuntimeManager(
    instance_type=INSTANCE_TYPE, 
    instance_count=EVAL_INSTANCE_COUNT, 
    execution_role=EXECUTION_ROLE
)
# Create baseline evaluator
baseline_customizer = NovaModelCustomizer(
    model=MODEL, 
    method=TrainingMethod.EVALUATION, 
    infra=runtime_manager, 
    data_s3_path=eval_s3_uri, 
    output_s3_path=f"s3://{S3_BUCKET}/{S3_PREFIX}/baseline-eval"
)

# Run evaluation
# GEN_QA task provides metrics like ROUGE, BLEU, F1, and Exact Match
baseline_result = baseline_customizer.evaluate( 
    job_name="blogpost-baseline",
    eval_task=EvaluationTask.GEN_QA # Use GEN_QA for classification
)

def classification_accuracy(samples): 
    """Extract predicted class via substring match and compute accuracy."""
    correct, total, no_pred = 0, 0, 0 
    for s in samples: 
        gold = s["gold"].strip().upper() 
        pred_raw = s["inference"][0] if isinstance(s["inference"], list) else s["inference"] 
        pred_cat = extract_category(pred_raw) 
        if pred_cat is None: 
            no_pred += 1 
            continue 
        total += 1 
        if pred_cat == gold: 
            correct += 1 
    acc = correct / total if total else 0 
    print(f"Classification Accuracy: {correct}/{total} ({acc*100:.1f}%)") 
    print(f" No valid prediction: {no_pred}/{total + no_pred}") 
    return acc

print("???? Baseline Classification Accuracy (extracted class labels):")
baseline_accuracy = classification_accuracy(baseline_samples)

loader = CSVDatasetLoader(
    question='Body', # Stack Overflow question text
    answer='Y', # Classification label (HQ, LQ_EDIT, LQ_CLOSE)
    system='system' # System prompt column
)

loader.load('sft.csv')
loader.transform(method=TrainingMethod.SFT_LORA, model=Model.NOVA_LITE_2)
loader.show(n=3)

{ 
    "system": [ 
        {"text": "<system prompt>"} 
    ], 
    "messages": [ 
        { 
            "role": "user", 
            "content": [ 
                {"text": "<input data>"} 
            ] 
        }, 
        { 
            "role": "assistant", 
            "content": [ 
                {"text": "<output data>"} 
            ] 
        } 
    ]
}

loader.validate(method=TrainingMethod.SFT_LORA, model=Model.NOVA_LITE_2)

# S3 へ保存 train_path = loader.save_data(f\"s3://{S3_BUCKET}/{S3_PREFIX}/data/train.jsonl\")

runtime = SMTJRuntimeManager( 
    instance_type=INSTANCE_TYPE, 
    instance_count=TRAIN_INSTANCE_COUNT, 
    execution_role=EXECUTION_ROLE
)

customizer = NovaModelCustomizer( 
    model=MODEL, 
    method=TrainingMethod.SFT_LORA, 
    infra=runtime, 
    data_s3_path=train_path, 
    output_s3_path=f\"s3://{S3_BUCKET}/{S3_PREFIX}/sft-output\"
)

training_config = {
    \"lr\": 5e-6, # 学習率
    \"warmup_steps\": 17, # 段階的な学習率の増加
    \"max_steps\": 100, # 総トレーニングステップ
    \"global_batch_size\": 64, # 勾配更新ごとのサンプル数
    \"max_length\": 8192, # トークンの最大シーケンス長
}

result = customizer.train( 
    job_name=\"blogpost-sft\", 
    overrides=training_config
)

result = customizer.train( 
    job_name=\"blogpost-sft\", 
    overrides=training_config, 
    dry_run=True
)

result = customizer.train( 
    job_name=\"blogpost-sft\", 
    overrides=training_config
)

# ファイルへ保存
result.dump(file_path=\".\", file_name=\"training_result.json\")

# ファイルから読み込み
result = TrainingResult.load(\"training_result.json\")

monitor = CloudWatchLogMonitor.from_job_result(result)
monitor.show_logs(limit=50)

customizer.get_logs(limit=20)

result.get_job_status() # 返されます (JobStatus.IN_PROGRESS, ...) または (JobStatus.COMPLETED, ...)

# ランタイムインフラの設定
runtime_manager = SMTJRuntimeManager( 
    instance_type=INSTANCE_TYPE, 
    instance_count=EVAL_INSTANCE_COUNT, 
    execution_role=EXECUTION_ROLE
)

# ベースライン評価者の作成
baseline_customizer = NovaModelCustomizer( 
    model=MODEL, 
    method=TrainingMethod.EVALUATION, 
    infra=runtime_manager, 
    data_s3_path=eval_s3_uri, 
    output_s3_path=f\"s3://{S3_BUCKET}/{S3_PREFIX}/sft-eval\"
)

# 評価の実行
baseline_result = baseline_customizer.evaluate( 
    job_name=\"blogpost-eval\", 
    eval_task=EvaluationTask.GEN_QA
    job_result=result, # トレーニング結果から自動的にチェックポイントパスを導出
)

print(\"Post-SFT Classification Accuracy (extracted class labels):\")
sft_accuracy = classification_accuracy(sft_samples)

"""Binary reward function for classification: +1 correct, -1 wrong.
Simple and clear signal:
- Correct prediction: +1.0
- Wrong prediction: -1.0
"""

def calculate_reward(prediction: str, ground_truth: str) -> float: 
    """ Calculates binary reward """ 
    extracted = extract_category(prediction) # Extracts category from prediction and normalize it 
    truth_norm = normalize_text(ground_truth) # Normalize the groundtruth

 # Correct prediction
    if extracted and extracted == truth_norm: return 1.0

 # Wrong prediction
    return -1.0

def lambda_handler(event, context): 
    """ Lambda handler with binary rewards. """ 
    scores: List[RewardOutput] = [] 
    
    for sample in event: 
        idx = sample.get("id", "no_id") 
        ground_truth = sample.get("reference_answer", "") 
        prediction = last_message.get("content", "")

 # Calculate binary reward
        reward = calculate_reward(prediction, ground_truth) 
 scores.append(RewardOutput(id=idx, aggregate_reward_score=reward)) 
 return [asdict(score) for score in scores]

RFT_DATA = './rft.csv' 

rft_loader = CSVDatasetLoader( 
    query='Body', 
    response='Y', 
    system='system'
) 

rft_loader.load(RFT_DATA)

# Transform for RFT
rft_loader.transform(method=TrainingMethod.RFT_LORA, model=MODEL)
rft_loader.validate(method=TrainingMethod.RFT_LORA, model=MODEL)

# Save to S3
rft_s3_uri = rft_loader.save_data( 
    f"s3://{S3_BUCKET}/{S3_PREFIX}/data/rft.jsonl"
)

{ 
    "system": [ 
        { 
            "text": "<system prompt>"
        } 
    ], 
    "messages": [ 
        { 
            "role": "user", 
            "content": [ 
                { 
                    "text": "<input data>" 
                } 
            ] 
        } 
    ], 
    "reference_answer": "<reference answer>"
    #any other metadata field you use in data loader mapping
}

REWARD_LAMBDA_ARN = "arn:aws:lambda:us-east-1:ACCOUNT:function:classification-reward"

# Configure RFT infrastructure
RFT_INSTANCE_COUNT = 2 

rft_runtime = SMTJRuntimeManager( 
    instance_type=INSTANCE_TYPE, 
    instance_count=RFT_INSTANCE_COUNT, 
    execution_role=EXECUTION_ROLE
)

# Create RFT customizer
rft_customizer = NovaModelCustomizer( 
    model=MODEL, 
    method=TrainingMethod.RFT_LORA, 
    infra=rft_runtime, 
    data_s3_path=rft_s3_uri, 
    output_s3_path=f"s3://{S3_BUCKET}/{S3_PREFIX}/rft-output",
    model_path=sft_checkpoint # Start from SFT checkpoint
)

rft_overrides = {
    "lr": 0.00001, # Learning rate
    "number_generation": 4, # N samples per prompt to estimate advantages (variance vs cost).
    "reasoning_effort": "null", # Enables reasoning mode High / Low / or null for  non-reasoning
    "max_new_tokens": 50, # This cuts off verbose outputs
    "kl_loss_coef": 0.02, # Weight on the KL penalty between the actor  (trainable policy) and a frozen reference model
    "temperature": 1, # Softmax temperature
    "ent_coeff": 0.01, # A bonus added to the policy loss that rewards  higher-output entropy
    "max_steps": 40, # Steps to train for. One Step = global_batch_size
    "save_steps": 30, # Steps after which a checkpoint will be saved
	"top_k": 5, # Sample only from top-K logits
    "global_batch_size": 64, # Total samples per optimizer step across all  replicas (16/32/64/128/256)
}

# Start RFT training
rft_result = rft_customizer.train( 
    job_name="stack-overflow-rft", 
    rft_lambda_arn=REWARD_LAMBDA_ARN, 
    overrides = rft_overrides
)

rft_result = CloudWatchLogMonitor.from_job_result(rft_result)
rft_result.show_logs()

ENDPOINT_NAME = "blogpost-sdkg6"
deployment_result = rft_customizer.deploy( 
    job_result = rft_result, 
    deploy_platform=DeployPlatform.SAGEMAKER, 
    unit_count=1, 
    endpoint_name= ENDPOINT_NAME, 
    execution_role_name="blogpost-sagemaker", 
    sagemaker_instance_type="ml.p5.48xlarge", 
    sagemaker_environment_variables={ 
        "CONTEXT_LENGTH": "12000", 
        "MAX_CONCURRENCY": "16" 
    }
)

streaming_chat_request = { 
    "messages": [{"role": "user", "content": "Tell me a short story"}], 
    "max_tokens": 200, 
    "stream": True,}
ENDPOINT_NAME = f"arn:aws:sagemaker:REGION:ACCOUNT_ID:endpoint/{ENDPOINT_NAME}"

inference_result = rft_customizer.invoke_inference( 
    request_body=streaming_chat_request, 
    endpoint_arn=ENDPOINT_NAME
)
inference_result.show()

import boto3

sagemaker_client = boto3.client('sagemaker')

# Delete endpoint
sagemaker_client.delete_endpoint(EndpointName='your-endpoint-name')

import boto3

iam_client = boto3.client('iam')
role_name = 'your-role-name'

# Detach managed policies
attached_policies = iam_client.list_attached_role_policies(RoleName=role_name)
for policy in attached_policies['AttachedPolicies']: 
    iam_client.detach_role_policy( 
        RoleName=role_name, 
        PolicyArn=policy['PolicyArn'] 
    )

# Delete inline policies
inline_policies = iam_client.list_role_policies(RoleName=role_name)
for policy_name in inline_policies['PolicyNames']: 
    iam_client.delete_role_policy( 
        RoleName=role_name, 
        PolicyName=policy_name 
)

# Remove from instance profiles
instance_profiles = iam_client.list_instance_profiles_for_role(RoleName=role_name)
for profile in instance_profiles['InstanceProfiles']: 
    iam_client.remove_role_from_instance_profile( 
        InstanceProfileName=profile['InstanceProfileName'], 
        RoleName=role_name 
    )

# Delete the role
iam_client.delete_role(RoleName=role_name)