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**
| API | Finalidad | ¿Cuándo llamar |
| --- | --- | --- |
| Sample | Utilice el modelo de políticas para obtener una respuesta de inferencia única | En cada turno, su agente necesita la salida del modelo |
| SampleWithResponseStream | Llame al modelo de políticas con respuesta de transmisión (SSE) | Igual que en Sample, cuando necesitas streaming token por token |
| CompleteRollout | Señala que el lanzamiento ha finalizado | Una vez que su agente complete todos los turnos para recibir un aviso |
| UpdateReward | Indica la puntuación de la recompensa durante todo el lanzamiento | Después de calcular la recompensa, normalmente se llama con CompleteRollout |
**nota**
Ambas `Sample` `SampleWithResponseStream` son OpenAI-compatible 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:
+ **Bedrock AgentCore:** despliegue su agente en Amazon Bedrock AgentCore para obtener un alojamiento totalmente gestionado. SageMaker La IA llama a su agente directamente durante la formación sin necesidad de configurar una infraestructura adicional. Esta opción funciona mejor para los agentes creados con el SDK de Strands.
+ **Traiga a su propio agente:** aloje a su agente en cualquier entorno, incluidos Amazon EKS, Amazon EC2 o su propia infraestructura, y conéctelo al ciclo de formación mediante un reenviador Lambda. El reenviador Lambda es una función de Lambda que actúa como puente entre la SageMaker IA y su agente, lo que proporciona a la SageMaker IA una forma coherente de comunicarse con su agente sin necesidad de acceso directo a su infraestructura o credenciales. Esta opción es ideal si desea tener el control total de su entorno de alojamiento o si desea utilizar el marco de agentes que prefiera.
## 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 **AmazonSageMakerJobRuntimeAccess**gestionada 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-train`](https://github.com/aws/sagemaker-python-sdk/tree/master/sagemaker-train)SDK 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:
+ Reciba una solicitud de lanzamiento que contenga un mensaje
+ Llame al modelo de políticas para obtener una respuesta
+ Ejecute acciones en su entorno, como llamadas a herramientas o solicitudes de API
+ Repite los dos pasos anteriores durante varios turnos hasta completar la tarea
+ Devuelve una puntuación de recompensa
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:::runtime/`
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:
+ Acepte las solicitudes de despliegue del reenviador Lambda
+ Llame al RFT Runtime para realizar inferencias de modelos utilizando los encabezados y el token proporcionados
+ Ejecute acciones en su entorno (herramientas, API, etc.)
+ Support múltiples llamadas de inferencia por despliegue (varios turnos)
+ Indique la trayectoria como completa cuando finalice la tarea
+ Envía una puntuación de recompensa al RFT Runtime
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 - <.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": "", "errorMessage": ""}
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
```
**nota**
Esta prueba confirma que Lambda está desplegada y que puede comunicarse con su agente. La implementación no se completará correctamente porque no hay ningún trabajo de formación activo, por lo que se espera que se produzca un error. Consulta los registros de tu agente para confirmar que recibió la solicitud.