Como Fazer Deploy e Avaliar Modelos com o Amazon SageMaker JumpStart

Por que avaliar modelos antes de escolher?

Escolher o modelo certo para uma aplicação vai muito além de ler benchmarks genéricos. O que realmente importa é testar o modelo com entradas representativas do seu caso de uso — e fazer isso de forma repetível. O Amazon SageMaker JumpStart oferece exatamente esse caminho: um catálogo amplo de modelos pré-treinados que podem ser implantados, testados e removidos de forma controlada, tudo via código ou interface gráfica.

A abordagem descrita pela AWS usa um script Python reutilizável que percorre três etapas: descoberta e deploy do modelo, execução de prompts de exemplo com medição de latência e taxa de geração, e remoção do endpoint ao final. Vale lembrar que o tempo de deploy varia conforme o modelo, o tipo de instância, a cota disponível e a Região AWS utilizada.

Novidades de 2026 no SageMaker JumpStart

Em 2026, o SageMaker JumpStart recebeu duas melhorias relevantes. A primeira é a seção Model Hub no console do Amazon SageMaker AI, que permite navegar pelo catálogo, visualizar notebooks de exemplo e abri-los diretamente no SageMaker Studio. A segunda é a disponibilização de configurações de endpoint pré-definidas, otimizadas para diferentes prioridades — baixo custo, baixa latência ou cargas de trabalho de uso geral — incluindo configurações otimizadas para chips Neuron, cobrindo uma ampla gama de modelos de IA Generativa (Generative AI).

Além disso, agora é possível ir da descoberta de um modelo no Hugging Face até o deploy no SageMaker AI com um único clique, graças à nova integração de deep link entre as duas plataformas.

Pré-requisitos

  • Conta AWS com permissões para o SageMaker AI
  • Python 3+ com o SageMaker Python SDK v3 instalado (pip install "sagemaker>=3,<4")
  • Cota suficiente de instâncias GPU para o modelo desejado — caso necessário, é possível solicitar aumento de cota

Passo 1a: Descobrindo modelos disponíveis

Além de navegar pelo catálogo pela interface do SageMaker Studio, é possível listar os modelos disponíveis diretamente via código. O exemplo abaixo lista os modelos de geração de texto disponíveis na região us-east-1:

from sagemaker.core.jumpstart.notebook_utils import (
    list_jumpstart_models,
)

REGION = "us-east-1"
models = list_jumpstart_models(
    filter="task == textgeneration",
    region=REGION,
    list_versions=True,
)
print(f"Found {len(models)} text generation models")
for model_id, model_version in models[:10]:
    print(
        f" - {model_id} "
        f"(version {model_version})"
    )

Também é possível buscar por nome de modelo específico:

from sagemaker.core.jumpstart.notebook_utils import list_jumpstart_models

ids = list_jumpstart_models(
    filter="model_id contains qwen",
    region=REGION,
    list_versions=True,
)
print(f"{len(ids)} models matched in {REGION}\n")
for model_id, version in ids:
    print(f"{model_id:<55} v{version}")

Passo 1b: Fazendo o deploy do modelo

Com o modelo identificado, o deploy pode ser feito via SDK. O código abaixo resolve automaticamente a configuração recomendada pelo JumpStart — incluindo o tipo de instância ideal — e sobe o endpoint:

import time
from sagemaker.core.jumpstart.accessors import JumpStartModelsAccessor
from sagemaker.core.jumpstart.configs import JumpStartConfig
from sagemaker.core.training.configs import Compute
from sagemaker.serve import ModelBuilder

MODEL_ID = "huggingface-vlm-qwen3-5-27b"
MODEL_VERSION = "1.13.0"
REGION = "us-east-1"
ACCEPT_EULA = False  # Set True only after reviewing the model terms

def get_field(value, name, default=None):
    if isinstance(value, dict):
        return value.get(name, default)
    return getattr(value, name, default)

# Resolve the JumpStart-recommended deployment configuration.
specs = JumpStartModelsAccessor().get_model_specs(
    region=REGION,
    model_id=MODEL_ID,
    version=MODEL_VERSION,
)
if isinstance(specs, list):
    if not specs:
        raise ValueError(
            f"No JumpStart specification found for {MODEL_ID}"
        )
    specs = specs[0]

recommended_instance_type = get_field(
    specs,
    "default_inference_instance_type",
)
if not recommended_instance_type:
    raise ValueError(
        f"No recommended instance type found for {MODEL_ID}"
    )

print(f"Model: {MODEL_ID}")
print(f"Region: {REGION}")
print(f"Recommended instance: {recommended_instance_type}")

start = time.time()
jumpstart_config = JumpStartConfig(
    model_id=MODEL_ID,
    model_version=MODEL_VERSION,
    accept_eula=ACCEPT_EULA,
)
model_builder = ModelBuilder.from_jumpstart_config(
    jumpstart_config=jumpstart_config,
    compute=Compute(
        instance_type=recommended_instance_type,
        instance_count=1,
    ),
)
model = model_builder.build()
endpoint = model_builder.deploy()
deploy_time = time.time() - start
print(f"Deployed in {deploy_time:.1f} seconds")
print(f"Endpoint: {endpoint.endpoint_name}")

Um ponto importante: a variável ACCEPT_EULA deve ser definida como True apenas após a leitura e aceite dos termos de uso do modelo. Alguns modelos no JumpStart exigem esse aceite explícito antes do deploy.

Passo 2: Executando prompts e medindo desempenho

Com o endpoint ativo, o próximo passo é executar os prompts de exemplo disponíveis para o modelo e medir latência, volume de saída e taxa de geração por segundo. O código abaixo faz exatamente isso:

from sagemaker.core.payloads import retrieve_all_examples
import json
import time

MODEL_ID = "huggingface-vlm-qwen3-5-27b"
MODEL_VERSION = "1.13.0"
REGION = "us-east-1"

example_payloads = retrieve_all_examples(
    region=REGION,
    model_id=MODEL_ID,
    model_version=MODEL_VERSION,
    serialize=True,
)

#you can re-create the endpoint pointer using the endpoint name:
#from sagemaker.core.resources import Endpoint
#endpoint = Endpoint.get(endpoint_name="<your_endpoint_name>",region=REGION)

def read_response_body(result):
    body = result.body
    if hasattr(body, "read"):
        body = body.read()
    if isinstance(body, bytes):
        body = body.decode("utf-8")
    if isinstance(body, str):
        try:
            return json.loads(body)
        except json.JSONDecodeError:
            return body
    return body

def extract_output(response):
    """Handle common JumpStart text-generation response formats."""
    if isinstance(response, list):
        response = response[0] if response else ""
    if isinstance(response, dict):
        choices = response.get("choices")
        if choices:
            choice = choices[0]
            message = choice.get("message")
            if isinstance(message, dict):
                return message.get("content", "")
            return choice.get("text", "")
        for field in (
            "generated_text",
            "output_text",
            "text",
        ):
            if field in response:
                return response[field]
    return response

results = []
for payload in example_payloads or []:
    start = time.perf_counter()
    result = endpoint.invoke(
        body=payload.body,
        content_type=payload.content_type,
        accept=payload.accept,
    )
    latency = time.perf_counter() - start
    response = read_response_body(result)
    output = extract_output(response)
    if not isinstance(output, str):
        output = json.dumps(output, indent=2)

    # This is an estimate based on whitespace-separated words.
    output_words = len(output.split())
    results.append(
        {
            "request_preview": payload.body[:50],
            "latency_s": round(latency, 2),
            "estimated_output_words": output_words,
            "words_per_second": round(
                output_words / latency, 1,
            ) if latency > 0 else 0,
        }
    )
    print("Input:\n", payload.body, end="\n\n")
    print("Output:\n", output, end="\n\n")
    print(
        f"Latency: {latency:.2f} seconds\n"
        f"Estimated output words: {output_words}\n"
        f"Estimated single-request output generation rate: "
        f"{output_words / latency:.1f} words/second\n"
    )
    print("-" * 80)

print("\nSummary:")
for result in results:
    print(result)

Um exemplo real de resultado obtido com o modelo huggingface-vlm-qwen3-5-27b, ao perguntar "What is deep learning?", retornou 153 palavras estimadas em 9,07 segundos — uma taxa de aproximadamente 16,9 palavras por segundo. Esses números servem como referência para comparação entre modelos diferentes.

Passo 3: Removendo o endpoint

Após os testes, é fundamental remover o endpoint para evitar cobranças desnecessárias. O processo é direto:

print(f"Deleting endpoint: {endpoint.endpoint_name}")
endpoint.delete()
endpoint.wait_for_delete()
print("Endpoint deleted")

Considerações finais

Esse fluxo funciona para a maioria dos modelos compatíveis com o SageMaker JumpStart, mas é importante estar atento a variações: o formato do payload, o tipo de instância necessário, os requisitos de Contrato de Licença de Usuário Final (EULA) e o esquema de resposta podem exigir ajustes específicos por modelo.

Ao usar prompts consistentes e medir latência, throughput e qualidade de saída em diferentes modelos, equipes conseguem comparar opções de forma objetiva e escolher aquela que melhor equilibra qualidade, velocidade e custo para cada aplicação.

Para aprofundar o tema, a AWS disponibiliza recursos adicionais: a documentação oficial do SageMaker Python SDK, um post sobre benchmark e otimização de deploys no SageMaker JumpStart, e um guia sobre a integração entre Hugging Face e SageMaker Studio em um clique.

Fonte

Deploy and evaluate models with Amazon SageMaker JumpStart (https://aws.amazon.com/blogs/machine-learning/deploy-and-evaluate-models-with-amazon-sagemaker-jumpstart/)

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *