Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.

Preparazione del tuo agente

Prima di avviare un lavoro di formazione, è necessario creare un agente che possa partecipare al ciclo di formazione. Questa sezione illustra come configurare e distribuire il tuo agente, sia che utilizzi Amazon Bedrock AgentCore per l'hosting gestito sia che utilizzi la tua infrastruttura con un agente personalizzato.

Panoramica sull'integrazione del codice dell'agente

Durante la formazione, l' SageMaker intelligenza artificiale invia al tuo agente i prompt dal tuo set di dati di addestramento. Il tuo agente elabora ogni richiesta, richiama il modello di policy per una risposta e intraprende azioni nel tuo ambiente, ad esempio chiamando gli strumenti. Il modello di policy è il modello che viene addestrato. Inizia come modello base e i suoi pesi vengono aggiornati nel tempo man mano che riceve feedback durante il ciclo di allenamento. In base al risultato, il tuo agente segnala una ricompensa all' SageMaker IA per aver completato il ciclo di formazione. Ciò si ripete per tutti i prompt del set di dati e il modello di policy migliora nel tempo in base ai premi raccolti.

SageMaker Servizio AI Job Runtime

Durante la formazione, il tuo agente comunica con l' SageMaker AI Job Runtime Service per richiamare il modello di policy per l'inferenza e riportare i risultati al trainer. Il decoratore SDK gestisce questa integrazione automaticamente, ma se il tuo agente ha requisiti personalizzati, puoi chiamare direttamente le API di Runtime.

Endpoint

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

API

Autenticazione

Il tuo agente si autentica con il servizio Runtime utilizzando token bearer. Genera un token utilizzando l'SDK:

from sagemaker.core.token_generator import generate_token

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

Passa questo token come chiave API quando chiami il Runtime Service.

Integrazione diretta delle API (avanzata)

Se il tuo framework di agenti non può utilizzare il decoratore SDK, chiama direttamente le 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.

Opzioni di distribuzione degli agenti

SageMaker L'intelligenza artificiale supporta due opzioni per collegare l'agente al ciclo di formazione:

Scenario 1: Agents on Bedrock Runtime AgentCore

Implementa il tuo agente su Amazon Bedrock AgentCore per un hosting completamente gestito. SageMaker L'intelligenza artificiale richiama il tuo agente durante la formazione del modello.

Prerequisiti

Prima di iniziare, completa i seguenti prerequisiti.

Ruolo esecutivo di Bedrock AgentCore

Durante l'addestramento, l' SageMaker IA assume il tuo ruolo di SageMaker esecuzione per invocare il tuo agente. L'agente ha inoltre bisogno di un proprio ruolo separato, chiamato ruolo di AgentCore esecuzione Bedrock, per chiamare RFT Runtime per l'inferenza dei modelli e la segnalazione dei premi. RFT Runtime è l'endpoint del servizio di SageMaker intelligenza artificiale con cui il tuo agente comunica durante la formazione. Si occupa di due aspetti: fornisce le risposte del modello di policy durante l'inferenza e riceve la ricompensa riportata dall'agente alla fine di ogni implementazione.

Crea il ruolo di AgentCore esecuzione Bedrock con la seguente politica di fiducia. Questa politica concede a Bedrock AgentCore il permesso di assumere il ruolo per tuo conto durante la formazione. Senza di essa, Bedrock AgentCore non può accedere all'immagine del contenitore o chiamare RFT Runtime.

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

Quindi collega la politica AmazonSageMakerJobRuntimeAccessgestita a questo ruolo. Ciò concede le autorizzazioni necessarie all'agente per richiamare il modello di policy per l'inferenza e riportare i risultati durante la formazione.

Scrivi o aggiorna il codice del tuo agente

Il tuo agente deve utilizzare l'sagemaker-trainSDK e applicare il @sagemaker_rft_handler decoratore alla funzione del punto di ingresso del tuo agente. Questo decoratore configura il server HTTP che viene AgentCore richiamato durante l'allenamento, ascolta le richieste di implementazione in arrivo e indirizza i prompt alla funzione entrypoint.

Installazione dell'SDK

Installa l'SDK nell'ambiente del tuo agente utilizzando uno dei seguenti metodi.

Opzione 1: installazione diretta

pip install sagemaker-train

Opzione 2: Costruisci e installa le ruote singolarmente. Utilizzatela se l'installazione diretta supera i limiti di dimensione del vostro 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

Autenticazione

Il tuo agente si autentica con SageMaker l'intelligenza artificiale utilizzando token bearer. Genera un token utilizzando il generate_token() metodo dell'SDK e passalo come chiave API quando richiami il modello di policy per l'inferenza.

Requisiti dell'agente

Il tuo agente deve eseguire le seguenti operazioni per ogni implementazione:

L'esempio seguente mostra un modello di agente di base che utilizza Strands SDK:

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

Distribuisci il tuo agente su Bedrock AgentCore

Distribuisci il tuo agente su Amazon Bedrock AgentCore seguendo la guida allo AgentCore sviluppo.

Dopo aver distribuito l'agente utilizzando la AgentCore CLI, annota l'ARN di Agent Runtime dall'output. Ne hai bisogno quando crei il tuo lavoro di formazione. Il Runtime ARN segue il formato di. arn:aws:bedrock-agentcore:<region>:<account-id>:runtime/<agent-name>

Verifica che il tuo agente sia distribuito e funzionante:

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

Scenario 2: agente personalizzato con Lambda Forwarder

Oltre a Bedrock AgentCore, l' SageMaker intelligenza artificiale supporta agenti personalizzati ospitati in qualsiasi ambiente. Il tuo agente si connette al ciclo di formazione tramite uno spedizioniere Lambda, offrendoti la flessibilità di utilizzare qualsiasi framework di agenti e piattaforma di hosting di tua scelta.

Il tuo agente può essere creato utilizzando qualsiasi framework o piattaforma, come Strands Agents SDK, o la tua implementazione personalizzata. Può essere eseguito su qualsiasi ambiente di elaborazione tra cui Amazon Bedrock, AgentCore Amazon EKS, Amazon EC2 AWS , Fargate o sulla tua infrastruttura.

La funzione Lambda riceve le richieste di rollout dall' SageMaker IA e le inoltra all'endpoint HTTP del tuo agente.

Le sezioni seguenti forniscono ulteriori informazioni sulla configurazione di un agente personalizzato con uno spedizioniere Lambda, con esempi di utilizzo di un agente personalizzato distribuito su Amazon EKS.

Prerequisiti

Ruolo di esecuzione Lambda

Durante l'addestramento, l' SageMaker IA assume il tuo ruolo di SageMaker esecuzione per richiamare il tuo spedizioniere Lambda. La funzione Lambda richiede il proprio ruolo di esecuzione in modo che AWS Lambda possa eseguirla. Questo ruolo consente a Lambda di eseguire e scrivere registri.

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

Scrivi il codice del tuo agente

L'agente deve esporre un endpoint HTTP che accetta le richieste di implementazione e interagisce con il servizio RFT Runtime, chiamandolo per l'inferenza del modello e segnalando i premi una volta completata l'attività.

Formato della richiesta di implementazione

Il tuo agente riceve le richieste nel seguente formato dallo spedizioniere Lambda:

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

Aspettative degli agenti

Affinché Amazon SageMaker AI completi correttamente le implementazioni, il tuo agente deve:

L'esempio seguente mostra un agente personalizzato con un 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"}

Installa le dipendenze richieste nell'ambiente del tuo agente:

pip install sagemaker-train fastapi uvicorn openai

Implementa il tuo agente

Il tuo agente può funzionare in qualsiasi ambiente di elaborazione. Ha bisogno dell'accesso alla rete in uscita per raggiungere SageMaker AI Runtime per la generazione di report di inferenza e ricompensa. Se il server di inoltro Lambda chiama direttamente il tuo agente, quest'ultimo deve anche esporre un endpoint HTTP raggiungibile da Lambda.

I seguenti passaggi distribuiscono un agente personalizzato in Amazon EKS.

Crea un cluster EKS

Per istruzioni complete sulla configurazione EKS, consulta la guida introduttiva di EKS. Di seguito è riportata una configurazione minima per questo esempio:

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

Aggiorna kubeconfig per connetterti al cluster:

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

Crea e invia l'immagine del tuo contenitore:

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

Implementa su 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

Ottieni l'endpoint del tuo agente:

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

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

Verifica la distribuzione:

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

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

Crea il tuo spedizioniere Lambda

Lo spedizioniere Lambda riceve le richieste di implementazione dall' SageMaker IA e le inoltra al tuo agente. La cosa principale da personalizzare è la _call_agent() funzione, che traduce la richiesta di implementazione nel formato API dell'agente.

Se il tuo agente non dispone di un endpoint HTTP pubblico, puoi sostituire la chiamata HTTP in entrata _call_agent() con un SQS send_message e fare in modo che l'agente effettui invece il polling della coda.

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

Implementa la funzione Lambda:

Package della funzione Lambda:

zip lambda_forwarder.zip lambda_forwarder.py

Crea la funzione:

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

Prova 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