Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异，请参阅 中国的 Amazon Web Services 服务入门 (PDF)。

本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

为多回合强化学习创建资产

提示数据集格式

您的训练数据集是 A SageMaker I 在训练期间发送给您的代理的一组提示。每个提示都会启动一个部署：你的代理会处理它，在一个或多个回合中采取行动，然后返回奖励。数据集的质量和结构直接影响模型学习的内容。

支持的文件格式

数据集架构

提示栏检测

RFT 服务使用以下规则按顺序检测提示列：

请务必为提示列命名prompt，以免出现歧义。您可以添加其他列以用于自己的跟踪目的，但是 RFT 服务只能读取提示列。

如何使用提示

RFT 服务读取提示列，并按原样将字符串值直接传递给您的代理。它不解析、验证或转换内容。使用哪种格式完全取决于您的代理的期望——简单的代理可能采用纯文本，而更复杂的代理可能需要包含对话历史记录、工具配置和奖励规格的 JSON 字符串。

数据保护

由于 RFT 服务无需检查即可通过提示，因此您有责任保护敏感内容。请考虑在存储提示数据之前对其进行编码或加密，并在代理中处理解码或解密。

常用方法：

示例 1：简单 Q & A 数据集（纯文本）

适用于带有纯文本提示的简单训练任务。

用例：基本问题解答，简单说明如下

实木复合地板（Python）

import pyarrow as pa
import pyarrow.parquet as pq

data = {
    "prompt": [
        "What is 2 + 2?",
        "Explain the concept of machine learning.",
        "Write a Python function to reverse a string.",
        "What is the capital of France?",
        "How does photosynthesis work?",
    ]
}

table = pa.table(data)
pq.write_table(table, "training_data.parquet")

JSON Lines (.jsonl)

{"prompt": "What is 2 + 2?"}
{"prompt": "Explain the concept of machine learning."}
{"prompt": "Write a Python function to reverse a string."}

示例 2： Search/Reasoning 使用工具

适用于在模型推理期间需要访问外部工具（例如搜索引擎）的任务。

用例：网络搜索 Fact-based 问答、检索增强推理

结构：

prompt (column) = JSON string (recommend encoded/encrypted) containing:
├── data_source: Dataset origin identifier
├── prompt: Conversation messages [system, user]
├── ability: Task category (e.g., "fact-reasoning")
├── env_class: "search"
├── reward_spec: Ground truth answer for evaluation
└── extra_info: Tool configuration and metadata

示例行：

import pyarrow as pa
import pyarrow.parquet as pq
import json

task_data = {
    "data_source": "searchR1_nq",
    "prompt": [
        {
            "role": "system",
            "content": "You are a helpful and harmless assistant."
        },
        {
            "role": "user",
            "content": "Answer the given question. You must conduct reasoning inside <think> and </think> first every time you get new information. After reasoning, if you find you lack some knowledge, you can call a search engine by <search> query </search> and it will return the top searched results between <information> and </information>. You can search as many times as you want. If you find no further external knowledge needed, you can directly provide the answer inside <answer> and </answer>, without detailed illustrations. For example, <answer> Beijing </answer>. Question: total number of death row inmates in the us?"
        }
    ],
    "ability": "fact-reasoning",
    "env_class": "search",
    "reward_spec": {
        "ground_truth": {
            "target": [
                "2,718"
            ]
        },
        "style": "rule"
    },
    "extra_info": {
        "index": 0,
        "question": "total number of death row inmates in the us?",
        "split": "train",
        "need_tools_kwargs": true,
        "tools_kwargs": {
            "search": {
                "create_kwargs": {
                    "question": "total number of death row inmates in the us?",
                    "ground_truth": {
                        "target": [
                            "2,718"
                        ]
                    },
                    "data_source": "searchR1_nq"
                }
            }
        }
    }
}

# Recommend: encode or encrypt before storing
data = {"prompt": [json.dumps(task_data)]}
table = pa.table(data)
pq.write_table(table, "search_training_data.parquet")

示例 3：SQL 生成（Multi-Turn 使用复杂上下文）

适用于需要数据库架构、多步推理和 SQL 执行反馈的代码生成任务。

用例： Text-to-SQL，通过执行验证生成代码

结构：

prompt (column) = JSON string (recommend encoded/encrypted) containing:
├── input_seq: Human-readable task description
├── prompt: Conversation messages [system, user]
├── env_class: "text2sql"
├── reward_spec: Ground truth SQL and evaluation config
├── instance_id: Unique task identifier
├── schema: Database schema definition
├── question: Natural language question
└── extra_info: Additional metadata

示例行：

import pyarrow as pa
import pyarrow.parquet as pq
import json

task_data = {
    "input_seq": "Task Overview:\nYou are a data science expert. Below, you are provided with a database schema\nand a natural language question. Your task is to understand the schema and\ngenerate a valid SQL query to answer the question.\n\nDatabase Engine: SQLite\n\nDatabase Schema:\nCREATE TABLE countries (\n    country_id INTEGER PRIMARY KEY,\n    english_name TEXT,\n    population INTEGER\n);\n\nCREATE TABLE country_metrics (\n    metric_id INTEGER PRIMARY KEY,\n    country_id INTEGER,\n    metric_type TEXT,\n    year INTEGER,\n    value REAL\n);\n\nQuestion: List all countries with their current population and average\npopulation over the last five years.",
    "prompt": [
        {
            "role": "system",
            "content": "Task Overview:\nYou are a data science expert. Your task is to understand the schema and generate\na valid SQL query to answer the question within limited turns.\n\nInstructions:\n- Make sure you only output the information asked in the question.\n- Think through the steps before generating the final SQL query.\n\nFormat:\n- Conduct thinking inside <think>...</think> blocks.\n- You can use SQL tool written within <sql>your sql</sql> to explore or verify.\n- SQL tool output will be shown inside <observation>...</observation>.\n- Provide the final SQL query inside <solution>...</solution>."
        },
        {
            "role": "user",
            "content": "Database Schema:\nCREATE TABLE countries (\n    country_id INTEGER PRIMARY KEY,\n    english_name TEXT,\n    population INTEGER\n);\n\nCREATE TABLE country_metrics (\n    metric_id INTEGER PRIMARY KEY,\n    country_id INTEGER,\n    metric_type TEXT,\n    year INTEGER,\n    value REAL\n);\n\nQuestion: List all countries with their current population and average\npopulation over the last five years."
        }
    ],
    "env_class": "text2sql",
    "instance_id": "sql_task_001",
    "reward_spec": {
        "ground_truth": "SELECT c.english_name, c.population, AVG(m.value) as avg_pop\nFROM countries c\nJOIN country_metrics m ON c.country_id = m.country_id\nWHERE m.metric_type = 'Population' AND m.year > strftime('%Y', 'now') - 5\nGROUP BY c.country_id;",
        "style": "rule"
    },
    "schema": "CREATE TABLE countries (...); CREATE TABLE country_metrics (...);",
    "question": "List all countries with their current population...",
    "extra_info": {
        "split": "train",
        "difficulty": "medium"
    }
}

# Recommend: encode or encrypt before storing
data = {"prompt": [json.dumps(task_data)]}
table = pa.table(data)
pq.write_table(table, "sql_training_data.parquet")

最佳实践

数据集大小

最少示例数至少等于training_batch_size。建议将批量大小提高到 10 倍以上，以实现多样性。

即时质量

数据保护

奖励功能设计

奖励功能设计对于在复杂的多步代理系统中提供有效的学习信号至关重要。在为多回合 RL 设计奖励函数时，请考虑以下准则。

奖励设计流程

实际上，奖励函数将剧集的完整消息历史记录作为输入并返回两个输出：标量奖励（衡量轨迹质量的浮点分数，值越高表示性能越好）和用于记录、调试和监控的指标字典。

示例：搜索代理奖励功能

以下示例显示了使用搜索回答问题的代理的奖励功能。它演示了结果评估、格式塑造和答案正确性检查。

class TextAnswerReward:
    """Reward function to check text answer against gold answers.

    formula: format_coef * (correct_format - 1) + correct_answer
    """

    gold_answers: list[str]
    format_coef: float = 0.1

    async def __call__(self, history: list[Message]) -> tuple[float, dict[str, float]]:
        """Grade the completed episode by checking the final assistant message."""
        final_message = None
        for msg in reversed(history):
            if msg.get("role") == "assistant":
                final_message = msg
                break

        if final_message is None:
            return 0.0, {"format": 0.0, "correct": 0.0}

        content = get_text_content(final_message)

        correct_format = float(self._extract_answer(content) is not None)
        correct_answer = float(self._check_answer(content))

        reward = self.format_coef * (correct_format - 1) + correct_answer
        return reward, {"format": correct_format, "correct": correct_answer}

    def _extract_answer(self, text: str) -> str | None:
        if "Answer:" not in text:
            return None
        parts = text.split("Answer:")
        if len(parts) != 2:
            return None
        return parts[1].strip()

    def _check_answer(self, text: str) -> bool:
        model_answer = self._extract_answer(text)
        if model_answer is None or len(self.gold_answers) == 0:
            return False
        for gold in self.gold_answers:
            if normalize_answer(model_answer) == normalize_answer(gold):
                return True
        return False

此奖励功能包括以下关键设计选择：