Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.

Preparar a su agente

Antes de poder iniciar un trabajo de formación, debe crear un agente que pueda participar en el ciclo de formación. En esta sección se explica cómo configurar e implementar su agente, tanto si utiliza Amazon Bedrock AgentCore para el alojamiento gestionado como si incorpora su propia infraestructura con un agente personalizado.

Descripción general de la integración del código del agente

Durante el entrenamiento, la SageMaker IA envía instrucciones desde tu conjunto de datos de entrenamiento a tu agente. El agente procesa cada solicitud, solicita una respuesta al modelo de políticas y toma medidas en su entorno, como llamar a las herramientas. El modelo de política es el modelo que se está entrenando. Comienza como tu modelo base y sus pesos se actualizan con el tiempo a medida que recibe información a lo largo del ciclo de entrenamiento. En función del resultado, tu agente envía una recompensa a la SageMaker IA por completar el ciclo de entrenamiento. Esto se repite para todas las solicitudes de tu conjunto de datos, y el modelo de políticas mejora con el tiempo en función de las recompensas acumuladas.

SageMaker Servicio AI Job Runtime

Durante la formación, su agente se comunica con el SageMaker AI Job Runtime Service para recurrir al modelo de políticas a efectos de inferencia e informar de los resultados al formador. El decorador del SDK gestiona esta integración automáticamente, pero si tu agente tiene requisitos personalizados, puedes llamar directamente a las API de Runtime.

Punto de conexión

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

API

Autenticación

Su agente se autentica con el Runtime Service mediante tokens portadores. Genera un token con el SDK:

from sagemaker.core.token_generator import generate_token

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

Pase este token como clave de API cuando llames al Runtime Service.

Integración directa de la API (avanzada)

Si el marco de su agente no puede usar el decorador del SDK, llame directamente a las API:

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.

Opciones de despliegue de agentes

SageMaker La IA admite dos opciones para conectar a tu agente al ciclo de formación:

Escenario 1: Agents on Bedrock Runtime AgentCore

Despliegue su agente en Amazon Bedrock AgentCore para obtener un alojamiento totalmente gestionado. SageMaker La IA invoca a su agente durante la formación del modelo.

Requisitos previos

Antes de comenzar, complete los siguientes requisitos previos.

Función fundamental de ejecución AgentCore

Durante el entrenamiento, la SageMaker IA asume tu función de SageMaker ejecución para invocar a tu agente. Su agente también necesita su propia función independiente, denominada función de AgentCore ejecución de Bedrock, que se denomina RFT Runtime para la inferencia de modelos y la generación de informes sobre recompensas. El RFT Runtime es el punto final del servicio de SageMaker IA con el que se comunica su agente durante la formación. Se encarga de dos cosas: entregar las respuestas del modelo de políticas durante la inferencia y recibir la recompensa que su agente reporta al final de cada implementación.

Cree el rol de AgentCore ejecución de Bedrock con la siguiente política de confianza. Esta política otorga a Bedrock AgentCore permiso para asumir el cargo en tu nombre durante la formación. Sin ella, Bedrock AgentCore no puede acceder a la imagen de su contenedor ni llamar al RFT Runtime.

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

A continuación, asocie la política AmazonSageMakerJobRuntimeAccessgestionada a este rol. Esto otorga los permisos que su agente necesita para utilizar el modelo de políticas con fines de inferencia e informar de los resultados durante la formación.

Escribe o actualiza tu código de agente

Su agente debe usar el sagemaker-trainSDK y aplicar el @sagemaker_rft_handler decorador a la función de punto de entrada de su agente. Este decorador configura el servidor HTTP que AgentCore invoca durante el entrenamiento y escucha las solicitudes de despliegue entrantes y las redirige a la función de punto de entrada.

Instalación del SDK

Instale el SDK en el entorno de su agente mediante uno de los siguientes métodos.

Opción 1: instalación directa

pip install sagemaker-train

Opción 2: Construya e instale las ruedas individualmente. Utilice esta opción si la instalación directa supera las restricciones de tamaño de su entorno.

# 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

Autenticación

Su agente se autentica con SageMaker IA mediante fichas portadoras. Genera un token con el generate_token() método del SDK y pásalo como clave de API cuando llames al modelo de políticas para realizar inferencias.

Requisitos del agente

El agente debe hacer lo siguiente en cada implementación:

El siguiente ejemplo muestra una plantilla de agente básica que utiliza el SDK de 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)}

Despliega tu agente en Bedrock AgentCore

Despliegue su agente en Amazon Bedrock AgentCore siguiendo la guía de AgentCore desarrollo.

Una vez que haya implementado el agente mediante la AgentCore CLI, anote el ARN de Agent Runtime en el resultado. Lo necesitará al crear su trabajo de formación. El ARN de ejecución sigue el formato de. arn:aws:bedrock-agentcore:<region>:<account-id>:runtime/<agent-name>

Compruebe que el agente esté desplegado y en buen estado:

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

Escenario 2: Agente personalizado con Lambda Forwarder

Además de Bedrock AgentCore, la SageMaker IA admite agentes personalizados alojados en cualquier entorno. Su agente se conecta al ciclo de formación a través de un reenviador Lambda, lo que le brinda la flexibilidad de utilizar cualquier marco de agentes y plataforma de alojamiento que elija.

Su agente puede crearse con cualquier marco o plataforma, como el SDK de Strands Agents, o con su propia implementación personalizada. Puede ejecutarse en cualquier entorno informático, incluidos Amazon Bedrock AgentCore, Amazon EKS, Amazon EC2, AWS Fargate o en su propia infraestructura.

La función Lambda recibe las solicitudes de implementación de la SageMaker IA y las reenvía al punto de enlace HTTP del agente.

En las siguientes secciones se proporciona más información sobre la configuración de un agente personalizado con un reenviador Lambda, con ejemplos del uso de un agente personalizado implementado en Amazon EKS.

Requisitos previos

Rol de ejecución de Lambda

Durante el entrenamiento, la SageMaker IA asume su función de SageMaker ejecución para invocar su reenviador Lambda. La función Lambda requiere su propia función de ejecución para que AWS Lambda pueda ejecutarla. Esta función permite a Lambda ejecutar y escribir 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

Escriba el código de su agente

Su agente debe mostrar un punto de enlace HTTP que acepte las solicitudes de despliegue e interactúe con el servicio RFT Runtime. Para ello, debe llamarlo para obtener una inferencia de modelos e informar de las recompensas una vez finalizada la tarea.

Formato de solicitud de implementación

Su agente recibe las solicitudes del reenviador Lambda en el siguiente formato:

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

Expectativas del agente

Para que Amazon SageMaker AI complete las implementaciones correctamente, su agente debe:

El siguiente ejemplo muestra un agente personalizado con un punto final de 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 las dependencias necesarias en el entorno de su agente:

pip install sagemaker-train fastapi uvicorn openai

Implementación de su agente

Su agente puede ejecutarse en cualquier entorno informático. Necesita acceso a la red saliente para acceder a SageMaker AI Runtime y poder generar informes de inferencias y recompensas. Si el reenviador Lambda llama directamente a su agente, este también debe exponer un punto final HTTP al que se pueda acceder desde Lambda.

Los siguientes pasos implementan un agente personalizado en Amazon EKS.

Cree un clúster de EKS

Para obtener instrucciones completas de configuración de EKS, consulte la guía de introducción de EKS. La siguiente es una configuración mínima para este ejemplo:

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

Actualiza tu kubeconfig para conectarte al clúster:

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

Crea y envía la imagen de tu contenedor:

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

Implemente en 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

Obtenga su terminal de agente:

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

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

Verifique el despliegue:

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

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

Cree su reenviador Lambda

El reenviador Lambda recibe las solicitudes de implementación de SageMaker AI y las reenvía a su agente. Lo principal que hay que personalizar es la _call_agent() función, que traduce la solicitud de despliegue al formato de API del agente.

Si tu agente no tiene un punto de conexión HTTP público, puedes sustituir la llamada HTTP por un SQS send_message y, en _call_agent() su lugar, hacer que tu agente sondee la cola.

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

Implemente la función Lambda:

Package la función Lambda:

zip lambda_forwarder.zip lambda_forwarder.py

Cree la función:

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

Pruebe la 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