기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다.

에이전트 준비

훈련 작업을 시작하려면 먼저 훈련 루프에 참여할 수 있는 에이전트를 설정해야 합니다. 이 섹션에서는 관리형 호스팅에 Amazon Bedrock AgentCore를 사용하거나 사용자 지정 에이전트를 통해 자체 인프라를 가져오는 등 에이전트를 구성하고 배포하는 방법을 안내합니다.

에이전트 코드 통합 개요

훈련 중에 SageMaker AI는 훈련 데이터 세트의 프롬프트를 에이전트에게 전송합니다. 에이전트는 각 프롬프트를 처리하고, 정책 모델을 호출하여 응답을 받고, 사용자 환경에서 도구 호출과 같은 작업을 수행합니다. 정책 모델은 훈련 중인 모델입니다. 기본 모델로 시작되며 훈련 루프를 통해 피드백을 받으면 시간이 지남에 따라 가중치가 업데이트됩니다. 결과에 따라 에이전트는 SageMaker AI에 보상을 보고하여 훈련 루프를 완료합니다. 이는 데이터 세트의 모든 프롬프트에 대해 반복되며, 정책 모델은 수집된 보상에 따라 시간이 지남에 따라 개선됩니다.

SageMaker AI 작업 런타임 서비스

훈련 중에 에이전트는 SageMaker AI 작업 런타임 서비스와 통신하여 추론을 위해 정책 모델을 호출하고 결과를 트레이너에게 다시 보고합니다. SDK 데코레이터는이 통합을 자동으로 처리하지만 에이전트에 사용자 지정 요구 사항이 있는 경우 런타임 APIs 직접 호출할 수 있습니다.

Endpoint

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

API

Authentication

에이전트는 보유자 토큰을 사용하여 런타임 서비스로 인증합니다. 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 실행 역할을 수임하여 에이전트를 호출합니다. 또한 에이전트는 모델 추론 및 보상 보고를 위해 RFT 런타임을 호출하기 위해 Bedrock AgentCore 실행 역할이라는 별도의 역할이 필요합니다. 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 설치

다음 방법 중 하나를 사용하여 에이전트 환경에 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

Authentication

에이전트는 보유자 토큰을 사용하여 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 전달자를 사용하는 사용자 지정 에이전트

SageMaker AI는 Bedrock AgentCore 외에도 모든 환경에서 호스팅되는 사용자 지정 에이전트를 지원합니다. 에이전트는 Lambda 전달자를 통해 훈련 루프에 연결되므로 원하는 에이전트 프레임워크와 호스팅 플랫폼을 유연하게 사용할 수 있습니다.

에이전트는 Strands Agents SDK 또는 사용자 지정 구현과 같은 모든 프레임워크 또는 플랫폼을 사용하여 빌드할 수 있습니다. Amazon Bedrock AgentCore, Amazon EKS, Amazon EC2, AWS Fargate 또는 자체 인프라를 포함한 모든 컴퓨팅 환경에서 실행할 수 있습니다.

Lambda 함수는 SageMaker AI로부터 롤아웃 요청을 수신하여 에이전트의 HTTP 엔드포인트로 전달합니다.

다음 섹션에서는 Amazon EKS에 배포된 사용자 지정 에이전트를 사용하는 예제와 함께 Lambda 전달자를 사용하여 사용자 지정 에이전트를 설정하는 방법에 대한 자세한 정보를 제공합니다.

사전 조건

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로부터 롤아웃 요청을 수신하여 에이전트에게 전달합니다. 사용자 지정해야 할 주요 사항은 롤아웃 요청을 에이전트의 API 형식으로 변환하는 _call_agent() 함수입니다.

에이전트에 퍼블릭 HTTP 엔드포인트가 없는 경우의 HTTP 호출을 SQS_call_agent()로 바꾸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