Deploy and consider fashions with Amazon SageMaker JumpStart

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

REGION = "us-east-1"

fashions = list_jumpstart_models(
    filter="activity == textgeneration",
    area=REGION,
    list_versions=True,
)

print(f"Discovered {len(fashions)} textual content technology fashions")

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

from sagemaker.core.jumpstart.notebook_utils import list_jumpstart_models

ids = list_jumpstart_models(
    filter="model_id incorporates qwen",
    area=REGION,
    list_versions=True,
)

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

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

import time

from sagemaker.core.jumpstart.accessors import JumpStartModelsAccessor
from sagemaker.core.jumpstart.configs import JumpStartConfig
from sagemaker.core.coaching.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 solely after reviewing the mannequin phrases

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

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

if isinstance(specs, record):
    if not specs:
        increase ValueError(
            f"No JumpStart specification discovered for {MODEL_ID}"
        )
    specs = specs[0]

recommended_instance_type = get_field(
    specs,
    "default_inference_instance_type",
)

if not recommended_instance_type:
    increase ValueError(
        f"No really useful occasion kind discovered for {MODEL_ID}"
    )

print(f"Mannequin: {MODEL_ID}")
print(f"Area: {REGION}")
print(f"Beneficial occasion: {recommended_instance_type}")

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

mannequin = model_builder.construct()
endpoint = model_builder.deploy()
deploy_time = time.time() - begin

print(f"Deployed in {deploy_time:.1f} seconds")
print(f"Endpoint: {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(
    area=REGION,
    model_id=MODEL_ID,
    model_version=MODEL_VERSION,
    serialize=True,
)

#you may re-create the endpoint pointer utilizing the endpoint identify:
#from sagemaker.core.sources import Endpoint
#endpoint = Endpoint.get(endpoint_name="",area=REGION)

def read_response_body(end result):
    physique = end result.physique
    if hasattr(physique, "learn"):
        physique = physique.learn()
    if isinstance(physique, bytes):
        physique = physique.decode("utf-8")
    if isinstance(physique, str):
        attempt:
            return json.masses(physique)
        besides json.JSONDecodeError:
            return physique
    return physique

def extract_output(response):
    """Deal with frequent JumpStart text-generation response codecs."""
    if isinstance(response, record):
        response = response[0] if response else ""
    if isinstance(response, dict):
        decisions = response.get("decisions")
        if decisions:
            selection = decisions[0]
            message = selection.get("message")
            if isinstance(message, dict):
                return message.get("content material", "")
            return selection.get("textual content", "")
        for area in (
            "generated_text",
            "output_text",
            "textual content",
        ):
            if area in response:
                return response[field]
    return response

outcomes = []

for payload in example_payloads or []:
    begin = time.perf_counter()
    end result = endpoint.invoke(
        physique=payload.physique,
        content_type=payload.content_type,
        settle for=payload.settle for,
    )
    latency = time.perf_counter() - begin
    response = read_response_body(end result)
    output = extract_output(response)
    if not isinstance(output, str):
        output = json.dumps(output, indent=2)
    # That is an estimate based mostly on whitespace-separated phrases.
    output_words = len(output.cut up())
    outcomes.append(
        {
            "request_preview": payload.physique[:50],
            "latency_s": spherical(latency, 2),
            "estimated_output_words": output_words,
            "words_per_second": spherical(
                output_words / latency,
                1,
            ) if latency > 0 else 0,
        }
    )
    print("Enter:n", payload.physique, finish="nn")
    print("Output:n", output, finish="nn")
    print(
        f"Latency: {latency:.2f} secondsn"
        f"Estimated output phrases: {output_words}n"
        f"Estimated single-request output technology charge: "
        f"{output_words / latency:.1f} phrases/secondn"
    )
    print("-" * 80)

print("nSummary:")
for lead to outcomes:
    print(end result)

Enter:
{"max_tokens": 256, "messages": [{"content": "What is deep learning?", "role": "user"}], "mannequin": "/decide/ml/mannequin"}
Output:
Here is a considering course of that results in the reason of deep studying:
1. **Deconstruct the Request:**
* **Query:** "What's deep studying?"
* **Intent:** The person needs a transparent, complete, but accessible definition and overview of deep studying.
* **Goal Viewers:** Seemingly somebody searching for a normal understanding, presumably a scholar, tech
fanatic, or skilled exploring the sector. The tone must be informative, structured, and
not overly jargon-heavy (although key phrases want rationalization).
2. **Preliminary Brainstorming & Key Ideas:**
* What's it? A subset of machine studying, which is a subset of AI.
* Core mechanism: Synthetic Neural Networks (ANNs).
* "Deep" means many layers (hidden layers).
* How does it work? Studying representations from information (characteristic studying).
* Key functions: Picture recognition, NLP, speech, self-driving vehicles.
* Why is it in style now? Large information, GPU energy, higher algorithms.
* Professionals/Cons: Excessive accuracy vs. black field, information starvation.
3. **Structuring the Reply:**
Latency: 9.07 seconds
Estimated output phrases: 153
Estimated single-request output technology charge: 16.9 phrases/second
--------------------------------------------------------------------------------
Abstract:
{'request_preview': '{"max_tokens": 256, "messages": {"content material": "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")