本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。 # 为多回合强化学习创建资产 ## 提示数据集格式 您的训练数据集是 A SageMaker I 在训练期间发送给您的代理的一组提示。每个提示都会启动一个部署：你的代理会处理它，在一个或多个回合中采取行动，然后返回奖励。数据集的质量和结构直接影响模型学习的内容。 ### 支持的文件格式 | Format | 扩展程序 | 注意 | | --- | --- | --- | | Apache Parquet | .parquet | 推荐用于大型数据集 — 高效存储和快速加载 | | JSON 行 | .jsonl | 每行一个 JSON 对象 — 易于创建且人类可读 | | JSON | .json | JSON 对象数组 | | CSV | .csv | Comma-separated 带有标题行的值 | ### 数据集架构 **提示栏检测** RFT 服务使用以下规则按顺序检测提示列： + 如果`prompt`存在名为的列，则使用该列。 + 否则，将使用数据集中的第一列。 请务必为提示列命名`prompt`，以免出现歧义。您可以添加其他列以用于自己的跟踪目的，但是 RFT 服务只能读取提示列。 **如何使用提示** RFT 服务读取提示列，并按原样将字符串值直接传递给您的代理。它不解析、验证或转换内容。使用哪种格式完全取决于您的代理的期望——简单的代理可能采用纯文本，而更复杂的代理可能需要包含对话历史记录、工具配置和奖励规格的 JSON 字符串。 **数据保护** 由于 RFT 服务无需检查即可通过提示，因此您有责任保护敏感内容。请考虑在存储提示数据之前对其进行编码或加密，并在代理中处理解码或解密。 常用方法： + Base64 编码 — 对非敏感数据进行简单混淆处理 + 加密 — 敏感或专有数据（例如，使用由您的代理管理密钥的 AES） ## 示例 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 and first every time you get new information. After reasoning, if you find you lack some knowledge, you can call a search engine by query and it will return the top searched results between and . You can search as many times as you want. If you find no further external knowledge needed, you can directly provide the answer inside and , without detailed illustrations. For example, Beijing . 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 ... blocks.\n- You can use SQL tool written within your sql to explore or verify.\n- SQL tool output will be shown inside ....\n- Provide the final SQL query inside ...." }, { "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 倍以上，以实现多样性。 **即时质量** + **完整上下文**：包括模型生成有用响应所需的所有信息 + **结构一致**：在所有提示中保持一致的格式 + **避免重复**：独特的提示可提供更好的训练信号 + **明确说明**：对于工具使用任务，请提供明确的格式说明 **数据保护** + **对提示内容进行编码或加密**以保护敏感数据 + 在您的部署服务器上安全地@@ **管理解密密钥** + RFT 服务无需检查即可通过提示，因此保护是您的责任 ## 奖励功能设计 奖励功能设计对于在复杂的多步代理系统中提供有效的学习信号至关重要。在为多回合 RL 设计奖励函数时，请考虑以下准则。 + **从基于结果的奖励开始。**首先对最终结果进行评分，以建立干净可靠的基准，然后再添加中间奖励或奖励形成。 + **考虑连续奖励而不是二进制奖励。**持续的奖励可以提供更清晰的部分积分信号，但很容易上手。当部分积分难以定义或需要一个干净的基线时，首选二元奖励。 + **谨慎使用塑造奖励。**塑造奖励可以指导学习，但应谨慎使用，因为过于强大或未对齐的塑形可能会教会捷径。 + **防范奖励黑客攻击。**使奖励难以利用，并验证模型是否在解决实际任务，而不是玩计分规则。 + **训练前进行验证。**在训练之前，在真实轨迹上测试奖励功能，以捕捉错误、漏洞或误导性信号。 + **监控行为指标，而不仅仅是奖励。**跟踪诸如完成率、转弯次数、工具使用情况和过度拟合差距等指标，以确保模型以预期的方式得到改进。 ### 奖励设计流程 1. 定义成功是什么样子，并确定是否可以自动评分。 1. 评估基础模型以确定基线成功率。 1. 设计奖励等级：成功奖励为正，失败为零奖励，堕落行为为负奖励。 1. 明确处理边缘情况，包括超时、环境错误、格式错误的输出和空响应。 1. 检查每个奖励部分是否存在潜在的奖励黑客攻击。 1. 训练前在真实轨迹上进行验证。 1. 在训练期间同时监控行为指标。 1. 根据初始结果进行迭代。 实际上，奖励函数将剧集的完整消息历史记录作为输入并返回两个输出：标量奖励（衡量轨迹质量的浮点分数，值越高表示性能越好）和用于记录、调试和监控的指标字典。 ### 示例：搜索代理奖励功能 以下示例显示了使用搜索回答问题的代理的奖励功能。它演示了结果评估、格式塑造和答案正确性检查。 ``` 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 ``` 此奖励功能包括以下关键设计选择： + **正确性占主导地位。**无论格式如何，正确答案的分数总是高于错误答案。 + **格式是一个小的整形信号。**格式系数（0.1）是结果奖励的10％，足够小，以至于模型无法仅从格式合规性中获利，但足够大，可以将其引向可解析的输出。 + **错误的格式和错误的答案会受到轻微的惩罚。**-0.1 的分数与完全非结构化的输出相去甚远，而不会压倒学习信号。 + **没有答案被视为不正确且格式不正确。**如果模型从未生成辅助消息，则该函数返回 0.0，将其与当前但格式错误的响应的主动惩罚 -0.1 区分开来。