As traduções são geradas por tradução automática. Em caso de conflito entre o conteúdo da tradução e da versão original em inglês, a versão em inglês prevalecerá.

Criação de ativos para aprendizado por reforço em vários turnos

Formato de conjunto de dados imediato

Seu conjunto de dados de treinamento é uma coleção de solicitações que a SageMaker IA envia ao seu agente durante o treinamento. Cada solicitação inicia uma implementação: seu agente a processa, executa ações em um ou mais turnos e retorna uma recompensa. A qualidade e a estrutura do seu conjunto de dados afetam diretamente o que o modelo aprende.

Formatos de arquivo compatíveis

Esquema do conjunto de dados

Detecção rápida de colunas

O serviço RFT detecta a coluna de prompt usando as seguintes regras, na ordem:

Sempre nomeie sua coluna de prompt prompt para evitar ambigüidades. Você pode incluir colunas adicionais para seus próprios fins de rastreamento, mas somente a coluna de prompt é lida pelo serviço RFT.

Como os prompts são usados

O serviço RFT lê a coluna de prompt e passa o valor da string diretamente para seu agente no estado em que se encontra. Ele não analisa, valida ou transforma o conteúdo. O formato a ser usado depende inteiramente do que seu agente espera: um agente simples pode usar texto sem formatação, enquanto um mais sofisticado pode esperar uma string JSON contendo histórico de conversas, configuração de ferramentas e especificações de recompensa.

Proteção de dados

Como o serviço RFT passa as solicitações sem inspeção, você é responsável por proteger o conteúdo confidencial. Considere codificar ou criptografar dados imediatos antes de armazená-los e manipular a decodificação ou a descriptografia em seu agente.

Abordagens comuns:

Exemplo 1: Conjunto de dados & QA simples (texto sem formatação)

Para tarefas de treinamento simples com instruções em texto simples.

Caso de uso: resposta básica a perguntas, instruções simples a seguir

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")

Linhas JSON (.jsonl)

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

Exemplo 2: Search/Reasoning com o uso de ferramentas

Para tarefas que exigem acesso a ferramentas externas (por exemplo, mecanismos de pesquisa) durante o raciocínio do modelo.

Caso de uso: Fact-based perguntas e respostas com pesquisa na web, raciocínio aumentado para recuperação

Estrutura:

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

Exemplo de linha:

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")

Exemplo 3: Geração de SQL (Multi-Turn com contexto complexo)

Para tarefas de geração de código que exigem esquemas de banco de dados, raciocínio em várias etapas e feedback de execução de SQL.

Caso de uso: Text-to-SQL geração de código com verificação de execução

Estrutura:

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

Exemplo de linha:

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")

Práticas recomendadas

Tamanho do conjunto de dados

Exemplos mínimos pelo menos iguais training_batch_size a. 10x+ do tamanho do seu lote para diversidade é recomendado.

Qualidade rápida

Proteção de dados

Design da função de recompensa

O design da função de recompensa é fundamental para fornecer sinais de aprendizado eficazes em sistemas de agentes complexos de várias etapas. Ao projetar funções de recompensa para RL de vários turnos, considere as diretrizes a seguir.

Processo de design de recompensas

Na prática, uma função de recompensa usa o histórico completo da mensagem de um episódio como entrada e retorna duas saídas: uma recompensa escalar (uma pontuação de ponto flutuante que mede a qualidade da trajetória, com valores mais altos indicando melhor desempenho) e um dicionário de métricas para registro, depuração e monitoramento.

Exemplo: função de recompensa do agente de pesquisa

O exemplo a seguir mostra uma função de recompensa para um agente que responde perguntas usando a pesquisa. Ele demonstra avaliação de resultados, modelagem de formato e verificação da exatidão das respostas.

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

Essa função de recompensa inclui as seguintes opções principais de design: