←
AITYBlog
MLOps

Otimizando LLMs com Nova Forge SDK: Um Estudo de Caso

Introdução

A jornada de customização de modelos de linguagem grandes (LLMs) e a transição entre plataformas tem sido tradicionalmente complexa, exigindo expertise técnica, configuração de infraestrutura e um investimento considerável de tempo. Essa desconexão entre o potencial e as aplicações práticas é precisamente o que o Nova Forge SDK visa resolver.

O Nova Forge SDK torna a customização de LLMs acessível, capacitando equipes a explorar todo o potencial dos modelos de linguagem sem os desafios de gerenciamento de dependências, seleção de imagens e configuração de "receitas". Enxergamos a customização como um continuum na escala de adaptação, por isso, o SDK suporta todas as opções, desde adaptações baseadas em Amazon SageMaker AI até a customização profunda usando os recursos do Amazon Nova Forge.

Neste artigo, detalharemos o processo de uso do Nova Forge SDK para treinar um modelo Amazon Nova usando Amazon SageMaker AI Training Jobs. Avaliaremos o desempenho baseline do modelo em um dataset do Stack Overflow, aplicaremos o Supervised Fine-Tuning (SFT) para refinar seu desempenho e, em seguida, o Reinforcement Fine-Tuning (RFT) no modelo customizado para melhorar ainda mais a qualidade das respostas. Após cada tipo de fine-tuning, avaliaremos o modelo para demonstrar suas melhorias ao longo do processo de customização e, finalmente, implantaremos o modelo customizado em um endpoint de inferência do Amazon SageMaker AI.

Caso de Estudo: Classificação Automática de Perguntas do Stack Overflow

Vamos entender os benefícios do Nova Forge SDK através de um cenário real: a classificação automática de perguntas do Stack Overflow em três categorias bem definidas (HQ, LQ_EDIT, LQ_CLOSE). O Stack Overflow possui milhares de perguntas com grande variação de qualidade. Classificar automaticamente a qualidade ajuda os moderadores a priorizar seus esforços e guiar os usuários a melhorarem suas postagens.

Esta solução demonstra como usar o Amazon Nova Forge SDK para construir um classificador de qualidade automatizado que pode distinguir entre: * HQ (High Quality): Postagens bem escritas sem edições. * LQ_EDIT (Low Quality – Edited): Postagens com pontuações negativas e múltiplas edições da comunidade, mas que permanecem abertas. * LQ_CLOSE (Low Quality – Closed): Postagens fechadas pela comunidade sem edições.

Para nossos experimentos, usamos o dataset Stack Overflow Question Quality, contendo 60.000 perguntas de 2016-2020. Amostramos aleatoriamente 4.700 perguntas e as dividimos da seguinte forma:

O experimento consiste em quatro etapas principais: avaliação baseline, Supervised Fine-Tuning (SFT), Reinforcement Fine-Tuning (RFT) e implantação. Cada etapa de fine-tuning se baseia na anterior, com melhorias mensuráveis a cada passo.

Utilizamos o seguinte system prompt comum para todos os datasets:

This is a stack overflow question from 2016-2020 and it can be classified into three categories:
* HQ: High-quality posts without a single edit.
* LQ_EDIT: Low-quality posts with a negative score, and multiple community edits. However, they remain open after those changes.
* LQ_CLOSE: Low-quality posts that were closed by the community without a single edit.
You are a technical assistant who will classify the question from users into any of above three categories. Respond with only the category name: HQ, LQ_EDIT, or LQ_CLOSE.
**Do not add any explanation, just give the category as output**.

Etapa 1: Estabelecer Desempenho Baseline

Antes do fine-tuning, estabelecemos um baseline avaliando o modelo pré-treinado Nova 2.0 em nosso conjunto de avaliação. Isso nos fornece uma base concreta para medir melhorias futuras. A avaliação baseline é crucial para entender as capacidades iniciais do modelo, identificar lacunas de desempenho, definir metas de melhoria mensuráveis e validar a necessidade de fine-tuning.

Para começar, instale o SDK:

pip install nova-forge-sdk

O Amazon Nova Forge SDK oferece utilitários poderosos de carregamento de dados que lidam automaticamente com validação e transformação. Começamos carregando nosso dataset de avaliação e transformando-o para o formato esperado pelos modelos Nova:

# General Configuration
MODEL = Model.NOVA_LITE_2
INSTANCE_TYPE = 'ml.p5.48xlarge'
EXECUTION_ROLE = '<YOUR_EXECUTION_ROLE_ARN>'
TRAIN_INSTANCE_COUNT = 4
EVAL_INSTANCE_COUNT = 1
S3_BUCKET = '<YOUR_S3_BUCKET>'
S3_PREFIX = 'stack-overflow'
EVAL_DATA = './eval.csv'

# Load data
loader = CSVDatasetLoader(
    query='Body',  # Coluna de texto da pergunta
    response='Y',  # Coluna do rótulo de classificação (HQ, LQ_EDIT, LQ_CLOSE)
    system='system' # Coluna do system prompt
)
loader.load(EVAL_DATA)

# Transformar para o formato Nova
loader.transform(method=TrainingMethod.EVALUATION, model=MODEL)

# Validar formato dos dados
loader.validate(method=TrainingMethod.EVALUATION, model=MODEL)

# Salvar no S3
eval_s3_uri = loader.save_data(
    f"s3://{S3_BUCKET}/{S3_PREFIX}/data/eval.jsonl"
)

Com os dados preparados, inicializamos o SMTJRuntimeManager para configurar a infraestrutura de tempo de execução e, em seguida, um objeto NovaModelCustomizer. Chamamos baseline_customizer.evaluate() para iniciar o job de avaliação baseline:

# Configurar infraestrutura de tempo de execução
runtime_manager = SMTJRuntimeManager(
    instance_type=INSTANCE_TYPE,
    instance_count=EVAL_INSTANCE_COUNT,
    execution_role=EXECUTION_ROLE
)

# Criar avaliador baseline
baseline_customizer = NovaModelCustomizer(
    model=MODEL,
    method=TrainingMethod.EVALUATION,
    infra=runtime_manager,
    data_s3_path=eval_s3_uri,
    output_s3_path=f"s3://{S3_BUCKET}/{S3_PREFIX}/baseline-eval"
)

# Rodar avaliação (GEN_QA para classificação)
baseline_result = baseline_customizer.evaluate(
    job_name="blogpost-baseline",
    eval_task=EvaluationTask.GEN_QA
)

Para tarefas de classificação, usamos a tarefa de avaliação GEN_QA, que trata a classificação como uma tarefa generativa. A métrica exact_match do GEN_QA corresponde diretamente à acurácia da classificação.

Os resultados baseline mostraram que o modelo base alcançou apenas 13.0% de acurácia exact-match nesta tarefa de classificação de 3 classes, enquanto um chute aleatório renderia 33.3%. Isso demonstra claramente a necessidade de fine-tuning.

Analisando as falhas do baseline, observamos que todas as respostas incluíam explicações verbosas e raciocínio, apesar do system prompt solicitar apenas o nome da categoria. Além disso, o LLM tinha um claro viés em classificar mensagens como "High Quality" independentemente de sua classificação real. Mesmo com uma métrica permissiva, que ignora a verbosidade, obtivemos apenas 52.2% de acurácia de classificação.

def classification_accuracy(samples):
    """Extrai a classe predita via correspondência de substring e calcula a acurácia."""
    correct, total, no_pred = 0, 0, 0
    for s in samples:
        gold = s["gold"].strip().upper()
        pred_raw = s["inference"][0] if isinstance(s["inference"], list) else s["inference"]
        pred_cat = extract_category(pred_raw) # assume extract_category existe e funciona
        if pred_cat is None:
            no_pred += 1
            continue
        total += 1
        if pred_cat == gold:
            correct += 1
    acc = correct / total if total else 0
    print(f"Classification Accuracy: {correct}/{total} ({acc*100:.1f}%)")
    print(f" No valid prediction: {no_pred}/{total + no_pred}")
    return acc

print("???? Baseline Classification Accuracy (extracted class labels):")
# baseline_accuracy = classification_accuracy(baseline_samples) # baseline_samples viria do output do job

Com base nesses resultados de falhas de instrução e viés de classificação, aplicaremos o Supervised Fine-Tuning (SFT) e, em seguida, o Reinforcement Learning (RL).

Etapa 2: Supervised Fine-Tuning (SFT)

Com o baseline estabelecido e a análise de falhas concluída, utilizaremos o Supervised Fine-Tuning para melhorar o desempenho, focando inicialmente em um Parametric Efficient Fine-Tuning (PEFT) para sinais iniciais de aprendizado.

A preparação de dados para SFT é simplificada pelo Nova Forge SDK, que oferece funções auxiliares e validações integradas. Carregamos nossos dados CSV de treinamento e os transformamos para o formato necessário, similar ao processo de avaliação:

# Carregamento e transformação do dataset para SFT
# loader = CSVDatasetLoader(...)
# loader.load(TRAIN_DATA)
# loader.transform(method=TrainingMethod.SFT_LORA, model=MODEL)
# train_path = loader.save_data(f"s3://{S3_BUCKET}/{S3_PREFIX}/data/train.jsonl")

Com os dados preparados e enviados ao Amazon S3, iniciamos o job de SFT. O Nova Forge SDK simplifica a especificação da infraestrutura e o lançamento dos jobs de treinamento, eliminando a necessidade de se preocupar com configurações de "receita" ou formatos de API. Para SFT, continuamos usando Amazon SageMaker Training Jobs com 4 instâncias ml.p5.48xlarge.

customizer = NovaModelCustomizer(
    model=MODEL,
    method=TrainingMethod.SFT_LORA,
    infra=runtime_manager, # Reutilizar runtime_manager configurado
    data_s3_path=train_path,
    output_s3_path=f"s3://{S3_BUCKET}/{S3_PREFIX}/sft-output"
)

training_config = {
    "lr": 5e-6,           # Learning rate
    "warmup_steps": 17,   # Rampa gradual de LR
    "max_steps": 100,     # Total de passos de treinamento
    "global_batch_size": 64, # Amostras por atualização de gradiente
    "max_length": 8192,   # Comprimento máximo da sequência em tokens
}

# Executar em dry_run para validar antes de iniciar
result = customizer.train(
    job_name="blogpost-sft",
    overrides=training_config,
    dry_run=True
)

# Lançar o job após a validação
result = customizer.train(
    job_name="blogpost-sft",
    overrides=training_config
)

Você pode monitorar os logs do job SFT, que são publicados no Amazon CloudWatch, para acompanhar métricas como perda e taxa de aprendizado em tempo real. O SDK possui utilitários para extrair e exibir esses logs diretamente em seu ambiente de notebook:

customizer.get_logs(limit=20)
result.get_job_status() # Retorna (JobStatus.IN_PROGRESS, ...) ou (JobStatus.COMPLETED, ...)

Após a conclusão do treinamento, avaliamos o modelo SFT no mesmo dataset usado para a avaliação baseline. Observamos melhorias significativas em todas as métricas, exceto BLEU (que dá pontuações baixas para respostas extremamente curtas), atingindo 77.2% de acurácia exact match. Nossa métrica de acurácia de classificação personalizada mostrou 79.0%.

A análise detalhada de desempenho revelou que o modelo aprendeu o formato de saída exigido e houve um aumento drástico na acurácia de classificação para as classes LQ_EDIT e LQ_CLOSE, reduzindo o viés do modelo em classificar tudo como HQ.

Etapa 3: Reinforcement Fine-Tuning (RFT)

Apesar do SFT ter sido eficaz em treinar o modelo no formato exigido, ainda há espaço para melhorar a acurácia dos rótulos gerados. O Reinforcement Fine-Tuning (RFT) é aplicado iterativamente sobre o checkpoint SFT treinado para otimizar a acurácia, especialmente em casos de uso complexos onde a tarefa pode ser enquadrada em termos de uma recompensa quantificável.

Para a classificação, criamos uma função AWS Lambda que recompensa as previsões corretas com uma pontuação positiva (+1) e as incorretas com uma pontuação negativa (-1). Esta estrutura de recompensa binária cria gradientes fortes e inequívocos, ajudando o modelo a distinguir entre as categorias de conteúdo de alta e baixa qualidade.

"""Função de recompensa binária para classificação: +1 correto, -1 errado.
Sinal simples e claro:
- Previsão correta: +1.0
- Previsão errada: -1.0
"""
def calculate_reward(prediction: str, ground_truth: str) -> float:
    """ Calcula a recompensa binária """
    # Assume extract_category e normalize_text existem
    extracted = extract_category(prediction)
    truth_norm = normalize_text(ground_truth)
    if extracted and extracted == truth_norm: return 1.0
    return -1.0

def lambda_handler(event, context):
    """ Handler Lambda com recompensas binárias. """
    scores: List[RewardOutput] = [] # Assume RewardOutput e asdict existem
    for sample in event:
        idx = sample.get("id", "no_id")
        ground_truth = sample.get("reference_answer", "")
        # Supondo que 'last_message' é extraído corretamente do 'sample'
        last_message = sample.get("messages", [])[-1]
        prediction = last_message.get("content", "")
        reward = calculate_reward(prediction, ground_truth)
        scores.append(RewardOutput(id=idx, aggregate_reward_score=reward))
    return [asdict(score) for score in scores]

Após implantar esta função Lambda na AWS e obter seu ARN, garantimos que as políticas de invocação Lambda sejam adicionadas à função IAM de customização, permitindo que o Amazon SageMaker AI a invoque durante o treinamento.

A preparação de dados para RFT, similar ao SFT, utiliza o Nova Forge SDK para curar e validar o dataset, transformando-o para o esquema OpenAI esperado pelo RFT.

RFT_DATA = './rft.csv'
rft_loader = CSVDatasetLoader(
    query='Body',
    response='Y',
    system='system'
)
rft_loader.load(RFT_DATA)

# Transformar para RFT
rft_loader.transform(method=TrainingMethod.RFT_LORA, model=MODEL)
rft_loader.validate(method=TrainingMethod.RFT_LORA, model=MODEL)

# Salvar no S3
rft_s3_uri = rft_loader.save_data(
    f"s3://{S3_BUCKET}/{S3_PREFIX}/data/rft.jsonl"
)

Lançamos o job RFT sobre o checkpoint SFT, fornecendo o dataset formatado e a função de recompensa.

rft_overrides = {
    "lr": 0.00001,      # Learning rate
    "number_generation": 4, # N amostras por prompt
    "reasoning_effort": "null", # Modo de raciocínio (null para não-raciocínio)
    "max_new_tokens": 50, # Limita saídas verbosas
    "kl_loss_coef": 0.02, # Peso para penalidade KL
    "temperature": 1,
    "ent_coeff": 0.01,  # Bônus para entropia de saída
    "max_steps": 40,    # Passos para treinar
    "save_steps": 30,   # Passos após os quais um checkpoint é salvo
    "top_k": 5,         # Amostra apenas dos top-K logits
    "global_batch_size": 64, # Total de amostras por passo do otimizador
}

# Iniciar treinamento RFT
rft_result = rft_customizer.train(
    job_name="stack-overflow-rft",
    rft_lambda_arn=REWARD_LAMBDA_ARN, # ARN da função Lambda de recompensa
    overrides = rft_overrides
)

Monitoramos os logs de treinamento RFT, que incluem estatísticas de recompensa, pontuações do crítico, métricas de gradiente de política e estatísticas de comprimento da resposta. Isso nos ajuda a identificar problemas e ajustar hiperparâmetros ou funções de recompensa.

Avaliamos o modelo customizado SFT+RFT usando a mesma configuração de avaliação anterior. Observamos uma melhoria consistente de cerca de 1% em todas as métricas em comparação com o modelo SFT, indicando que o RFT calibra padrões diferentes no modelo, complementando o SFT. As classificações se tornaram mais precisas, com uma diminuição nas classificações incorretas.

Análise de Resultado Final

A abordagem de customização em duas etapas (SFT + RFT) demonstrou melhorias consistentes e significativas:

Etapa 4: Implantação em um Endpoint de Inferência do Amazon SageMaker AI

Com o modelo final pronto, o Nova Forge SDK facilita a implantação para servir previsões em tempo real. Escolhemos o Amazon SageMaker AI Inference para esta demonstração, que oferece flexibilidade para tipos de instância e configurações de ambiente personalizadas.

# Supondo que 'rft_customizer' contém o modelo final treinado
# Endpoint será criado automaticamente ou você pode especificar um existente
# rft_customizer.deploy(
#     target=DeploymentTarget.SAGEMAKER_AI,
#     instance_type=INSTANCE_TYPE,
#     initial_instance_count=1,
#     endpoint_name="blogpost-rft-endpoint"
# )

# Exemplo de invocação do endpoint
streaming_chat_request = {
    "messages": [{"role": "user", "content": "Qual a melhor forma de otimizar consultas SQL em grande escala?"}],
    "max_tokens": 200,
    "stream": False,
}
ENDPOINT_NAME = f"arn:aws:sagemaker:REGION:ACCOUNT_ID:endpoint/blogpost-rft-endpoint" # Substituir REGION e ACCOUNT_ID

inference_result = rft_customizer.invoke_inference(
    request_body=streaming_chat_request,
    endpoint_arn=ENDPOINT_NAME
)
inference_result.show()

Etapa 5: Limpeza

Após testar a implantação, é crucial limpar os recursos para evitar cobranças contínuas da AWS.

import boto3

iam_client = boto3.client('iam')
role_name = 'your-role-name' # Substitua pelo nome da sua role

# Desanexar políticas gerenciadas
attached_policies = iam_client.list_attached_role_policies(RoleName=role_name)
for policy in attached_policies['AttachedPolicies']:
    iam_client.detach_role_policy(
        RoleName=role_name,
        PolicyArn=policy['PolicyArn']
    )

# Excluir políticas inline
inline_policies = iam_client.list_role_policies(RoleName=role_name)
for policy_name in inline_policies['PolicyNames']:
    iam_client.delete_role_policy(
        RoleName=role_name,
        PolicyName=policy_name
    )

# Remover de perfis de instância
instance_profiles = iam_client.list_instance_profiles_for_role(RoleName=role_name)
for profile in instance_profiles['InstanceProfiles']:
    iam_client.remove_role_from_instance_profile(
        InstanceProfileName=profile['InstanceProfileName'],
        RoleName=role_name
    )

# Excluir a role
iam_client.delete_role(RoleName=role_name)

# Lembre-se de deletar também endpoints do SageMaker, jobs de treinamento, e dados S3.

Conclusão

O Nova Forge SDK transforma a customização de modelos de um processo complexo e pesado em infraestrutura para um fluxo de trabalho acessível e amigável ao desenvolvedor. Através do nosso estudo de caso de classificação do Stack Overflow, demonstramos como as equipes podem usar o SDK para alcançar melhorias mensuráveis através de treinamento iterativo, movendo de uma acurácia baseline de 13% para 79% após SFT e atingindo 80.6% com RFT adicional.

Ao remover as barreiras tradicionais à customização de LLMs, como a exigência de expertise técnica e o investimento de tempo, o Nova Forge SDK capacita as organizações a construir modelos que compreendem seu contexto exclusivo sem sacrificar as capacidades gerais que tornam os modelos de fundação valiosos. O SDK lida com a configuração de recursos de computação, orquestração de todo o pipeline de customização, monitoramento de jobs de treinamento e implantação de endpoints. O resultado é uma IA empresarial que é especializada e inteligente, tanto especialista no domínio quanto amplamente capaz.

Comentários

Interações
Seu Perfil

Aguardando Login...