本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

準備您的代理程式

您必須先設定可以參與訓練循環的代理程式，才能啟動訓練任務。本節將逐步解說如何設定和部署您的代理程式，無論您是使用 Amazon Bedrock AgentCore 進行受管託管，還是使用自訂代理程式使用您自己的基礎設施。

客服人員程式碼整合概觀

在訓練期間，SageMaker AI 會從訓練資料集傳送提示給您的代理程式。您的客服人員會處理每個提示、呼叫政策模型以進行回應，並在您的環境中採取動作，例如呼叫工具。政策模型是正在訓練的模型。它從基礎模型開始，其權重會在透過訓練迴圈接收意見回饋時隨著時間更新。根據結果，您的代理程式會將獎勵回報給 SageMaker AI 以完成訓練迴圈。這會重複資料集中的所有提示，而政策模型會根據收集的獎勵隨著時間而改善。

SageMaker AI 任務執行期服務

在訓練期間，您的代理程式會與 SageMaker AI Job Runtime Service 通訊，以呼叫政策模型進行推論，並將結果回報給培訓人員。SDK 裝飾項目會自動處理此整合，但如果您的代理程式有自訂需求，您可以直接呼叫執行時間 APIs。

端點

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

API

身分驗證

您的代理程式會使用承載字符透過執行期服務進行驗證。使用 SDK 產生權杖：

from sagemaker.core.token_generator import generate_token

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

呼叫 執行期服務時，將此字符做為 API 金鑰傳遞。

直接 API 整合 （進階）

如果您的代理程式架構無法使用 SDK 裝飾項目，請直接呼叫 APIs：

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.

代理程式部署選項

SageMaker AI 支援將代理程式連接到訓練迴圈的兩個選項：

案例 1：Bedrock AgentCore 執行期上的代理程式

將您的代理程式部署到 Amazon Bedrock AgentCore 以進行全受管託管。SageMaker AI 會在模型訓練期間叫用您的代理程式。

先決條件

開始之前，請先完成下列先決條件。

Bedrock AgentCore 執行角色

在訓練期間，SageMaker AI 會擔任您的 SageMaker 執行角色來叫用您的代理程式。您的代理程式也需要自己的獨立角色，稱為 Bedrock AgentCore 執行角色，以呼叫 RFT 執行期進行模型推論和獎勵報告。RFT 執行期是您的代理程式在訓練期間與 通訊的 SageMaker AI 服務端點。它會處理兩件事：在推論期間提供政策模型的回應，並在每次推展結束時收到客服人員報告的獎勵。

使用下列信任政策建立 Bedrock AgentCore 執行角色。此政策授予 Bedrock AgentCore 在訓練期間代表您擔任角色的許可。如果沒有它， Bedrock AgentCore 無法存取您的容器映像或呼叫 RFT 執行期。

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

然後將 AmazonSageMakerJobRuntimeAccess 受管政策連接至此角色。這會授予您的代理程式在訓練期間呼叫政策模型以進行推論和報告結果所需的許可。

撰寫或更新代理程式程式碼

您的代理程式必須使用 sagemaker-train SDK，並將@sagemaker_rft_handler裝飾項目套用至代理程式的進入點函數。此裝飾項目會設定 AgentCore 在訓練期間叫用的 HTTP 伺服器，接聽傳入的推展請求，並將提示路由到您的進入點函數。

安裝軟體開發套件

使用下列其中一種方法，在您的代理程式環境中安裝 SDK。

選項 1：直接安裝

pip install sagemaker-train

選項 2：個別建置和安裝車輪。如果直接安裝超過您環境的大小限制，請使用此選項。

# 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

身分驗證

您的代理程式使用承載字符向 SageMaker AI 進行驗證。使用 SDK 的 generate_token()方法產生權杖，並在呼叫政策模型進行推論時將其做為 API 金鑰傳遞。

客服人員需求

您的代理程式必須針對每個推展執行下列動作：

下列範例顯示使用 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)}

將您的代理程式部署到 Bedrock AgentCore

遵循 AgentCore 開發指南，將您的代理程式部署到 Amazon Bedrock AgentCore。

使用 AgentCore CLI 部署代理程式後，請注意輸出中的代理程式執行期 ARN。您在建立訓練任務時需要此項目。執行期 ARN 遵循 的格式arn:aws:bedrock-agentcore:<region>:<account-id>:runtime/<agent-name>。

確認您的代理程式已部署且運作狀態良好：

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

案例 2：使用 Lambda 轉送器的自訂代理程式

除了 Bedrock AgentCore 之外，SageMaker AI 還支援在任何環境中託管的自訂代理程式。您的代理程式會透過 Lambda 轉寄站連線到訓練迴圈，讓您靈活地使用任何代理程式架構和託管平台。

您可以使用任何架構或平台來建置您的代理程式，例如 Strands Agents SDK，或您自己的自訂實作。它可以在任何運算環境中執行，包括 Amazon Bedrock AgentCore、Amazon EKS、Amazon EC2、 AWS Fargate 或您自己的基礎設施。

Lambda 函數會從 SageMaker AI 接收推展請求，並將其轉送至代理程式的 HTTP 端點。

下列各節提供使用 Lambda 轉送器設定自訂代理程式的詳細資訊，以及使用部署在 Amazon EKS 上的自訂代理程式的範例。

先決條件

Lambda 執行角色

在訓練期間，SageMaker AI 會擔任您的 SageMaker 執行角色來叫用 Lambda 轉寄站。Lambda 函數需要自己的執行角色，以便 AWS Lambda 可以執行它。此角色允許 Lambda 執行和寫入日誌。

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

撰寫您的代理程式程式碼

您的代理程式必須公開接受推展請求並與 RFT 執行期服務互動的 HTTP 端點，方法是呼叫它進行模型推論，並在任務完成後報告獎勵。

推展請求格式

您的代理程式會從 Lambda 轉送器收到下列格式的請求：

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

客服人員期望

若要讓 Amazon SageMaker AI 成功完成推展，您的代理程式必須：

下列範例顯示具有 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"}

在您的代理程式環境中安裝必要的相依性：

pip install sagemaker-train fastapi uvicorn openai

部署您的代理程式

您的代理程式可以在任何運算環境中執行。它需要傳出網路存取權才能到達 SageMaker AI 執行期，以進行推論和獎勵報告。如果 Lambda 轉送器直接呼叫您的代理程式，您的代理程式也必須公開可從 Lambda 連線的 HTTP 端點。

下列步驟會將自訂代理程式部署到 Amazon EKS。

建立 EKS 叢集

如需完整的 EKS 設定說明，請參閱 EKS 入門指南。以下是此範例的最低設定：

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

更新您的 kubeconfig 以連線至叢集：

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

建置和推送您的容器映像：

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

部署至 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

取得您的代理程式端點：

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

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

驗證部署：

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

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

建立 Lambda 轉送器

Lambda 轉送器會收到來自 SageMaker AI 的推展請求，並將其轉送到您的代理程式。要自訂的主要內容是 _call_agent()函數，其會將推展請求轉譯為代理程式的 API 格式。

如果您的客服人員沒有公有 HTTP 端點，您可以將 中的 HTTP 呼叫取代_call_agent()為 SQS，send_message並讓客服人員輪詢佇列。

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

部署 Lambda 函數：

封裝 Lambda 函數：

zip lambda_forwarder.zip lambda_forwarder.py

建立 函數：

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

測試 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