Amazon SageMaker Python SDK v3 将生成式 AI 推理建议直接带进了 Notebook 工作流。开发者可以围绕同一个模型依次完成端点基准测试、生成数据驱动的部署建议,并部署推荐配置,不必在 Notebook、控制台和零散脚本之间反复切换。
这项集成真正解决的并不是“少点几次鼠标”,而是让性能数据、部署决策和可复现代码处在同一条工程链路中。
从经验选型转向基准数据
LLM 推理部署通常要同时平衡几个互相牵制的指标:
- 首个 token 延迟,直接影响交互体验。
- 持续生成吞吐量,决定长文本输出效率。
- 并发请求下的尾部延迟,例如 P95 和 P99。
- 实例数量、实例类型以及由此产生的运行成本。
- 模型大小、精度和推理优化方式带来的资源约束。
仅凭模型参数量或者历史经验选择实例,很容易忽略真实请求的输入长度、输出长度和并发模式。SageMaker SDK v3 暴露的生成式 AI 推理建议,可以把候选配置放到实际基准测试中比较,再根据测量结果给出部署建议。
这意味着 Notebook 不再只是模型实验环境,也可以保存一份可审查的部署决策记录:测试了哪些负载、接受什么延迟阈值、为何选择某个配置,以及最终部署了什么。
一条更完整的 Notebook 工作流
一个实用的优化流程可以拆成四步:
- 明确服务目标,例如最大 P95 延迟、目标并发量和预算上限。
- 准备有代表性的请求数据,覆盖短问答、长上下文和不同输出长度。
- 让 SageMaker 对候选端点配置执行基准测试并生成建议。
- 部署推荐配置,再使用独立流量进行验证。
关键是不要让基准数据只包含几个理想化 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 的价值在于缩短反馈链路,但实例选择、成本边界和生产验收标准仍然需要由团队明确控制。