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á.

Preparando seu agente

Antes de iniciar um trabalho de treinamento, você precisa configurar um agente que possa participar do ciclo de treinamento. Esta seção explica como configurar e implantar seu agente, esteja você usando o Amazon Bedrock AgentCore para hospedagem gerenciada ou trazendo sua própria infraestrutura com um agente personalizado.

Visão geral da integração do código do agente

Durante o treinamento, a SageMaker IA envia solicitações do seu conjunto de dados de treinamento para seu agente. Seu agente processa cada solicitação, chama o modelo de política para obter uma resposta e executa ações em seu ambiente, como chamar ferramentas. O modelo político é o modelo que está sendo treinado. Ele começa quando seu modelo básico e seus pesos são atualizados ao longo do tempo, à medida que ele recebe feedback por meio do ciclo de treinamento. Com base no resultado, seu agente reporta uma recompensa à SageMaker IA para concluir o ciclo de treinamento. Isso se repete para todas as solicitações em seu conjunto de dados, e o modelo de política melhora com o tempo com base nas recompensas coletadas.

SageMaker Serviço AI Job Runtime

Durante o treinamento, seu agente se comunica com o SageMaker AI Job Runtime Service para chamar o modelo de política para inferência e reportar os resultados ao instrutor. O decorador do SDK lida com essa integração automaticamente, mas se seu agente tiver requisitos personalizados, você poderá chamar as APIs do Runtime diretamente.

Endpoint

https://job-runtime.sagemaker.region.api.aws

APIs

Autenticação

Seu agente se autentica com o Runtime Service usando tokens portadores. Gere um token usando o SDK:

from sagemaker.core.token_generator import generate_token

token = generate_token(region="us-west-2")

Passe esse token como chave de API ao chamar o Runtime Service.

Integração direta de API (avançada)

Se sua estrutura de agente não puder usar o decorador do SDK, chame as APIs diretamente:

import requests
import openai
from sagemaker.core.token_generator import generate_token
from sagemaker.train.rft.headers import make_inference_headers

token = generate_token(region="us-west-2")
endpoint = "https://job-runtime.sagemaker.us-west-2.api.aws"
headers = {
    "Authorization": f"Bearer {token}",
    "Content-Type": "application/json",
    **make_inference_headers(metadata),  # injects jobArn, rolloutId tracking
}

# Sample (inference call)

# Option 1: Create OpenAI-compatible client pointing to the Runtime Service (recommended)
client = openai.OpenAI(
    base_url=endpoint + "/v1",
    api_key=token,
    default_headers=headers,
)

# Option 2: direct request call
response = requests.post(
    f"{endpoint}/v1/chat/completions",
    headers=headers,
    json={
        "model": "default",
        "messages": [{"role": "user", "content": "..."}],
        "max_tokens": 256,
        "temperature": 1.0,
    }
)

# CompleteRollout + UpdateReward (after all turns complete)
# The decorator handles this; for direct calls, use the SDK's
# RolloutFeedbackClient or call the APIs per the Smithy model.

Opções de implantação do agente

SageMaker A IA oferece suporte a duas opções para conectar seu agente ao ciclo de treinamento:

Cenário 1: Agentes no Bedrock AgentCore Runtime

Implante seu agente no Amazon Bedrock AgentCore para uma hospedagem totalmente gerenciada. SageMaker A IA invoca seu agente durante o treinamento do modelo.

Pré-requisitos

Antes de começar, conclua os pré-requisitos a seguir.

Papel fundamental na AgentCore execução

Durante o treinamento, a SageMaker IA assume sua função de SageMaker execução para invocar seu agente. Seu agente também precisa de sua própria função separada, chamada de função de AgentCore execução Bedrock, para chamar o RFT Runtime para inferência de modelos e relatórios de recompensas. O RFT Runtime é o terminal do serviço de SageMaker IA com o qual seu agente se comunica durante o treinamento. Ele lida com duas coisas: fornecer respostas do modelo de política durante a inferência e receber a recompensa que seu agente relata no final de cada lançamento.

Crie a função de AgentCore execução do Bedrock com a seguinte política de confiança. Esta política concede AgentCore permissão à Bedrock para assumir a função em seu nome durante o treinamento. Sem isso, o Bedrock AgentCore não pode acessar sua imagem de contêiner nem chamar o RFT Runtime.

{
    "Version": "2012-10-17",
    "Statement": [{
        "Effect": "Allow",
        "Principal": {"Service": "bedrock-agentcore.amazonaws.com"},
        "Action": "sts:AssumeRole"
    }]
}

Em seguida, anexe a política AmazonSageMakerJobRuntimeAccessgerenciada a essa função. Isso concede as permissões que seu agente precisa para chamar o modelo de política para inferência e relatar os resultados durante o treinamento.

Escreva ou atualize seu código de agente

Seu agente deve usar o sagemaker-trainSDK e aplicar o @sagemaker_rft_handler decorador à função de ponto de entrada do seu agente. Esse decorador configura o servidor HTTP que AgentCore invoca durante o treinamento, ouvindo as solicitações de implantação recebidas e encaminhando as solicitações para sua função de ponto de entrada.

Instalar o SDK

Instale o SDK no ambiente do seu agente usando um dos métodos a seguir.

Opção 1: instalação direta

pip install sagemaker-train

Opção 2: Construa e instale rodas individualmente. Use isso se a instalação direta exceder as restrições de tamanho do seu ambiente.

# Clone the SDK repository
git clone https://github.com/aws/sagemaker-python-sdk.git

# Build the wheels
cd sagemaker-python-sdk/sagemaker-train
python -m build --wheel --outdir ./dist/

cd sagemaker-python-sdk/sagemaker-core
python -m build --wheel --outdir ./dist/

# Install
pip install ./sagemaker-train/dist/sagemaker_train-*.whl
pip install ./sagemaker-core/dist/sagemaker_core-*.whl

Autenticação

Seu agente se autentica com SageMaker IA usando tokens de portador. Gere um token usando o generate_token() método do SDK e passe-o como chave de API ao chamar o modelo de política para inferência.

Requisitos do agente

Seu agente deve fazer o seguinte para cada lançamento:

O exemplo a seguir mostra um modelo básico de agente usando o SDK Strands:

from sagemaker.train.rft import sagemaker_rft_handler, RolloutFeedbackClient
from sagemaker.train.rft.adapters.strands import wrap_model
from sagemaker.core.token_generator import generate_token
from strands import Agent, OpenAIModel
import os

@sagemaker_rft_handler
def handle_rollout(payload):
    metadata = payload.get("metadata", {})
    prompt = payload.get("prompt", "")
    endpoint = metadata.get("endpoint", os.environ.get("RFT_RUNTIME_ENDPOINT", ""))

    # Generate bearer token for authenticating with the RFT Runtime
    token = generate_token(region=os.environ.get("AWS_REGION", "us-west-2"))

    # Create client for model inference
    model = OpenAIModel(
        model_id="default",
        client_args={
            "api_key": token,
            "base_url": endpoint + "/v1",
        },
    )

    # Wrap model to auto-inject RFT tracking headers and inference parameters
    model = wrap_model(model)

    # Execute actions in your environment (tools, APIs, etc.)
    agent = Agent(model=model, tools=[..., ...])
    result = agent(prompt)

    # Return reward -   decorator handles CompleteRollout + UpdateReward
    return {"reward": compute_reward(result)}

Envie seu agente para Bedrock AgentCore

Implante seu agente no Amazon Bedrock AgentCore seguindo o guia de AgentCore desenvolvimento.

Depois de implantar seu agente usando a AgentCore CLI, observe o ARN do Agent Runtime na saída. Você precisa disso ao criar seu trabalho de treinamento. O ARN do Runtime segue o formato de. arn:aws:bedrock-agentcore:<region>:<account-id>:runtime/<agent-name>

Verifique se seu agente está implantado e em bom estado:

aws bedrock-agentcore-control list-agent-runtimes --region us-west-2

Cenário 2: Agente personalizado com Lambda Forwarder

Além do Bedrock AgentCore, a SageMaker IA oferece suporte a agentes personalizados hospedados em qualquer ambiente. Seu agente se conecta ao ciclo de treinamento por meio de um encaminhador Lambda, oferecendo a flexibilidade de usar qualquer estrutura de agente e plataforma de hospedagem de sua escolha.

Seu agente pode ser criado usando qualquer estrutura ou plataforma, como o Strands Agents SDK ou sua própria implementação personalizada. Ele pode ser executado em qualquer ambiente computacional, incluindo Amazon Bedrock, AgentCore Amazon EKS, Amazon EC2 AWS , Fargate ou em sua própria infraestrutura.

A função Lambda recebe solicitações de implantação da SageMaker IA e as encaminha para o endpoint HTTP do seu agente.

As seções a seguir fornecem mais informações sobre como configurar um agente personalizado com um encaminhador Lambda, com exemplos de uso de um agente personalizado implantado no Amazon EKS.

Pré-requisitos

Função de execução do Lambda

Durante o treinamento, a SageMaker IA assume sua função de SageMaker execução para invocar seu encaminhador Lambda. A função Lambda requer sua própria função de execução para que o AWS Lambda possa executá-la. Essa função permite que o Lambda execute e grave registros.

aws iam create-role \
  --role-name RFTLambdaForwarderRole \
  --assume-role-policy-document '{
    "Version": "2012-10-17",
    "Statement": [{
      "Effect": "Allow",
      "Principal": {"Service": "lambda.amazonaws.com"},
      "Action": "sts:AssumeRole"
    }]
  }'

aws iam attach-role-policy \
  --role-name RFTLambdaForwarderRole \
  --policy-arn arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole

Escreva seu código de agente

Seu agente deve expor um endpoint HTTP que aceite solicitações de implantação e interaja com o serviço RFT Runtime, chamando-o para inferência do modelo e relatando as recompensas quando a tarefa for concluída.

Formato de solicitação de implantação

Seu agente recebe solicitações no seguinte formato do encaminhador Lambda:

{
    "prompt": "string",
    "metadata": {
        "jobArn": "string",
        "rolloutId": "string"
    },
    "inferenceParams": {
        "temperature": 1.0,
        "topP": 1.0,
        "maxTokens": 256
    }
}

expectativas do agente

Para que a Amazon SageMaker AI conclua com sucesso os lançamentos, seu agente deve:

O exemplo a seguir mostra um agente personalizado com um endpoint FastAPI:

"""
Custom Lambda Agent for SageMaker RFT.

This is a minimal example showing the required integration points.
You can use any framework (Strands, LangChain, raw OpenAI, etc.)
or any HTTP framework (FastAPI, Flask, Django, etc.) for your agent.
"""
import os
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from typing import Any, Dict
from sagemaker.train.rft import sagemaker_rft_handler, set_rollout_context
from sagemaker.train.rft.headers import make_inference_headers
from sagemaker.core.token_generator import generate_token

app = FastAPI()

@app.post("/rollout")
def rollout_endpoint(payload: dict):
    handle_rollout(payload)
    return {"status": "accepted"}

@sagemaker_rft_handler
def handle_rollout(payload):
    """Handle a rollout request from RFT."""
    prompt = payload.get("prompt", "")
    metadata = payload.get("metadata", {})
    inference_params = payload.get("inferenceParams", {})
    endpoint = metadata.get("endpoint", os.environ.get("RFT_RUNTIME_ENDPOINT", ""))

    # 1. Set rollout context and capture inference headers
    set_rollout_context(metadata)
    headers = make_inference_headers(metadata)

    # 2. Get bearer token for RFT Runtime authentication
    token = generate_token(region=os.environ.get("AWS_REGION", "us-west-2"))

    # 3. Run your agent logic
    result = run_agent(prompt, token, endpoint, inference_params, headers)

    # 4. Compute reward based on your task's success criteria
    reward = compute_reward(result)

    # 5. Report completion with reward to RFT Runtime
    # decorator handles CompleteRollout + UpdateReward
    return {"reward": reward}

def run_agent(prompt: str, token: str, endpoint: str, inference_params: dict, headers: dict) -> str:
    """Your agent logic goes here."""
    import openai
    client = openai.OpenAI(
        base_url=(endpoint or os.environ.get("RFT_RUNTIME_ENDPOINT", "")) + "/v1",
        api_key=token,
        default_headers=headers,
    )

    response = client.chat.completions.create(
        model="default",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=inference_params.get("maxTokens", 256),
        temperature=inference_params.get("temperature", 1.0),
        top_p=inference_params.get("topP", 1.0),
    )

    return response.choices[0].message.content

def compute_reward(result: str) -> float:
    """Implement your task-specific reward function here."""

@app.get("/health")
def health():
    return {"status": "ok"}

Instale as dependências necessárias em seu ambiente de agente:

pip install sagemaker-train fastapi uvicorn openai

Implante seu agente

Seu agente pode ser executado em qualquer ambiente computacional. Ele precisa de acesso externo à rede para acessar o SageMaker AI Runtime para relatórios de inferência e recompensa. Se o encaminhador do Lambda ligar diretamente para seu agente, ele também deverá expor um endpoint HTTP acessível pelo Lambda.

As etapas a seguir implantam um agente personalizado no Amazon EKS.

Crie um cluster EKS

Para obter instruções completas de configuração do EKS, consulte o guia de introdução do EKS. Veja a seguir uma configuração mínima para este exemplo:

eksctl create cluster \
  --name external-agent \
  --region us-west-2 \
  --nodegroup-name agent-nodes \
  --node-type t3.medium \
  --nodes 1 \
  --nodes-min 1 \
  --nodes-max 2 \
  --managed

Atualize seu kubeconfig para se conectar ao cluster:

aws eks update-kubeconfig --name external-agent --region us-west-2

Crie e envie sua imagem de contêiner:

aws ecr create-repository --repository-name my-external-agent --region us-west-2

aws ecr get-login-password --region us-west-2 | \
  docker login --username AWS --password-stdin account-id.dkr.ecr.us-west-2.amazonaws.com

docker build -t my-external-agent .
docker tag my-external-agent:latest account-id.dkr.ecr.us-west-2.amazonaws.com/my-external-agent:latest
docker push account-id.dkr.ecr.us-west-2.amazonaws.com/my-external-agent:latest

Implante no EKS:

kubectl apply -f - <<EOF
apiVersion: apps/v1
kind: Deployment
metadata:
  name: external-agent
spec:
  replicas: 1
  selector:
    matchLabels:
      app: external-agent
  template:
    metadata:
      labels:
        app: external-agent
    spec:
      containers:
      - name: agent
        image: <account-id>.dkr.ecr.us-west-2.amazonaws.com/external-agent:latest
        ports:
        - containerPort: 8080
        env:
        - name: AWS_REGION
          value: us-west-2
        - name: RFT_RUNTIME_ENDPOINT
          value: https://job-runtime.sagemaker.us-west-2.api.aws
---
apiVersion: v1
kind: Service
metadata:
  name: external-agent
spec:
  selector:
    app: external-agent
  ports:
  - port: 80
    targetPort: 8080
  type: LoadBalancer
EOF

Obtenha o endpoint do seu agente:

AGENT_ENDPOINT=$(kubectl get svc external-agent -o jsonpath='{.status.loadBalancer.ingress[0].hostname}')

echo "Agent endpoint: http://$AGENT_ENDPOINT"

Verifique a implantação:

curl -s "http://$AGENT_ENDPOINT/health"

# Expected: {"status": "ok"}

Crie seu encaminhador Lambda

O encaminhador do Lambda recebe solicitações de implantação da SageMaker IA e as encaminha para seu agente. A principal coisa a ser personalizada é a _call_agent() função, que traduz a solicitação de lançamento para o formato de API do seu agente.

Se seu agente não tiver um endpoint HTTP público, você poderá substituir a chamada HTTP por um SQS send_message e, em _call_agent() vez disso, fazer com que seu agente pesquise a fila.

"""
Lambda Template

Bridges SageMaker Job rollout requests to any agent platform with a public endpoint.
Implement _call_agent() with your platform-specific logic.

If your agent environment does not have a public endpoint, you can
replace the HTTP call with an SQS send_message to enqueue the request,
and have your agent poll the queue for work.

Env vars:
    AGENT_ENDPOINT  - target agent base URL
    AGENT_API_KEY   - API key for the target agent (prefer Secrets Manager)
"""

import json
import logging
import os
import re
import urllib.error
import urllib.request

logger = logging.getLogger()
logger.setLevel(os.environ.get("LOG_LEVEL", "INFO"))

AGENT_ENDPOINT = os.environ.get("AGENT_ENDPOINT", "")
AGENT_API_KEY = os.environ.get("AGENT_API_KEY", "")

_SAFE_ID = re.compile(r"^[\w\-.]+$")

# ---------------------------------------------------------------------------
# CUSTOMIZE THIS - translate rollout request to your platform's API
# ---------------------------------------------------------------------------
def _call_agent(prompt: str, metadata: dict, inference_params: dict):
    """
    Forward the prompt to your agent platform.
    Replace the body below with your platform's request format.
    """
    payload = json.dumps({
        "prompt": prompt,
        "metadata": metadata,
        "inferenceParams": inference_params,
    }).encode()

    req = urllib.request.Request(
        AGENT_ENDPOINT,
        data=payload,
        headers={
            "Content-Type": "application/json",
            "Authorization": f"Bearer {AGENT_API_KEY}",
        },
        method="POST",
    )

    urllib.request.urlopen(req, timeout=120)


# ---------------------------------------------------------------------------
# Validation - no changes needed below
# ---------------------------------------------------------------------------
def _validate(event: dict) -> dict:
    body = json.loads(event["body"]) if isinstance(event.get("body"), str) else event

    prompt = body.get("prompt")
    if not isinstance(prompt, str) or not prompt.strip():
        raise ValueError("'prompt' is required and must be a non-empty string")

    meta = body.get("metadata")
    if not isinstance(meta, dict):
        raise ValueError("'metadata' is required")
    for key in ("jobArn", "rolloutId"):
        val = meta.get(key)
        if not isinstance(val, str) or not val.strip():
            raise ValueError(f"metadata.{key} must match [a-zA-Z0-9_\\-.]")

    params = body.get("inferenceParams") or {}
    if not isinstance(params, dict):
        raise ValueError("'inferenceParams' must be an object")

    return {
        "prompt": prompt.strip(),
        "metadata": meta,
        "inferenceParams": params,
    }


# ---------------------------------------------------------------------------
# CUSTOMIZE THIS - handle errors thrown from your agent environment
# ---------------------------------------------------------------------------
def _handle_agent_error(exc: Exception) -> dict:
    """
    Called when _call_agent() raises an exception. Customize this to map
    platform-specific errors to appropriate error types and messages.

    The return payload must follow this structure:
        {"errorType": "<type>", "errorMessage": "<description>"}

    Supported errorType values:
        ValidationError, InternalServerError, AccessDenied

    Examples:
          return {"errorType": "AccessDenied","errorMessage": "Agent denied access"}
          return {"errorType": "ValidationError","errorMessage": "Missing required field"}
          return {"errorType":"InternalServerError", "errorMessage": str(exc)}
    """

def handler(event, context):
    try:
        body = _validate(event)
    except ValueError as exc:
        logger.warning("Validation error: %s", exc)
        return {"errorType": "ValidationError", "errorMessage": str(exc)}

    try:
        result = _call_agent(body["prompt"], body["metadata"], body["inferenceParams"])
        logger.info("Rollout %s completed", body["metadata"]["rolloutId"])
        return {}
    except Exception as exc:
        return _handle_agent_error(exc)

Implante a função Lambda:

Package a função Lambda:

zip lambda_forwarder.zip lambda_forwarder.py

Crie a função:

aws lambda create-function \
  --function-name rft-agent-forwarder \
  --runtime python3.12 \
  --handler lambda_forwarder.handler \
  --role arn:aws:iam::account-id:role/RFTLambdaForwarderRole \
  --zip-file fileb://lambda_forwarder.zip \
  --timeout 600 \
  --environment "Variables={AGENT_ENDPOINT=http://$AGENT_ENDPOINT}" \
  --region us-west-2

Teste o Lambda:

aws lambda invoke \
  --function-name rft-agent-forwarder \
  --cli-binary-format raw-in-base64-out \
  --payload '{"prompt": "Plan a 3-day itinerary for a trip to Seattle", "metadata": {"jobArn": "arn:aws:sagemaker:us-west-2:123456789012:job/AgentRFT/test", "rolloutId": "roll-1"}}' \
  --region us-west-2 \
  /tmp/response.json && cat /tmp/response.json