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

建立資產以進行多迴轉強化學習

提示資料集格式

您的訓練資料集是 SageMaker AI 在訓練期間傳送給客服人員的提示集合。每個提示都會啟動一個推展：您的代理程式會處理它、跨一或多個回合採取動作，並傳回獎勵。資料集的品質和結構會直接影響模型學習的內容。

支援檔案格式

資料集結構描述

提示欄偵測

RFT 服務會使用下列規則依序偵測提示欄：

一律為您的提示欄命名prompt，以避免模棱兩可的情況。您可以為自己的追蹤目的包含其他資料欄，但 RFT 服務只會讀取提示資料欄。

如何使用提示

RFT 服務會讀取提示欄，並依原狀將字串值直接傳遞給您的代理程式。它不會剖析、驗證或轉換內容。要使用的格式完全取決於代理程式的預期 — 簡單代理程式可能會採用純文字，而更複雜的代理程式可能會預期包含對話歷史記錄、工具組態和獎勵規格的 JSON 字串。

資料保護

由於 RFT 服務會在未經檢查的情況下傳遞提示，因此您必須負責保護敏感內容。在儲存提示資料之前，請考慮對其進行編碼或加密，並在您的代理程式中處理解碼或解密。

常見方法：

範例 1：簡單問答資料集 （純文字）

對於使用純文字提示的直接訓練任務。

使用案例：基本問題回答、簡單的說明

Parquet (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：使用工具使用搜尋/重新排序

對於模型推理期間需要外部工具存取的任務 （例如搜尋引擎）。

使用案例：以事實為基礎的問答搭配 Web 搜尋、擷取擴增推理

結構：

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 產生 （具有複雜內容的多轉）

對於需要資料庫結構描述、多步驟推理和 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

此獎勵函數包含下列關鍵設計選擇：