Deploy and evaluate models with Amazon SageMaker JumpStart

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}")

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  # 仅在查看模型条款后设置为True

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

# 获取JumpStart推荐的部署配置
specs = JumpStartModelsAccessor().get_model_specs(
    region=REGION,
    model_id=MODEL_ID,
    version=MODEL_VERSION,
)

if isinstance(specs, list):
    if not specs:
        raise ValueError(
            f"未找到 {MODEL_ID} 的JumpStart配置"
        )
    specs = specs[0]

recommended_instance_type = get_field(
    specs,
    "default_inference_instance_type",
)

if not recommended_instance_type:
    raise ValueError(
        f"未找到 {MODEL_ID} 的推荐实例类型"
    )

print(f"模型: {MODEL_ID}")
print(f"区域: {REGION}")
print(f"推荐实例: {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"部署耗时 {deploy_time:.1f} 秒")
print(f"端点: {endpoint.endpoint_name}")

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,
)

# 可以使用端点名称重新创建端点指针：
#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):
    """处理常见的JumpStart文本生成响应格式"""
    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)
    # 这是基于空格分隔单词的估算
    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("输入:\n", payload.body, end="\n\n")
    print("输出:\n", output, end="\n\n")
    print(
        f"延迟: {latency:.2f} 秒\n"
        f"预估输出单词数: {output_words}\n"
        f"单次请求输出生成速率: "
        f"{output_words / latency:.1f} 单词/秒\n"
    )
    print("-" * 80)

print("\n摘要:")
for result in results:
    print(result)

输出结果:

输入:
{"max_tokens": 256, "messages": [{"content": "What is deep learning?", "role": "user"}], "model": "/opt/ml/model"}
输出:
这是一个引导至深度学习解释的思考过程：
1. **拆解请求：**
* **问题：** "什么是深度学习？"
* **意图：** 用户希望获得清晰、全面且易懂的定义和概述。
* **目标受众：** 可能是寻求一般性理解的人，可能是学生、科技爱好者或正在探索该领域的人士。语气应为信息性、结构化且不过度使用术语（但关键术语需要解释）。
2. **初步头脑风暴与核心概念：**
* 什么是深度学习？机器学习的子集，而机器学习又是人工智能的子集。
* 核心机制：人工神经网络（ANN）。
* "深度"意味着许多层（隐藏层）。
* 它如何运作？从数据中学习表示（特征学习）。
* 关键应用：图像识别、自然语言处理、语音识别、自动驾驶汽车。
* 为什么现在流行？大数据、GPU算力、更好的算法。
* 优缺点：高准确性 vs 黑箱问题、数据饥渴。
3. **构建答案：**
延迟: 9.07 秒
预估输出单词数: 153
预估单次请求输出生成速率: 16.9 单词/秒
--------------------------------------------------------------------------------
摘要:
{'request_preview': '{"max_tokens": 256, "messages": {"content": "What', 'latency_s': 9.07, 'estimated_output_words': 153, 'words_per_second': 16.9}

print(f"正在删除端点: {endpoint.endpoint_name}")
endpoint.delete()
endpoint.wait_for_delete()
print("端点已删除")