本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

为经纪人做好准备

在启动训练作业之前，您需要设置一个可以参与训练循环的代理。本节将向您介绍如何配置和部署代理，无论您是使用 Amazon Bedrock AgentCore 进行托管托管，还是使用自定义代理自带基础设施。

代理代码集成概述

在训练期间， SageMaker AI 会将您的训练数据集中的提示发送给您的代理。您的代理会处理每个提示，调用策略模型进行响应，并在您的环境中执行操作，例如调用工具。策略模型是正在训练的模型。它从你的基础模型开始，当它通过训练循环收到反馈时，它的权重会随着时间的推移而更新。根据结果，您的代理会向 SageMaker AI 报告奖励以完成训练循环。数据集中的所有提示都会重复此操作，并且策略模型会根据收集的奖励随着时间的推移而改进。

SageMaker AI Job 运行时服务

在训练期间，您的代理会与 SageMaker AI Job Runtime Service 通信，调用策略模型进行推理，并将结果报告给培训师。SDK 装饰器会自动处理此集成，但如果您的代理有自定义要求，则可以直接调用 Runtime API。

Endpoint

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 装饰器，请直接调用 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.

代理部署选项

SageMaker AI 支持两种将您的代理连接到训练循环的选项：

场景 1：基岩 AgentCore 运行时上的代理

将您的代理部署到 Amazon Bedrock AgentCore 以实现完全托管的主机。 SageMaker AI 会在模型训练期间调用你的代理。

先决条件

开始之前，请满足以下先决条件。

基岩 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-trainSDK 并将@sagemaker_rft_handler装饰器应用于代理的入口点函数。这个装饰器设置了 HTTP 服务器，该服务器在训练期间 AgentCore 调用，监听传入的部署请求并将提示路由到你的 entrypoint 函数。

安装开发工具包

使用以下方法之一在代理环境中安装 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 按照 AgentCore 开发指南将您的代理部署到 Amazon Bedrock。

使用 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、Far AWS gate 或你自己的基础设施。

Lambda 函数接收来自 SageMaker AI 的部署请求并将其转发到您的代理的 HTTP 终端节点。

以下各节提供了有关使用 Lambda 转发器设置自定义代理的更多信息，并提供了使用部署在 Amazon EKS 上的自定义代理的示例。

先决条件

Lambda 执行角色

在训练期间， SageMaker AI 会扮演您的 SageMaker 执行角色来调用您的 Lambda 转发器。Lambda 函数需要自己的执行角色，这样 Lambda 才能运行 AWS 它。此角色允许 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 Runtime 服务交互的 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 终端节点，则可以用 SQS 替换 HTTP 调用send_message，改为让您的代理轮询队列。_call_agent()

"""
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 函数：

Package 将 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