翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

RFT 評価

RFT 評価とは

RFT 評価では、強化学習トレーニングの前、最中、または後にカスタム報酬関数を使用してモデルのパフォーマンスを評価できます。事前定義されたメトリクスを使用する標準評価とは異なり、RFT 評価では、特定の要件に基づいてモデル出力をスコアリングする Lambda 関数を通じて独自の成功基準を定義できます。

RFT で評価する理由

RL ファインチューニングプロセスに以下があるかどうかを判断するには、評価が不可欠です。

RFT 評価を使用するタイミング

RFT 評価は、以下のシナリオで使用します。

データ形式の要件

入力データ構造

RFT 評価入力データは、OpenAI 強化ファインチューニング形式に従う必要があります。各例は、以下を含む JSON オブジェクトです。

データ形式の例

次の例は、必要な形式を示しています。

{
  "messages": [
    {
      "role": "user",
      "content": [
        {
          "type": "text",
          "text": "Solve for x. Return only JSON like {\"x\": <number>}. Equation: 2x + 5 = 13"
        }
      ]
    }
  ],
  "reference_answer": {
    "x": 4
  }
}

現在の制限事項

RFT 評価には、次の制限が適用されます。

評価レシピの準備

サンプルレシピ設定

次の例は、完全な 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 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:<region>:<account_id>:function:<reward-function-name>

プリセット報酬関数

オープンソースの verl ライブラリから 2 つのプリセット報酬関数 (prime_code、Prime_math) を Lambda レイヤーに用意しました。ここでは、使用する RFT Lambda に簡単にバンドルできます。

概要:

これらのプリセット関数は、以下に対してout-of-the-box評価機能を提供します。

Quick Setup

プリセット報酬関数を使用するには:

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"}

主な機能:

ベストプラクティス

プリセット報酬関数を使用する場合は、次のベストプラクティスに従ってください。

報酬関数の作成

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."
    }
  }
]

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"
      }
    ]
  }
]

レスポンスフィールド:

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 実行ロールに追加します。

評価ジョブの実行

評価結果について

出力形式

評価ジョブは、指定した 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"
    }
  ]
}

結果の解釈

報酬スコアの集計:

個々のメトリクス:

パフォーマンスベンチマーク

「良い」パフォーマンスを構成するものは、ユースケースによって異なります。