如何用 Amazon SageMaker HyperPod Data Capture 构建模型改进飞轮

部署在 Amazon SageMaker HyperPod 上的基础模型会遇到任何预先策划的数据集都未曾覆盖的输入：领域专用术语、意外格式、新产品以及不断变化的用户行为。如果没有生产推理数据的记录，团队就无法检测漂移、发现失败模式，或利用真实交互来改进模型。

Amazon SageMaker HyperPod 现在提供 Data Capture（数据捕获）功能，这是一项内置功能，可将实时推理输入和输出记录到 Amazon Simple Storage Service (Amazon S3)，提供三个独立层级，对延迟的影响极小，且对模型可用性零影响。

在本文中，我们将通过单个 SageMaker HyperPod 集群上的一个完整示例进行讲解。一个家电支持路由器捕获其生产流量，Amazon Athena 查询发现一个新产品的盲点，一条 SQL 策略为捕获的记录打标签，然后基于这些数据进行微调，让路由器学会这个新产品。我们在模型部署之前不可能存在的数据上，测量微调前后的准确率。推理运维方在一个节点上运行模型服务，训练运维方在另一个节点上运行微调。

我们在演练中使用合成数据。脚本、清单、SQL、合成数据集和预期输出可在随附的 GitHub 仓库中获取。

三层 Data Capture 架构

HyperPod Data Capture 提供三个独立层级，各自位于请求路径的不同位置。你可以在同一个 dataCapture 块中开启任意组合。

层级 1 和层级 3 都存储完整负载，因此同时开启两者会将每条记录存储两次。将层级 1 用于合规和 Model Monitor，当你只需要流量元数据时使用层级 2，当你打算使用这些负载时使用层级 3。本文仅使用层级 3。由于层级 3 记录完整的负载，其中可能包含姓名、地址和订单号，因此在这些数据成为训练数据之前要加以保护。使用采样、AWS Key Management Service (AWS KMS) 加密（kmsKeyId）、S3 生命周期规则以及个人身份信息 (PII) 脱敏（例如使用 Amazon Comprehend）。

前提条件

• 一个由 Amazon Elastic Kubernetes Service (Amazon EKS) 编排的 SageMaker HyperPod 集群，已安装推理运维方（Setting up your HyperPod clusters for model deployment）以及训练运维方 EKS 附加组件 amazon-sagemaker-hyperpod-training-operator（Installing the training operator）。
• 两块可用的 NVIDIA GPU，一块用于服务，一块用于训练。我们使用 ml.g6e.8xlarge 进行服务，使用 ml.g6e.xlarge 进行训练（各一块 48 GB NVIDIA L40S）。服务 Pod 在端点的整个生命周期内占用其 GPU，因此如果只有一块 GPU，训练 Pod 将一直处于 Pending 状态。
• 已为集群配置 kubectl。
• 同一区域内的 S3 存储桶。

确认两个操作器均已安装。该命令会返回两行：

kubectl get crd | grep -E 'inferenceendpointconfigs|hyperpodpytorchjobs'
hyperpodpytorchjobs.sagemaker.amazonaws.com                         2026-09-21T10:57:14Z
inferenceendpointconfigs.inference.sagemaker.aws.amazon.com         2026-08-06T11:58:53Z

解决方案概览

下图展示了在一个 SageMaker HyperPod 集群上演练的五个步骤：将每个请求和响应捕获到 Amazon S3，用 Athena 分析捕获内容，用 SQL 策略为其打标，在第二个 GPU 上用训练操作器进行微调，然后在捕获功能仍然开启的情况下并排重新部署。

图 1：单个 SageMaker HyperPod 集群上的五步飞轮，从捕获到 Athena 分析、打标、微调和重新部署

推理操作器为模型 pod 添加三个 sidecar。每个经过 NGINX 代理的请求都会被复制到 OpenTelemetry 收集器，进行批处理，并通过 Mountpoint-S3 卷写入 S3。Athena 直接就地读取这些文件。训练操作器在第二个 GPU 上将微调作为 HyperPodPyTorchJob 运行，新模型由第二个端点提供服务，该端点同样开启了捕获功能。

步骤 0：设置变量

后续每个步骤都会读取这些变量。编辑第一个代码块，然后粘贴整个内容：

# --- edit these ---
export EKS_CLUSTER_NAME="my-hyperpod-eks"
export AWS_REGION="us-east-1"
export BUCKET="amzn-s3-demo-bucket"
export SERVE_INSTANCE_TYPE="ml.g6e.8xlarge"     # node type for the first endpoint
export TRAIN_INSTANCE_TYPE="ml.g6e.xlarge"      # node type for the fine-tuning job
# --- leave as-is ---
export AWS_DEFAULT_REGION="$AWS_REGION"
export ENDPOINT_NAME="appliance-router"
export MODEL_LOCATION="models/qwen2.5-3b-instruct"
export MODEL_V2_LOCATION="models/qwen2.5-3b-router-v2"
export TRAIN_SA="flywheel"                     # service account with S3 access via EKS Pod Identity
aws eks update-kubeconfig --name "$EKS_CLUSTER_NAME" --region "$AWS_REGION"
render() { python3 -c 'import os,sys; sys.stdout.write(os.path.expandvars(sys.stdin.read()))'; }

步骤 1：将基础模型暂存到 S3

我们使用 Qwen2.5-3B-Instruct，因为它足够小，大约一分钟内即可完成微调，并且能装进单个 L40S。

pip install -q huggingface_hub
python3 -c "from huggingface_hub import snapshot_download; \
  snapshot_download('Qwen/Qwen2.5-3B-Instruct', local_dir='qwen2.5-3b-instruct')"
aws s3 sync qwen2.5-3b-instruct "s3://${BUCKET}/${MODEL_LOCATION}/" --exclude ".cache/*"

步骤 2：IAM 权限

四个身份需要访问权限。有关策略声明，请参阅 HyperPod 上的推理数据捕获。

• Mountpoint-S3 CSI 驱动程序角色。Tier 3 捕获通过 Mountpoint-S3 卷写入，驱动程序以自己的 IAM roles for service accounts (IRSA) 角色进行身份验证，而不是以 pod 的服务账户身份。它需要对您的捕获存储桶拥有 s3:PutObject、s3:AbortMultipartUpload 和 s3:DeleteObject 权限，如果您设置了 kmsKeyId，还需要 kms:Decrypt、kms:GenerateDataKey 和 kms:DescribeKey 权限。如果缺少这些权限，端点将保持健康且请求成功，但不会出现任何捕获内容。请参阅 Mountpoint for Amazon S3 CSI driver。
• 推理操作器执行角色。它需要对您的捕获存储桶拥有 s3:PutObject 权限，如果您设置了 kmsKeyId，还需要 kms:Decrypt 和 kms:GenerateDataKey 权限。捕获文件本身是通过 CSI 驱动程序角色写入的，因此当没有捕获内容出现时，请先检查该角色。
• 训练 pod 的服务账户（$TRAIN_SA）。通过 EKS Pod Identity 获得对基础模型和训练数据的读取权限，以及对合并后模型的写入权限。
• Athena。对捕获前缀的读取权限，以及一个查询结果位置。

启用数据捕获

接下来的三个步骤部署开启了 Tier 3 捕获的路由器，检查 pod 是否运行捕获 sidecar，并确认记录到达 Amazon S3。

步骤 3：部署开启了 Tier 3 捕获的模型

这就是整个功能的核心：dataCapture.modelPod.enabled: true 块。其余部分都是普通的 vLLM 部署。

cat > endpoint.yaml <<'YAML'
apiVersion: inference.sagemaker.aws.amazon.com/v1
kind: InferenceEndpointConfig
metadata:
  name: ${ENDPOINT_NAME}
  namespace: default
spec:
  replicas: 1
  modelName: ${ENDPOINT_NAME}
  endpointName: ${ENDPOINT_NAME}
  instanceType: ${SERVE_INSTANCE_TYPE}
  invocationEndpoint: v1/chat/completions
  modelSourceConfig:
    modelSourceType: s3
    modelLocation: ${MODEL_LOCATION}
    s3Storage:
      bucketName: ${BUCKET}
      region: ${AWS_REGION}

  # >>> This is all you need to turn on Tier 3 data capture <<<
  dataCapture:
    s3Uri: "s3://${BUCKET}/capture/"
    modelPod:
      enabled: true
      initialSamplingPercentage: 100
      bufferConfig:
        batchSize: 50
        flushIntervalSeconds: 60
      payloadConfig:
        maxPayloadSizeKB: 1024

  worker:
    image: vllm/vllm-openai:v0.10.0
    modelInvocationPort:
      containerPort: 8000
      name: http
    modelVolumeMount:
      mountPath: /opt/ml/model
      name: model-weights
    args: ["--model", "/opt/ml/model", "--served-model-name", "${ENDPOINT_NAME}",
           "--port", "8000", "--max-model-len", "2048", "--gpu-memory-utilization", "0.90"]
    resources:
      requests: { cpu: "2", memory: 16Gi, nvidia.com/gpu: 1 }
      limits:   { nvidia.com/gpu: 1 }
YAML
render < endpoint.yaml | kubectl apply -f -

前面的 modelPod 块设置了 100% 采样率并捕获输入和输出，这些是默认值。要进行调整，请在 modelPod 下添加以下任一可选字段：

modelPod:
  enabled: true
  initialSamplingPercentage: 50          # capture 50% of requests (default 100)
  bufferConfig:
    batchSize: 100                       # records buffered per upload batch (default 10)
    flushIntervalSeconds: 60             # max seconds before a flush (default 60)
  payloadConfig:
    maxPayloadSizeKB: 1024               # truncate payloads larger than this (default: no limit)
  kmsKeyId: "arn:aws:kms:...:key/..."    # encrypt captures at rest

步骤 4：等待 pod

开启了捕获功能的 pod 运行四个容器：模型、sidecar-reverse-proxy（NGINX）、otel-collector 和 s3-uploader。模型容器和 pod 的 app 标签都以 modelName 命名。

kubectl get pods -l app=$ENDPOINT_NAME -o custom-columns=\
'NAME:.metadata.name,READY:.status.containerStatuses[*].ready,CONTAINERS:.spec.containers[*].name,NODE:.spec.nodeName'
NAME                            READY                 CONTAINERS                                                              NODE
appliance-router-849f466476-4f5v8   true,true,true,true   appliance-router,s3-uploader,sidecar-reverse-proxy,otel-collector   hyperpod-i-0f3fe2684afd6a583

在我们的运行中，pod 在 kubectl apply 之后大约五分钟就绪，其中大部分时间用于下载权重。

步骤 5：发送请求并确认捕获内容落盘到 S3

通过端口 8081 上的捕获代理发送一个推理请求，然后，在刷新间隔之后，列出捕获前缀。模型本身监听 8000 端口。发送到该端口的请求会绕过代理，因此不会被捕获。

POD=$(kubectl get pod -l app=$ENDPOINT_NAME -o jsonpath='{.items[0].metadata.name}')
kubectl port-forward "pod/$POD" 18081:8081 >/dev/null 2>&1 &
sleep 5

curl -s http://localhost:18081/v1/chat/completions \
  -H 'Content-Type: application/json' \
  -d '{"model":"'"$ENDPOINT_NAME"'","messages":[{"role":"user","content":"my fridge is not cooling"}],"max_tokens":24}'

echo "Waiting for the capture buffer to flush..."; sleep 75
aws s3 ls "s3://${BUCKET}/capture/" --recursive | grep '\.jsonl$' | head
2026-09-28 15:47:14     634356 capture/20e4425092f90c7f/pod/AllTraffic/2026/09/28/15/28-15-47-12-999-ac31d31b-5156-46ad-9ccc-e8a040ab9ab2.jsonl

您会在 {s3Uri}/{uniqueID}/pod/{variant}/{YYYY}/{MM}/{DD}/{HH}/ 下方看到一个或多个 .jsonl 对象。uniqueID 是按终端节点生成的。每一行都是一条采用下文所示的 SageMaker 数据捕获格式的推理记录。

捕获记录的格式

请求和响应正文以 base64 编码存储，如 encoding 字段所示，因此读取它们需要先进行 base64 解码，再解析 JSON：

{
  "captureData": {
    "endpointInput":  { "observedContentType": "application/json", "mode": "INPUT",
                        "data": "eyJtb2RlbCI6ICJyb3V0ZXItZnJlc2giLCAidGVtcGVyYXR1cmUiOiAwLCAi...", "encoding": "BASE64" },
    "endpointOutput": { "observedContentType": "application/json", "mode": "OUTPUT",
                        "data": "eyJpZCI6ImNoYXRjbXBsLTctMTc5MDYxMDM2OS4wMDUtMjMxMiIsIm9iamVj...", "encoding": "BASE64" }
  },
  "eventMetadata": { "eventId": "26859f07-d60b-4683-94aa-d21e7f19e746", "inferenceTime": "2026-09-28T15:46:09Z" },
  "eventVersion": "0"
}

解码后，endpointInput.data 是客户端发送的请求正文，endpointOutput.data 是模型返回的响应。

场景：工单路由器遇上新品发布

部署的模型是一家家电零售商的售后工单路由器。它使用 vLLM 引导式 JSON 解码，将每张工单分类到六个类别之一：ProductFridge、ProductWasher、ProductOven、ProductDishwasher、ProductMicrowave，或兜底的 Other。引导式解码意味着它只能返回包含这些标签之一的有效对象：

{"category": "ProductWasher", "confidence": 1}

模型部署之后，这家零售商推出了自主品牌 Arctis 系列空调（Arctis 9K、12K、Mini 等）。路由器没有与之对应的类别。它无法说“这是新东西”，只能挑选它所知的最接近的标签。除非每个请求和响应都被记录下来，否则证据会在响应返回的那一刻消失。

合成流量

我们生成了两个合成数据集：

• 生产流量：900 张工单，通过捕获代理在 15 分钟内按顺序重放，使用路由器原有的六标签提示词。前 400 张出现在发布之前，不涉及空调。在剩余的 500 张中，Arctis 工单的占比从 15% 逐步上升到 55%。
• 评估集：280 张已标注工单，每类 40 张，包括新增的 ProductAC。它按内容进行留出：使用了流量中从未出现过的症状、句式模板和 Arctis 型号名称。

每个生产请求如下所示（系统提示词已缩略）：

{"model": "appliance-router", "temperature": 0, "max_tokens": 40,
 "messages": [{"role": "system", "content": "You are a support-ticket router for a home-appliance retailer. ... category is one of: ProductFridge, ProductWasher, ProductOven, ProductDishwasher, ProductMicrowave, Other. ..."},
              {"role": "user", "content": "Hi, our Arctis 12K blows warm air since last week. Order #66759."}],
 "guided_json": {"type": "object", "required": ["category", "confidence"],
                 "properties": {"category": {"enum": ["ProductFridge", "ProductWasher", "ProductOven",
                                                      "ProductDishwasher", "ProductMicrowave", "Other"]},
                                "confidence": {"type": "number"}}}}

要重放流量，请克隆随附的代码仓库，生成数据集，并通过捕获代理发送它们。scripts/pod.sh 会在模型容器内部运行客户端，因此不需要负载均衡器或 DNS。客户端会打印 Athena 步骤所需的三个时间戳：RUN_START、LAUNCH_TS（发布后的第一张工单）和 RUN_END。

git clone https://github.com/awslabs/awsome-distributed-ai.git
cd awsome-distributed-ai/examples/use-cases/sagemaker-hyperpod-data-capture-flywheel
(cd data && python3 ../scripts/gen_dataset.py)
scripts/pod.sh $ENDPOINT_NAME drive /tmp/tickets_traffic.jsonl --seconds 900
RUN_START=2026-09-28 15:52:40
  sent 100/900
...
LAUNCH_TS=2026-09-28 15:59:20
...
RUN_END=2026-09-28 16:07:40

导出您运行时打印的三个值，并等待一个刷新间隔：

export RUN_START="2026-09-28 15:52:40" LAUNCH_TS="2026-09-28 15:59:20" RUN_END="2026-09-28 16:07:40"
sleep 75    # one flush interval

用 Amazon Athena 发现问题

发布之后，离线基准测试看起来仍然没有变化，因为它们凝固在过去。这种偏移只在捕获的生产流量中可见。本节将展示数据科学家如何在无需预置任何基础设施的情况下，使用 Athena 发现它。

第 6 步：设置变量并配置一个 Athena 辅助函数

export ATHENA_DB="flywheel_demo"
export ATHENA_OUTPUT="s3://${BUCKET}/athena-results/"
# Each endpoint writes under its own uniqueID. Pick the prefix whose captured requests name this
# endpoint, not whichever prefix sorts first (the Step 15 endpoint adds a second one).
for id in $(aws s3 ls "s3://${BUCKET}/capture/" | awk '/PRE/{print $2}'); do
  key=$(aws s3 ls "s3://${BUCKET}/capture/${id}pod/AllTraffic/" --recursive | awk '/\.jsonl$/{print $4; exit}')
  [ -n "$key" ] || continue
  model=$(aws s3 cp "s3://${BUCKET}/${key}" - 2>/dev/null | head -n 1 | python3 -c 'import sys, json, base64; r = json.loads(sys.stdin.readline()); print(json.loads(base64.b64decode(r["captureData"]["endpointInput"]["data"]))["model"])')
  if [ "$model" = "$ENDPOINT_NAME" ]; then export CAPTURE_LOCATION="s3://${BUCKET}/capture/${id}pod/AllTraffic/"; break; fi
done
echo "$CAPTURE_LOCATION"     # s3://<bucket>/capture/<uniqueID>/pod/AllTraffic/

# run_athena "<SQL>": runs one statement, prints tab-separated rows, and sets LAST_QID
run_athena() {
  local qid st
  qid=$(aws athena start-query-execution --query-string "$1" \
        --query-execution-context "Database=$ATHENA_DB" \
        --result-configuration "OutputLocation=$ATHENA_OUTPUT" \
        --output text --query 'QueryExecutionId') || return 1
  while :; do
    st=$(aws athena get-query-execution --query-execution-id "$qid" \
         --output text --query 'QueryExecution.Status.State')
    case "$st" in
      SUCCEEDED) break ;;
      FAILED|CANCELLED) aws athena get-query-execution --query-execution-id "$qid" \
          --output text --query 'QueryExecution.Status.StateChangeReason' >&2; return 1 ;;
    esac
    sleep 2
  done
  LAST_QID=$qid
  aws athena get-query-results --query-execution-id "$qid" \
    --output text --query 'ResultSet.Rows[*].Data[*].VarCharValue'
}
run_athena "CREATE DATABASE IF NOT EXISTS ${ATHENA_DB}"

第 7 步：向 Athena 声明捕获数据（两层结构）

第一层是信封。一个外部表描述每条记录的外层 JSON，而对 year/month/day/hour 进行分区投影，可以在没有爬网程序或 ADD PARTITION 的情况下将按小时划分的 S3 前缀映射为分区。data 字段被声明为普通字符串，因为它们是 base64 文本。

run_athena "
CREATE EXTERNAL TABLE IF NOT EXISTS inference_captures (
  captureData struct<
    endpointInput:  struct<observedContentType: string, mode: string, data: string, encoding: string>,
    endpointOutput: struct<observedContentType: string, mode: string, data: string, encoding: string>
  >,
  eventMetadata struct<eventId: string, inferenceTime: string>,
  eventVersion string
)
PARTITIONED BY (year string, month string, day string, hour string)
ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe'
LOCATION '${CAPTURE_LOCATION}'
TBLPROPERTIES (
  'projection.enabled'='true',
  'projection.year.type'='integer',  'projection.year.range'='2026,2035',
  'projection.month.type'='integer', 'projection.month.range'='1,12', 'projection.month.digits'='2',
  'projection.day.type'='integer',   'projection.day.range'='1,31',   'projection.day.digits'='2',
  'projection.hour.type'='integer',  'projection.hour.range'='0,23',  'projection.hour.digits'='2',
  'storage.location.template'='${CAPTURE_LOCATION}\${year}/\${month}/\${day}/\${hour}/'
)"

第二层是一个视图，对每条记录解码一次。from_utf8(from_base64(...)) 将载荷转回 JSON。没有它，每次提取都会返回 NULL。工单是用户消息（messages[1]，因为 messages[0] 是系统提示词）。模型的回答是响应 JSON 内部的一个 JSON 字符串，因此需要第二次提取。

run_athena "
CREATE OR REPLACE VIEW preds AS
WITH decoded AS (
  SELECT CAST(from_iso8601_timestamp(eventMetadata.inferenceTime) AS timestamp) AS ts,
         from_utf8(from_base64(captureData.endpointInput.data))  AS req,
         from_utf8(from_base64(captureData.endpointOutput.data)) AS resp
  FROM inference_captures
  WHERE year = '${RUN_START:0:4}' AND month = '${RUN_START:5:2}' AND day = '${RUN_START:8:2}'
)
SELECT ts,
       json_extract_scalar(req, '\$.messages[1].content') AS ticket_text,
       json_extract_scalar(json_extract_scalar(resp, '\$.choices[0].message.content'), '\$.category') AS predicted_category,
       CAST(json_extract_scalar(json_extract_scalar(resp, '\$.choices[0].message.content'), '\$.confidence') AS double) AS confidence
FROM decoded
WHERE ts >= timestamp '${RUN_START}' AND ts < timestamp '${RUN_END}'
"
run_athena "SELECT count(*) AS n, count_if(predicted_category IS NULL) AS unparsed FROM preds"
n   unparsed
900 0

关于该视图的两点说明：

• 务必基于分区进行过滤。Athena 读取的是原始 JSON，因此没有分区过滤条件的查询会扫描您捕获过的每一个对象。该视图过滤到一天，而时间界限将其限定在我们的流量运行范围内：同一个前缀还包含来自第 5 步的请求。
• messages[1] 假设为一条系统消息后跟一条用户消息。捕获的流量就是客户端实际发送的内容，因此对于多轮或 few-shot 请求，需相应调整提取逻辑。

第 8 步：类别组合是否在变化？

run_athena "
SELECT date_trunc('minute', ts) AS minute, count(*) AS tickets,
       count_if(predicted_category = 'Other') AS other,
       round(100.0 * count_if(predicted_category = 'Other') / count(*), 1) AS other_pct
FROM preds GROUP BY 1 ORDER BY 1"
minute                   tickets  other  other_pct
2026-09-28 15:52:00.000       20      0        0.0
2026-09-28 15:53:00.000       60      3        5.0
2026-09-28 15:54:00.000       60      4        6.7
2026-09-28 15:55:00.000       60      4        6.7
2026-09-28 15:56:00.000       60      2        3.3
2026-09-28 15:57:00.000       60      1        1.7
2026-09-28 15:58:00.000       60      3        5.0
2026-09-28 15:59:00.000       60      9       15.0
2026-09-28 16:00:00.000       60     17       28.3
2026-09-28 16:01:00.000       60     14       23.3
2026-09-28 16:02:00.000       60     19       31.7
2026-09-28 16:03:00.000       60     21       35.0
2026-09-28 16:04:00.000       60     20       33.3
2026-09-28 16:05:00.000       60     32       53.3
2026-09-28 16:06:00.000       60     28       46.7
2026-09-28 16:07:00.000       40     26       65.0

（输出为制表符分隔，这里为了可读性做了对齐。我们这次运行把“几周”压缩成了 15 分钟。在生产环境中，应改用 date_trunc('day', ts) 或 'week' 来分桶。）

下图以柱状图形式绘制了每分钟路由到 Other 的工单占比，并以折线图绘制累计占比，同时标注了发布时间点。

图 2：Arctis 发布前后每分钟及累计路由到 Other 的工单占比

发布之前，Other 每分钟始终低于 7%。在发布那一分钟跃升至 15%，并持续攀升，在最后三分钟达到 47–65%。图中的累计折线只到 22.6%，因为它是把发布后的工单与发布前的全部流量放在一起平均的结果。这就是为什么应该用每分钟比率（而不是累计占比）来触发告警。

按发布时间点拆分后，整个类别组合如下：

run_athena "
SELECT IF(ts < timestamp '${LAUNCH_TS}', '1-before launch', '2-after launch') AS period,
       predicted_category, count(*) AS n
FROM preds GROUP BY 1, 2 ORDER BY 1, 3 DESC"
period           predicted_category    n
1-before launch  ProductFridge        91
1-before launch  ProductWasher        83
1-before launch  ProductDishwasher    74
1-before launch  ProductOven          74
1-before launch  ProductMicrowave     59
1-before launch  Other                19
2-after launch   Other               184
2-after launch   ProductFridge        84
2-after launch   ProductOven          66
2-after launch   ProductDishwasher    64
2-after launch   ProductWasher        58
2-after launch   ProductMicrowave     44

Other 从发布前占工单的 4.8%（400 中的 19 条）变为发布后的 36.8%（500 中的 184 条），并成为最大的类别。

第 9 步：Other 这个桶里装的是什么？

run_athena "
SELECT word, count(*) AS freq
FROM preds CROSS JOIN UNNEST(regexp_extract_all(lower(ticket_text), '[a-z0-9-]{4,}')) AS t(word)
WHERE predicted_category = 'Other'
  AND word NOT IN ('this','that','with','from','have','since','week','help','need','again','time',
                   'month','second','take','look','someone','order','last','after','when','onto')
GROUP BY 1 ORDER BY 2 DESC, 1 LIMIT 10"
word      freq
arctis     141
unit        40
mini        33
cooling     30
link        30
lost        30
loudly      30
rattles     30
wi-fi       30
blows       28

Other 桶并不是一堆互不相关的工单。它被一个产品名称 arctis 以及制冷设备的词汇所主导，这表明路由器遇到了一个它没有标签的新类别。

第 10 步：规模有多大？模型的置信度标记出它了吗？

在整个流量流中搜索新产品，而不仅仅是 Other 桶：

run_athena "
SELECT IF(ts < timestamp '${LAUNCH_TS}', '1-before launch', '2-after launch') AS period,
       predicted_category, count(*) AS n
FROM preds WHERE regexp_like(lower(ticket_text), 'arctis|air condition')
GROUP BY 1, 2 ORDER BY 1, 3 DESC"
period          predicted_category    n
2-after launch  Other               164
2-after launch  ProductFridge        14

共有 178 条空调工单，占全部流量的 19.8%。发布之前一条都没有出现。它们分为两组：

• 164 条进入了 Other。在六标签的模式下，这是正确的答案：Other 正是为这种情况而存在的。路由器没有出问题。是分类体系缺少一个类别，导致这些工单进入了一个通用队列，而不是支持该产品的团队。
• 14 条进入了 ProductFridge。这些是真正的误路由：冰箱团队现在收到了空调工单。

现在测试模型在这些已知误路由上的置信度：

run_athena "
SELECT predicted_category, confidence, count(*) AS n FROM preds
WHERE regexp_like(lower(ticket_text), 'arctis|air condition')
GROUP BY 1, 2 ORDER BY 1, 2 DESC"
predicted_category  confidence    n
Other                      1.0  157
Other                      0.9    7
ProductFridge             0.95    1
ProductFridge              0.9    9
ProductFridge              0.8    2
ProductFridge              0.7    1
ProductFridge              0.0    1

在 14 条误路由工单中，有 10 条报告的置信度为 0.9 或更高。诸如“保留置信度 confidence >= 0.9 的预测”这样的标注规则，会保留 71% 的已知错误标签。这是因为 confidence 是我们要求模型写出的一个字段，而不是经过校准的概率。如果你需要一个真正的不确定性信号，应向 vLLM 请求 token logprobs，读取类别 token 处的概率，而不是模型的自报置信度。要为额外的载荷大小预留预算，因为 logprobs 会随每次响应一起被捕获。

从诊断到带标签的训练数据

数据科学家不会手工标注数百条工单。相反，他们会把 Athena 的发现写成一项标注策略：一个为每条捕获记录分配修正标签的 SQL 视图。我们把误路由的空调工单重新标注为这个新类别。

第 11 步：标注策略

run_athena "
CREATE OR REPLACE VIEW labeled_training_data AS
WITH t AS (SELECT ticket_text, predicted_category, lower(ticket_text) AS txt FROM preds)
SELECT ticket_text, predicted_category,
  CASE
    -- 1. The new class, identified by keyword evidence, not by what the model predicted
    WHEN regexp_like(txt, '\\barctis\\b|air condition|\\bac unit\\b|mini split')
     AND NOT regexp_like(txt, 'fridge|refrigerat|freezer|washer|\\boven\\b|\\brange\\b|dishwasher|microwave')
      THEN 'ProductAC'
    -- 2. Keep the production decision when nothing hints at the new product
    WHEN NOT regexp_like(txt, '\\barctis\\b|air condition|\\bac\\b|mini split')
      THEN predicted_category
    -- 3. Mixed signals: send to human review (Amazon SageMaker Ground Truth)
    ELSE NULL
  END AS corrected_category
FROM t"

双反斜杠（\\b）是双引号字符串中的 shell 转义。Athena 收到的是 \b，一个正则表达式的单词边界。

检查该策略产生了什么：

run_athena "
SELECT coalesce(corrected_category, '(human review)') AS label, count(*) AS n,
       count(DISTINCT ticket_text) AS distinct_texts,
       count_if(corrected_category <> predicted_category) AS relabeled
FROM labeled_training_data GROUP BY 1 ORDER BY 2 DESC"
label               n    distinct_texts  relabeled
ProductAC         178               114        178
ProductFridge     161               110          0
ProductWasher     141                96          0
ProductOven       140                97          0
ProductDishwasher 138                95          0
ProductMicrowave  103                82          0
Other              39                 8          0

这个启发式方法之所以有效，是因为新类别界限分明（可通过关键词识别）。对于模糊的工单或重叠的类别，NULL 行会被路由到 Ground Truth 进行人工标注。

第 12 步：构建训练文件

导出带标注的视图并将其转换为聊天格式的 JSONL。允许的类别集合现在包含 ProductAC，也就是重新部署后客户端将发送的同样的七类提示词。将其保存为 build_training_file.py：

cat > build_training_file.py <<'PY'
import csv, json, sys
CATS = ["ProductFridge", "ProductWasher", "ProductOven", "ProductDishwasher", "ProductMicrowave", "ProductAC", "Other"]
PROMPT = ("You are a support-ticket router for a home-appliance retailer. Route each ticket to the product it "
          'concerns. Respond with ONLY a JSON object {"category": "...", "confidence": <0-1>} where category is '
          "one of: " + ", ".join(CATS) + '. Use "Other" if the ticket does not clearly concern one of these products.')
n = 0
with open(sys.argv[1], newline="") as fin, open(sys.argv[2], "w") as fout:
    for row in csv.DictReader(fin):
        fout.write(json.dumps({"messages": [
            {"role": "system", "content": PROMPT},
            {"role": "user", "content": row["ticket_text"]},
            {"role": "assistant", "content": json.dumps({"category": row["corrected_category"], "confidence": 1.0})},
        ]}) + "\n"); n += 1
print(f"wrote {n} training examples -> {sys.argv[2]}")
PY

run_athena "SELECT ticket_text, corrected_category FROM labeled_training_data
            WHERE corrected_category IS NOT NULL" > /dev/null
aws s3 cp "${ATHENA_OUTPUT}${LAST_QID}.csv" labeled.csv --only-show-errors   # Athena's own CSV: header + proper quoting
python3 build_training_file.py labeled.csv train.jsonl
aws s3 cp train.jsonl "s3://${BUCKET}/training/train.jsonl"
wrote 900 training examples -> train.jsonl

使用 HyperPod 训练算子在捕获的数据上进行微调

我们在服务模型的同一集群上进行微调，在第二个 GPU 上使用 HyperPod 训练算子。第一个端点在整个过程中持续提供服务。

第 13 步：训练脚本

该脚本从 S3 下载基础模型和训练数据，并训练一个低秩适应（LoRA）适配器。损失只在助手回合上计算，因此模型学会生成标签而不是重复提示词。脚本随后将适配器合并回基础权重，这样 vLLM 无需改动即可服务合并后的模型，并将合并后的模型上传到 S3。该任务通过 ConfigMap 将脚本挂载到训练 Pod 中。

cat > train.py <<'PY'
import os, boto3, torch
from datasets import load_dataset
from peft import LoraConfig, get_peft_model
from transformers import AutoModelForCausalLM, AutoTokenizer, Trainer, TrainingArguments

s3 = boto3.client("s3")
def split(uri): b, _, k = uri[5:].partition("/"); return b, k

# 1. Stage base model + training data from S3
bucket, prefix = split(os.environ["BASE_MODEL_S3"])
for page in s3.get_paginator("list_objects_v2").paginate(Bucket=bucket, Prefix=prefix):
    for o in page.get("Contents", []):
        dst = os.path.join("/work/base", os.path.relpath(o["Key"], prefix))
        os.makedirs(os.path.dirname(dst), exist_ok=True); s3.download_file(bucket, o["Key"], dst)
s3.download_file(*split(os.environ["TRAIN_FILE_S3"]), "/work/train.jsonl")
print("[stage] base model and training data downloaded", flush=True)

# 2. LoRA SFT, loss only on the assistant turn (the label), not the prompt
tok = AutoTokenizer.from_pretrained("/work/base")
def encode(ex):
    full = tok.apply_chat_template(ex["messages"], tokenize=False)
    prompt = tok.apply_chat_template(ex["messages"][:-1], tokenize=False, add_generation_prompt=True)
    ids = tok(full, add_special_tokens=False)["input_ids"]
    n = len(tok(prompt, add_special_tokens=False)["input_ids"])
    return {"input_ids": ids, "labels": [-100] * n + ids[n:]}
ds = load_dataset("json", data_files="/work/train.jsonl", split="train")
print(f"[data] {len(ds)} examples", flush=True)
ds = ds.map(encode, remove_columns=ds.column_names)

def collate(batch):
    m = max(len(b["input_ids"]) for b in batch)
    pad = lambda xs, v: [x + [v] * (m - len(x)) for x in xs]
    return {"input_ids": torch.tensor(pad([b["input_ids"] for b in batch], tok.pad_token_id)),
            "labels": torch.tensor(pad([b["labels"] for b in batch], -100)),
            "attention_mask": torch.tensor(pad([[1] * len(b["input_ids"]) for b in batch], 0))}

model = AutoModelForCausalLM.from_pretrained("/work/base", torch_dtype=torch.bfloat16, device_map="cuda:0")
model = get_peft_model(model, LoraConfig(r=16, lora_alpha=32, lora_dropout=0.05, task_type="CAUSAL_LM",
    target_modules=["q_proj", "k_proj", "v_proj", "o_proj", "gate_proj", "up_proj", "down_proj"]))
model.print_trainable_parameters()
Trainer(model=model, train_dataset=ds, data_collator=collate, args=TrainingArguments(
    output_dir="/tmp/ckpt", num_train_epochs=float(os.getenv("EPOCHS", "2")), learning_rate=1e-4,
    per_device_train_batch_size=8, bf16=True, logging_steps=10, save_strategy="no", report_to=[],
    warmup_ratio=0.1, lr_scheduler_type="cosine")).train()

# 3. Merge the adapter into the base weights (a plain model vLLM can serve) and upload it
model.merge_and_unload().save_pretrained("/work/out", safe_serialization=True)
tok.save_pretrained("/work/out")
bucket, prefix = split(os.environ["OUTPUT_S3"])
for f in os.listdir("/work/out"):
    s3.upload_file(os.path.join("/work/out", f), bucket, f"{prefix.rstrip('/')}/{f}")
print(f"[done] merged model uploaded to {os.environ['OUTPUT_S3']}", flush=True)
PY
kubectl create configmap router-train --from-file=train.py

第 14 步：提交一个 HyperPodPyTorchJob

cat > finetune-job.yaml <<'YAML'
apiVersion: sagemaker.amazonaws.com/v1
kind: HyperPodPyTorchJob
metadata:
  name: router-ft
  namespace: default
spec:
  nprocPerNode: "1"
  runPolicy:
    jobMaxRetryCount: 2
    cleanPodPolicy: All              # release the GPU as soon as the job finishes
    logMonitoringConfiguration:      # restart a job that never starts, or goes quiet mid-run
      - name: JobStart
        logPattern: '.*loss.*'
        expectedStartCutOffInSeconds: 1200
      - name: JobHangingDetection
        logPattern: '.*loss.*'
        expectedRecurringFrequencyInSeconds: 600
  replicaSpecs:
    - name: pod
      replicas: 1
      template:
        metadata:
          labels: { app: router-ft }
        spec:
          serviceAccountName: ${TRAIN_SA}        # EKS Pod Identity -> S3 read/write
          nodeSelector:
            node.kubernetes.io/instance-type: ${TRAIN_INSTANCE_TYPE}
          restartPolicy: Always                  # required by the operator; retries live in runPolicy
          containers:
            - name: pytorch
              image: vllm/vllm-openai:v0.10.0    # already has torch + transformers + boto3
              command: ["bash", "-c"]
              args:
                - pip install -q peft==0.14.0 datasets==3.2.0 hyperpod-elastic-agent==1.1.3 &&
                  hyperpodrun --nnodes=1 --nproc_per_node=1 /scripts/train.py
              env:
                - { name: BASE_MODEL_S3, value: "s3://${BUCKET}/${MODEL_LOCATION}/" }
                - { name: TRAIN_FILE_S3, value: "s3://${BUCKET}/training/train.jsonl" }
                - { name: OUTPUT_S3,     value: "s3://${BUCKET}/${MODEL_V2_LOCATION}/" }
              resources:
                requests: { cpu: "2", memory: 24Gi, nvidia.com/gpu: 1 }
                limits:   { nvidia.com/gpu: 1 }
              volumeMounts:
                - { name: scripts, mountPath: /scripts }
                - { name: work,    mountPath: /work }
                - { name: shm,     mountPath: /dev/shm }
          volumes:
            - { name: scripts, configMap: { name: router-train } }
            - { name: work,    emptyDir: {} }
            - { name: shm,     emptyDir: { medium: Memory, sizeLimit: 8Gi } }
YAML
render < finetune-job.yaml | kubectl apply -f -
kubectl logs -l app=router-ft -f

• logMonitoringConfiguration 是该训练算子的特色功能。如果启动后 20 分钟内没有出现 loss 行，或者运行中进程仍然存活但日志静默了 10 分钟，算子会重启任务，而不是让它白白消耗 GPU 时间。
• 该任务在启动时会安装三个 Python 包。训练节点需要出站互联网访问（例如通过 NAT 网关），从 Docker Hub 拉取 vllm/vllm-openai 镜像也是如此。在没有互联网出站的集群上，可以将这些包预先打包进 Amazon Elastic Container Registry (Amazon ECR) 中的自定义镜像，并去掉 pip install。

任务日志摘录：

{'loss': 0.2484, 'grad_norm': 3.387, 'learning_rate': 3.91e-05, 'epoch': 0.09}
{'loss': 0.018, 'grad_norm': 0.0037, 'learning_rate': 8.26e-05, 'epoch': 0.18}
...
{'loss': 0.0, 'grad_norm': 0.000105, 'learning_rate': 2.93e-07, 'epoch': 1.95}
{'train_runtime': 59.68, 'train_samples_per_second': 30.16, 'train_loss': 0.0123, 'epoch': 2.0}
[done] merged model uploaded to s3://amzn-s3-demo-bucket/models/qwen2.5-3b-router-v2/
[HyperPodElasticAgent] ... Transitioned from AgentState.RUNNING to AgentState.COMPLETED
kubectl get hyperpodpytorchjob router-ft -o jsonpath='{range .status.conditions[*]}=  {end}'
Created=True  Completed=True

训练在一个 L40S 上花了 60 秒。rank 16 的 LoRA 训练约 2990 万个参数，不到这个 31 亿参数模型的 1%。平均训练损失为 0.0123，每步损失在第一个 epoch 内就达到了 0.0，这表明模型很容易记住这 900 个模板化的示例。在真实数据上，应保留一个验证集并进行早停。这里我们依靠留出的评估集。

第 15 步：将微调后的模型并行部署，同时保持捕获开启

与其编辑正在运行的端点，不如从同一个清单将新模型部署为第二个端点，部署到训练任务刚刚释放的 GPU 上。在你切换客户端之前，第一个端点持续提供服务，因此重新部署不会造成停机，而回滚只需将客户端指回第一个端点。

export ENDPOINT_NAME="appliance-router-v2" MODEL_LOCATION="$MODEL_V2_LOCATION" SERVE_INSTANCE_TYPE="$TRAIN_INSTANCE_TYPE"
render < endpoint.yaml | kubectl apply -f -
kubectl get pods -l app=$ENDPOINT_NAME -o custom-columns='NAME:.metadata.name,READY:.status.containerStatuses[*].ready,NODE:.spec.nodeName'
NAME                                  READY                 NODE
appliance-router-v2-c96594c5-fhs2x    true,true,true,true   hyperpod-i-0e0b756209b5e497a

新端点在自己的 uniqueID 下写入捕获数据。由于 ENDPOINT_NAME 现在是 appliance-router-v2，重新执行第 6 步的 CAPTURE_LOCATION 查找即可找到该前缀，因此你可以将第二个 Athena 表指向它，用于下一轮分析。在切换客户端时，请在同一次发布中更新端点、标签集和 guided-JSON 枚举。

衡量提升效果

我们用同样的 280 个留出工单对微调后的端点进行了测试：

overall 280/280 = 100.00%
  ProductFridge      40/40
  ProductWasher      40/40
  ProductOven        40/40
  ProductDishwasher  40/40
  ProductMicrowave   40/40
  ProductAC          40/40
  Other              40/40

微调让路由器学会了新类别，并且在五个既有类别上没有造成任何性能回退。它还修复了上线前就存在的五个 Other 错误，上线时的路由器曾将这些工单发送到 ProductFridge。

由于我们的评估集来自与训练数据相同的生成器，我们还手工编写了 24 个探针工单，其风格是生成器从未产生过的。微调后的模型得分为 20/24：

  MISS ProductFridge -> ProductAC: The cooling fan in my fridge is squealing.
  MISS Other -> ProductAC: When will the Arctis 12K be back in stock?
  MISS Other -> ProductAC: Can I get a quote to install central air in my house?
  MISS Other -> ProductAC: Do you price match the Arctis at other stores?
overall 20/24 = 83.33%

它把全部八个空调探针都路由正确，但它学到了两条捷径：“cooling”意味着空调，而任何提到 Arctis 的内容都意味着 ProductAC，包括库存和价格匹配问题。两者都源于训练数据：标注策略把所有 Arctis 工单都发给了 ProductAC，而捕获的流量中没有售前问题。开启新端点的捕获后，这些工单恰恰就是下一轮捕获会收集的内容。

捕获会增加多少延迟？

我们直接进行了测量：同样 400 个路由请求，从 pod 内部发送，要么发到端口 8000（无捕获），要么发到端口 8081（经过捕获代理）。我们以 8 个并发请求运行了三轮，外加一轮串行：

并发 8 的各行是三轮的平均值。轮次间的波动（p50 约 ±1 ms）大于两条路径之间的差异。

捕获数据的其他用途

推理数据落入 S3 之后，同样的记录可以支撑多种工作流：

• 改进模型：使用监督微调（本文所示）、将大模型的输出蒸馏到较小的模型，或为直接偏好优化（DPO）构建偏好对。
• 理解模型：运行 Athena 分析（本文所示）、从真实提示构建黄金回归集、设置漂移告警、发现新兴意图，并检测格式错误的输出和拒答。
• 运营与治理：将模糊样本路由到 Ground Truth 进行标注、保留审计记录（推荐 Tier 1）、重放事故背后的确切输入，并分析流量和负载以进行成本归因。

成本考量

有两项成本随使用量扩展：

• S3。我们的记录平均每条约 2.3 KB（9.0 MB 中有 3,955 条记录）。更长的提示和响应成本成比例增加。上传器每个刷新间隔每个 pod 写入一个对象，在 60 秒间隔下约为每个 pod 每天 1,440 次 PUT。
• Athena 按扫描的数据量收费，因此务必在分区列上进行过滤，并在需要反复分析时用 CREATE TABLE AS SELECT（CTAS）查询将原始 JSON 转换为 Parquet。

有关当前价格，请参阅 Amazon S3 定价和 Amazon Athena 定价。

清理

为避免产生不必要的费用，请清理本演练中创建的资源：

kubectl delete inferenceendpointconfig appliance-router appliance-router-v2
kubectl delete hyperpodpytorchjob router-ft
kubectl delete configmap router-train
run_athena "DROP VIEW IF EXISTS labeled_training_data"; run_athena "DROP VIEW IF EXISTS preds"
run_athena "DROP TABLE IF EXISTS inference_captures"; run_athena "DROP DATABASE IF EXISTS ${ATHENA_DB}"
aws s3 rm "s3://${BUCKET}/capture/" --recursive
aws s3 rm "s3://${BUCKET}/athena-results/" --recursive
aws s3 rm "s3://${BUCKET}/training/" --recursive
aws s3 rm "s3://${BUCKET}/${MODEL_V2_LOCATION}/" --recursive

结论

在这篇文章中，我们展示了如何使用 SageMaker HyperPod Data Capture 记录实时推理数据并将其转化为训练数据。捕获支持三个独立层级，你可以：

• 以对服务延迟无可测量的影响收集真实世界的训练数据。
• 使用 Amazon Athena 大规模查询推理模式，无需管理任何基础设施。
• 把一次 Athena 诊断转化为可复现的 SQL 标注策略。
• 在采集到的生产数据上进行微调，并衡量前后明确的提升效果。
• 与 HyperPod 训练操作符和 SageMaker Ground Truth 集成，以完成循环的其余部分。

我们在一个家电支持路由场景运行了完整一轮循环。采集的流量揭示了一个新产品的盲点，一条 SQL 标注策略将这些采集数据转化为训练数据。使用 HyperPod 训练操作符进行的 60 秒 LoRA 微调，就在为该模型提供服务的同一个集群上运行。它将新类别的召回率从 0% 提升到完整召回，且没有在任何现有类别上出现回归。所有这些训练数据在模型部署之前都不存在。

Data Capture 现已在带有推理操作符的 SageMaker HyperPod 集群上可用。要开始使用，请将 dataCapture 部分添加到您的 InferenceEndpointConfig 或 JumpStartModel 资源中并进行部署。要在同一集群上闭环，请添加 amazon-sagemaker-hyperpod-training-operator EKS 附加组件。本演练的完整代码位于随附的 GitHub 仓库中。

更多信息，请参阅SageMaker HyperPod 推理文档和 HyperPod 训练操作符文档。

关于作者