在 Notebook 中完成 LLM 推理基准测试与 SageMaker 部署优化

2026-08-07 58 预计阅读时间: 1 分钟
来源: aws.amazon.com AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间:8 分钟

Amazon SageMaker Python SDK v3 将生成式 AI 推理建议直接带进了 Notebook 工作流。开发者可以围绕同一个模型依次完成端点基准测试、生成数据驱动的部署建议,并部署推荐配置,不必在 Notebook、控制台和零散脚本之间反复切换。

这项集成真正解决的并不是“少点几次鼠标”,而是让性能数据、部署决策和可复现代码处在同一条工程链路中。

从经验选型转向基准数据

LLM 推理部署通常要同时平衡几个互相牵制的指标:

  • 首个 token 延迟,直接影响交互体验。
  • 持续生成吞吐量,决定长文本输出效率。
  • 并发请求下的尾部延迟,例如 P95 和 P99。
  • 实例数量、实例类型以及由此产生的运行成本。
  • 模型大小、精度和推理优化方式带来的资源约束。

仅凭模型参数量或者历史经验选择实例,很容易忽略真实请求的输入长度、输出长度和并发模式。SageMaker SDK v3 暴露的生成式 AI 推理建议,可以把候选配置放到实际基准测试中比较,再根据测量结果给出部署建议。

这意味着 Notebook 不再只是模型实验环境,也可以保存一份可审查的部署决策记录:测试了哪些负载、接受什么延迟阈值、为何选择某个配置,以及最终部署了什么。

一条更完整的 Notebook 工作流

一个实用的优化流程可以拆成四步:

  1. 明确服务目标,例如最大 P95 延迟、目标并发量和预算上限。
  2. 准备有代表性的请求数据,覆盖短问答、长上下文和不同输出长度。
  3. 让 SageMaker 对候选端点配置执行基准测试并生成建议。
  4. 部署推荐配置,再使用独立流量进行验证。

关键是不要让基准数据只包含几个理想化 Prompt。生产负载中的 token 分布、请求突发和生成长度,往往比模型名称更能影响最终配置。

下面的 YAML 可以作为一次优化实验的输入清单。它不是 SageMaker 的固定配置格式,而是一份可以纳入版本控制、再由 Notebook 读取的项目配置:

# inference-experiment.yaml
experiment_name: support-assistant-v1
model:
  artifact_uri: s3://my-ml-bucket/models/support-assistant/
  role_arn: arn:aws:iam::123456789012:role/SageMakerExecutionRole

benchmark:
  dataset_uri: s3://my-ml-bucket/benchmarks/support-prompts.jsonl
  target_concurrency: [1, 4, 8]
  max_p95_latency_ms: 2500
  minimum_requests_per_minute: 120

selection:
  objective: lowest_cost_meeting_slo
  allowed_instance_types:
    - ml.g5.2xlarge
    - ml.g5.12xlarge
    - ml.p4d.24xlarge

deployment:
  endpoint_name: support-assistant-optimized
  initial_instance_count: 1

运行前需要替换 S3 地址、IAM Role、实例类型和性能目标。候选实例还应与当前区域的供应情况以及模型资源需求保持一致。

可以这样组织 SDK v3 代码

来源摘要没有给出 SDK v3 中具体类名和方法签名,因此下面采用适配模板:SageMakerOptimizationAdapter 代表项目中对实际 SDK v3 优化接口的一层薄封装。替换其中三个方法后,Notebook 主流程可以保持稳定。

先安装依赖并确认使用的是 v3:

python -m pip install --upgrade 'sagemaker>=3,<4' boto3 pyyaml
python -c "import sagemaker; print(sagemaker.__version__)"

然后把以下代码放进 Notebook 单元格或 optimize_endpoint.py。示例默认先以 DRY_RUN=1 运行,因此可以直接检查配置;接入实际 SDK 方法后再关闭 dry-run。

import json
import os
from dataclasses import dataclass
from pathlib import Path
from typing import Any

import boto3
import yaml


@dataclass
class Recommendation:
    instance_type: str
    instance_count: int
    metrics: dict[str, Any]


class SageMakerOptimizationAdapter:
    """Replace these methods with the corresponding SageMaker SDK v3 APIs."""

    def __init__(self, region_name: str):
        self.region_name = region_name
        self.session = boto3.Session(region_name=region_name)

    def benchmark(self, config: dict[str, Any]) -> str:
        # Integration point: submit a generative AI inference benchmark job.
        raise NotImplementedError("Connect this method to the SDK v3 benchmark API")

    def recommend(self, benchmark_job: str) -> Recommendation:
        # Integration point: fetch the data-driven deployment recommendation.
        raise NotImplementedError("Connect this method to the SDK v3 recommendation API")

    def deploy(self, config: dict[str, Any], recommendation: Recommendation) -> str:
        # Integration point: deploy the recommended endpoint configuration.
        raise NotImplementedError("Connect this method to the SDK v3 deployment API")


def load_config(path: str) -> dict[str, Any]:
    with Path(path).open(encoding="utf-8") as file:
        return yaml.safe_load(file)


def validate(config: dict[str, Any]) -> None:
    required = [
        ("model", "artifact_uri"),
        ("model", "role_arn"),
        ("benchmark", "dataset_uri"),
        ("deployment", "endpoint_name"),
    ]
    missing = [f"{section}.{key}" for section, key in required if not config.get(section, {}).get(key)]
    if missing:
        raise ValueError(f"Missing configuration: {', '.join(missing)}")


def main() -> None:
    config = load_config("inference-experiment.yaml")
    validate(config)

    if os.getenv("DRY_RUN", "1") == "1":
        print(json.dumps(config, indent=2))
        print("Dry run complete. Replace adapter methods, then set DRY_RUN=0.")
        return

    optimizer = SageMakerOptimizationAdapter(
        region_name=os.getenv("AWS_REGION", "us-east-1")
    )
    job = optimizer.benchmark(config)
    recommendation = optimizer.recommend(job)
    endpoint_name = optimizer.deploy(config, recommendation)
    print(f"Deployed endpoint: {endpoint_name}")


if __name__ == "__main__":
    main()

先执行安全的配置检查:

export AWS_REGION=us-east-1
export DRY_RUN=1
python optimize_endpoint.py

接入实际 SageMaker Python SDK v3 接口时,建议只在适配器内部处理 SDK 对象、轮询状态和响应结构。这样即使 SDK 接口升级,实验配置和 Notebook 的业务流程也不需要整体重写。

部署建议不是最终答案

推荐配置来自特定数据集和测试条件,不能自动代表所有生产流量。部署之后仍应执行一轮独立验证,并至少检查以下内容:

  • 推荐结果是否满足 P95、P99 和吞吐量目标。
  • 测试是否包含真实的输入与输出 token 分布。
  • 端点预热、模型加载和扩缩容期间是否出现延迟尖峰。
  • 优化后的模型质量是否与基线一致。
  • 基准测试、临时端点和最终端点各自产生了多少费用。
  • 测试数据是否包含敏感信息,以及 S3、日志和 IAM 权限是否符合要求。

更稳妥的采用方式,是先把这套流程用于一个已有性能基线的非关键模型。将实验配置、SDK 版本、推荐结果和部署指标一起保存,确认其可复现后,再扩展到更多模型和环境。SDK v3 的价值在于缩短反馈链路,但实例选择、成本边界和生产验收标准仍然需要由团队明确控制。


相关推荐