기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다.

멀티턴 강화 학습을 위한 자산 생성

프롬프트 데이터 세트 형식

훈련 데이터 세트는 SageMaker AI가 훈련 중에 에이전트에게 보내는 프롬프트 모음입니다. 각 프롬프트는 하나의 롤아웃을 시작합니다. 즉, 에이전트가 이를 처리하고, 하나 이상의 차례로 작업을 수행하고, 보상을 반환합니다. 데이터 세트의 품질과 구조는 모델이 학습하는 내용에 직접적인 영향을 미칩니다.

지원되는 파일 형식

데이터 세트 스키마

프롬프트 열 감지

RFT 서비스는 다음 규칙을 순서대로 사용하여 프롬프트 열을 감지합니다.

모호하지 prompt 않도록 항상 프롬프트 열의 이름을 지정합니다. 자체 추적 목적으로 추가 열을 포함할 수 있지만 프롬프트 열만 RFT 서비스에서 읽습니다.

프롬프트 사용 방법

RFT 서비스는 프롬프트 열을 읽고 문자열 값을 에이전트에 있는 그대로 직접 전달합니다. 콘텐츠를 구문 분석, 검증 또는 변환하지 않습니다. 사용할 형식은 전적으로 에이전트가 기대하는 것에 따라 달라집니다. 간단한 에이전트는 일반 텍스트를 사용할 수 있고, 보다 정교한 에이전트는 대화 기록, 도구 구성 및 보상 사양이 포함된 JSON 문자열을 기대할 수 있습니다.

데이터 보호

RFT 서비스는 검사 없이 프롬프트를 전달하므로 민감한 콘텐츠를 보호할 책임은 사용자에게 있습니다. 프롬프트 데이터를 저장하기 전에 인코딩 또는 암호화하고 에이전트에서 디코딩 또는 복호화를 처리하는 것이 좋습니다.

일반적인 접근 방식:

예제 1: 단순 Q&A 데이터 세트(일반 텍스트)

일반 텍스트 프롬프트가 있는 간단한 훈련 작업용입니다.

사용 사례: 기본 질문 답변, 간단한 지침 준수

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 라인(.jsonl)

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

예제 2: 도구 사용을 통한 검색/이유

모델 추론 중에 외부 도구 액세스(예: 검색 엔진)가 필요한 작업의 경우.

사용 사례: 웹 검색, 검색 증강 추론이 포함된 사실 기반 Q&A

구조:

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

이 보상 함수에는 다음과 같은 주요 설계 선택 사항이 포함됩니다.