# RFT 評価 **注記** 独自の AWS 環境でのリモート報酬関数による評価は、Amazon Nova Forge のお客様のみが使用できます。 **重要** `rl_env` 設定フィールドは、トレーニングではなく評価にのみ使用されます。トレーニング中に、`reward_lambda_arn` (シングルターン) または BYOO インフラストラクチャと `rollout.delegate: true` (マルチターン) を使用して報酬関数を設定します。 **RFT 評価とは?** RFT 評価では、強化学習トレーニングの前、最中、または後にカスタム報酬関数を使用してモデルのパフォーマンスを評価できます。事前定義されたメトリクスを使用する標準評価とは異なり、RFT 評価では、特定の要件に基づいてモデル出力をスコアリングする Lambda 関数を通じて独自の成功基準を定義できます。 **RFT で評価する理由** RL ファインチューニングプロセスで以下を実現できたかどうかを判断するには、評価が不可欠です。 + 特定のユースケースと人間の価値観とのモデルアライメントを改善 + 主要なタスクでモデル機能を維持または改善 + 事実性の低下、冗長性の向上、他のタスクのパフォーマンスの低下などの意図しない副作用を回避 + 報酬関数で定義されたカスタム成功基準を充足 **RFT 評価を使用するタイミング** RFT 評価は、以下のシナリオで使用します。 + RFT トレーニング前: 評価データセットにベースラインメトリクスを確立する + RFT トレーニング中: 中間チェックポイントでトレーニングの進行状況をモニタリングする + RFT トレーニング後: 最終モデルが要件を満たしていることを確認する + モデルの比較: 一貫した報酬基準を使用して複数のモデルバージョンを評価する **注記** ドメイン固有のカスタムメトリクスが必要な場合は、RFT 評価を使用します。汎用評価 (精度、Perplexity、BLEU) には、標準的な評価方法を使用します。 **Topics** + [データ形式の要件](#nova-hp-evaluate-rft-data-format) + [評価レシピの準備](#nova-hp-evaluate-rft-recipe) + [プリセット報酬関数](#nova-hp-evaluate-rft-preset) + [報酬関数を作成する](#nova-hp-evaluate-rft-create-function) + [IAM アクセス許可](#nova-hp-evaluate-rft-iam) + [評価ジョブを実行する](#nova-hp-evaluate-rft-execution) + [評価結果を理解する](#nova-hp-evaluate-rft-results) ## データ形式の要件 **入力データ構造** RFT 評価入力データは、OpenAI 強化ファインチューニング形式に従う必要があります。各例は、以下を含む JSON オブジェクトです。 + `messages`: `system` および `user` ロールとの会話ターンの配列 + reference\_answer などのオプションの他のメタデータ **データ形式の例** 次の例は、必要な形式を示しています。 ``` { "messages": [ { "role": "user", "content": [ { "type": "text", "text": "Solve for x. Return only JSON like {\"x\": }. Equation: 2x + 5 = 13" } ] } ], "reference_answer": { "x": 4 } } ``` **現在の制限事項** RFT 評価には以下の制限が適用されます。 + テキストのみ: マルチモーダル入力 (画像、音声、動画) はサポートされていません + シングルターン会話: シングルユーザーメッセージのみをサポート (マルチターンダイアログなし) + JSON 形式: 入力データは JSONL 形式である必要があります (1 行あたり 1 つの JSON オブジェクト) + モデル出力: 指定されたモデルによって生成された補完結果に対して評価が実行されます ## 評価レシピの準備 **サンプルレシピ設定** 次の例は、完全な RFT 評価レシピを示しています。 ``` run: name: nova-lite-rft-eval-job model_type: amazon.nova-lite-v1:0:300k model_name_or_path: s3://escrow_bucket/model_location # [MODIFIABLE] S3 path to your model or model identifier replicas: 1 # [MODIFIABLE] For SageMaker Training jobs only; fixed for SageMaker HyperPod jobs data_s3_path: "" # [REQUIRED FOR HYPERPOD] Leave empty for SageMaker Training jobs output_s3_path: "" # [REQUIRED] Output artifact S3 path for evaluation results evaluation: task: rft_eval # [FIXED] Do not modify strategy: rft_eval # [FIXED] Do not modify metric: all # [FIXED] Do not modify # Inference Configuration inference: max_new_tokens: 8196 # [MODIFIABLE] Maximum tokens to generate top_k: -1 # [MODIFIABLE] Top-k sampling parameter top_p: 1.0 # [MODIFIABLE] Nucleus sampling parameter temperature: 0 # [MODIFIABLE] Sampling temperature (0 = deterministic) top_logprobs: 0 # Evaluation Environment Configuration (NOT used in training) rl_env: reward_lambda_arn: arn:aws:lambda:::function: ``` ## プリセット報酬関数 RFT Lambda 関数と簡単に統合できるように、2 つのプリセット報酬関数 (`prime_code` および `prime_math`) を Lambda レイヤーとして使用できます。 **概要** これらのプリセット関数は、複雑な設定なしで以下を評価できます。 + **prime\_code**: コードの生成と正確性の評価 + **prime\_math**: 数学的な推論と問題解決の評価 **Quick Setup** プリセット報酬関数を使用するには: 1. [nova-custom-eval-sdk リリース](https://github.com/aws/nova-custom-eval-sdk/releases)から Lambda レイヤーをダウンロードする 1. AWS CLI を使用して Lambda レイヤーを公開する: ``` aws lambda publish-layer-version \ --layer-name preset-function-layer \ --description "Preset reward function layer with dependencies" \ --zip-file fileb://universal_reward_layer.zip \ --compatible-runtimes python3.9 python3.10 python3.11 python3.12 \ --compatible-architectures x86_64 arm64 ``` 1. AWS コンソールで Lambda 関数にレイヤーを追加する (カスタムレイヤーから preset-function-layer を選択し、numpy の依存関係のために AWSSDKPandas-Python312 を追加する) 1. Lambda コードで次をインポートして使用します: ``` from prime_code import compute_score # For code evaluation from prime_math import compute_score # For math evaluation ``` **prime\_code 関数** **目的**: テストケースに対してコードを実行し、正確性を測定することで、Python コード生成タスクを評価します。 **評価からの入力データセット形式の例**: ``` {"messages":[{"role":"user","content":"Write a function that returns the sum of two numbers."}],"reference_answer":{"inputs":["3\n5","10\n-2","0\n0"],"outputs":["8","8","0"]}} {"messages":[{"role":"user","content":"Write a function to check if a number is even."}],"reference_answer":{"inputs":["4","7","0","-2"],"outputs":["True","False","True","True"]}} ``` **主な特徴**: + マークダウンコードブロックからの自動コード抽出 + 関数検出と呼び出しベースのテスト + タイムアウト保護を受けたテストケースの実行 + 構文の検証とコンパイルのチェック + トレースバックを使用した詳細なエラーレポート **prime\_math 関数** **目的**: シンボリック数学のサポートにより、数学的な推論と問題解決能力を評価します。 **入力形式**: ``` {"messages":[{"role":"user","content":"What is the derivative of x^2 + 3x?."}],"reference_answer":"2*x + 3"} ``` **主な特徴**: + SymPy を使用したシンボリック数学評価 + 複数の回答形式 (LaTeX、プレーンテキスト、シンボリック) + 数学的同等性チェック + 式の正規化と簡素化 **ベストプラクティス** プリセット報酬関数を使用する場合は、次のベストプラクティスに従ってください。 + テストケースで適切なデータ型を使用する (整数と文字列、ブール値と「True」) + コードの問題で明確な関数署名を提供する + テスト入力にエッジケースを含める (ゼロ、負の数、空の入力) + 参照回答で数式を一貫してフォーマットする + デプロイ前にサンプルデータを使用して報酬関数をテストする ## 報酬関数を作成する **Lambda ARN** Lambda ARN では、次の形式を参照する必要があります。 ``` "arn:aws:lambda:*:*:function:*SageMaker*" ``` Lambda にこの命名スキームがない場合、ジョブは次のエラーで失敗します: ``` [ERROR] Unexpected error: lambda_arn must contain one of: ['SageMaker', 'sagemaker', 'Sagemaker'] when running on SMHP platform (Key: lambda_arn) ``` **Lambda 関数構造** Lambda 関数はモデル出力のバッチを受け取り、報酬スコアを返します。次に示すのは実装の例です。 ``` from typing import List, Any import json import re from dataclasses import asdict, dataclass @dataclass class MetricResult: """Individual metric result.""" name: str value: float type: str @dataclass class RewardOutput: """Reward service output.""" id: str aggregate_reward_score: float metrics_list: List[MetricResult] def lambda_handler(event, context): """ Main lambda handler """ return lambda_grader(event) def lambda_grader(samples: list[dict]) -> list[dict]: """ Core grader function """ scores: List[RewardOutput] = [] for sample in samples: print("Sample: ", json.dumps(sample, indent=2)) # Extract components idx = sample.get("id", "no id") if not idx or idx == "no id": print(f"ID is None/empty for sample: {sample}") ground_truth = sample.get("reference_answer") if "messages" not in sample: print(f"Messages is None/empty for id: {idx}") continue if ground_truth is None: print(f"No answer found in ground truth for id: {idx}") continue # Get model's response (last turn is assistant turn) last_message = sample["messages"][-1] if last_message["role"] != "nova_assistant": print(f"Last message is not from assistant for id: {idx}") continue if "content" not in last_message: print(f"Completion text is empty for id: {idx}") continue model_text = last_message["content"] # --- Actual scoring logic (lexical overlap) --- ground_truth_text = _extract_ground_truth_text(ground_truth) # Calculate main score and individual metrics overlap_score = _lexical_overlap_score(model_text, ground_truth_text) # Create two separate metrics as in the first implementation accuracy_score = overlap_score # Use overlap as accuracy fluency_score = _calculate_fluency(model_text) # New function for fluency # Create individual metrics metrics_list = [ MetricResult(name="accuracy", value=accuracy_score, type="Metric"), MetricResult(name="fluency", value=fluency_score, type="Reward") ] ro = RewardOutput( id=idx, aggregate_reward_score=overlap_score, metrics_list=metrics_list ) print(f"Response for id: {idx} is {ro}") scores.append(ro) # Convert to dict format result = [] for score in scores: result.append({ "id": score.id, "aggregate_reward_score": score.aggregate_reward_score, "metrics_list": [asdict(metric) for metric in score.metrics_list] }) return result def _extract_ground_truth_text(ground_truth: Any) -> str: """ Turn the `ground_truth` field into a plain string. """ if isinstance(ground_truth, str): return ground_truth if isinstance(ground_truth, dict): # Common patterns: { "explanation": "...", "answer": "..." } if "explanation" in ground_truth and isinstance(ground_truth["explanation"], str): return ground_truth["explanation"] if "answer" in ground_truth and isinstance(ground_truth["answer"], str): return ground_truth["answer"] # Fallback: stringify the whole dict return json.dumps(ground_truth, ensure_ascii=False) # Fallback: stringify anything else return str(ground_truth) def _tokenize(text: str) -> List[str]: # Very simple tokenizer: lowercase + alphanumeric word chunks return re.findall(r"\w+", text.lower()) def _lexical_overlap_score(model_text: str, ground_truth_text: str) -> float: """ Simple lexical overlap score in [0, 1]: score = |tokens(model) ∩ tokens(gt)| / |tokens(gt)| """ gt_tokens = _tokenize(ground_truth_text) model_tokens = _tokenize(model_text) if not gt_tokens: return 0.0 gt_set = set(gt_tokens) model_set = set(model_tokens) common = gt_set & model_set return len(common) / len(gt_set) def _calculate_fluency(text: str) -> float: """ Calculate a simple fluency score based on: - Average word length - Text length - Sentence structure Returns a score between 0 and 1. """ # Simple implementation - could be enhanced with more sophisticated NLP words = _tokenize(text) if not words: return 0.0 # Average word length normalized to [0,1] range # Assumption: average English word is ~5 chars, so normalize around that avg_word_len = sum(len(word) for word in words) / len(words) word_len_score = min(avg_word_len / 10, 1.0) # Text length score - favor reasonable length responses ideal_length = 100 # words length_score = min(len(words) / ideal_length, 1.0) # Simple sentence structure check (periods, question marks, etc.) sentence_count = len(re.findall(r'[.!?]+', text)) + 1 sentence_ratio = min(sentence_count / (len(words) / 15), 1.0) # Combine scores fluency_score = (word_len_score + length_score + sentence_ratio) / 3 return fluency_score ``` **Lambda リクエスト形式** Lambda 関数は次の形式でデータを受け取ります。 ``` [ { "id": "sample-001", "messages": [ { "role": "user", "content": [ { "type": "text", "text": "Do you have a dedicated security team?" } ] }, { "role": "nova_assistant", "content": [ { "type": "text", "text": "As an AI developed by Company, I don't have a dedicated security team in the traditional sense. However, the development and deployment of AI systems like me involve extensive security measures, including data encryption, user privacy protection, and other safeguards to ensure safe and responsible use." } ] } ], "reference_answer": { "compliant": "No", "explanation": "As an AI developed by Company, I do not have a traditional security team. However, the deployment involves stringent safety measures, such as encryption and privacy safeguards." } } ] ``` **注記** メッセージ構造には、入力データ形式に一致するネストされた `content` 配列が含まれます。ロール `nova_assistant` を含む最後のメッセージには、モデルが生成したレスポンスが含まれます。 **Lambda レスポンス形式** Lambda 関数は次の形式でデータを返す必要があります。 ``` [ { "id": "sample-001", "aggregate_reward_score": 0.75, "metrics_list": [ { "name": "accuracy", "value": 0.85, "type": "Metric" }, { "name": "fluency", "value": 0.90, "type": "Reward" } ] } ] ``` **レスポンスフィールド**: + `id`: 入力サンプル ID と一致する必要があります + `aggregate_reward_score`: 総合スコア (通常は 0.0～1.0) + `metrics_list`: 以下を含む個々のメトリクスの配列: + `name`: メトリクス識別子 (「精度」、「流暢さ」など) + `value`: メトリクススコア (通常は 0.0～1.0) + `type`: 「メトリクス」 (レポート用) または「報酬」 (トレーニングに使用) ## IAM アクセス許可 **必要なアクセス許可** SageMaker AI 実行ロールには、Lambda 関数を呼び出すためのアクセス許可が必要です。この SageMaker AI 実行ロールにこのポリシーを追加します。 ``` { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "lambda:InvokeFunction" ], "Resource": "arn:aws:lambda:region:account-id:function:function-name" } ] } ``` **Lambda 実行ロール** Lambda 関数の実行ロールには、基本的な Lambda 実行アクセス許可が必要です。 ``` { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents" ], "Resource": "arn:aws:logs:*:*:*" } ] } ``` **追加のアクセス許可**: Lambda 関数が他の AWS サービス (参照データ用の Amazon S3、ログ記録用の DynamoDB など) にアクセスする場合は、このようなアクセス許可を Lambda 実行ロールに追加します。 ## 評価ジョブを実行する 1. **データを準備する** + データ形式の要件に従って評価データをフォーマットする + JSONL ファイルを Amazon S3 にアップロードします: `s3://your-bucket/eval-data/eval_data.jsonl` 1. **レシピを設定する** 次のように、サンプルレシピを設定で更新します。 + `model_name_or_path` をモデルの場所に設定する + `lambda_arn` を報酬関数 ARN に設定する + `output_s3_path` を必要な出力場所に設定する + 必要に応じて `inference` パラメータを調整する レシピを `rft_eval_recipe.yaml` として保存する 1. **評価を実行する** 提供されたノートブックである [Nova モデル評価ノートブック](https://docs.aws.amazon.com/sagemaker/latest/dg/nova-model-evaluation.html#nova-model-evaluation-notebook)を使用して評価ジョブを実行する 1. **進行状況をモニタリングする** 以下を通じて評価ジョブをモニタリングします。 + SageMaker AI コンソール: ジョブのステータスとログを確認する + CloudWatch Logs: 詳細な実行ログを表示する + Lambda ログ: 報酬関数の問題をデバッグする ## 評価結果を理解する **出力形式** 評価ジョブは、指定された Amazon S3 の場所に JSONL 形式で結果を出力します。各行には、1 つのサンプルの評価結果が含まれます。 ``` { "id": "sample-001", "aggregate_reward_score": 0.75, "metrics_list": [ { "name": "accuracy", "value": 0.85, "type": "Metric" }, { "name": "fluency", "value": 0.90, "type": "Reward" } ] } ``` **注記** RFT 評価ジョブの出力は、Lambda レスポンス形式と同じです。評価サービスは、Lambda 関数の応答を変更せずにパススルーし、報酬の計算と最終結果の一貫性を確保します。 **結果の解釈** **報酬スコアの集計**: + 範囲: 通常 0.0 (最悪)～1.0 (最良) ですが、実装によって異なります + 目的: 全体的なパフォーマンスをまとめた 1 つの数値 + 使用法: モデルを比較し、トレーニングに対する改善を追跡する **個々のメトリクス**: + メトリクスタイプ: 分析のための情報メトリクス + 報酬タイプ: RFT トレーニング中に使用されるメトリクス + 解釈: 値が大きいほど一般的にパフォーマンスが向上する (逆メトリクスを設計しない限り) **パフォーマンスベンチマーク** 何が「良好」なパフォーマンスとなるかは、ユースケースによって異なります。 | スコア範囲 | 解釈 | アクション | | --- |--- |--- | | 0.8～1.0 | 優秀 | デプロイの準備ができたモデル | | 0.6～0.8 | 良好 | 軽微な改善が有益である可能性があります | | 0.4～0.6 | 可 | 大幅な改善が必要 | | 0.0～0.4 | 不良 | トレーニングデータと報酬関数を確認する | **重要** これらは一般的なガイドラインです。ビジネス要件、ベースラインモデルのパフォーマンス、ドメイン固有の制約、さらなるトレーニングの費用対効果分析に基づいて、独自のしきい値を定義します。