本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

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>

預設獎勵函數

我們已將 2 個預設獎勵函數 (prime_code， prime_math) 從開放原始碼 verl 程式庫提供到 lambda 層，您可以輕鬆地與 RFT lambda 綁定使用。

概觀

這些預設函數提供out-of-the-box評估功能：

快速設定

若要使用預設獎勵函數：

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 執行角色。

執行評估任務

了解評估結果

輸出格式

評估任務會以 JSONL 格式將結果輸出至您指定的 Amazon S3 位置。每行包含一個範例的評估結果：

{
  "id": "sample-001",
  "aggregate_reward_score": 0.75,
  "metrics_list": [
    {
      "name": "accuracy",
      "value": 0.85,
      "type": "Metric"
    },
    {
      "name": "fluency",
      "value": 0.90,
      "type": "Reward"
    }
  ]
}

解譯結果

彙總獎勵分數：

個別指標：

效能基準

什麼構成「良好」效能取決於您的使用案例：