Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

Préparation de votre agent

Avant de lancer une tâche de formation, vous devez configurer un agent capable de participer à la boucle de formation. Cette section explique comment configurer et déployer votre agent, que vous utilisiez Amazon Bedrock AgentCore pour l'hébergement géré ou que vous utilisiez votre propre infrastructure avec un agent personnalisé.

Vue d'ensemble de l'intégration du code agent

Pendant l'entraînement, l' SageMaker IA envoie des instructions à votre agent depuis votre ensemble de données de formation. Votre agent traite chaque demande, appelle le modèle de politique pour obtenir une réponse et prend des mesures dans votre environnement, par exemple en appelant des outils. Le modèle de politique est le modèle en cours de formation. Cela commence lorsque votre modèle de base et ses poids sont mis à jour au fil du temps à mesure qu'il reçoit des commentaires tout au long de la boucle d'entraînement. Sur la base du résultat, votre agent rapporte une récompense à l' SageMaker IA pour terminer la boucle d'entraînement. Cela se répète pour toutes les instructions de votre ensemble de données, et le modèle de politique s'améliore au fil du temps en fonction des récompenses collectées.

SageMaker Service d'exécution de tâches AI

Pendant la formation, votre agent communique avec le service SageMaker AI Job Runtime pour appeler le modèle de politique à des fins d'inférence et communiquer les résultats au formateur. Le décorateur du SDK gère automatiquement cette intégration, mais si votre agent a des exigences personnalisées, vous pouvez appeler directement les API Runtime.

Point de terminaison

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

API

Authentification

Votre agent s'authentifie auprès du Runtime Service à l'aide de jetons porteurs. Générez un jeton à l'aide du SDK :

from sagemaker.core.token_generator import generate_token

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

Transmettez ce jeton comme clé d'API lorsque vous appelez le service Runtime.

Intégration directe à l'API (avancée)

Si votre infrastructure d'agents ne peut pas utiliser le décorateur du SDK, appelez directement les 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.

Options de déploiement de l'agent

SageMaker L'IA propose deux options pour connecter votre agent à la boucle d'entraînement :

Scénario 1 : Agents sur Bedrock Runtime AgentCore

Déployez votre agent sur Amazon Bedrock AgentCore pour un hébergement entièrement géré. SageMaker L'IA invoque votre agent pendant la formation du modèle.

Conditions préalables

Avant de commencer, remplissez les conditions préalables suivantes.

Rôle d' AgentCore exécution de Bedrock

Pendant la formation, l' SageMaker IA assume votre rôle SageMaker d'exécution pour invoquer votre agent. Votre agent a également besoin de son propre rôle distinct, appelé rôle d' AgentCore exécution Bedrock, pour appeler le RFT Runtime à des fins d'inférence de modèles et de rapports sur les récompenses. Le RFT Runtime est le point de terminaison du service d' SageMaker intelligence artificielle avec lequel votre agent communique pendant l'entraînement. Il gère deux choses : envoyer les réponses issues du modèle de politique lors de l'inférence et recevoir la récompense que votre agent rapporte à la fin de chaque déploiement.

Créez le rôle AgentCore d'exécution Bedrock avec la politique de confiance suivante. Cette politique autorise Bedrock AgentCore à assumer le rôle en votre nom pendant la formation. Sans cela, Bedrock AgentCore ne peut pas accéder à l'image de votre conteneur ni appeler le RFT Runtime.

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

Attachez ensuite la politique AmazonSageMakerJobRuntimeAccessgérée à ce rôle. Cela donne les autorisations dont votre agent a besoin pour appeler le modèle de politique à des fins d'inférence et pour communiquer les résultats pendant la formation.

Rédigez ou mettez à jour votre code d'agent

Votre agent doit utiliser le sagemaker-trainSDK et appliquer le @sagemaker_rft_handler décorateur à la fonction de point d'entrée de votre agent. Ce décorateur configure le serveur HTTP qui est AgentCore invoqué pendant l'entraînement, écoute les demandes de déploiement entrantes et achemine les instructions vers votre fonction de point d'entrée.

Installer le SDK

Installez le SDK dans l'environnement de votre agent à l'aide de l'une des méthodes suivantes.

Option 1 : installation directe

pip install sagemaker-train

Option 2 : Construisez et installez les roues individuellement. Utilisez-le si l'installation directe dépasse les contraintes de taille de votre environnement.

# 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

Authentification

Votre agent s'authentifie auprès de l' SageMaker IA à l'aide de jetons porteurs. Générez un jeton à l'aide de la generate_token() méthode du SDK et transmettez-le comme clé d'API lorsque vous appelez le modèle de politique à des fins d'inférence.

Exigences relatives aux agents

Votre agent doit effectuer les opérations suivantes pour chaque déploiement :

L'exemple suivant montre un modèle d'agent de base utilisant le 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)}

Déployez votre agent sur Bedrock AgentCore

Déployez votre agent sur Amazon Bedrock en AgentCore suivant le guide de AgentCore développement.

Une fois que vous avez déployé votre agent à l'aide de la AgentCore CLI, notez l'ARN d'exécution de l'agent indiqué dans la sortie. Vous en avez besoin lors de la création de votre poste de formation. L'ARN d'exécution suit le format dearn:aws:bedrock-agentcore:<region>:<account-id>:runtime/<agent-name>.

Vérifiez que votre agent est déployé et qu'il fonctionne correctement :

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

Scénario 2 : agent personnalisé avec le redirecteur Lambda

Outre Bedrock AgentCore, SageMaker AI prend en charge les agents personnalisés hébergés dans n'importe quel environnement. Votre agent se connecte à la boucle d'entraînement par le biais d'un redirecteur Lambda, ce qui vous donne la flexibilité d'utiliser le framework d'agent et la plateforme d'hébergement de votre choix.

Votre agent peut être créé à l'aide de n'importe quel framework ou plate-forme, tel que le SDK Strands Agents, ou de votre propre implémentation personnalisée. Il peut fonctionner sur n'importe quel environnement informatique, y compris Amazon Bedrock AgentCore, Amazon EKS, Amazon EC2, AWS Fargate, ou sur votre propre infrastructure.

La fonction Lambda reçoit les demandes de déploiement de l' SageMaker IA et les transmet au point de terminaison HTTP de votre agent.

Les sections suivantes fournissent plus d'informations sur la configuration d'un agent personnalisé avec un redirecteur Lambda, avec des exemples d'utilisation d'un agent personnalisé déployé sur Amazon EKS.

Conditions préalables

Rôle d’exécution Lambda

Pendant la formation, l' SageMaker IA assume votre rôle SageMaker d'exécution pour appeler votre redirecteur Lambda. La fonction Lambda nécessite son propre rôle d'exécution afin que AWS Lambda puisse l'exécuter. Ce rôle permet à Lambda d'exécuter et d'écrire des journaux.

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

Rédigez votre code d'agent

Votre agent doit exposer un point de terminaison HTTP qui accepte les demandes de déploiement et interagit avec le service RFT Runtime, en l'appelant pour l'inférence du modèle et en signalant les récompenses une fois la tâche terminée.

Format de demande de déploiement

Votre agent reçoit des demandes au format suivant de la part du redirecteur Lambda :

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

Les attentes des agents

Pour qu'Amazon SageMaker AI puisse mener à bien ses déploiements, votre agent doit :

L'exemple suivant montre un agent personnalisé doté d'un point de terminaison 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"}

Installez les dépendances requises dans votre environnement d'agent :

pip install sagemaker-train fastapi uvicorn openai

Déployez votre agent

Votre agent peut fonctionner sur n'importe quel environnement informatique. Il a besoin d'un accès au réseau sortant pour atteindre l' SageMaker AI Runtime à des fins d'inférence et de génération de rapports sur les récompenses. Si le redirecteur Lambda appelle directement votre agent, celui-ci doit également exposer un point de terminaison HTTP accessible depuis Lambda.

Les étapes suivantes permettent de déployer un agent personnalisé sur Amazon EKS.

Création d'un cluster EKS

Pour obtenir des instructions complètes de configuration d'EKS, consultez le guide de démarrage d'EKS. Voici une configuration minimale pour cet exemple :

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

Mettez à jour votre kubeconfig pour vous connecter au cluster :

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

Créez et diffusez l'image de votre conteneur :

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

Déployez sur 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

Obtenez le point de terminaison de votre agent :

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

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

Vérifiez le déploiement :

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

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

Créez votre redirecteur Lambda

Le redirecteur Lambda reçoit les demandes de déploiement de l' SageMaker IA et les transmet à votre agent. La principale chose à personnaliser est la _call_agent() fonction, qui traduit la demande de déploiement dans le format API de votre agent.

Si votre agent ne possède pas de point de terminaison HTTP public, vous pouvez remplacer l'appel HTTP entrant par un SQS send_message et demander _call_agent() à votre agent d'interroger la file d'attente à la place.

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

Déployez la fonction Lambda :

Package de la fonction Lambda :

zip lambda_forwarder.zip lambda_forwarder.py

Créez la fonction :

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

Testez le 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