探索用于强化微调的模型评分器
本文以医学问答任务为例，系统讲解如何为 OpenAI o4-mini 的强化微调（RFT）设计评分器。内容涵盖数据集准备、基础模型基准测试、字符串评分器与模型评分器的实现、奖励黑客的识别与修复、RFT 训练启动及微调模型评估。重点展示评分器如何决定 RFT 的学习信号，并通过对比实验说明模型评分器在语义保真度上的优势与风险。
技术背景
强化微调（RFT）通过在模型输出上运行强化学习，强化能获得高奖励的推理策略，从而提升推理模型在专家级任务上的表现。RFT 特别适合需要精确、可靠、可重复推理的场景，如 ICD-10 编码、账本规范化、欺诈风险分级等。与监督微调不同，RFT 不依赖大量标注样本，而是通过评分器提供奖励信号，让模型在探索中学习正确行为并避免错误模式。评分器设计是 RFT 成功的关键，它直接决定模型学到的行为方向。
核心原理
RFT 的核心是奖励函数，即评分器。评分器接收模型输出和参考答案，返回 0 到 1 的分数。训练时模型采样多条轨迹，评分器对每条轨迹打分，策略梯度更新使高奖励轨迹概率上升。评分器可分为基于字符串的评分器（精确匹配、token 相似度）和基于模型的评分器（LLM 按评分标准打分）。字符串评分器信号稀疏，容易导致奖励停滞；模型评分器能捕捉语义同义、疾病分类等细微差别，但若评分标准不严，模型会通过填充同义词、剂量、完整方案等方式进行奖励黑客，抬高词汇相似度却不提升临床准确性。因此评分器需要明确输出约束、降低词汇权重、提高临床相似度权重，并用对比示例校准。
架构分析
整体流程分为六步：1）环境设置与 API 密钥配置；2）从 Hugging Face 加载 medical-o1-verifiable-problem 数据集，过滤年龄相关问题，随机抽取 100 训练样本和 100 测试样本，归一化答案并转为 jsonl；3）用 o4-mini 和 o3 在简单系统提示下生成预测，用组合评分器（0.85 模糊 + 0.15 精确）基准测试，发现约 0.6 初始准确率；4）定义评分器：先尝试字符串评分器，再引入 gpt-4.1 模型评分器，发现奖励黑客后迭代出更严格的评分标准，降低词汇权重至 0.10，临床权重 0.90，并加入关键修饰词惩罚；5）通过 /v1/fine_tuning/alpha/graders/validate 验证评分器，上传训练/验证文件，创建 RFT 任务，设置 compute_multiplier、eval_samples、eval_interval、n_epochs、reasoning_effort 等超参数；6）训练完成后获取 fine_tuned_model_id，在验证集上对比基础模型与微调模型的评分和推理摘要。
完整实战教程
*本指南面向已经熟悉 OpenAI API、对强化微调（RFT）有基本了解，并希望将微调后的模型用于研究或其他适当用途的开发者和机器学习从业者。OpenAI 的服务不适用于任何医疗状况的个性化治疗或诊断，并受我们的适用条款约束。*
强化微调（RFT）推理模型的过程是在模型之上运行强化学习，通过探索解决方案空间并强化能带来更高奖励的策略，来提升其推理性能。RFT 帮助模型做出更精准的决策，并更有效地理解上下文。
在本指南中，我们将通过生命科学研究领域的一个任务——从医患对话记录和描述中预测结果（这是许多健康研究中必要的评估）——来演示如何将 RFT 应用于 OpenAI 的 o4-mini 推理模型。我们将使用 medical-o1-verifiable-problem 数据集的一个子集。你将学到为你的用例成功运行 RFT 任务所需的关键步骤。
以下是我们要涵盖的内容：
- --
1. 设置
即使是强大的推理模型，在处理专家级行为时也可能出错——尤其是在医学等需要细微差别和精确性的领域。想象一下，一个模型试图从对话记录中提取 ICD-10 代码：即使它理解了要点，也可能不会使用医学专业人员所期望的精确术语。
其他适合 RFT 的优秀候选场景包括账本规范化或欺诈风险分级——在这些场景中，你需要精确、可靠且可重复的推理。查看我们的 RFT 用例指南 获取优秀示例。
在我们的案例中，我们将专注于教 o4-mini 更好地预测临床对话和描述的结果。具体来说，我们想看看 RFT 是否能提高预测的准确性。
在此过程中，我们将讨论如何编写有效的评分器、它们如何引导模型的学习，以及如何警惕经典的奖励黑客陷阱。
- --
2. 收集数据集
让我们从 Hugging Face 加载数据集开始。我们感兴趣的是那些以患者病例描述为框架、附带相关问题并随后给出正确答案的样本。这些样本代表了真实世界的记录，其中医生会总结病例并给出结果。对于任何用例而言，验证金标准答案的准确性都至关重要，需要仔细考量。在这里，我们将信任数据集的质量。
import re
from datasets import load_dataset
ds = load_dataset("FreedomIntelligence/medical-o1-verifiable-problem")
def is_age_question(sample):
question = sample.get('Open-ended Verifiable Question', '')
# Match "A 88-year-old", "An 8-year-old", "A 23-year-old", etc. at the start
return re.match(r"^(A|An) \d{1,2}-year-old", question) is not None
filtered_samples = [s for s in ds["train"] if is_age_question(s)]
print(f"Filtered samples: {len(filtered_samples)}")
RFT 的优势之一是，它不需要成千上万的样本就能开始产生效果。得益于轨迹采样和训练过程中的反馈循环，模型不仅学习正确的行为，还能学习需要避免的模式。这意味着即使使用小数据集，我们也能看到显著的提升。
在这次运行中，我们将随机抽取 100 个训练样本和 100 个测试样本，并对它们进行轻微归一化。
import random
# Set a random seed for reproducibility
random.seed(42)
# Randomly select 100 training samples from filtered_samples
train_samples = random.sample(filtered_samples, min(100, len(filtered_samples)))
# Remove training samples from filtered_samples to avoid overlap
remaining_samples = [s for s in filtered_samples if s not in train_samples]
# Randomly select 100 test samples from the remaining samples (no overlap)
test_samples = random.sample(remaining_samples, min(100, len(remaining_samples)))
print(f"Number of training samples: {len(train_samples)}")
print(f"Number of test samples: {len(test_samples)}")
# Standardize the 'Ground-True Answer' fields to all lowercase in train and test samples
for sample in train_samples:
if 'Ground-True Answer' in sample and isinstance(sample['Ground-True Answer'], str):
sample['Ground-True Answer'] = sample['Ground-True Answer'].lower()
for sample in test_samples:
if 'Ground-True Answer' in sample and isinstance(sample['Ground-True Answer'], str):
sample['Ground-True Answer'] = sample['Ground-True Answer'].lower()
我们将把这些样本转换为 jsonl 格式，以符合强化微调 API 的要求。
import json
def convert_to_jsonl_format(samples, filename):
with open(filename, "w") as f:
for sample in samples:
user_content = sample.get("Open-ended Verifiable Question", "")
reference_answer = sample.get("Ground-True Answer", "")
json_obj = {
"messages": [
{"role": "user", "content": user_content}
],
"reference_answer": reference_answer
}
f.write(json.dumps(json_obj) + "\n")
def load_jsonl(filename):
samples = []
with open(filename, "r") as f:
for line in f:
samples.append(json.loads(line))
return samples
# Save the datasets to jsonl files
convert_to_jsonl_format(train_samples, "data/medical_01_verifiable_problem_train.jsonl")
convert_to_jsonl_format(test_samples, "data/medical_01_verifiable_problem_val.jsonl")
# Load the datasets back from jsonl files
train_samples_loaded = load_jsonl("data/medical_01_verifiable_problem_train.jsonl")
test_samples_loaded = load_jsonl("data/medical_01_verifiable_problem_val.jsonl")
接下来：我们将看看基础模型开箱即用的表现——以及哪里还有提升空间。
- --
3. 对基础模型进行基准测试
在微调任何内容之前，我们需要知道起点在哪里。基准测试让我们清晰地了解模型初始的优势和劣势——这样之后我们就能衡量它进步了多少。
我们首先依赖两个简单却强大的评估器：
clinical_phrase_binary_grader—— 一个精确匹配检查器。clinical_phrase_grader—— 一个更宽松的、基于 token 的相似度评分器。
from rapidfuzz import fuzz, utils
def clinical_phrase_grader(sample: dict, item: dict) -> float:
from rapidfuzz import fuzz, utils
score = fuzz.token_set_ratio(sample["output_text"], item["reference_answer"], processor=utils.default_process)
return score / 100.0
def clinical_phrase_binary_grader(sample: dict, item: dict) -> float:
return 1.0 if sample["output_text"] == item["reference_answer"] else 0.0
def combined_grader(sample: dict, item: dict, weights: list[float] = [0.85, 0.15]) -> float:
clinical_phrase_score = clinical_phrase_grader(sample, item)
binary_score = clinical_phrase_binary_grader(sample, item)
return weights[0] * clinical_phrase_score + weights[1] * binary_score
这种组合让我们既能追踪严格正确性，也能追踪部分词汇重叠。二元评分器给出明确的 0 或 1：模型是否产生了完全匹配？较宽松的评分器则提供更多细微差别——输出与标准答案有多接近？我们同时使用两者，因为结果往往有多种有效表述方式。例如，模型可能回答“痛风性关节炎”而不是“痛风”。虽然人类评估者可能认为这在一定程度上可以接受，但严格的字符串匹配则不会。结合精确评分和模糊评分，可以确保对模型输出进行更准确、更公平的评估。
我们构建一个辅助函数，在示例前添加系统提示。
def prepend_system_prompt_to_first_user_message(samples, system_prompt, path=None):
new_samples = []
for sample in samples:
# Deep copy to avoid mutating the original
sample_copy = json.loads(json.dumps(sample))
messages = sample_copy.get("messages", [])
if messages and messages[0].get("role") == "user" and isinstance(messages[0].get("content"), str):
if not messages[0]["content"].startswith(system_prompt):
messages[0]["content"] = f"{system_prompt}\n\n{messages[0]['content']}"
new_samples.append(sample_copy)
if path is not None:
with open(path, "w", encoding="utf-8") as f:
for item in new_samples:
f.write(json.dumps(item, ensure_ascii=False) + "\n")
return new_samples
simple_prompt = """You are an expert clinician. For each clinical vignette, respond with exactly one phrase: the single most likely outcome or phenomenon, all in lowercase.
- Do not add punctuation, articles, explanations, or commentary - output only the term itself.
- Sometimes, the expected answer can be a synonym of what you think.
- Use the standard clinical name (e.g. “thought withdrawal”, “Toxoplasma encephalitis”)."""
train_samples_loaded_simple_sys_prompt = prepend_system_prompt_to_first_user_message(
train_samples_loaded, simple_prompt, path="data/medical_01_verifiable_problem_train_simple_prompt.jsonl"
)
test_samples_loaded_simple_sys_prompt = prepend_system_prompt_to_first_user_message(
test_samples_loaded, simple_prompt, path="data/medical_01_verifiable_problem_val_simple_prompt.jsonl"
)
然后构建一个辅助函数来生成并存储模型的预测结果。
from openai import OpenAI
import concurrent.futures
from tqdm import tqdm
import os
client = OpenAI()
def generate_model_predictions(
subset,
prompt_type,
model_name="o4-mini-2025-04-16",
reasoning_effort="medium",
n_runs=1,
verbose=False,
):
if isinstance(subset, str):
samples_path = f"data/medical_01_verifiable_problem_{subset}_{prompt_type}_prompt.jsonl"
with open(samples_path, "r", encoding="utf-8") as f:
test_samples = [json.loads(line) for line in f if line.strip()]
else:
test_samples = [subset]
def run_inference(item):
resp = client.responses.create(
model=model_name,
input=item["messages"],
reasoning={"effort": reasoning_effort, "summary": "detailed"},
)
model_prediction = {'output_text': resp.output_text}
reasoning_tokens_used = resp.usage.output_tokens_details.reasoning_tokens
summaries = [seg.text for item in resp.output if item.type == "reasoning" for seg in item.summary]
summaries_string = "\n".join(summaries)
if verbose:
print("Prompt: {}".format(item["messages"][0]["content"]))
print(f"Model Sample: {model_prediction}\nSolution: {item['reference_answer']}\n")
return {
"model_prediction": model_prediction["output_text"],
"input": item,
"reasoning_tokens_used": reasoning_tokens_used,
"reference_answer": item["reference_answer"],
"summaries": summaries_string
}
# Ensure the predictions directory exists before any file operations
predictions_dir = os.path.join("data", "rft", "predictions")
os.makedirs(predictions_dir, exist_ok=True)
# Check if results already exist for all runs
results_per_run = []
for run_idx in range(n_runs):
run_save_path = os.path.join(
predictions_dir,
f"{subset}_{prompt_type}_{model_name}_{reasoning_effort}_predictions_run{run_idx+1}.json"
)
if os.path.exists(run_save_path):
print(f"Results for run {run_idx+1} already exist at {run_save_path}. Loading results.")
with open(run_save_path, "r", encoding="utf-8") as f:
run_results = json.load(f)
results_per_run.append(run_results)
else:
if len(test_samples) == 1:
run_results = [run_inference(test_samples[0])]
else:
run_results = []
with concurrent.futures.ThreadPoolExecutor() as executor:
futures = [executor.submit(run_inference, item) for item in test_samples]
for future in tqdm(futures, total=len(futures), desc=f"Generating predictions (run {run_idx+1})"):
result = future.result()
run_results.append(result)
with open(run_save_path, "w", encoding="utf-8") as f:
json.dump(run_results, f, ensure_ascii=False, indent=2)
results_per_run.append(run_results)
# Return a flat list for backward compatibility
if n_runs == 1:
return results_per_run[0]
else:
return results_per_run
要生成预测，首先确保你的 API 密钥已设置：
export OPENAI_API_KEY=...
# OpenAI o4-mini model
results_simple_o4mini = generate_model_predictions(
subset="train",
prompt_type="simple",
model_name="o4-mini",
reasoning_effort="medium",
n_runs=3
)
# OpenAI o3 model
results_simple_o3 = generate_model_predictions(
subset="train",
prompt_type="simple",
model_name="o3",
reasoning_effort="medium",
n_runs=3
)
我们现在已经有了可以评估的预测结果。<br>
我们将构建一个辅助函数，以便轻松替换不同的评分方法，
import functools
def evaluate_predictions_with_grader(
predictions,
grader_func=combined_grader,
):
results = []
if isinstance(predictions, dict):
predictions = [predictions]
def run_grading(pred):
model_prediction = {"output_text": pred["model_prediction"]}
item = pred["input"]
score = grader_func(model_prediction, item)
result = pred.copy()
result["score"] = score
return result
if len(predictions) == 1:
result = run_grading(predictions[0])
results.append(result)
else:
with concurrent.futures.ThreadPoolExecutor() as executor:
futures = [executor.submit(run_grading, pred) for pred in predictions]
for future in tqdm(concurrent.futures.as_completed(futures), total=len(futures), desc="Grading predictions"):
results.append(future.result())
total = len(results)
correct = sum(r["score"] for r in results)
accuracy = correct / total if total else 0.0
metrics = {
"total_samples": total,
"accuracy": accuracy,
}
print(metrics)
return metrics, results
def run_prediction_evaluation(
model_name="o4-mini",
reasoning_effort="medium",
prompt_type="simple",
subset="train",
grader_func=combined_grader,
num_runs=3,
):
if isinstance(grader_func, functools.partial):
name = grader_func.func.__name__
mg = grader_func.keywords["model_grader"]
mg_name = mg["name"]
name = f"{name}_{mg_name}"
else:
name = getattr(grader_func, "__name__", getattr(grader_func, "__class__", type(grader_func)).__name__)
grader_func_name = name.replace(" ", "_").replace(":", "_").replace("/", "_").replace(",", "_")
for i in range(num_runs):
preds_path = f"data/rft/predictions/{subset}_{prompt_type}_{model_name}_{reasoning_effort}_predictions_run{i+1}.json"
with open(preds_path, "r") as f:
preds = json.load(f)
metrics, results_with_scores = evaluate_predictions_with_grader(preds, grader_func=grader_func)
# Save the scored results
with open(f"data/rft/predictions/{subset}_{prompt_type}_{model_name}_{reasoning_effort}_{grader_func_name}_predictions_run_{i+1}_scored.json", "w") as f:
json.dump(results_with_scores, f, indent=2)
# Save the metrics
with open(f"data/rft/predictions/{subset}_{prompt_type}_{model_name}_{reasoning_effort}_{grader_func_name}_predictions_run_{i+1}_metrics.json", "w") as f:
json.dump(metrics, f, indent=2)
# Save the scores (if present in results_with_scores)
scores = [item.get("score") for item in results_with_scores if "score" in item]
with open(f"data/rft/predictions/{subset}_{prompt_type}_{model_name}_{reasoning_effort}_{grader_func_name}_predictions_run_{i+1}_scores.json", "w") as f:
json.dump(scores, f, indent=2)
def load_predictions(
model_name="o4-mini",
reasoning_effort="medium",
prompt_type="simple",
subset="train",
grader_func_name="clinical_phrase_grader",
num_runs=3
):
all_predictions = []
all_metrics = []
for run in range(1, num_runs + 1):
pred_path = f"data/rft/predictions/{subset}_{prompt_type}_{model_name}_{reasoning_effort}_{grader_func_name}_predictions_run_{run}_scored.json"
metrics_path = f"data/rft/predictions/{subset}_{prompt_type}_{model_name}_{reasoning_effort}_{grader_func_name}_predictions_run_{run}_metrics.json"
try:
with open(pred_path, "r") as f:
predictions = json.load(f)
except FileNotFoundError:
predictions = None
try:
with open(metrics_path, "r") as f:
metrics = json.load(f)
except FileNotFoundError:
metrics = None
all_predictions.append(predictions)
all_metrics.append(metrics)
return all_predictions, all_metrics
然后运行评估。
model_name = "o4-mini"
reasoning_effort = "medium"
prompt_type = "simple"
subset = "train"
grader_func = combined_grader
grader_func_name = "combined_grader"
num_runs = 3
run_prediction_evaluation(
model_name=model_name,
reasoning_effort=reasoning_effort,
prompt_type=prompt_type,
subset=subset,
grader_func=grader_func,
num_runs=num_runs
)
predictions_o4mini_medium_simple_prompt, metrics_o4mini_medium_simple_prompt = load_predictions(model_name=model_name, reasoning_effort=reasoning_effort, prompt_type=prompt_type, subset=subset, grader_func_name=grader_func_name, num_runs=num_runs)
将结果可视化有助于我们发现趋势和失败模式。
# Print mistakes where the model did not get the correct answer (score < 1.0)
mistakes = [
{"index": i, **res}
for i, res in enumerate(predictions_o4mini_medium_simple_prompt[0])
if res["score"] < 1.0
]
print(f"\nTotal mistakes: {len(mistakes)}")
for m in mistakes[15:20]:
print(f"\n[Sample {m['index']}]")
print(f" Model prediction: {m['model_prediction']}")
print(f" Reference answer: {m['reference_answer']}")
print(f" Score: {m['score']}")
如上所述，典型的失败模式可分为三类：
- 微小差异和格式问题，得分 >=0.8。
- 部分词汇匹配，0.3 < 得分 < 0.8。
- 词汇完全不匹配，得分 < 0.3。
我们可以在训练集上可视化完整的得分分布。
注意：在实践中，大规模分析模型错误通常涉及人工审查和自动化方法的结合，例如标记失败类型或按得分和内容对预测进行聚类。该工作流程超出了本指南的范围，但一旦你识别出广泛的模式，这是一个有价值的下一步。
import matplotlib.pyplot as plt
scores_distribution = [m['score'] for m in predictions_o4mini_medium_simple_prompt[0]]
plt.hist(scores_distribution, alpha=0.6, label='o4-mini medium simple prompt')
plt.legend()
让我们与其他模型和提示词进行比较，并将分数可视化。
# OpenAI o3 model
model_name = "o3"
reasoning_effort = "medium"
prompt_type = "simple"
subset = "train"
grader_func = combined_grader
grader_func_name = "combined_grader"
num_runs = 3
run_prediction_evaluation(model_name=model_name, reasoning_effort=reasoning_effort, prompt_type=prompt_type, subset=subset, grader_func=grader_func, num_runs=num_runs)
predictions_o3_medium_simple_prompt, metrics_o3_medium_simple_prompt = load_predictions(model_name=model_name, reasoning_effort=reasoning_effort, prompt_type=prompt_type, subset=subset, grader_func_name=grader_func_name, num_runs=num_runs)
import numpy as np
import pandas as pd
import seaborn as sns
def average_and_std_metrics(metrics_list):
"""Returns dicts of mean and std for a list of metrics dicts."""
if not metrics_list: return {}, {}
keys = metrics_list[0].keys()
arr = {k: np.array([m[k] for m in metrics_list]) for k in keys}
mean = {k: float(np.mean(arr[k])) for k in keys}
std = {k: float(np.std(arr[k])) for k in keys}
return mean, std
def plot_model_accuracies(model_metrics_avg, model_metrics_std, grader_title="Combined Grader Accuracy", sharey: bool = True) -> None:
"""Plots model accuracies with standard deviation error bars."""
# Convert the nested dicts into tidy DataFrames
df_avg = pd.DataFrame(model_metrics_avg).T.reset_index().rename(columns={"index": "Model"})
df_std = pd.DataFrame(model_metrics_std).T.reset_index().rename(columns={"index": "Model"})
# Long-form for Seaborn
long_df_avg = df_avg.melt(id_vars="Model", value_vars=["accuracy"], var_name="Metric", value_name="Accuracy")
long_df_std = df_std.melt(id_vars="Model", value_vars=["accuracy"], var_name="Metric", value_name="Std")
# Merge avg and std for error bars
long_df = pd.merge(long_df_avg, long_df_std, on=["Model", "Metric"])
pretty_names = {"accuracy": grader_title}
# Create a separate figure for each metric
for metric_key in ["accuracy"]:
metric_df = long_df[long_df["Metric"] == metric_key].copy()
plt.figure(figsize=(8, 5))
# Plot bars with error bars
ax = sns.barplot(data=metric_df, x="Model", y="Accuracy", hue="Model", palette="tab10", legend=False, errorbar=None)
bars = ax.patches
# Add error bars manually
for i, row in enumerate(metric_df.itertuples()):
bar = bars[i]
x = bar.get_x() + bar.get_width() / 2
y = row.Accuracy
yerr = row.Std
ax.errorbar(x=x, y=y, yerr=yerr, fmt='none', ecolor='black', capsize=5, elinewidth=2, capthick=2, zorder=10)
plt.title(pretty_names[metric_key])
plt.ylabel("Accuracy")
plt.xlabel("")
if sharey: plt.ylim(0, 1)
# Annotate bars with exact values
for bar in bars:
height = bar.get_height()
ax.annotate(f"{height:.2f}", xy=(bar.get_x() + bar.get_width() / 2, height), xytext=(0, 6), textcoords="offset points", ha='center', va='bottom', fontsize=10, fontweight='bold')
plt.xticks(rotation=15, ha="right")
plt.tight_layout()
plt.show()
avg_metrics_o4mini_medium_simple_prompt, std_metrics_o4mini_medium_simple_prompt = average_and_std_metrics(metrics_o4mini_medium_simple_prompt)
avg_metrics_o3_medium_simple_prompt, std_metrics_o3_medium_simple_prompt = average_and_std_metrics(metrics_o3_medium_simple_prompt)
model_metrics_avg = {
"o4-mini-medium-simple-prompt": avg_metrics_o4mini_medium_simple_prompt,
"o3-medium-simple-prompt": avg_metrics_o3_medium_simple_prompt,
}
model_metrics_std = {
"o4-mini-medium-simple-prompt": std_metrics_o4mini_medium_simple_prompt,
"o3-medium-simple-prompt": std_metrics_o3_medium_simple_prompt,
}
plot_model_accuracies(model_metrics_avg, model_metrics_std, grader_title="Combined Grader Accuracy")
我们可以看到，模型的性能存在明显的上限。在实践中，对提示词进行迭代通常有助于提升基线结果，并更充分地发挥基础模型的潜力。然而，在本例中，我们的提示词工程并未带来有意义的改进——因此我们将这些运行结果排除在分析之外。
RFT 能够奏效的一个关键要求是，基础模型从一开始就展现出它至少能在部分示例上成功完成任务。约 0.6 的初始准确率是一个强烈的信号，表明 RFT 可以提升性能。如果模型在你的任务上从未成功过，就没有可供爬坡的训练信号。
这一评估过程为我们下一步做好了准备：用来自评分器的结构化、高质量反馈来引导模型。
- --
4. 定义你的评分器
评分器定义了在 RFT 过程中塑造模型行为的奖励函数。它提供期望输出的示例——并对不期望的输出进行惩罚。设计一个有效的评分器既需要原则性的结构，也需要深思熟虑的领域洞察，这或许是成功进行 RFT 最重要的任务。
在本节中，我们将展示 3 个评分器，说明如何设置它们以适配 API，并讨论它们所产生的结果。然后，我们将展示如何实际启动一个 RFT 任务。
基于字符串的评分器
我们首先使用之前的评估函数构建了一个双重评分器，因为它提供了与预测答案和参考答案之间词汇接近度对齐的分数分布。它提供了一个起点，但信号不够丰富，不足以让 o4-mini 真正学习和改进，并且第一次实验显示在 RFT 运行期间奖励停滞。对于 API 调用，您应该按照以下所示构建 Python 评分函数。
import inspect
# --- Utility functions ---
def build_python_grader_payload(grader_fn) :
"""Build a payload for a python grader."""
grader_source = inspect.getsource(grader_fn)
# Enforce function name to be `grade`
grader_source = grader_source.replace(grader_fn.__name__, "grade", 1)
return {
"type": "python",
"source": grader_source,
}
multi_python_grader_tool_call = {
"type": "multi",
"graders": {
"clinical_phrase": {
"name": "clinical_phrase_grader",
"image_tag": "2025-05-08",
**build_python_grader_payload(clinical_phrase_grader),
},
"clinical_phrase_binary": {
"name": "clinical_phrase_binary_grader",
"image_tag": "2025-05-08",
**build_python_grader_payload(clinical_phrase_binary_grader),
},
},
"calculate_output": "0.85 * clinical_phrase + 0.15 * clinical_phrase_binary",
}
以下是其训练曲线的快照，其中绿色曲线是训练集奖励，蓝色曲线是测试集奖励：
Model Grader 1
为了解决这一局限，我们引入了一种更先进的方法：模型评分器。基于模型的评分器让我们能够将语义理解和细微差别嵌入反馈中。当涉及领域特定的同义词或模糊推理时，这一点尤其强大。
我们使用 gpt-4.1 作为评分模型，并以一个强调语义保真度的评分标准为指导：临床同义性、正确的疾病分类以及概念一致性。评分器并非关注表面措辞——例如“这是同一个字符串吗？”——而是旨在回答：“这是否反映了正确的结果或现象？”
为了确保评分器与专家期望一致，我们在基础模型预测的一个子集上对其进行了评估。对于任何生产用例，领域专家评审员都应验证模型给出的分数是否反映了首选的答案排序，并与领域判断保持一致。这通常涉及确认模型评分器是否根据预测的有效性正确地对它们进行排序。在本 cookbook 的范围内，我们通过使用 OpenAI o3 来检查更高质量的预测是否相对于其替代方案持续获得奖励，从而近似实现了这一评估。
基于这些关于 o3 的讨论，我们迭代更新模型评分器，直到结果对齐。
GRADER_PROMPT_1 = """
System:
You are an expert medical grader. Compare the **Reference Answer** to the **Model's Answer** and produce **only** a JSON object with:
• **result**: a float between 0.0 and 1.0
• **steps**: a list of reasoning steps (each with a `"description"` and a `"conclusion"`)
Scoring rubric (start at 0.0, then add or subtract):
1. Exact lexical match: **+0.15**
2. Clinical synonym (e.g. “withdrawal of thought” ↔ “thought withdrawal”): **+0.35**
3. Same disease family (e.g. two viral encephalitides): **+0.35**
4. Partial term overlap (e.g. “ulcer” in both phrases): **+0.15**
5. Completely unrelated: **-0.10**
• If multiple criteria apply, sum their weights (max 1.0).
• Cap the final score to the [0.0, 1.0] range.
• In your **steps**, show which rule you applied and the running subtotal.
"""
要通过 API 提交，字典就是这样构建的。
model_grader_1 = {
"type": "score_model",
"name": "gpt41_score_model_1",
"input": [
{
"role": "system",
"content": GRADER_PROMPT_1
},
{
"role": "user",
"content": "Reference Answer: {{item.reference_answer}}. Model's Answer: {{sample.output_text}}"
}
],
"pass_threshold": 0.75,
"model": "gpt-4.1-2025-04-14",
"range": [0, 1],
"sampling_params": {
"seed": 42,
"temperature": 0,
},
}
相应地，我们在本地设置了模型评分器，以检查我们接下来将要微调的模型的结果。
response_format = {
"name": "float_score_classification",
"strict": True,
"schema": {
"type": "object",
"properties": {
"steps": {
"type": "array",
"description": "A sequence of steps outlining the reasoning process.",
"items": {
"type": "object",
"properties": {
"description": {
"type": "string",
"description": "Detailed description of the reasoning in this step."
},
"conclusion": {
"type": "string",
"description": "The conclusion of the reasoning in this step."
}
},
"required": ["description", "conclusion"],
"additionalProperties": False
}
},
"result": {
"type": "number",
"description": "The float score assigned to the response. This should be in inclusive range RANGE_MIN to RANGE_MAX."
}
},
"required": ["steps", "result"],
"additionalProperties": False
}
}
# for completions
response_format = {
"type": "json_schema",
"json_schema": response_format
}
# Adapted python_model_grader to match the other graders' interface
def python_model_grader(sample, item, model_grader=model_grader_1):
"""
Calls an OpenAI model to grade the model output against the reference answer.
Expects sample to have "output_text", item to have "reference_answer".
Returns a float score (parsed from the model's JSON response).
"""
# Prepare the prompt as the grader expects
system_prompt = model_grader["input"][0]["content"]
user_prompt = model_grader["input"][1]["content"]
user_prompt_filled = user_prompt.replace("{{item.reference_answer}}", item["reference_answer"]).replace("{{sample.output_text}}", sample["output_text"])
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt_filled}
]
# Call the OpenAI API with the grader's model
response = client.chat.completions.create(
model=model_grader["model"],
messages=messages,
seed=model_grader.get("sampling_params", {}).get("seed", None),
temperature=model_grader.get("sampling_params", {}).get("temperature", 0),
response_format=response_format,
)
# Parse the float score from the model's JSON response
parsed = json.loads(response.choices[0].message.content)
return float(parsed["result"])
虽然评分标准最初给出了合理的反馈，但模型很快发现了一个漏洞，并开始奖励黑客行为。分数飙升——有时提高20-30个百分点——并非因为临床准确性提高，而是因为模型用同义词、剂量和完整管理方案填充了其“一个短语”的答案。你可能会看到 开始华法林治疗 并 继续普通肝素至少5天，重叠直至INR达到治疗范围（2–3） 或 咀嚼阿司匹林325毫克立即加硝酸甘油……，而不是分别要求的 继续普通肝素 或 阿司匹林。尽管系统提示明确说明——*“仅用一个短语回答：最可能的结果或现象”*——这些冗长的输出抬高了*lexical_similarity*分数，却没有精确增加预测价值。这一经验凸显了持续检查模型输出并警惕可能悄然扭曲评估指标的奖励黑客行为的必要性。
以下是其训练曲线的快照（绿色为训练奖励，蓝色为测试奖励）：
Model Grader 2
为了缓解这种奖励黑客行为，我们通过明确预期、实施更严格的输出约束，并提供正确与错误行为的对比示例，改进了评分器提示。我们再次使用 o3 进行迭代，利用基础 o4-mini 的预测以及之前微调模型的作弊示例，来设计和验证我们的评分器。这个更新版评分器的另一个重要点是降低了 *lexical_similarity* 的权重，以确保 *clinical_similarity* 占主导地位。
GRADER_PROMPT_2 = """You are an expert medical grader.
Compare the reference_answer (gold standard) with the model_prediction
and return **exactly** this JSON object:
{
"steps": [ // each: {"description": "...", "conclusion": "..."}
…
],
"result": <float 0-1 rounded to 3 decimals>
}
──────────────── Input placeholders ───────────────
reference_answer:
model_prediction:
──────────── Normalisation steps ────────────
• lowercase, strip punctuation / excess whitespace
• expand common abbreviations (e.g. cll → chronic lymphocytic leukemia)
• map both strings to ICD-10 / SNOMED concepts when possible
──────────── Clinical layer rubric ───────────
L1 exact concept or universally accepted synonym
L2 same concept but benign modifier differs (e.g. “acute”, “left”)
L3 same disease / drug family but wrong subtype or variant
L4 same organ system but entirely different disease / intervention
L5 only partial mechanistic overlap (e.g. both vasodilators)
L6 unrelated or nonsensical
──────────── Scoring parameters ─────────────
clinical_weight = 0.90
lexical_weight = 0.10
clinical_similarity = {1:1.00, 2:0.85, 3:0.45, 4:0.30, 5:0.10, 6:0.00}
lexical_similarity = normalized_levenshtein(reference_answer,
model_prediction)
# Optional penalty if a clinically critical adjective is missing
critical_modifiers = [
"wide", "narrow", "acute", "chronic", "posteromedial",
"oxidized", "oxidised", "left", "right"
]
modifier_pen = -0.05 if any(
w in reference_answer and w not in model_prediction
for w in critical_modifiers
) else 0.0
# Determine layer L (1-6) per rubric above using ontology + judgment.
if L == 6:
score = 0.0
else:
score = (clinical_weight * clinical_similarity[L] +
lexical_weight * lexical_similarity) + modifier_pen
Clamp to [0,1] and round to 3 decimals.
Output **only** the JSON.
──────────────── Worked examples ─────────────
reference_answer: beta-thalassemia major
model_prediction: beta-thalassemia minor
reasoning: Both involve β-globin chain synthesis, but “major” causes
transfusion-dependent anemia while “minor” is largely benign;
same family, wrong subtype → **L3**. Lexical ≈ 0.83.
score = 0.90·0.45 + 0.10·0.83 = 0.488 → **0.488**
reference_answer: ACE inhibitor
model_prediction: angiotensin-receptor blocker
reasoning: Both act on the renin–angiotensin axis yet on different
targets; only partial mechanistic overlap → **L5**.
Lexical ≈ 0.31.
score = 0.90·0.10 + 0.10·0.31 = 0.121 → **0.121**
reference_answer: acute pancreatitis
model_prediction: pancreatitis
reasoning: Same disorder but missing timing adjective “acute”;
benign modifier difference → **L2**. Lexical ≈ 0.78.
score = 0.90·0.85 + 0.10·0.78 = 0.843 → **0.843**
reference_answer: valproate
model_prediction: valproic acid
reasoning: Valproic acid is the active moiety of valproate; mechanisms
and indications are identical → **L1**. Lexical ≈ 0.82.
score = 0.90·1.00 + 0.10·0.82 = 0.982 → **0.982**
reference_answer: riboflavin
model_prediction: riboflavin deficiency
reasoning: Adds “deficiency” but refers to the same vitamin (B₂);
benign modifier difference → **L2**. Lexical ≈ 0.60.
score = 0.90·0.85 + 0.10·0.60 = 0.825 → **0.825**
reference_answer: splenectomy
model_prediction: acetaminophen overdose
reasoning: Surgical removal of the spleen has no mechanistic or anatomic
relationship to toxic drug ingestion → **L6**.
score = **0.000**
reference_answer: ulcerative colitis
model_prediction: Crohn disease
reasoning: Both are inflammatory-bowel diseases but differ in location,
histology and management; same organ system, different disease
→ **L4**. Lexical ≈ 0.38.
score = 0.90·0.30 + 0.10·0.38 = 0.308 → **0.308**"""
model_grader_2 = {
"type": "score_model",
"name": "gpt41_score_model_2",
"input": [
{
"role": "system",
"content": GRADER_PROMPT_2
},
{
"role": "user",
"content": "Reference Answer: {{item.reference_answer}}. Model's Answer: {{sample.output_text}}"
}
],
"pass_threshold": 0.75,
"model": "gpt-4.1-2025-04-14",
"range": [0, 1],
"sampling_params": {
"seed": 42,
"temperature": 0,
},
}
最终结果是一个高信号、领域敏感的评分器，它引导模型做出更合适、更简洁的预测。
关于成本的说明： LLM 评分器除了训练计算之外，还会产生 token 使用费用。为了有效管理成本，我们建议：
- 在基础模型补全（以及可选的合成补全）上本地测试你的评分器，以确保其符合你的评分标准或人类偏好。在可用的情况下，使用 flex processing 进行更高效的评估。
- 从小规模 RFT 运行开始，以验证评分器对齐情况并在扩大规模之前检测潜在的奖励黑客行为。
接下来让我们看看如何在下一步启动训练！
- --
5. 训练
一旦你的提示词和评分器最终确定，就可以开始训练了。本节展示如何使用最终的评分器启动 RFT——当然，在试验早期版本的评分器以评估其性能时，你其实已经运行过类似的命令了。
我们确保评分器通过了 API 测试，
import requests
API_KEY = os.environ["OPENAI_API_KEY"]
HEADERS = {"Authorization": f"Bearer {API_KEY}"}
# Validate a grader configuration for fine-tuning
payload = {"grader": model_grader_2}
try:
response = requests.post(
"https://api.openai.com/v1/fine_tuning/alpha/graders/validate",
json=payload,
headers=HEADERS,
)
response.raise_for_status()
print("Grader validated")
except requests.exceptions.RequestException as e:
print(f"Error validating grader: {e}")
if 'response' in locals():
print(f"Response: {response.text}")
并将训练集和测试集上传到 OpenAI 文件系统。
# Set your training and test file paths
train_file = "data/medical_01_verifiable_problem_train_simple_prompt.jsonl"
test_file = "data/medical_01_verifiable_problem_val_simple_prompt.jsonl"
def upload_file(file_path: str) -> str:
"""Upload a file to the OpenAI platform for fine-tuning."""
print(f"Uploading file: {file_path}")
with open(file_path, 'rb') as f:
response = requests.post(
"https://api.openai.com/v1/files",
headers=HEADERS,
files={"file": f},
data={"purpose": "fine-tune"}
)
response.raise_for_status()
file_id = response.json()["id"]
print(f"File uploaded successfully. File ID: {file_id}")
return file_id
train_file_id = train_file
if train_file.endswith("jsonl"):
print(f"Training file detected: {train_file}")
train_file_id = upload_file(train_file)
test_file_id = test_file
if test_file and test_file.endswith("jsonl"):
print(f"test file detected: {test_file}")
test_file_id = upload_file(test_file)
现在我们来定义本次运行的超参数。我们将对 o4-mini 进行微调，使用 medium 推理强度。该参数会通过限制模型用于推理的 token 数量来影响运行时长。我们以适中的计算量倍数和合理的 epoch 数量进行调优，优先考虑效率和快速迭代。此外，我们将 eval_samples 参数设为 3，以便在 o4-mini 输出具有随机性的情况下使验证曲线更加稳健。对多个样本取平均可以降低噪声，并有助于揭示一致的学习模式。
你需要根据预算、期望的泛化能力以及数据集难度来调整这些参数。
# Set the model and other parameters
model = "o4-mini-2025-04-16"
suffix = "medical_01_verifiable_problem_gpt41_grader"
reasoning_effort = "medium"
n_epochs = 5
seed = 42
grader = model_grader_2
response_format_predictions = None
compute_multiplier = 1.0
eval_samples = 3
eval_interval = 5
我们现在准备好启动运行了！
# Launch the RFT job
payload = dict(
training_file=train_file_id,
validation_file=test_file_id,
model=model,
suffix=suffix,
method=dict(
type="reinforcement",
reinforcement=dict(
grader=grader,
response_format=response_format_predictions,
hyperparameters=dict(
compute_multiplier=compute_multiplier,
eval_samples=eval_samples,
eval_interval=eval_interval,
n_epochs=n_epochs,
reasoning_effort=reasoning_effort,
)
)
),
seed=seed
)
try:
response = requests.post(
"https://api.openai.com/v1/fine_tuning/jobs",
json=payload,
headers=HEADERS,
)
response.raise_for_status()
job_id = response.json().get("id")
if job_id:
print("Training job created with ID:", job_id)
print(
f"View the job details at: https://platform.openai.com/finetune/{job_id}")
else:
print("Failed to retrieve job ID from response.")
except requests.exceptions.RequestException as e:
print(f"An error occurred while creating the training job: {e}")
if 'response' in locals():
print(f"Response: {response.text}")
在仪表盘上，你可以观察奖励曲线——它们让你看到整体性能随步骤提升，而在 *multi_grader* 的情况下，每个评分器的图表会分解具体组件。推理 token 使用趋势（通常随着模型变得更加自信而减少）和步骤持续时间指标提供了对效率的洞察。评分器延迟和错误计数图表有助于确保你的评分器在运行期间保持高性能且无错误。
以下是我们的训练曲线快照，其中绿色和橙色曲线对应训练集，而蓝色和红色曲线对应测试子集：
在训练过程中，测试集上的评估运行会直接记录到 Evaluation API。你可以前往那里跟踪你的样本表现，并了解预测如何随时间演变。
- --
6. 使用你的微调模型
训练完成后，你可以通过 model_id 调用你的新模型，并评估其改进效果。期待更精准的预测！
# To retrieve information about a fine-tuning job (including the fine-tuned model id), use the job_id:
response = requests.get(
f"https://api.openai.com/v1/fine_tuning/jobs/{job_id}",
headers=HEADERS,
)
if response.ok:
data = response.json()
if data.get("status") == "succeeded":
fine_tuned_model_id = data.get("fine_tuned_model")
else:
fine_tuned_model_id = None
else:
raise Exception(f"Request failed: {response.status_code} - {response.text}")
print("Fine-tuned model id:", fine_tuned_model_id)
模型的预测得分
让我们计算基础模型和微调模型的得分以进行比较。
from functools import partial
model_name = fine_tuned_model_id
reasoning_effort = "medium"
prompt_type = "simple"
subset = "val"
grader_func = partial(python_model_grader, model_grader=model_grader_2)
grader_func_name = "python_model_grader_gpt41_score_model_2"
num_runs = 3
results_ft_model_grader_2 = generate_model_predictions(
subset=subset,
prompt_type=prompt_type,
model_name=model_name,
reasoning_effort=reasoning_effort,
n_runs=num_runs
)
run_prediction_evaluation(
model_name=model_name,
reasoning_effort=reasoning_effort,
prompt_type=prompt_type,
subset=subset,
grader_func=grader_func,
num_runs=num_runs
)
predictions_ftmodel_medium_simple_prompt_model_grader_2, metrics_ftmodel_medium_simple_prompt_model_grader_2 = load_predictions(
model_name=model_name,
reasoning_effort=reasoning_effort,
prompt_type=prompt_type,
subset=subset,
grader_func_name=grader_func_name,
num_runs=num_runs
)
model_name = "o4-mini"
reasoning_effort = "medium"
prompt_type = "simple"
subset = "val"
grader_func = partial(python_model_grader, model_grader=model_grader_2)
grader_func_name = "python_model_grader_gpt41_score_model_2"
num_runs = 3
results_o4mini_model_grader_2 = generate_model_predictions(
subset=subset,
prompt_type=prompt_type,
model_name=model_name,
reasoning_effort=reasoning_effort,
n_runs=num_runs
)
run_prediction_evaluation(
model_name=model_name,
reasoning_effort=reasoning_effort,
prompt_type=prompt_type,
subset=subset,
grader_func=grader_func,
num_runs=num_runs
)
predictions_o4mini_medium_simple_prompt_model_grader_2, metrics_o4mini_medium_simple_prompt_model_grader_2 = load_predictions(
model_name=model_name,
reasoning_effort=reasoning_effort,
prompt_type=prompt_type,
subset=subset,
grader_func_name=grader_func_name,
num_runs=num_runs
)
model_name = "o3"
reasoning_effort = "medium"
prompt_type = "simple"
subset = "val"
grader_func = partial(python_model_grader, model_grader=model_grader_2)
grader_func_name = "python_model_grader_gpt41_score_model_2"
num_runs = 3
results_o3_model_grader_2 = generate_model_predictions(
subset=subset,
prompt_type=prompt_type,
model_name=model_name,
reasoning_effort=reasoning_effort,
n_runs=num_runs
)
run_prediction_evaluation(
model_name=model_name,
reasoning_effort=reasoning_effort,
prompt_type=prompt_type,
subset=subset,
grader_func=grader_func,
num_runs=num_runs
)
predictions_o3_medium_simple_prompt_model_grader_2, metrics_o3_medium_simple_prompt_model_grader_2 = load_predictions(
model_name=model_name,
reasoning_effort=reasoning_effort,
prompt_type=prompt_type,
subset=subset,
grader_func_name=grader_func_name,
num_runs=num_runs
)
我们现在可以将它们可视化了！
avg_metrics_o4mini_medium_simple_prompt_model_grader_2, std_metrics_o4mini_medium_simple_prompt_model_grader_2 = average_and_std_metrics(metrics_o4mini_medium_simple_prompt_model_grader_2)
avg_metrics_o3_medium_simple_prompt_model_grader_2, std_metrics_o3_medium_simple_prompt_model_grader_2 = average_and_std_metrics(metrics_o3_medium_simple_prompt_model_grader_2)
avg_metrics_ftmodel_medium_simple_prompt_model_grader_2, std_metrics_ftmodel_medium_simple_prompt_model_grader_2 = average_and_std_metrics(metrics_ftmodel_medium_simple_prompt_model_grader_2)
model_metrics_avg = {
"o4-mini-medium-simple-prompt": avg_metrics_o4mini_medium_simple_prompt_model_grader_2,
"o3-medium-simple-prompt": avg_metrics_o3_medium_simple_prompt_model_grader_2,
"ftmodel-medium-simple-prompt": avg_metrics_ftmodel_medium_simple_prompt_model_grader_2
}
model_metrics_std = {
"o4-mini-medium-simple-prompt": std_metrics_o4mini_medium_simple_prompt_model_grader_2,
"o3-medium-simple-prompt": std_metrics_o3_medium_simple_prompt_model_grader_2,
"ftmodel-medium-simple-prompt": std_metrics_ftmodel_medium_simple_prompt_model_grader_2
}
plot_model_accuracies(model_metrics_avg, model_metrics_std, grader_title="Model Grader 2 Accuracy")
# Print mistakes where the model did not get the correct answer (score < 1.0)
mistakes = [
{"index": i, **res}
for i, res in enumerate(predictions_ftmodel_medium_simple_prompt_model_grader_2[0])
if res["score"] < 1.0
]
print(f"\nTotal mistakes: {len(mistakes)}")
for m in mistakes[5:10]:
print(f"\n[Sample {m['index']}]")
print(f" Model prediction: {m['model_prediction']}")
print(f" Reference answer: {m['reference_answer']}")
print(f" Score: {m['score']}")
我们看到微调后准确率大约提升了 5 个百分点。查看前几个错误，模型倾向于严厉惩罚那些接近但临床上不完全相同的答案——比如*普通肝素*与*依诺肝素*。它也会对较长的答案扣分，即使这些答案是正确的，比如*腹部增强计算机断层扫描*。
scores_o4 = [p['score'] for p in predictions_o4mini_medium_simple_prompt_model_grader_2[0]]
scores_ft = [p['score'] for p in predictions_ftmodel_medium_simple_prompt_model_grader_2[0]]
# Determine common bins for both histograms
all_scores = scores_o4 + scores_ft
bins = plt.hist(all_scores, bins=5, alpha=0)[1]
# Plot histograms and capture the counts
counts_o4, _, _ = plt.hist(
scores_o4,
bins=bins,
alpha=0.6,
label='o4-mini-medium-simple-prompt'
)
counts_ft, _, _ = plt.hist(
scores_ft,
bins=bins,
alpha=0.6,
label='ftmodel-medium-simple-prompt'
)
plt.title("Model Grader 2 Score Distribution by Model")
plt.xlabel("Score")
plt.ylabel("Count")
plt.ylim(top=75)
plt.legend()
# Print the bin counts
print("o4-mini-medium-simple-prompt bin counts:", counts_o4)
print("ftmodel-medium-simple-prompt bin counts:", counts_ft)
print("Max bin count (y-axis):", max(max(counts_o4), max(counts_ft)))
分数分布显示，RFT 将模型的预测从中低区间（0.2-0.6）转移到了高区间（0.8-1.0）。由于评分器更强调临床相似性而非词汇匹配，我们的*专家*评分器将这一转变归因于更强的医学推理能力，而非更好的措辞。少数原本就较弱的预测落入了 0.0-0.1 区间，表明仍存在知识缺口。
请注意，由于之前的 combined_grader 旨在奖励词汇正确性，其准确率并没有太大提升——这是意料之中的。这一差距进一步说明了为什么验证模型评分器至关重要，以及为什么你应该监控奖励黑客行为。在我们的案例中，我们使用 o3 来抽查评分行为，但领域专家审查是必不可少的。
模型的推理过程
分析微调模型时的另一个重要方面是推理摘要。模型可能会在这些摘要中提供关键信息，通过探索这些摘要来理解模型在何处失败，可以推动模型和评分器系统提示的更新。下面，我们展示了模型生成的此类思维链摘要示例，以展示其回答问题的方式：
# Flatten the list of lists into a single list of dicts
predictions = {
"o4-mini": predictions_o4mini_medium_simple_prompt_model_grader_2,
"o3": predictions_o3_medium_simple_prompt_model_grader_2,
"ftmodel": predictions_ftmodel_medium_simple_prompt_model_grader_2,
}
for model_name, predictions in predictions.items():
all_preds = [item for sublist in predictions for item in sublist]
reasoning_tokens = [p['reasoning_tokens_used'] for p in all_preds if 'reasoning_tokens_used' in p]
mean_reasoning_tokens = np.mean(reasoning_tokens)
print(f"Mean reasoning_tokens_used {model_name}: {mean_reasoning_tokens:.0f}")
微调后的模型会花费更多的推理 token 来思考问题。让我们借助推理摘要来可视化一个示例。
from IPython.display import Markdown, display
markdown_text = results_o4mini_model_grader_2[0][30]["summaries"]
display(Markdown(markdown_text))
markdown_text = results_ft_model_grader_2[0][30]["summaries"]
display(Markdown(markdown_text))
基础版 o4‑mini 的推理直接跳到“腹部 CT 扫描”，主要担心小写格式问题，并且只给出了一个粗略的“通常是首选”的理由。而 finetuned model 则首先将患者的脂肪泻和饮酒史与慢性胰腺炎或癌症联系起来，权衡了 CT 与 MRCP 和 EUS 的利弊，并解释了为什么对比增强腹部 CT 最能显示钙化和结构变化。后者似乎更加细致，并且似乎学会了更深入地分解病例描述。
进一步提升分数
基线 o3 和我们微调后的 o4-mini 有时会在相同的样本上得零分——这是一个危险信号，表明参考标签可能有误。在增加更多计算资源之前，先投资于数据质量：让领域专家重新标注噪声较大的数据切片，分析模型的推理过程，然后收紧评分器的提示词。干净、可信的数据加上有条理的更新，几乎总能比增加训练轮次带来更高的准确率。
（未提供需要翻译的 Markdown 正文，请补充内容。）
结论
我们已经探讨了如何设计评分器，以便在 RFT 过程中为 o4-mini 提供所需的详细反馈。正是这种信号帮助模型真正学习并超越基线水平。模型评分器在这方面可以非常强大——但前提是它们必须被精心设计。草率的评分器或草率的数据会发出错误的信号，并将模型引向错误的方向。
现在，你已经准备好使用 OpenAI API 在你自己的模型上应用强化微调了。我们非常期待看到你如何通过自定义评分器和更智能的模型行为来突破推理和工具使用的边界！
如需故障排除或后续步骤，请参阅 OpenAI 微调文档。
常见问题
常见错误包括：评分器未通过 API 验证，需检查 Python 函数名是否为 grade、source 是否正确；奖励黑客表现为分数飙升但输出冗长，需收紧评分标准、降低词汇权重、增加输出约束；评分器延迟或错误计数高，需检查模型评分器的 token 成本和超时；训练奖励停滞，可能是评分器信号不足或基础模型初始准确率过低，需先确保模型在部分样本上能成功；参考标签噪声导致模型和 o3 同时得零分，需领域专家重新标注；API 密钥未设置或文件上传失败，需检查 OPENAI_API_KEY 和文件格式；评估样本数不足导致验证曲线噪声大，可增大 eval_samples。
企业应用扩展
生产环境优化建议：1）评分器本地测试与人工校准，使用 flex processing 降低成本；2）从小规模 RFT 运行开始验证评分器对齐，再扩大规模；3）监控奖励曲线、推理 token 使用趋势、评分器延迟和错误计数；4）结合领域专家审查评分器排序是否符合临床判断；5）分析推理摘要定位失败模式，迭代更新模型和评分器提示；6）投资数据质量，重新标注噪声切片，优先于增加训练轮次；7）探索多评分器组合、动态权重调整、以及将评分器用于其他专家任务如法律、金融等。
总结
本文以医学问答任务为例，系统讲解如何为 OpenAI o4-mini 的强化微调（RFT）设计评分器。内容涵盖数据集准备、基础模型基准测试、字符串评分器与模型评分器的实现、奖励黑客的识别与修复、RFT 训练启动及微调模型评估。重点展示评分器如何决定 RFT 的学习信号，并通过对比实验说明模型评分器在语义保真度上的优势与风险。
- --
内容来源与验证
- 内容等级：PREMIUM_A
- 质量等级：A
- 难度：高级
- 验证等级：V1
- 原文作者：theophile-oai
- 原文日期：2025-05-23
- 来源许可：MIT
- 来源版本：
39bb10aea4c0aeca05d8b95511ee3e765c83e000 - 查看原文
站内还有更多完整实战教程
加入会员，解锁全部教程、完整代码与持续更新内容。
加入会员