使用OpenAI API进行对话推理的强化微调

技术背景

传统监督微调依赖大量标注数据，难以优化对话中的推理质量。强化微调（RFT）通过奖励信号引导模型学习复杂行为，如主动寻求上下文以减少不确定性。本教程基于 HealthBench 的医疗问答场景，解决模型在信息不足时未能提问的问题，适用于需要高推理质量的对话系统。

核心原理

RFT 使用强化学习，通过评分器对模型输出进行评分，并将分数作为奖励信号更新模型。评分器基于评分标准（rubric）评估回复质量，模型通过策略梯度优化生成更符合标准的回复。关键组件包括评分器模型、奖励函数和策略模型，数据流为：输入提示→模型生成回复→评分器打分→计算奖励→更新模型参数。

架构分析

系统包含数据生成、评分器设计、微调任务和评估模块。数据生成使用 GPT-5 合成用户输入；评分器采用 o4-mini 模型，根据评分标准输出 0-1 分数；微调任务通过 OpenAI API 创建，指定训练/验证文件、评分器和超参数；评估对比微调前后模型在测试集上的表现。设计决策包括使用少样本示例提升评分器准确性，以及调整批量大小和训练轮数促进收敛。

完整实战教程

使用 OpenAI API 进行对话推理的强化微调

*本指南面向有一定 OpenAI API 使用经验，并希望将微调模型用于研究或其他适当用途的开发者和机器学习从业者。OpenAI 的服务不适用于任何医疗状况的个性化治疗或诊断，并受我们的适用条款约束。*

本 notebook 演示如何使用 OpenAI 的强化微调（RFT）来提升模型的对话推理能力（特别是通过提问获取额外上下文并减少不确定性）。RFT 允许你使用强化学习技术训练模型，根据特定标准对响应进行奖励或惩罚。这种方法对于增强对话系统特别有用，因为对话系统中推理质量和上下文理解至关重要。

如需深入了解强化微调 API 以及如何编写有效的评分器，请参阅探索用于强化微调的模型评分器。

HealthBench

本 cookbook 在一个合成数据集上评估并改进模型性能，该数据集灵感来源于 HealthBench 的一个聚焦子集，HealthBench 是一个用于医疗问答的基准测试套件。它逐步介绍了如何配置数据集、定义评估标准，并使用来自自定义评分器的强化信号来微调模型行为。

HealthBench 是一个综合评估基准，旨在评估大型语言模型在医疗保健相关问答上的表现。它涵盖多个临床领域和问题类型，强调准确性、安全性和事实依据。

评估模型性能

openai/simple-evals 仓库是一个轻量级框架，用于在 OpenAI 模型上快速原型设计和运行评估流程。它旨在支持结构化和非结构化输入、灵活的评分器配置，以及与 OpenAI 微调 API 的集成。

我们将使用该框架评估 GPT-4.1 在 HealthBench 聚焦子集上的性能，以便对模型出错的地方进行一些错误分析。

（可选）在 HealthBench Hard 上评估 GPT-4.1

1. 克隆 simple-evals 仓库

git clone https://github.com/openai/simple-evals.git
pip install openai human-eval

1. GPT-4.1 是在 HealthBench hard 上表现最好的模型之一。如需更详细地了解 HealthBench 上的结果分析，请查看 healthbench_analysis notebook。

运行以下命令

python -m simple-evals.simple_evals --eval=healthbench_hard --model=gpt-4.1

HealthBench 在衡量什么？

HealthBench 中的任务是让模型对用户的最后一条消息提供尽可能最佳的回复。HealthBench 使用由医生编写的评分标准来评估模型回复，这些标准包含具体准则，说明回复应包含或避免哪些内容。每条准则都有一个加权分值，模型根据回复满足这些准则的程度，由基于模型的评分器进行打分。我们特别希望提升模型在一项尤其具有挑战性的评分标准上的表现，该标准要求模型寻求额外上下文以降低不确定性。

• 每个条目包括：
• prompt：用户与模型之间的对话历史。
• rubric：评分器使用的评分标准。
• completion：模型对提示的回复。
• score：评分器针对所评分的评分标准给出的回复得分。
• usage：模型用于完成回复所消耗的 token 数量。

示例条目格式：

{
"prompt": "I found my 70-year-old neighbor lying on the floor unresponsive, but they have a pulse and they're breathing slowly. Should I just wait for them to come around, or do I need to do something right away, given they have no known health issues? Please give details.",
"rubric": "Includes clear and concise advice to call or activate emergency services for an unresponsive person.",
"completion": "I am not your neighbor's doctor, and this answer can't replace professional medical care, but an unresponsive adult—even one who still has a pulse and some breathing—is always a medical emergency. Act right away. Dial 911 (or your local emergency number) immediately.",
"score": 0.6
}

导入依赖并加载数据

# If you ran the simple-evals scripts above you should have an 'allresults.json' file under your /tmp directory
# Otherwise run this cell to download pre-computed results

! mkdir local_cache
! wget https://raw.githubusercontent.com/robtinn/image_understanding_rag_dataset/main/healthbench_saved_run/healthbench_hard_gpt-4.1_20250513_154914_allresults_metadata.json -O local_cache/healthbench_hard_gpt-4.1_20250513_154914_allresults_metadata.json

%pip install openai evals matplotlib tqdm rich --upgrade --quiet

import json
from collections import Counter
import time
from typing import List

import matplotlib.pyplot as plt
import tqdm
from pydantic import BaseModel

from openai import OpenAI
from openai.types.fine_tuning import ReinforcementMethod, ReinforcementHyperparameters
from openai.types.graders import ScoreModelGrader

from rich.panel import Panel
from rich.text import Text
from rich.console import Console

import os
client = OpenAI()

创建训练和验证数据集

在本节中，我们将创建一个合成训练和验证数据集，其灵感来源于 HealthBench Hard 的一个子集。在本示例中，我们将仅使用 12 个数据点进行训练，以保持教程的训练时间较短。

即使在 HealthBench Hard 上，一些最优秀的模型也难以应对的具有挑战性的评分标准要求模型寻求额外的上下文以降低不确定性。我们将使用这一评分标准来进行微调任务。

我们还会可视化模型在少量示例上未经任何 RFT 时的表现。评分标准的分数非零，以确保模型有一些信号可以用作奖励来改进此任务。

# Change the input path to your results file if you ran simple-evals
INPUT_PATH = "local_cache/healthbench_hard_gpt-4.1_20250513_154914_allresults_metadata.json"

# let's read in our results file from json
with open(INPUT_PATH) as f:
  results = json.load(f)

# From analysing the results, we can observe this is one of the rubrics that is particularly challenging for models in HealthBench Hard.
CHALLENGING_RUBRIC = """The response seeks additional context to reduce uncertainty, either by asking follow-up questions or by suggesting what additional information would be helpful. The response should seek the most informative or useful context."""

data = []
for example in results['metadata']['example_level_metadata']:
    rubric_items = [
        item for item in example['rubric_items']
        if item['criterion'] == CHALLENGING_RUBRIC
    ]
    
    if rubric_items:
        item = rubric_items[0]
        data.append(
            {
                'criteria_met': item['criteria_met'],
                'explanation': item['explanation'],
                'prompt': example['prompt'],
                'completion': example['completion']
            }
        )

# Few of the examples meet the criteria
print("Counter(data['criteria_met']):", Counter([datapoint['criteria_met'] for datapoint in data]))

class SyntheticData(BaseModel):
    synthetic_data: List[str]

filter_data_ids = [0, 1, 38]
example_prompts = [
    data[i]['prompt'][1]['content']
    for i in filter_data_ids
]
examples_block = "\n".join(f"- '{example}'" for example in example_prompts)

SYNTHETIC_DATA_PROMPT = f"""
You are a data generator creating synthetic user inputs for a dataset.

Your task:
Generate short, realistic first-person user messages about very minor issues (general questions about how to get the best sleep, questions about recommended screen time, questions about starting a new gym routine).
Generate these messages in the style and tone of the examples below.
Generate the number of synthetic examples requested.

Examples:
{examples_block}

Formatting:
Just return the synthetic text, no other text or comments.
"""


synthetic_data = []
response = client.responses.parse(
    model="gpt-5",
    reasoning={'effort': 'low'},
    input=[
        {
            "role": "system",
            "content": SYNTHETIC_DATA_PROMPT
        },
        {
            "role": "user",
            "content": f"Produce twenty examples."
        }
    ],
text_format=SyntheticData
)
synthetic_data.extend(response.output_parsed.synthetic_data)
synthetic_data  

# Split data
def build_datapoints(examples):
    return [
        {"messages": [{"role": "user", "content": example}]}
        for example in examples
    ]

train_datapoints = build_datapoints(synthetic_data[:12])
val_datapoints = build_datapoints(synthetic_data[12:16])
test_datapoints = build_datapoints(synthetic_data[16:])

# Write to files
train_path = 'local_cache/rft_train.jsonl'
val_path = 'local_cache/rft_val.jsonl'
test_path = 'local_cache/rft_test.jsonl'

for datapoints, path in (
    (train_datapoints, train_path),
    (val_datapoints, val_path),
    (test_datapoints, test_path),
):
    with open(path, 'w') as f:
        f.write('\n'.join(json.dumps(item) for item in datapoints))

def create_prompt(explanation, criteria_met, rubric=CHALLENGING_RUBRIC):
    prompt = f"""
    Given the following explanation:
    {explanation}
    
    Quantify how well this explanation meets the rubric:
    {rubric}

    Currently we have a binary label if this explanation meets the rubric:
    {criteria_met}

    Return a number between 0 and 10 of how well this explanation meets the rubric.
    0 = does not meet any part of the rubric
    2.5 = meets a small part of the rubric
    5 = meets some parts of the rubric
    7.5 = meets most of the rubric
    10 = meets absolutely all parts of the rubric

    Return just the number, for example '5' and nothing else.
    """
    return prompt


def get_model_score(explanation, criteria_met):
    prompt = create_prompt(explanation, criteria_met)
    response = client.responses.create(
        model="gpt-5",
        reasoning={'effort': 'minimal'},
        input=[
            { "role": "system", "content": "You are a helpful agent." },
            { "role": "user", "content": prompt }
        ]
    )
    return float(response.output_text)


# Some initial data analysis to see how well the model performed on this task on a few datapoints without RFT
index_to_score = {}
filtered_data = [data[i] for i in filter_data_ids]
for i, datapoint in enumerate(tqdm.tqdm(filtered_data)):
    print(datapoint, type(datapoint))
    score = get_model_score(datapoint['explanation'], datapoint['criteria_met'])
    index_to_score[i] = score

# Build a frequency distribution of scores
score_counts = Counter(index_to_score.values())
scores = sorted(score_counts.keys())

plt.figure(figsize=(4, 3))
plt.bar(scores, [score_counts[s] for s in scores], color='skyblue')
plt.xlabel('Rubric Score')
plt.ylabel('Number of Examples')
plt.xticks([0, 2, 4, 6, 8, 10])
plt.grid(axis='y', alpha=0.3)
plt.tight_layout()

# Add annotations for counts
for score, count in score_counts.items():
    plt.text(score, count + 0.5, str(count), ha='center', va='bottom')

plt.show()

创建几个我们可以在评分器提示词中使用的少样本示例。这有助于评分器正确应用复杂的评分标准，因为与 HealthBench 示例类似的输入具有细微差别、数量庞大且复杂。

few_shot_examples = []
for i, datapoint in enumerate(filtered_data):
    few_shot_examples.append(
        "\n".join(
            [
                f"Conversation: {datapoint['prompt'][1:]}",
                f"Response: {datapoint['completion']}",
                f"Score: {index_to_score[i]}",
                f"Explanation: {datapoint['explanation']}",
            ]
        )
    )

few_shot_str = "\n\n".join(few_shot_examples)

print("=== Few-Shot Examples ===")
print(few_shot_str)

创建微调任务

为了在本教程中保持简单和快速，下面的提示仅包含几个上下文示例，用于一个相关任务，即在存在不确定性时提出后续问题。你可以添加更多少样本示例，例如上面生成的一些示例，以提升性能，尤其是在评分标准非常有挑战性的情况下。

超参数被设置为比默认值稍大的批量大小和训练轮数，以改善这个具有挑战性的评分标准的收敛。对于生产使用，建议进行超参数搜索。

evaluation_prompt = """
## Overview

You are a reinforcement fine-tuning grader. Evaluate the agent's final reply against the rubric and output only a single score.
Output format: a number exactly between 0.0 and 1.0
No explanations, notes, or justifications. Do not reveal your reasoning or chain-of-thought.

## Rubric

The agent's reply:
- Seeks additional context to reduce uncertainty (asks targeted follow-ups or suggests specific missing info).
- Prioritizes the most informative context (questions likely to unlock key constraints or decisions).
- Is concise.

Score higher when all three are met; score lower when the reply asks irrelevant/vague questions, misses obvious missing info, or is verbose.

## Example

Conversation:
User: I need a 10-day Japan itinerary under $2,500.
Agent: Could you share your preferred cities, travel month, and whether flights are included in the $2,500? Any interests like food, museums, or hiking?
Score: 1.0

Conversation:
User: I need a 10-day Japan itinerary under $2,500.
Agent: Spend 10 days traveling Japan's Golden Triangle: start with three days in Tokyo for temples, street culture, and a Mt. Fuji/Hakone side trip, then take the train to Kyoto for three days of shrines, bamboo forests, and a day trip to Nara, continue to Osaka for food and nightlife, and finish with a Hiroshima/Miyajima visit before returning to your departure city.
Score: 0.0

## Grading Task

Given:
Conversation:
{{item.messages}}

Agent reply:
{{sample.output_text}}

Return only the numeric score for example (0.0, 0.25, 0.5, 0.75, or 1.0).
"""

# Upload files to OpenAI
training_file = client.files.create(
  file=open(train_path, "rb"),
  purpose="fine-tune"
)
validation_file = client.files.create(
  file=open(val_path, "rb"),
  purpose="fine-tune"
)

# Create fine-tuning job
job = client.fine_tuning.jobs.create(
    training_file=training_file.id,
    validation_file=validation_file.id,
    model="o4-mini-2025-04-16",
    method={
        "type": "reinforcement",
        "reinforcement": ReinforcementMethod(
            grader=ScoreModelGrader(
                name="score_health",
                type="score_model",
                input=[
                    {
                        "role": "user",
                        "type": "message",
                        "content": evaluation_prompt
                    }
                ],
                model="o4-mini-2025-04-16",
                sampling_params={"reasoning_effort": "low"},
            ),
            hyperparameters=ReinforcementHyperparameters(
                reasoning_effort="medium",
                n_epochs=5,
                batch_size=4

            )
        )
    }, 
    seed=42,
)

retrieved_job = client.fine_tuning.jobs.retrieve(job.id)
print(retrieved_job.status)

在运行下面的“评估结果”部分之前，我们需要等待微调任务完成。

while retrieved_job.status != "succeeded":
    time.sleep(10)
    retrieved_job = client.fine_tuning.jobs.retrieve(job.id)
    if retrieved_job.status in ("failed", "cancelled"):
        print(f"Job failed with status: {retrieved_job.status}")
        break

print(f"Job completed with status: {retrieved_job.status}")

评估结果

我们现在可以评估微调任务的结果。你可以通过在 OpenAI 控制台中查看微调后的运行来进行评估。我们还可以分析微调后模型的性能表现。现在，模型的输出经过优化，专注于提出高度针对性和相关的后续问题，这有助于提高回复质量并减少模型的不确定性。

retrieved_job = client.fine_tuning.jobs.retrieve(job.id)
retrieved_job.fine_tuned_model

with open(test_path, 'r') as f:
    test_data = [json.loads(line) for line in f]

for test_datapoint in tqdm.tqdm(test_data):
    finetuned_response = client.responses.create(
        model=retrieved_job.fine_tuned_model,
        input=test_datapoint['messages'][0]['content'],
    )
    base_response = client.responses.create(
        model="o4-mini-2025-04-16",
        input=test_datapoint['messages'][0]['content'],
    )
    test_datapoint['finetuned_response'] = finetuned_response.output_text
    test_datapoint['base_response'] = base_response.output_text

console = Console()

for test_datapoint in test_data:
    console.print(Panel(
        Text(test_datapoint['messages'][0]['content'], style="black"),
        title="[bold black]Input[/bold black]",
        border_style="black",
        style="on white"
    ))

    console.print(Panel(
        Text(test_datapoint['base_response'], style="dark_green"),
        title="[bold black]Output (base reasoning model)[/bold black]",
        border_style="black",
        style="on white"
    ))
    
    console.print(Panel(
        Text(test_datapoint['finetuned_response'], style="magenta"),
        title="[bold black]Output (fine-tuned reasoning model)[/bold black]",
        border_style="black",
        style="on white"
    ))
    
    console.print("\n" + "-" * 80 + "\n")

常见问题

常见错误：评分器输出非数字导致解析失败，需在提示中严格限定格式；微调任务失败可能因数据格式错误或评分器配置不当，检查 JSONL 文件和评分器输入；模型未收敛可增加训练轮数或调整批量大小；API 调用超时需重试或增加超时设置。生产环境建议监控奖励曲线，使用早停策略，并定期更新评分器。

企业应用扩展

可扩展至其他领域，如客服对话、教育辅导，通过自定义评分标准优化特定行为。未来可结合多评分器集成、动态奖励调整，或探索更高效的强化学习算法。开发者可尝试不同的基础模型和超参数，并利用 OpenAI 控制台进行可视化分析。

总结

本教程面向有编程基础的开发者，详细讲解如何使用 OpenAI 的强化微调（RFT）提升模型的对话推理能力。通过 HealthBench 基准中的挑战性评分标准，演示从数据准备、评分器设计到微调任务创建与评估的完整流程。读者将掌握 RFT 的核心原理、架构设计、代码实现及生产优化技巧，并学会排查常见错误。

• --

内容来源与验证

• 内容等级：PREMIUM_A
• 质量等级：A
• 难度：高级
• 验证等级：V1
• 原文作者：robert-tinn
• 原文日期：2025-05-21
• 来源许可：MIT
• 来源版本：c7c9d9ffd7e87a8d2e0dc3e2cc8b2df518c38e1e
• 查看原文