用 JumpStart 完成模型部署与性能测试

为具体应用选择合适的模型，需要一种实用的方法，用有代表性的输入进行测试。Amazon SageMaker JumpStart 提供了丰富的预训练模型目录。本文将演示如何部署 SageMaker JumpStart 中的可用模型，使用 SageMaker Python SDK 运行轻量、可重复的性能测试，并在测试完成后删除 endpoint。对于受支持的模型，在有可用容量的情况下，这套流程提供了一条可重复的评估路径，不过部署耗时会因模型、实例类型、配额和 AWS Region 而异。

2026 年，SageMaker JumpStart 引入了两项改进，让模型发现和部署更加简单。在 Amazon SageMaker AI 控制台的 Model Hub 中，你可以浏览 SageMaker JumpStart 模型目录、查看示例 notebook，并直接在 SageMaker Studio 中打开它们。预配置的 endpoint 部署选项提供了多种具名配置，针对低成本、低延迟或通用工作负载等不同优先目标进行优化，覆盖多种生成式 AI 模型，也包括针对 Neuron 优化的配置。此外，通过新的深度链接集成，你现在可以一键从 Hugging Face 上发现的模型跳转到 SageMaker AI 部署。

我们采用一个可复用的 Python 脚本，完成以下三个步骤：

1. 从 SageMaker JumpStart 中查找并部署模型。
2. 运行模型专用的示例 prompt，测量延迟、近似输出长度和单次请求的生成速率。
3. 清理 endpoint。

这套模式适用于许多兼容的 SageMaker JumpStart 模型，不过 payload 格式、所需实例类型、最终用户许可协议（EULA）要求和响应结构，可能需要根据具体模型进行调整。

• 一个拥有 SageMaker AI 权限的 AWS 账户。
• Python 3+，并已安装 SageMaker Python SDK v3（pip install "sagemaker>=3,<4"）。
• 拥有足够的 GPU 实例配额来测试目标模型，或者申请提高配额。

步骤 1.a：查找可用模型

在 SageMaker JumpStart 中查找模型的 UI 路径：

图 1：在 SageMaker Studio 的 SageMaker JumpStart Models 页面浏览模型目录

使用 SageMaker Python SDK 的代码方式：

from sagemaker.core.jumpstart.notebook_utils import (
    list_jumpstart_models,
)

REGION = "us-east-1"

models = list_jumpstart_models(
    filter="task == textgeneration",
    region=REGION,
    list_versions=True,
)

print(f"Found {len(models)} text generation models")

for model_id, model_version in models[:10]:
    print(
        f" - {model_id} "
        f"(version {model_version})"
    )

也可以按模型名称搜索：

from sagemaker.core.jumpstart.notebook_utils import list_jumpstart_models

ids = list_jumpstart_models(
    filter="model_id contains qwen",
    region=REGION,
    list_versions=True,
)

print(f"{len(ids)} models matched in {REGION}\n")

for model_id, version in ids:
    print(f"{model_id:<55} v{version}")

步骤 1.b：部署模型

从 SageMaker JumpStart 部署模型的 UI 路径：

图 2：在 SageMaker Studio 中，从 SageMaker JumpStart 模型详情页部署模型

使用 SageMaker Python SDK 的代码方式：

import time

from sagemaker.core.jumpstart.accessors import JumpStartModelsAccessor
from sagemaker.core.jumpstart.configs import JumpStartConfig
from sagemaker.core.training.configs import Compute
from sagemaker.serve import ModelBuilder

MODEL_ID = "huggingface-vlm-qwen3-5-27b"
MODEL_VERSION = "1.13.0"
REGION = "us-east-1"
ACCEPT_EULA = False  # Set True only after reviewing the model terms

def get_field(value, name, default=None):
    if isinstance(value, dict):
        return value.get(name, default)
    return getattr(value, name, default)

# Resolve the JumpStart-recommended deployment configuration.
specs = JumpStartModelsAccessor().get_model_specs(
    region=REGION,
    model_id=MODEL_ID,
    version=MODEL_VERSION,
)

if isinstance(specs, list):
    if not specs:
        raise ValueError(
            f"No JumpStart specification found for {MODEL_ID}"
        )
    specs = specs[0]

recommended_instance_type = get_field(
    specs,
    "default_inference_instance_type",
)

if not recommended_instance_type:
    raise ValueError(
        f"No recommended instance type found for {MODEL_ID}"
    )

print(f"Model: {MODEL_ID}")
print(f"Region: {REGION}")
print(f"Recommended instance: {recommended_instance_type}")

start = time.time()

jumpstart_config = JumpStartConfig(
    model_id=MODEL_ID,
    model_version=MODEL_VERSION,
    accept_eula=ACCEPT_EULA,
)

model_builder = ModelBuilder.from_jumpstart_config(
    jumpstart_config=jumpstart_config,
    compute=Compute(
        instance_type=recommended_instance_type,
        instance_count=1,
    ),
)

model = model_builder.build()
endpoint = model_builder.deploy()
deploy_time = time.time() - start

print(f"Deployed in {deploy_time:.1f} seconds")
print(f"Endpoint: {endpoint.endpoint_name}")

步骤 2：运行示例 prompt 并测量性能

from sagemaker.core.payloads import retrieve_all_examples
import json
import time

MODEL_ID = "huggingface-vlm-qwen3-5-27b"
MODEL_VERSION = "1.13.0"
REGION = "us-east-1"

example_payloads = retrieve_all_examples(
    region=REGION,
    model_id=MODEL_ID,
    model_version=MODEL_VERSION,
    serialize=True,
)

#you can re-create the endpoint pointer using the endpoint name:
#from sagemaker.core.resources import Endpoint
#endpoint = Endpoint.get(endpoint_name="<your_endpoint_name>",region=REGION)

def read_response_body(result):
    body = result.body
    if hasattr(body, "read"):
        body = body.read()
    if isinstance(body, bytes):
        body = body.decode("utf-8")
    if isinstance(body, str):
        try:
            return json.loads(body)
        except json.JSONDecodeError:
            return body
    return body

def extract_output(response):
    """Handle common JumpStart text-generation response formats."""
    if isinstance(response, list):
        response = response[0] if response else ""
    if isinstance(response, dict):
        choices = response.get("choices")
        if choices:
            choice = choices[0]
            message = choice.get("message")
            if isinstance(message, dict):
                return message.get("content", "")
            return choice.get("text", "")
        for field in (
            "generated_text",
            "output_text",
            "text",
        ):
            if field in response:
                return response[field]
    return response

results = []

for payload in example_payloads or []:
    start = time.perf_counter()
    result = endpoint.invoke(
        body=payload.body,
        content_type=payload.content_type,
        accept=payload.accept,
    )
    latency = time.perf_counter() - start
    response = read_response_body(result)
    output = extract_output(response)
    if not isinstance(output, str):
        output = json.dumps(output, indent=2)
    # This is an estimate based on whitespace-separated words.
    output_words = len(output.split())
    results.append(
        {
            "request_preview": payload.body[:50],
            "latency_s": round(latency, 2),
            "estimated_output_words": output_words,
            "words_per_second": round(
                output_words / latency,
                1,
            ) if latency > 0 else 0,
        }
    )
    print("Input:\n", payload.body, end="\n\n")
    print("Output:\n", output, end="\n\n")
    print(
        f"Latency: {latency:.2f} seconds\n"
        f"Estimated output words: {output_words}\n"
        f"Estimated single-request output generation rate: "
        f"{output_words / latency:.1f} words/second\n"
    )
    print("-" * 80)

print("\nSummary:")
for result in results:
    print(result)

Input:
{"max_tokens": 256, "messages": [{"content": "What is deep learning?", "role": "user"}], "model": "/opt/ml/model"}
Output:
Here's a thinking process that leads to the explanation of deep learning:
1. **Deconstruct the Request:**
* **Question:** "What is deep learning?"
* **Intent:** The user wants a clear, comprehensive, yet accessible definition and overview of deep learning.
* **Target Audience:** Likely someone looking for a general understanding, possibly a student, tech
enthusiast, or professional exploring the field. The tone should be informative, structured, and
not overly jargon-heavy (though key terms need explanation).
2. **Initial Brainstorming & Key Concepts:**
* What is it? A subset of machine learning, which is a subset of AI.
* Core mechanism: Artificial Neural Networks (ANNs).
* "Deep" means many layers (hidden layers).
* How does it work? Learning representations from data (feature learning).
* Key applications: Image recognition, NLP, speech, self-driving cars.
* Why is it popular now? Big data, GPU power, better algorithms.
* Pros/Cons: High accuracy vs. black box, data hunger.
3. **Structuring the Answer:**
Latency: 9.07 seconds
Estimated output words: 153
Estimated single-request output generation rate: 16.9 words/second
--------------------------------------------------------------------------------
Summary:
{'request_preview': '{"max_tokens": 256, "messages": {"content": "What', 'latency_s': 9.07, 'estimated_output_words': 153, 'words_per_second': 16.9}

print(f"Deleting endpoint: {endpoint.endpoint_name}")
endpoint.delete()
endpoint.wait_for_delete()
print("Endpoint deleted")

这套模式适用于许多可用的 SageMaker JumpStart 模型，但 payload 格式、所需实例类型、EULA 和响应解析方式，可能需要根据具体模型进行调整。通过使用一致的测试 prompt，并测量各个模型的延迟、吞吐量和输出质量，你可以比较不同选项，选择最适合工作负载的模型。

试着用这套方法测试 Amazon SageMaker JumpStart 中的不同模型，为你的应用找到质量、速度和成本之间的合适平衡。

SageMaker Python SDK 文档

SageMaker JumpStart 部署基准测试与优化（博客文章）

一键从 Hugging Face 跳转到 SageMaker Studio（博客文章）

Amazon SageMaker JumpStart